You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by at...@apache.org on 2026/06/19 15:23:56 UTC
(iceberg-go) branch main updated: refactor(catalog): Minimize functions in interface for Hadoop catalog (#1185)
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 e3cd70dc refactor(catalog): Minimize functions in interface for Hadoop catalog (#1185)
e3cd70dc is described below
commit e3cd70dc19e696c0e8fb6b21a005d7ebe6e31692
Author: Colton Loftus <70...@users.noreply.github.com>
AuthorDate: Fri Jun 19 11:23:51 2026 -0400
refactor(catalog): Minimize functions in interface for Hadoop catalog (#1185)
Addresses https://github.com/apache/iceberg-go/issues/1096 (as the next
step; blob storage support is out of scope for this PR and would be
present in a follow up)
- Removes interfaces for `MkdirIO` and `ReadDirIO` to prepare for hadoop
cloud storage backends.
- Refactors hadoop catalog to use `MkdirAll` and `WalkDir` in their
place.
- Cleans up the comments for those interfaces to reflect current usage.
- Adds on new helper function `CheckedMkdirAll`, which before calling
`MkdirAll`, walks the directory and ensures that each subdirectory of
the full path exists.
- Currently this is only used once since only one previous `Mkdir` call
checked for errors if the directory already existed. That being said, I
think it helps to abstract logic a bit and I'm thinking that it likely
may be used in other parts of the Hadoop catalog.
---
catalog/hadoop/hadoop.go | 186 +++++++++++++++++++++---------
catalog/hadoop/hadoop_integration_test.go | 9 +-
catalog/hadoop/hadoop_test.go | 62 ++++++++--
catalog/hadoop/io.go | 35 ++----
4 files changed, 198 insertions(+), 94 deletions(-)
diff --git a/catalog/hadoop/hadoop.go b/catalog/hadoop/hadoop.go
index 8f5ba081..ef0b737f 100644
--- a/catalog/hadoop/hadoop.go
+++ b/catalog/hadoop/hadoop.go
@@ -188,25 +188,38 @@ func (c *Catalog) defaultTableLocation(ident table.Identifier) string {
func isTableDir(filesystem HadoopCatalogFS, path string) bool {
metaDir := filepath.Join(path, "metadata")
- entries, err := filesystem.ReadDir(metaDir)
- if err != nil {
- return false
- }
+ foundMetadata := false
+ err := filesystem.WalkDir(metaDir, func(path string, d fs.DirEntry, err error) error {
+ if err != nil {
+ return err
+ }
+
+ // Skip the root itself.
+ if path == metaDir {
+ return nil
+ }
- for _, e := range entries {
- if e.IsDir() {
- continue
+ // Don't descend into subdirectories.
+ if d.IsDir() {
+ return fs.SkipDir
}
- name := e.Name()
+ name := d.Name()
if versionPattern.MatchString(name) ||
uuidMetadataPattern.MatchString(name) ||
name == "version-hint.text" {
- return true
+ foundMetadata = true
+
+ return fs.SkipAll
}
+
+ return nil
+ })
+ if err != nil {
+ return false
}
- return false
+ return foundMetadata
}
func (c *Catalog) readVersionHint(ident table.Identifier) int {
@@ -275,25 +288,34 @@ func (c *Catalog) findVersion(ident table.Identifier) (int, error) {
dir := c.metadataDir(ident)
- entries, err := c.filesystem.ReadDir(dir)
- if err != nil {
- return 0, fmt.Errorf("hadoop catalog: cannot read metadata directory for %s: %w",
- strings.Join(ident, "."), catalog.ErrNoSuchTable)
- }
-
maxVer := 0
- for _, e := range entries {
- if e.IsDir() {
- continue
+ // Walk just one directory level to find metadata files,
+ // ignoring any subdirectories that may exist
+ err := c.filesystem.WalkDir(dir, func(path string, d fs.DirEntry, err error) error {
+ if err != nil {
+ return err
+ }
+ if path == dir {
+ return nil
+ }
+ if d.IsDir() {
+ return fs.SkipDir
}
- matches := versionPattern.FindStringSubmatch(e.Name())
+ name := d.Name()
+ matches := versionPattern.FindStringSubmatch(name)
if len(matches) == 2 {
v, _ := strconv.Atoi(matches[1])
if v > maxVer {
maxVer = v
}
}
+
+ return nil
+ })
+ if err != nil {
+ return 0, fmt.Errorf("hadoop catalog: cannot read metadata directory for %s: %w",
+ strings.Join(ident, "."), catalog.ErrNoSuchTable)
}
if maxVer == 0 {
@@ -526,29 +548,39 @@ func (c *Catalog) ListTables(_ context.Context, ns table.Identifier) iter.Seq2[t
return
}
- entries, err := c.filesystem.ReadDir(nsPath)
- if err != nil {
- yield(nil, fmt.Errorf("hadoop catalog: failed to read namespace directory: %w", err))
-
- return
- }
+ err = c.filesystem.WalkDir(nsPath, func(path string, d fs.DirEntry, err error) error {
+ if err != nil {
+ return err
+ }
- for _, e := range entries {
- if !e.IsDir() {
- continue
+ // Anything that is a file is not a namespace or table, so skip it.
+ if !d.IsDir() {
+ return nil
}
- child := filepath.Join(nsPath, e.Name())
- if !isTableDir(c.filesystem, child) {
- continue
+ // Skip the namespace directory itself.
+ if path == nsPath {
+ return nil
}
+ // Skip anything that is not a table directory.
+ if !isTableDir(c.filesystem, path) {
+ return fs.SkipDir
+ }
ident := make(table.Identifier, len(ns)+1)
copy(ident, ns)
- ident[len(ns)] = e.Name()
+ ident[len(ns)] = d.Name()
if !yield(ident, nil) {
- return
+ 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
}
}
}
@@ -594,9 +626,8 @@ func (c *Catalog) CreateNamespace(_ context.Context, ns table.Identifier, props
return errors.New("hadoop catalog: namespace properties are not supported")
}
- path := c.namespaceToPath(ns)
-
- if err := c.filesystem.Mkdir(path); err != nil {
+ // 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, "."))
}
@@ -619,7 +650,28 @@ func (c *Catalog) DropNamespace(_ context.Context, ns table.Identifier) error {
path := c.namespaceToPath(ns)
- entries, err := c.filesystem.ReadDir(path)
+ info, err := c.filesystem.Stat(path)
+ if errors.Is(err, fs.ErrNotExist) || (err == nil && !info.IsDir()) {
+ return fmt.Errorf("%w: %s", catalog.ErrNoSuchNamespace, strings.Join(ns, "."))
+ }
+ if err != nil {
+ return fmt.Errorf("hadoop catalog: failed to stat namespace directory: %w", err)
+ }
+
+ foundEntries := false
+ err = c.filesystem.WalkDir(path, func(p string, d fs.DirEntry, err error) error {
+ if err != nil {
+ return err
+ }
+
+ if p == path {
+ 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, "."))
@@ -628,7 +680,7 @@ func (c *Catalog) DropNamespace(_ context.Context, ns table.Identifier) error {
return fmt.Errorf("hadoop catalog: failed to read namespace directory: %w", err)
}
- if len(entries) > 0 {
+ if foundEntries {
return fmt.Errorf("%w: %s", catalog.ErrNamespaceNotEmpty, strings.Join(ns, "."))
}
@@ -674,23 +726,29 @@ func (c *Catalog) ListNamespaces(_ context.Context, parent table.Identifier) ([]
}
}
- entries, err := c.filesystem.ReadDir(path)
- if err != nil {
- return nil, fmt.Errorf("hadoop catalog: failed to read directory: %w", err)
- }
-
result := []table.Identifier{}
- for _, e := range entries {
- if !e.IsDir() {
- continue
+ err := c.filesystem.WalkDir(path, func(p string, d fs.DirEntry, err error) error {
+ if err != nil {
+ return err
}
-
- child := filepath.Join(path, e.Name())
- if isTableDir(c.filesystem, child) {
- continue
+ if p == path {
+ return nil
}
- result = append(result, table.Identifier{e.Name()})
+ if !d.IsDir() {
+ // skip plain files
+ return nil
+ }
+ if isTableDir(c.filesystem, p) {
+ // if a table, not a namespace, don't descend
+ return fs.SkipDir
+ }
+ result = append(result, table.Identifier{d.Name()})
+ // 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
@@ -723,3 +781,27 @@ func (c *Catalog) LoadNamespaceProperties(_ context.Context, ns table.Identifier
func (c *Catalog) UpdateNamespaceProperties(_ context.Context, _ table.Identifier, _ []string, _ iceberg.Properties) (catalog.PropertiesUpdateSummary, error) {
return catalog.PropertiesUpdateSummary{}, errors.New("hadoop catalog: UpdateNamespaceProperties not yet implemented")
}
+
+// 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_integration_test.go b/catalog/hadoop/hadoop_integration_test.go
index ffddf42d..fb164217 100644
--- a/catalog/hadoop/hadoop_integration_test.go
+++ b/catalog/hadoop/hadoop_integration_test.go
@@ -92,7 +92,8 @@ func (s *HadoopIntegrationSuite) createFakeTable(ident table.Identifier) {
metaDir := filepath.Join(tablePath, "metadata")
s.Require().NoError(os.MkdirAll(metaDir, 0o755))
s.Require().NoError(os.WriteFile(
- filepath.Join(metaDir, "v1.metadata.json"), []byte("{}"), 0o644))
+ filepath.Join(metaDir, "v1.metadata.json"), []byte("{}"), 0o644,
+ ))
}
// TestListTablesSparkCreated creates tables in Spark and verifies that
@@ -240,7 +241,8 @@ func (s *HadoopIntegrationSuite) TestGoCreateSparkDescribe() {
err := s.cat.CreateNamespace(s.ctx, []string{"describe_ns"}, nil)
s.Require().NoError(err)
- schema := iceberg.NewSchema(1,
+ schema := iceberg.NewSchema(
+ 1,
iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
iceberg.NestedField{ID: 2, Name: "name", Type: iceberg.PrimitiveTypes.String, Required: false},
)
@@ -281,7 +283,8 @@ func (s *HadoopIntegrationSuite) TestGoCreateSparkSelect() {
err := s.cat.CreateNamespace(s.ctx, []string{"select_ns"}, nil)
s.Require().NoError(err)
- schema := iceberg.NewSchema(1,
+ schema := iceberg.NewSchema(
+ 1,
iceberg.NestedField{ID: 1, Name: "value", Type: iceberg.PrimitiveTypes.Int32, Required: true},
)
diff --git a/catalog/hadoop/hadoop_test.go b/catalog/hadoop/hadoop_test.go
index 93fad424..ca37a75f 100644
--- a/catalog/hadoop/hadoop_test.go
+++ b/catalog/hadoop/hadoop_test.go
@@ -478,6 +478,16 @@ func (s *HadoopCatalogTestSuite) TestDropNamespaceNotEmptyWithChildNamespace() {
s.ErrorIs(err, catalog.ErrNamespaceNotEmpty)
}
+func (s *HadoopCatalogTestSuite) TestDropNameSpaceFileInsteadofDir() {
+ filePath := filepath.Join(s.warehouse, "not_a_dir")
+ s.Require().NoError(os.WriteFile(filePath, nil, 0o644))
+ err := s.cat.DropNamespace(context.Background(), []string{"not_a_dir"})
+ s.ErrorIs(err, catalog.ErrNoSuchNamespace)
+ info, err := os.Stat(filePath)
+ s.NoError(err)
+ s.False(info.IsDir())
+}
+
// CheckNamespaceExists tests
func (s *HadoopCatalogTestSuite) TestCheckNamespaceExistsTrue() {
@@ -811,6 +821,25 @@ func (s *HadoopCatalogTestSuite) TestListTablesNestedNamespace() {
s.Equal(table.Identifier{"a", "b", "tbl1"}, tables[0])
}
+func (s *HadoopCatalogTestSuite) TestListTablesDoesNotYieldTablesInChildNamespace() {
+ // When a namespace is passed into ListTables, only tables in that direct
+ // namespace should be yielded. Not the additionaltables in the children namespaces.
+ ctx := context.Background()
+ err := s.cat.CreateNamespace(ctx, []string{"ns"}, nil)
+ s.Require().NoError(err)
+ s.createFakeTable([]string{"ns", "tbl1"})
+ err = s.cat.CreateNamespace(ctx, []string{"ns", "child_ns"}, nil)
+ s.Require().NoError(err)
+ s.createFakeTable([]string{"ns", "child_ns", "tbl2"})
+ var tables []table.Identifier
+ for ident, err := range s.cat.ListTables(ctx, []string{"ns"}) {
+ s.Require().NoError(err)
+ tables = append(tables, ident)
+ }
+ s.Len(tables, 1)
+ s.Equal(table.Identifier{"ns", "tbl1"}, tables[0])
+}
+
// DropTable tests
func (s *HadoopCatalogTestSuite) TestDropTable() {
@@ -857,7 +886,8 @@ func (s *HadoopCatalogTestSuite) TestDropTableShortIdentifier() {
// CreateTable tests
func (s *HadoopCatalogTestSuite) testSchema() *iceberg.Schema {
- return iceberg.NewSchema(1,
+ return iceberg.NewSchema(
+ 1,
iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
iceberg.NestedField{ID: 2, Name: "data", Type: iceberg.PrimitiveTypes.String, Required: false},
)
@@ -1150,7 +1180,8 @@ func (s *HadoopCatalogTestSuite) TestCommitTableSingleUpdate() {
ctx := context.Background()
tbl := s.createTestTable("ns", "tbl")
- meta, metaLoc, err := s.cat.CommitTable(ctx, []string{"ns", "tbl"},
+ meta, metaLoc, err := s.cat.CommitTable(
+ ctx, []string{"ns", "tbl"},
[]table.Requirement{
table.AssertTableUUID(tbl.Metadata().TableUUID()),
},
@@ -1177,7 +1208,8 @@ func (s *HadoopCatalogTestSuite) TestCommitTableMultipleSequential() {
loaded, err := s.cat.LoadTable(ctx, ident)
s.Require().NoError(err)
- _, metaLoc, err := s.cat.CommitTable(ctx, ident,
+ _, metaLoc, err := s.cat.CommitTable(
+ ctx, ident,
[]table.Requirement{
table.AssertTableUUID(loaded.Metadata().TableUUID()),
},
@@ -1207,7 +1239,8 @@ func (s *HadoopCatalogTestSuite) TestCommitTableNoChanges() {
ctx := context.Background()
tbl := s.createTestTable("ns", "tbl")
- meta, metaLoc, err := s.cat.CommitTable(ctx, []string{"ns", "tbl"},
+ meta, metaLoc, err := s.cat.CommitTable(
+ ctx, []string{"ns", "tbl"},
[]table.Requirement{
table.AssertTableUUID(tbl.Metadata().TableUUID()),
},
@@ -1233,7 +1266,8 @@ func (s *HadoopCatalogTestSuite) TestCommitTableConflictDetection() {
ident := []string{"ns", "tbl"}
for i := 2; i <= 5; i++ {
- _, metaLoc, err := s.cat.CommitTable(ctx, ident,
+ _, metaLoc, err := s.cat.CommitTable(
+ ctx, ident,
nil,
[]table.Update{
table.NewSetPropertiesUpdate(iceberg.Properties{
@@ -1252,7 +1286,8 @@ func (s *HadoopCatalogTestSuite) TestCommitTableRequirementFailure() {
s.createTestTable("ns", "tbl")
wrongUUID := uuid.New()
- _, _, err := s.cat.CommitTable(ctx, []string{"ns", "tbl"},
+ _, _, err := s.cat.CommitTable(
+ ctx, []string{"ns", "tbl"},
[]table.Requirement{
table.AssertTableUUID(wrongUUID),
},
@@ -1268,7 +1303,8 @@ func (s *HadoopCatalogTestSuite) TestCommitTableLocationChange() {
ctx := context.Background()
s.createTestTable("ns", "tbl")
- _, _, err := s.cat.CommitTable(ctx, []string{"ns", "tbl"},
+ _, _, err := s.cat.CommitTable(
+ ctx, []string{"ns", "tbl"},
nil,
[]table.Update{
table.NewSetLocationUpdate("/some/other/location"),
@@ -1282,7 +1318,8 @@ func (s *HadoopCatalogTestSuite) TestCommitTableWriteMetadataLocation() {
ctx := context.Background()
s.createTestTable("ns", "tbl")
- _, _, err := s.cat.CommitTable(ctx, []string{"ns", "tbl"},
+ _, _, err := s.cat.CommitTable(
+ ctx, []string{"ns", "tbl"},
nil,
[]table.Update{
table.NewSetPropertiesUpdate(iceberg.Properties{
@@ -1301,7 +1338,8 @@ func (s *HadoopCatalogTestSuite) TestCommitTableNoOrphanedTempFiles() {
// Do several commits and verify no temp files are left behind.
for i := 0; i < 3; i++ {
- _, _, err := s.cat.CommitTable(ctx, ident,
+ _, _, err := s.cat.CommitTable(
+ ctx, ident,
nil,
[]table.Update{
table.NewSetPropertiesUpdate(iceberg.Properties{
@@ -1328,7 +1366,8 @@ func (s *HadoopCatalogTestSuite) TestCommitTableVersionHintUpdated() {
s.createTestTable("ns", "tbl")
ident := []string{"ns", "tbl"}
- _, _, err := s.cat.CommitTable(ctx, ident,
+ _, _, err := s.cat.CommitTable(
+ ctx, ident,
nil,
[]table.Update{
table.NewSetPropertiesUpdate(iceberg.Properties{"k": "v"}),
@@ -1345,7 +1384,8 @@ func (s *HadoopCatalogTestSuite) TestCommitTableCreateViaCommit() {
ident := []string{"ns", "tbl"}
loc := s.cat.defaultTableLocation(ident)
- meta, metaLoc, err := s.cat.CommitTable(ctx, ident,
+ meta, metaLoc, err := s.cat.CommitTable(
+ ctx, ident,
[]table.Requirement{
table.AssertCreate(),
},
diff --git a/catalog/hadoop/io.go b/catalog/hadoop/io.go
index ed5c2c6e..19694552 100644
--- a/catalog/hadoop/io.go
+++ b/catalog/hadoop/io.go
@@ -33,20 +33,18 @@ type HadoopCatalogFS interface {
RenameIO
RemoveAllIO
MkdirAllIO
- MkdirIO
- ReadDirIO
}
var _ HadoopCatalogFS = (*icebergio.LocalFS)(nil)
// StatIO is an extension of IO interface that includes the Stat
-// method for retrieving file information without reading the file
+// method for retrieving file or directory information
type StatIO interface {
icebergio.IO
- // The Stat method returns a FileInfo describing the named file, or an error
- // satisfying errors.Is(err, fs.ErrNotExist) if the file does not exist
- Stat(name string) (fs.FileInfo, error)
+ // The Stat method returns a FileInfo describing the named file or directory,
+ // or an error satisfying errors.Is(err, fs.ErrNotExist) if the file does not exist
+ Stat(path string) (fs.FileInfo, error)
}
// RenameIO is an extension of IO interface that includes the Rename
@@ -66,32 +64,13 @@ type RemoveAllIO interface {
RemoveAll(name string) error
}
-// MkdirIO is an extension of IO interface that includes the Mkdir
-// method for creating a directory
-type MkdirIO interface {
- icebergio.IO
-
- // Mkdir creates a new directory or returns an error
- // satisfying errors.Is(err, fs.ErrExist) if the directory already exists or
- // errors.Is(err, fs.ErrNotExist) if it does not create parent directories
- Mkdir(path string) error
-}
-
// MkdirAllIO is an extension of IO interface that includes the MkdirAll
// method for creating a directory path recursively
type MkdirAllIO interface {
icebergio.IO
+ // MkdirAll should not raise an error if the directory already exists,
+ // and should create any parent directories in the path if they
+ // do not already exist.
MkdirAll(path string) error
}
-
-// ReadDirIO is an extension of IO interface that includes the ReadDir
-// method for reading the contents of a directory and returning a slice of
-// DirEntry values
-type ReadDirIO interface {
- icebergio.IO
-
- // ReadDir returns a slice of DirEntry values for the named directory
- // or an error satisfying errors.Is(err, fs.ErrNotExist) if the directory does not exist
- ReadDir(name string) ([]fs.DirEntry, error)
-}