You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "C-Loftus (via GitHub)" <gi...@apache.org> on 2026/07/01 15:35:47 UTC
[PR] feat(catalog/hadoop): Add support for hadoop catalog fs interface methods to blob filesystem [iceberg-go]
C-Loftus opened a new pull request, #1350:
URL: https://github.com/apache/iceberg-go/pull/1350
- Move io interfaces needed for HadoopCatalogFS into the `io.go` file since they are now presumed to be relatively stable.
- Add implementations for any methods not previously fulfilled for the blob storage struct in order to have it fulfill HadoopCatalogFS
- Add an integration test for using the hadoop catalog with minio locally
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(catalog/hadoop): Add support for hadoop catalog fs interface methods to blob filesystem [iceberg-go]
Posted by "tanmayrauth (via GitHub)" <gi...@apache.org>.
tanmayrauth commented on code in PR #1350:
URL: https://github.com/apache/iceberg-go/pull/1350#discussion_r3509019419
##########
io/gocloud/blob.go:
##########
@@ -256,10 +319,178 @@ func (bfs *blobFileIO) DeleteFiles(ctx context.Context, paths []string) ([]strin
return deleted, errs
}
+// MkdirAll mimics creating a directory by creating a zero-length object for each component of the path
+func (bfs *BlobFileIO) MkdirAll(path string) error {
+ key, err := bfs.preprocess(path)
+ if err != nil {
+ return &fs.PathError{Op: "mkdir", Path: path, Err: err}
+ }
+
+ key = strings.Trim(key, "/")
+ if key == "" || key == "." {
+ return nil
+ }
+
+ parts := strings.Split(key, "/")
+ for idx := range parts {
+ marker := strings.Join(parts[:idx+1], "/") + "/"
+ if err := bfs.WriteAll(bfs.ctx, marker, nil, nil); err != nil {
+ return &fs.PathError{Op: "mkdir", Path: path, Err: err}
+ }
+ }
+
+ return nil
+}
+
+// ReadFile reads the contents of the file at the given path and returns it as a byte slice.
+func (bfs *BlobFileIO) ReadFile(path string) ([]byte, error) {
+ key, err := bfs.preprocess(path)
+ if err != nil {
+ return nil, &fs.PathError{Op: "ReadFile", Path: path, Err: err}
+ }
+
+ data, err := bfs.ReadAll(bfs.ctx, key)
+ if err != nil {
+ return nil, blobErrToFsErr("ReadFile", path, err)
+ }
+
+ return data, nil
+}
+
+// Stat interprets the input path as a directory or file and returns the corresponding FileInfo.
+// If the path does not exist, it returns fs.ErrNotExist
+func (bfs *BlobFileIO) Stat(path string) (fs.FileInfo, error) {
+ key, err := bfs.preprocess(path)
+ if err != nil {
+ return nil, &fs.PathError{Op: "Stat", Path: path, Err: err}
+ }
+
+ attrs, err := bfs.Attributes(bfs.ctx, key)
+ // if there is no error and we have attributes, we can return a FileInfo for object
+ if err == nil {
+ return blobFileInfo{
+ name: filepath.Base(key),
+ size: attrs.Size,
+ mode: fs.ModeIrregular,
+ modTime: attrs.ModTime,
+ sys: attrs,
+ }, nil
+ }
+
+ if gcerrors.Code(err) != gcerrors.NotFound {
+ return nil, blobErrToFsErr("Stat", path, err)
+ }
+
+ marker := directoryMarker(key)
+ // if the marker exists, we can return a FileInfo for the directory
+ if marker != "" {
+ if attrs, markerErr := bfs.Attributes(bfs.ctx, marker); markerErr == nil {
+ return blobFileInfo{
+ name: directoryName(key),
+ mode: fs.ModeDir,
+ modTime: attrs.ModTime,
+ sys: attrs,
+ }, nil
+ } else if gcerrors.Code(markerErr) != gcerrors.NotFound {
+ return nil, blobErrToFsErr("Stat", path, markerErr)
+ }
+ }
+
+ prefix := marker
+ if prefix == "" {
+ prefix = key
+ }
+
+ iter := bfs.List(&blob.ListOptions{Prefix: prefix})
+ if obj, listErr := iter.Next(bfs.ctx); listErr == nil {
+ return blobFileInfo{
+ name: directoryName(key),
+ mode: fs.ModeDir,
+ sys: obj,
+ }, nil
+ } else if listErr != io.EOF {
+ return nil, blobErrToFsErr("Stat", path, listErr)
+ }
+
+ return nil, &fs.PathError{Op: "Stat", Path: path, Err: fs.ErrNotExist}
+}
+
+// Rename renames a file or directory from oldpath to newpath, replacing newpath if it already exists.
+func (bfs *BlobFileIO) Rename(oldpath, newpath string) error {
+ oldKey, err := bfs.preprocess(oldpath)
+ if err != nil {
+ return &fs.PathError{Op: "Rename", Path: oldpath, Err: err}
+ }
+
+ newKey, err := bfs.preprocess(newpath)
+ if err != nil {
+ return &fs.PathError{Op: "Rename", Path: newpath, Err: err}
+ }
+
+ if err := bfs.Copy(bfs.ctx, newKey, oldKey, nil); err != nil {
+ return blobErrToFsErr("Rename", oldpath, err)
+ }
+
+ if err := bfs.Delete(bfs.ctx, oldKey); err != nil && gcerrors.Code(err) != gcerrors.NotFound {
+ return blobErrToFsErr("Rename", oldpath, err)
+ }
+
+ return nil
+}
+
+// RenameNoReplace renames a file or directory from oldpath to newpath, returning an error if newpath already exists.
+func (bfs *BlobFileIO) RenameNoReplace(oldpath, newpath string) error {
Review Comment:
Just noting for the record since this is the commit primitive: this is Stat-then-Copy+Delete, so two concurrent writers can both pass the Stat check and both Copy — the losing commit is silently dropped rather than rejected. That's correctly gated behind allow-unsafe-commits with a fail-closed default, so no change needed here. One precision worth capturing in the follow-up tracking issue: the hazard is concurrent lost-update, not single-writer crash corruption (S3 CopyObject is atomic at the destination). The real fix is a conditional-write primitive (S3 If-None-Match:*, GCS if-generation-match:0). Is that follow-up issue tracked anywhere yet?
##########
io/gocloud/blob.go:
##########
@@ -256,10 +319,178 @@ func (bfs *blobFileIO) DeleteFiles(ctx context.Context, paths []string) ([]strin
return deleted, errs
}
+// MkdirAll mimics creating a directory by creating a zero-length object for each component of the path
+func (bfs *BlobFileIO) MkdirAll(path string) error {
+ key, err := bfs.preprocess(path)
+ if err != nil {
+ return &fs.PathError{Op: "mkdir", Path: path, Err: err}
+ }
+
+ key = strings.Trim(key, "/")
+ if key == "" || key == "." {
+ return nil
+ }
+
+ parts := strings.Split(key, "/")
+ for idx := range parts {
+ marker := strings.Join(parts[:idx+1], "/") + "/"
+ if err := bfs.WriteAll(bfs.ctx, marker, nil, nil); err != nil {
+ return &fs.PathError{Op: "mkdir", Path: path, Err: err}
+ }
+ }
+
+ return nil
+}
+
+// ReadFile reads the contents of the file at the given path and returns it as a byte slice.
+func (bfs *BlobFileIO) ReadFile(path string) ([]byte, error) {
+ key, err := bfs.preprocess(path)
+ if err != nil {
+ return nil, &fs.PathError{Op: "ReadFile", Path: path, Err: err}
+ }
+
+ data, err := bfs.ReadAll(bfs.ctx, key)
+ if err != nil {
+ return nil, blobErrToFsErr("ReadFile", path, err)
+ }
+
+ return data, nil
+}
+
+// Stat interprets the input path as a directory or file and returns the corresponding FileInfo.
+// If the path does not exist, it returns fs.ErrNotExist
+func (bfs *BlobFileIO) Stat(path string) (fs.FileInfo, error) {
+ key, err := bfs.preprocess(path)
+ if err != nil {
+ return nil, &fs.PathError{Op: "Stat", Path: path, Err: err}
+ }
+
+ attrs, err := bfs.Attributes(bfs.ctx, key)
+ // if there is no error and we have attributes, we can return a FileInfo for object
+ if err == nil {
+ return blobFileInfo{
+ name: filepath.Base(key),
+ size: attrs.Size,
+ mode: fs.ModeIrregular,
+ modTime: attrs.ModTime,
+ sys: attrs,
+ }, nil
+ }
+
+ if gcerrors.Code(err) != gcerrors.NotFound {
+ return nil, blobErrToFsErr("Stat", path, err)
+ }
+
+ marker := directoryMarker(key)
+ // if the marker exists, we can return a FileInfo for the directory
+ if marker != "" {
+ if attrs, markerErr := bfs.Attributes(bfs.ctx, marker); markerErr == nil {
+ return blobFileInfo{
+ name: directoryName(key),
+ mode: fs.ModeDir,
+ modTime: attrs.ModTime,
+ sys: attrs,
+ }, nil
+ } else if gcerrors.Code(markerErr) != gcerrors.NotFound {
+ return nil, blobErrToFsErr("Stat", path, markerErr)
+ }
+ }
+
+ prefix := marker
+ if prefix == "" {
+ prefix = key
+ }
+
+ iter := bfs.List(&blob.ListOptions{Prefix: prefix})
+ if obj, listErr := iter.Next(bfs.ctx); listErr == nil {
+ return blobFileInfo{
+ name: directoryName(key),
+ mode: fs.ModeDir,
+ sys: obj,
+ }, nil
+ } else if listErr != io.EOF {
+ return nil, blobErrToFsErr("Stat", path, listErr)
+ }
+
+ return nil, &fs.PathError{Op: "Stat", Path: path, Err: fs.ErrNotExist}
+}
+
+// Rename renames a file or directory from oldpath to newpath, replacing newpath if it already exists.
Review Comment:
Doc says "renames a file or directory", but the body is a single Copy(newKey, oldKey) — with the marker layout, a directory's children (and its own marker) wouldn't move. Current callers only rename single files so it's fine in practice; can we tighten the comment to single-object rename to avoid surprising a future caller?
##########
io/gocloud/blob.go:
##########
@@ -102,13 +104,65 @@ type blobFileIO struct {
newRangeReader func(ctx context.Context, key string, offset, length int64) (io.ReadCloser, error)
}
-var _ icebergio.ListableIO = (*blobFileIO)(nil)
+var _ icebergio.ListableIO = (*BlobFileIO)(nil)
-func (bfs *blobFileIO) preprocess(path string) (string, error) {
+type blobFileInfo struct {
+ name string
+ size int64
+ mode fs.FileMode
+ modTime time.Time
+ sys any
+}
+
+// blobFileInfo implements the fs.FileInfo interface for a blob object
+var _ fs.FileInfo = (*blobFileInfo)(nil)
+
+func (f blobFileInfo) Name() string { return f.name }
+func (f blobFileInfo) Size() int64 { return f.size }
+func (f blobFileInfo) Mode() fs.FileMode { return f.mode }
+func (f blobFileInfo) ModTime() time.Time { return f.modTime }
+func (f blobFileInfo) IsDir() bool { return f.mode.IsDir() }
+func (f blobFileInfo) Sys() any { return f.sys }
+
+// preprocess returns the object key from an input path
+func (bfs *BlobFileIO) preprocess(path string) (string, error) {
return bfs.keyExtractor(path)
}
-func (bfs *blobFileIO) Open(path string) (icebergio.File, error) {
+// blobErrToFsErr converts a gocloud blob error to an error from the fs package; this is
+// necessary for comply with certain iceberg-go/io interface functions that expect fs.ErrNotExist
+// for missing files, rather than the gocloud error. If there is no corresponding fs error, or the
+// mapped error wouldn't be needed, the original error is returned.
+func blobErrToFsErr(op, name string, err error) error {
Review Comment:
Small doc mismatch: the comment says the original error is returned when there's no fs mapping, but this always wraps in *fs.PathError. Behavior's fine (errors.Is still works through the wrap) — just the comment.
##########
catalog/hadoop/hadoop_minio_integration_test.go:
##########
@@ -0,0 +1,181 @@
+// 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.
+
+//go:build integration
+
+package hadoop_test
+
+import (
+ "context"
+ "fmt"
+ "strings"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/catalog"
+ "github.com/apache/iceberg-go/catalog/hadoop"
+ icebergio "github.com/apache/iceberg-go/io"
+ _ "github.com/apache/iceberg-go/io/gocloud"
+ "github.com/apache/iceberg-go/table"
+ "github.com/google/uuid"
+ "github.com/stretchr/testify/suite"
+)
+
+type HadoopMinIOIntegrationSuite struct {
+ suite.Suite
+
+ ctx context.Context
+ cat *hadoop.Catalog
+ props iceberg.Properties
+ warehouse string
+}
+
+func (s *HadoopMinIOIntegrationSuite) SetupTest() {
+ s.ctx = context.Background()
+ s.warehouse = fmt.Sprintf("s3a://warehouse/hadoop-minio/%s", uuid.NewString())
+ s.props = iceberg.Properties{
+ "allow-unsafe-commits": "true",
+ icebergio.S3Region: "local",
+ icebergio.S3AccessKeyID: "admin",
+ icebergio.S3SecretAccessKey: "password",
+ // Endpoint is provided by the env var AWS_S3_ENDPOINT
+ // from `make integration-env` or recipe.Start.
+ }
+
+ cat, err := hadoop.NewCatalog("hadoop_minio_test", s.warehouse, s.props)
+ s.Require().NoError(err)
+ s.cat = cat
+}
+
+func (s *HadoopMinIOIntegrationSuite) testSchema() *iceberg.Schema {
+ return 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},
+ )
+}
+
+func (s *HadoopMinIOIntegrationSuite) TearDownTest() {
+ fs, err := icebergio.LoadFS(s.ctx, s.props, s.warehouse)
+ if err != nil {
+ return
+ }
+
+ if remover, ok := fs.(icebergio.RemoveAllIO); ok {
+ _ = remover.RemoveAll(s.warehouse)
+ }
+}
+
+func (s *HadoopMinIOIntegrationSuite) TestNamespaceTableRoundTrip() {
+ ns := table.Identifier{"minio_ns"}
+ ident := table.Identifier{"minio_ns", "tbl"}
+
+ s.Require().NoError(s.cat.CreateNamespace(s.ctx, ns, nil))
+
+ exists, err := s.cat.CheckNamespaceExists(s.ctx, ns)
+ s.Require().NoError(err)
+ s.True(exists)
+
+ tbl, err := s.cat.CreateTable(s.ctx, ident, s.testSchema())
+ s.Require().NoError(err)
+ s.NotNil(tbl)
+ s.True(strings.HasPrefix(tbl.MetadataLocation(), s.warehouse+"/minio_ns/tbl/metadata/v1.metadata.json"))
+
+ loaded, err := s.cat.LoadTable(s.ctx, ident)
+ s.Require().NoError(err)
+ s.Equal(tbl.MetadataLocation(), loaded.MetadataLocation())
+ s.Equal(2, len(loaded.Schema().Fields()))
+
+ var tables []table.Identifier
+ for found, err := range s.cat.ListTables(s.ctx, ns) {
+ s.Require().NoError(err)
+ tables = append(tables, found)
+ }
+ s.Equal([]table.Identifier{ident}, tables)
+
+ s.Require().NoError(s.cat.DropTable(s.ctx, ident))
+
+ tableExists, err := s.cat.CheckTableExists(s.ctx, ident)
+ s.Require().NoError(err)
+ s.False(tableExists)
+}
+
+func (s *HadoopMinIOIntegrationSuite) TestListTablesEmptyNamespace() {
+ ns := table.Identifier{"empty_ns"}
+ s.Require().NoError(s.cat.CreateNamespace(s.ctx, ns, nil))
+
+ var tables []table.Identifier
+ for found, err := range s.cat.ListTables(s.ctx, ns) {
+ s.Require().NoError(err)
+ tables = append(tables, found)
+ }
+ s.Empty(tables)
+}
+
+func (s *HadoopMinIOIntegrationSuite) TestListTablesMissingNamespace() {
+ for _, err := range s.cat.ListTables(s.ctx, table.Identifier{"missing_ns"}) {
+ s.ErrorIs(err, catalog.ErrNoSuchNamespace)
+
+ break
+ }
+}
+
+func (s *HadoopMinIOIntegrationSuite) TestListTablesOnlyDirectNamespaceTables() {
+ parent := table.Identifier{"parent_ns"}
+ child := table.Identifier{"parent_ns", "child_ns"}
+ directTable := table.Identifier{"parent_ns", "direct_tbl"}
+ childTable := table.Identifier{"parent_ns", "child_ns", "nested_tbl"}
+
+ s.Require().NoError(s.cat.CreateNamespace(s.ctx, parent, nil))
+ s.Require().NoError(s.cat.CreateNamespace(s.ctx, child, nil))
+ _, err := s.cat.CreateTable(s.ctx, directTable, s.testSchema())
+ s.Require().NoError(err)
+ _, err = s.cat.CreateTable(s.ctx, childTable, s.testSchema())
+ s.Require().NoError(err)
+
+ var tables []table.Identifier
+ for found, err := range s.cat.ListTables(s.ctx, parent) {
+ s.Require().NoError(err)
+ tables = append(tables, found)
+ }
+ s.Equal([]table.Identifier{directTable}, tables)
+}
+
+func (s *HadoopMinIOIntegrationSuite) TestDropTableRemovesTableFromListingAndLoad() {
+ ns := table.Identifier{"drop_ns"}
+ ident := table.Identifier{"drop_ns", "tbl"}
+
+ s.Require().NoError(s.cat.CreateNamespace(s.ctx, ns, nil))
+ _, err := s.cat.CreateTable(s.ctx, ident, s.testSchema())
+ s.Require().NoError(err)
+
+ s.Require().NoError(s.cat.DropTable(s.ctx, ident))
+
+ _, err = s.cat.LoadTable(s.ctx, ident)
+ s.ErrorIs(err, catalog.ErrNoSuchTable)
+
+ var tables []table.Identifier
+ for found, err := range s.cat.ListTables(s.ctx, ns) {
+ s.Require().NoError(err)
+ tables = append(tables, found)
+ }
+ s.Empty(tables)
+}
+
+func TestHadoopMinIOIntegration(t *testing.T) {
Review Comment:
The repo's `integration-hadoop` make target runs `-run="^TestHadoopIntegration"`, and this name (TestHadoopMinIOIntegration) doesn't match that anchored pattern — it diverges at "Hadoop*M*inIO". So this whole suite is skipped in CI and the green checks aren't exercising any of the new blob layout. Could we rename this to TestHadoopIntegrationMinIO so the existing target picks it up (or otherwise wire it into the integration run)? Everything else here really wants this suite executing.
##########
io/gocloud/blob.go:
##########
@@ -256,10 +319,178 @@ func (bfs *blobFileIO) DeleteFiles(ctx context.Context, paths []string) ([]strin
return deleted, errs
}
+// MkdirAll mimics creating a directory by creating a zero-length object for each component of the path
+func (bfs *BlobFileIO) MkdirAll(path string) error {
+ key, err := bfs.preprocess(path)
+ if err != nil {
+ return &fs.PathError{Op: "mkdir", Path: path, Err: err}
+ }
+
+ key = strings.Trim(key, "/")
+ if key == "" || key == "." {
+ return nil
+ }
+
+ parts := strings.Split(key, "/")
+ for idx := range parts {
+ marker := strings.Join(parts[:idx+1], "/") + "/"
+ if err := bfs.WriteAll(bfs.ctx, marker, nil, nil); err != nil {
+ return &fs.PathError{Op: "mkdir", Path: path, Err: err}
+ }
+ }
+
+ return nil
+}
+
+// ReadFile reads the contents of the file at the given path and returns it as a byte slice.
+func (bfs *BlobFileIO) ReadFile(path string) ([]byte, error) {
+ key, err := bfs.preprocess(path)
+ if err != nil {
+ return nil, &fs.PathError{Op: "ReadFile", Path: path, Err: err}
+ }
+
+ data, err := bfs.ReadAll(bfs.ctx, key)
+ if err != nil {
+ return nil, blobErrToFsErr("ReadFile", path, err)
+ }
+
+ return data, nil
+}
+
+// Stat interprets the input path as a directory or file and returns the corresponding FileInfo.
+// If the path does not exist, it returns fs.ErrNotExist
+func (bfs *BlobFileIO) Stat(path string) (fs.FileInfo, error) {
+ key, err := bfs.preprocess(path)
+ if err != nil {
+ return nil, &fs.PathError{Op: "Stat", Path: path, Err: err}
+ }
+
+ attrs, err := bfs.Attributes(bfs.ctx, key)
+ // if there is no error and we have attributes, we can return a FileInfo for object
+ if err == nil {
+ return blobFileInfo{
+ name: filepath.Base(key),
Review Comment:
filepath.Base here vs path.Base in directoryName for the dir case — object keys are always '/'-separated, so filepath.Base won't split them on Windows. Suggest path.Base for consistency.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(catalog/hadoop): Add support for hadoop catalog fs interface methods to blob filesystem [iceberg-go]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka merged PR #1350:
URL: https://github.com/apache/iceberg-go/pull/1350
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(catalog/hadoop): Add support for hadoop catalog fs interface methods to blob filesystem [iceberg-go]
Posted by "C-Loftus (via GitHub)" <gi...@apache.org>.
C-Loftus commented on PR #1350:
URL: https://github.com/apache/iceberg-go/pull/1350#issuecomment-4861303206
@tanmayrauth thank you very much for your feedback. I believe I have addressed it and at your convenience, it should be ready to review again. (Accidentally changed the test suite struct name instead of the test itself at first, but should be all set now!)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(catalog/hadoop): Add support for hadoop catalog fs interface methods to blob filesystem [iceberg-go]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on code in PR #1350:
URL: https://github.com/apache/iceberg-go/pull/1350#discussion_r3520952842
##########
io/gocloud/blob.go:
##########
@@ -256,10 +317,178 @@ func (bfs *blobFileIO) DeleteFiles(ctx context.Context, paths []string) ([]strin
return deleted, errs
}
+// MkdirAll mimics creating a directory by creating a zero-length object for each component of the path
+func (bfs *BlobFileIO) MkdirAll(path string) error {
+ key, err := bfs.preprocess(path)
+ if err != nil {
+ return &fs.PathError{Op: "mkdir", Path: path, Err: err}
+ }
+
+ key = strings.Trim(key, "/")
+ if key == "" || key == "." {
+ return nil
+ }
+
+ parts := strings.Split(key, "/")
+ for idx := range parts {
+ marker := strings.Join(parts[:idx+1], "/") + "/"
+ if err := bfs.WriteAll(bfs.ctx, marker, nil, nil); err != nil {
+ return &fs.PathError{Op: "mkdir", Path: path, Err: err}
+ }
+ }
+
+ return nil
+}
+
+// ReadFile reads the contents of the file at the given path and returns it as a byte slice.
+func (bfs *BlobFileIO) ReadFile(path string) ([]byte, error) {
+ key, err := bfs.preprocess(path)
+ if err != nil {
+ return nil, &fs.PathError{Op: "ReadFile", Path: path, Err: err}
+ }
+
+ data, err := bfs.ReadAll(bfs.ctx, key)
+ if err != nil {
+ return nil, blobErrToFsErr("ReadFile", path, err)
+ }
+
+ return data, nil
+}
+
+// Stat interprets the input path as a directory or file and returns the corresponding FileInfo.
+// If the path does not exist, it returns fs.ErrNotExist
+func (bfs *BlobFileIO) Stat(path string) (fs.FileInfo, error) {
+ key, err := bfs.preprocess(path)
+ if err != nil {
+ return nil, &fs.PathError{Op: "Stat", Path: path, Err: err}
+ }
+
+ attrs, err := bfs.Attributes(bfs.ctx, key)
+ // if there is no error and we have attributes, we can return a FileInfo for object
+ if err == nil {
+ return blobFileInfo{
+ name: pathpkg.Base(key),
+ size: attrs.Size,
+ mode: fs.ModeIrregular,
+ modTime: attrs.ModTime,
+ sys: attrs,
+ }, nil
+ }
+
+ if gcerrors.Code(err) != gcerrors.NotFound {
+ return nil, blobErrToFsErr("Stat", path, err)
+ }
+
+ marker := directoryMarker(key)
+ // if the marker exists, we can return a FileInfo for the directory
+ if marker != "" {
+ if attrs, markerErr := bfs.Attributes(bfs.ctx, marker); markerErr == nil {
+ return blobFileInfo{
+ name: directoryName(key),
+ mode: fs.ModeDir,
+ modTime: attrs.ModTime,
+ sys: attrs,
+ }, nil
+ } else if gcerrors.Code(markerErr) != gcerrors.NotFound {
+ return nil, blobErrToFsErr("Stat", path, markerErr)
+ }
+ }
+
+ prefix := marker
+ if prefix == "" {
+ prefix = key
+ }
+
+ iter := bfs.List(&blob.ListOptions{Prefix: prefix})
+ if obj, listErr := iter.Next(bfs.ctx); listErr == nil {
+ return blobFileInfo{
+ name: directoryName(key),
+ mode: fs.ModeDir,
+ sys: obj,
+ }, nil
+ } else if listErr != io.EOF {
+ return nil, blobErrToFsErr("Stat", path, listErr)
+ }
+
+ return nil, &fs.PathError{Op: "Stat", Path: path, Err: fs.ErrNotExist}
+}
+
+// Rename renames one file from oldpath to newpath, replacing newpath if it already exists.
+func (bfs *BlobFileIO) Rename(oldpath, newpath string) error {
+ oldKey, err := bfs.preprocess(oldpath)
+ if err != nil {
+ return &fs.PathError{Op: "Rename", Path: oldpath, Err: err}
+ }
+
+ newKey, err := bfs.preprocess(newpath)
+ if err != nil {
+ return &fs.PathError{Op: "Rename", Path: newpath, Err: err}
+ }
+
+ if err := bfs.Copy(bfs.ctx, newKey, oldKey, nil); err != nil {
+ return blobErrToFsErr("Rename", oldpath, err)
+ }
+
+ if err := bfs.Delete(bfs.ctx, oldKey); err != nil && gcerrors.Code(err) != gcerrors.NotFound {
+ return blobErrToFsErr("Rename", oldpath, err)
+ }
+
+ return nil
+}
+
+// RenameNoReplace renames a file or directory from oldpath to newpath, returning an error if newpath already exists.
+func (bfs *BlobFileIO) RenameNoReplace(oldpath, newpath string) error {
+ if _, err := bfs.Stat(newpath); err == nil {
Review Comment:
This is the one thing I'd want settled before merge. RenameNoReplace here is a check-then-act — Stat(newpath), then fall through to Copy+Delete — with no atomicity between the two steps, but RenameNoReplaceIO's godoc (io/io.go:190) promises the move happens "atomically" and the hadoop commit path leans on that. Two concurrent commits writing the same v1.metadata.json both pass the Stat("not found") check, both Copy, and the second silently overwrites the first — the losing writer's snapshot chain is gone. And commitMetadataFile's fs.ErrExist branch never fires, because nothing on the blob side ever returns it.
The `allow-unsafe-commits` gate is the right instinct and does mean nobody hits this without opting in, but the interface doc still over-promises for blob backends. I'd either soften the RenameIO/RenameNoReplaceIO godoc to state blob implementations aren't atomic and make the `allow-unsafe-commits` error spell out the last-writer-wins risk, or enforce the no-replace server-side with a conditional put (If-None-Match: * on S3/GCS). wdyt?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(catalog/hadoop): Add support for hadoop catalog fs interface methods to blob filesystem [iceberg-go]
Posted by "zeroshade (via GitHub)" <gi...@apache.org>.
zeroshade commented on code in PR #1350:
URL: https://github.com/apache/iceberg-go/pull/1350#discussion_r3553758274
##########
io/gocloud/blob.go:
##########
@@ -368,10 +455,194 @@ func (bfs *blobFileIO) DeleteFiles(ctx context.Context, paths []string) ([]strin
return deleted, errs
}
+// MkdirAll mimics creating a directory by creating a zero-length object for each component of the path
+func (bfs *BlobFileIO) MkdirAll(path string) error {
+ key, err := bfs.preprocess(path)
+ if err != nil {
+ return &fs.PathError{Op: "mkdir", Path: path, Err: err}
+ }
+
+ key = strings.Trim(key, "/")
+ if key == "" || key == "." {
+ return nil
+ }
+
+ parts := strings.Split(key, "/")
+ for idx := range parts {
+ marker := strings.Join(parts[:idx+1], "/") + "/"
+ if err := bfs.WriteAll(bfs.ctx, marker, nil, nil); err != nil {
+ return &fs.PathError{Op: "mkdir", Path: path, Err: err}
+ }
+ }
+
+ return nil
+}
+
+// ReadFile reads the contents of the file at the given path and returns it as a byte slice.
+func (bfs *BlobFileIO) ReadFile(path string) ([]byte, error) {
+ key, err := bfs.preprocess(path)
+ if err != nil {
+ return nil, &fs.PathError{Op: "ReadFile", Path: path, Err: err}
+ }
+
+ data, err := bfs.ReadAll(bfs.ctx, key)
+ if err != nil {
+ return nil, blobErrToFsErr("ReadFile", path, err)
+ }
+
+ return data, nil
+}
+
+// Stat interprets the input path as a directory or file and returns the corresponding FileInfo.
+// If the path does not exist, it returns fs.ErrNotExist
+func (bfs *BlobFileIO) Stat(path string) (fs.FileInfo, error) {
+ key, err := bfs.preprocess(path)
+ if err != nil {
+ return nil, &fs.PathError{Op: "Stat", Path: path, Err: err}
+ }
+
+ // If the path ends with a slash, we know by path semantics that we can
+ // check for a directory immediately and skip needing to check it as both an object
+ // or a directory marker.
+ if strings.HasSuffix(key, "/") {
+ if attrs, markerErr := bfs.Attributes(bfs.ctx, key); markerErr == nil {
+ return blobFileInfo{
+ name: directoryName(key),
+ mode: fs.ModeDir,
+ modTime: attrs.ModTime,
+ sys: attrs,
+ }, nil
+ } else if gcerrors.Code(markerErr) != gcerrors.NotFound {
+ return nil, blobErrToFsErr("Stat", path, markerErr)
+ }
+ }
+
+ attrs, err := bfs.Attributes(bfs.ctx, key)
+ // if there is no error and we have attributes, we can return a FileInfo for object
+ if err == nil {
+ return blobFileInfo{
+ name: pathpkg.Base(key),
+ size: attrs.Size,
+ mode: fs.ModeIrregular,
+ modTime: attrs.ModTime,
+ sys: attrs,
+ }, nil
+ }
+
+ if gcerrors.Code(err) != gcerrors.NotFound {
+ return nil, blobErrToFsErr("Stat", path, err)
+ }
+
+ marker := directoryMarker(key)
+ // if the marker exists, we can return a FileInfo for the directory
+ if marker != "" {
+ if attrs, markerErr := bfs.Attributes(bfs.ctx, marker); markerErr == nil {
+ return blobFileInfo{
+ name: directoryName(key),
+ mode: fs.ModeDir,
+ modTime: attrs.ModTime,
+ sys: attrs,
+ }, nil
+ } else if gcerrors.Code(markerErr) != gcerrors.NotFound {
+ return nil, blobErrToFsErr("Stat", path, markerErr)
+ }
+ }
+
+ prefix := marker
+ if prefix == "" {
+ prefix = key
+ }
+
+ iter := bfs.List(&blob.ListOptions{Prefix: prefix})
+ if obj, listErr := iter.Next(bfs.ctx); listErr == nil {
+ return blobFileInfo{
+ name: directoryName(key),
+ mode: fs.ModeDir,
+ sys: obj,
+ }, nil
+ } else if listErr != io.EOF {
+ return nil, blobErrToFsErr("Stat", path, listErr)
+ }
+
+ return nil, &fs.PathError{Op: "Stat", Path: path, Err: fs.ErrNotExist}
+}
+
+// Rename renames one file from oldpath to newpath, replacing newpath if it already exists.
+func (bfs *BlobFileIO) Rename(oldpath, newpath string) error {
+ oldKey, err := bfs.preprocess(oldpath)
+ if err != nil {
+ return &fs.PathError{Op: "Rename", Path: oldpath, Err: err}
+ }
+
+ newKey, err := bfs.preprocess(newpath)
+ if err != nil {
+ return &fs.PathError{Op: "Rename", Path: newpath, Err: err}
+ }
+
+ if err := bfs.Copy(bfs.ctx, newKey, oldKey, nil); err != nil {
+ return blobErrToFsErr("Rename", oldpath, err)
+ }
+
+ if err := bfs.Delete(bfs.ctx, oldKey); err != nil && gcerrors.Code(err) != gcerrors.NotFound {
+ return blobErrToFsErr("Rename", oldpath, err)
+ }
+
+ return nil
+}
+
+// RenameNoReplace renames a file or directory from oldpath to newpath, returning an error if newpath already exists.
Review Comment:
nit: this still says it renames a file or directory, but the implementation delegates to single-object `Rename`. Suggest wording like 'one file/object (non-recursive)' to avoid implying a recursive directory rename.
##########
catalog/hadoop/hadoop_minio_integration_test.go:
##########
@@ -0,0 +1,197 @@
+// 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.
+
+//go:build integration
+
+package hadoop_test
+
+import (
+ "context"
+ "fmt"
+ "strings"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/catalog"
+ "github.com/apache/iceberg-go/catalog/hadoop"
+ icebergio "github.com/apache/iceberg-go/io"
+ _ "github.com/apache/iceberg-go/io/gocloud"
+ "github.com/apache/iceberg-go/table"
+ "github.com/google/uuid"
+ "github.com/stretchr/testify/suite"
+)
+
+type HadoopMinIOIntegrationSuite struct {
+ suite.Suite
+
+ ctx context.Context
+ cat *hadoop.Catalog
+ props iceberg.Properties
+ warehouse string
+}
+
+func (s *HadoopMinIOIntegrationSuite) SetupTest() {
+ s.ctx = context.Background()
+ s.warehouse = fmt.Sprintf("s3a://warehouse/hadoop-minio/%s", uuid.NewString())
+ s.props = iceberg.Properties{
+ "allow-unsafe-commits": "true",
+ icebergio.S3Region: "local",
+ icebergio.S3AccessKeyID: "admin",
+ icebergio.S3SecretAccessKey: "password",
+ // Endpoint is provided by the env var AWS_S3_ENDPOINT
+ // from `make integration-env` or recipe.Start.
+ }
+
+ cat, err := hadoop.NewCatalog("hadoop_minio_test", s.warehouse, s.props)
+ s.Require().NoError(err)
+ s.cat = cat
+}
+
+func (s *HadoopMinIOIntegrationSuite) testSchema() *iceberg.Schema {
+ return 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},
+ )
+}
+
+func (s *HadoopMinIOIntegrationSuite) TearDownTest() {
+ fs, err := icebergio.LoadFS(s.ctx, s.props, s.warehouse)
+ if err != nil {
+ return
+ }
+
+ if remover, ok := fs.(icebergio.RemoveAllIO); ok {
+ _ = remover.RemoveAll(s.warehouse)
+ }
+}
+
+func (s *HadoopMinIOIntegrationSuite) TestNamespaceTableRoundTrip() {
+ ns := table.Identifier{"minio_ns"}
+ ident := table.Identifier{"minio_ns", "tbl"}
+
+ s.Require().NoError(s.cat.CreateNamespace(s.ctx, ns, nil))
+
+ exists, err := s.cat.CheckNamespaceExists(s.ctx, ns)
+ s.Require().NoError(err)
+ s.True(exists)
+
+ tbl, err := s.cat.CreateTable(s.ctx, ident, s.testSchema())
+ s.Require().NoError(err)
+ s.NotNil(tbl)
+ s.True(strings.HasPrefix(tbl.MetadataLocation(), s.warehouse+"/minio_ns/tbl/metadata/v1.metadata.json"))
+
+ loaded, err := s.cat.LoadTable(s.ctx, ident)
+ s.Require().NoError(err)
+ s.Equal(tbl.MetadataLocation(), loaded.MetadataLocation())
+ s.Equal(2, len(loaded.Schema().Fields()))
+
+ var tables []table.Identifier
+ for found, err := range s.cat.ListTables(s.ctx, ns) {
+ s.Require().NoError(err)
+ tables = append(tables, found)
+ }
+ s.Equal([]table.Identifier{ident}, tables)
+
+ s.Require().NoError(s.cat.DropTable(s.ctx, ident))
+
+ tableExists, err := s.cat.CheckTableExists(s.ctx, ident)
+ s.Require().NoError(err)
+ s.False(tableExists)
+}
+
+func (s *HadoopMinIOIntegrationSuite) TestListTablesEmptyNamespace() {
+ ns := table.Identifier{"empty_ns"}
+ s.Require().NoError(s.cat.CreateNamespace(s.ctx, ns, nil))
+
+ var tables []table.Identifier
+ for found, err := range s.cat.ListTables(s.ctx, ns) {
+ s.Require().NoError(err)
+ tables = append(tables, found)
+ }
+ s.Empty(tables)
+}
+
+func (s *HadoopMinIOIntegrationSuite) TestListTablesMissingNamespace() {
+ for _, err := range s.cat.ListTables(s.ctx, table.Identifier{"missing_ns"}) {
+ s.ErrorIs(err, catalog.ErrNoSuchNamespace)
+
+ break
+ }
+}
+
+func (s *HadoopMinIOIntegrationSuite) TestListTablesOnlyDirectNamespaceTables() {
+ parent := table.Identifier{"parent_ns"}
+ child := table.Identifier{"parent_ns", "child_ns"}
+ directTable := table.Identifier{"parent_ns", "direct_tbl"}
+ childTable := table.Identifier{"parent_ns", "child_ns", "nested_tbl"}
+
+ s.Require().NoError(s.cat.CreateNamespace(s.ctx, parent, nil))
+ s.Require().NoError(s.cat.CreateNamespace(s.ctx, child, nil))
+ _, err := s.cat.CreateTable(s.ctx, directTable, s.testSchema())
+ s.Require().NoError(err)
+ _, err = s.cat.CreateTable(s.ctx, childTable, s.testSchema())
+ s.Require().NoError(err)
+
+ var tables []table.Identifier
+ for found, err := range s.cat.ListTables(s.ctx, parent) {
+ s.Require().NoError(err)
+ tables = append(tables, found)
+ }
+ s.Equal([]table.Identifier{directTable}, tables)
+}
+
+func (s *HadoopMinIOIntegrationSuite) TestDropTableRemovesTableFromListingAndLoad() {
+ ns := table.Identifier{"drop_ns"}
+ ident := table.Identifier{"drop_ns", "tbl"}
+
+ s.Require().NoError(s.cat.CreateNamespace(s.ctx, ns, nil))
+ _, err := s.cat.CreateTable(s.ctx, ident, s.testSchema())
+ s.Require().NoError(err)
+
+ s.Require().NoError(s.cat.DropTable(s.ctx, ident))
+
+ _, err = s.cat.LoadTable(s.ctx, ident)
+ s.ErrorIs(err, catalog.ErrNoSuchTable)
+
+ var tables []table.Identifier
+ for found, err := range s.cat.ListTables(s.ctx, ns) {
+ s.Require().NoError(err)
+ tables = append(tables, found)
+ }
+ s.Empty(tables)
+}
+
+func (s *HadoopMinIOIntegrationSuite) TestDropNamespaceAfterDropTable() {
+ ns := table.Identifier{"drop_table_then_ns"}
+ ident := table.Identifier{"drop_table_then_ns", "tbl"}
+
+ s.Require().NoError(s.cat.CreateNamespace(s.ctx, ns, nil))
+ _, err := s.cat.CreateTable(s.ctx, ident, s.testSchema())
+ s.Require().NoError(err)
+
+ s.Require().NoError(s.cat.DropTable(s.ctx, ident))
+
+ // Ensure that the namespace can be dropped after dropping the table
+ // and that driectory marker logic is handled properly, not preventing
Review Comment:
nit: typo — 'driectory' should be 'directory'.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(catalog/hadoop): Add support for hadoop catalog fs interface methods to blob filesystem [iceberg-go]
Posted by "C-Loftus (via GitHub)" <gi...@apache.org>.
C-Loftus commented on code in PR #1350:
URL: https://github.com/apache/iceberg-go/pull/1350#discussion_r3509825722
##########
io/gocloud/blob.go:
##########
@@ -256,10 +319,178 @@ func (bfs *blobFileIO) DeleteFiles(ctx context.Context, paths []string) ([]strin
return deleted, errs
}
+// MkdirAll mimics creating a directory by creating a zero-length object for each component of the path
+func (bfs *BlobFileIO) MkdirAll(path string) error {
+ key, err := bfs.preprocess(path)
+ if err != nil {
+ return &fs.PathError{Op: "mkdir", Path: path, Err: err}
+ }
+
+ key = strings.Trim(key, "/")
+ if key == "" || key == "." {
+ return nil
+ }
+
+ parts := strings.Split(key, "/")
+ for idx := range parts {
+ marker := strings.Join(parts[:idx+1], "/") + "/"
+ if err := bfs.WriteAll(bfs.ctx, marker, nil, nil); err != nil {
+ return &fs.PathError{Op: "mkdir", Path: path, Err: err}
+ }
+ }
+
+ return nil
+}
+
+// ReadFile reads the contents of the file at the given path and returns it as a byte slice.
+func (bfs *BlobFileIO) ReadFile(path string) ([]byte, error) {
+ key, err := bfs.preprocess(path)
+ if err != nil {
+ return nil, &fs.PathError{Op: "ReadFile", Path: path, Err: err}
+ }
+
+ data, err := bfs.ReadAll(bfs.ctx, key)
+ if err != nil {
+ return nil, blobErrToFsErr("ReadFile", path, err)
+ }
+
+ return data, nil
+}
+
+// Stat interprets the input path as a directory or file and returns the corresponding FileInfo.
+// If the path does not exist, it returns fs.ErrNotExist
+func (bfs *BlobFileIO) Stat(path string) (fs.FileInfo, error) {
+ key, err := bfs.preprocess(path)
+ if err != nil {
+ return nil, &fs.PathError{Op: "Stat", Path: path, Err: err}
+ }
+
+ attrs, err := bfs.Attributes(bfs.ctx, key)
+ // if there is no error and we have attributes, we can return a FileInfo for object
+ if err == nil {
+ return blobFileInfo{
+ name: filepath.Base(key),
+ size: attrs.Size,
+ mode: fs.ModeIrregular,
+ modTime: attrs.ModTime,
+ sys: attrs,
+ }, nil
+ }
+
+ if gcerrors.Code(err) != gcerrors.NotFound {
+ return nil, blobErrToFsErr("Stat", path, err)
+ }
+
+ marker := directoryMarker(key)
+ // if the marker exists, we can return a FileInfo for the directory
+ if marker != "" {
+ if attrs, markerErr := bfs.Attributes(bfs.ctx, marker); markerErr == nil {
+ return blobFileInfo{
+ name: directoryName(key),
+ mode: fs.ModeDir,
+ modTime: attrs.ModTime,
+ sys: attrs,
+ }, nil
+ } else if gcerrors.Code(markerErr) != gcerrors.NotFound {
+ return nil, blobErrToFsErr("Stat", path, markerErr)
+ }
+ }
+
+ prefix := marker
+ if prefix == "" {
+ prefix = key
+ }
+
+ iter := bfs.List(&blob.ListOptions{Prefix: prefix})
+ if obj, listErr := iter.Next(bfs.ctx); listErr == nil {
+ return blobFileInfo{
+ name: directoryName(key),
+ mode: fs.ModeDir,
+ sys: obj,
+ }, nil
+ } else if listErr != io.EOF {
+ return nil, blobErrToFsErr("Stat", path, listErr)
+ }
+
+ return nil, &fs.PathError{Op: "Stat", Path: path, Err: fs.ErrNotExist}
+}
+
+// Rename renames a file or directory from oldpath to newpath, replacing newpath if it already exists.
+func (bfs *BlobFileIO) Rename(oldpath, newpath string) error {
+ oldKey, err := bfs.preprocess(oldpath)
+ if err != nil {
+ return &fs.PathError{Op: "Rename", Path: oldpath, Err: err}
+ }
+
+ newKey, err := bfs.preprocess(newpath)
+ if err != nil {
+ return &fs.PathError{Op: "Rename", Path: newpath, Err: err}
+ }
+
+ if err := bfs.Copy(bfs.ctx, newKey, oldKey, nil); err != nil {
+ return blobErrToFsErr("Rename", oldpath, err)
+ }
+
+ if err := bfs.Delete(bfs.ctx, oldKey); err != nil && gcerrors.Code(err) != gcerrors.NotFound {
+ return blobErrToFsErr("Rename", oldpath, err)
+ }
+
+ return nil
+}
+
+// RenameNoReplace renames a file or directory from oldpath to newpath, returning an error if newpath already exists.
+func (bfs *BlobFileIO) RenameNoReplace(oldpath, newpath string) error {
Review Comment:
Makes sense
> Is that follow-up issue tracked anywhere yet?
My understanding is that #1096 is currently scoped to baseline hadoop catalog blob store support and that there is not an existing issue for conditional writes. I think it probably makes sense to make that once #1096 is closed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(catalog/hadoop): Add support for hadoop catalog fs interface methods to blob filesystem [iceberg-go]
Posted by "C-Loftus (via GitHub)" <gi...@apache.org>.
C-Loftus commented on code in PR #1350:
URL: https://github.com/apache/iceberg-go/pull/1350#discussion_r3507823385
##########
catalog/hadoop/io.go:
##########
@@ -29,59 +28,15 @@ type HadoopCatalogFS interface {
icebergio.ListableIO
icebergio.ReadFileIO
icebergio.WriteFileIO
- StatIO
- RenameIO
- RenameNoReplaceIO
- RemoveAllIO
- MkdirAllIO
+ icebergio.StatIO
+ icebergio.RenameIO
+ icebergio.RenameNoReplaceIO
+ icebergio.RemoveAllIO
+ icebergio.MkdirAllIO
}
+// LocalFS can be used to implement a Hadoop catalog with a local filesystem.
var _ HadoopCatalogFS = (*icebergio.LocalFS)(nil)
-// StatIO is an extension of IO interface that includes the Stat
-// method for retrieving file or directory information
-type StatIO interface {
- icebergio.IO
-
- // 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
-// method for renaming (moving) files or directories; this must be
-// atomic and can be used for committing metadata updates
-type RenameIO interface {
- icebergio.IO
-
- Rename(oldpath, newpath string) error
-}
-
-// RenameNoReplaceIO is an extension of IO interface that includes the
-// RenameNoReplace method for atomically moving files only when the destination
-// path does not already exist. Implementations may require oldpath and newpath
-// to live on the same filesystem.
-type RenameNoReplaceIO interface {
- icebergio.IO
-
- RenameNoReplace(oldpath, newpath string) error
-}
-
-// RemoveAllIO is an extension of IO interface that includes the RemoveAll
-// method for removing a file path and any children recursively
-type RemoveAllIO interface {
- icebergio.IO
-
- RemoveAll(name 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
-}
+// BlobFileIO can be used to implement a Hadoop catalog with a blob storage bucket.
+var _ HadoopCatalogFS = (*gocloud.BlobFileIO)(nil)
Review Comment:
I have made `BlobFileIO` a public export now so that we can have a compile time interface assertion in the `hadoop` package.
##########
io/gocloud/blob.go:
##########
@@ -256,10 +319,178 @@ func (bfs *blobFileIO) DeleteFiles(ctx context.Context, paths []string) ([]strin
return deleted, errs
}
+// MkdirAll mimics creating a directory by creating a zero-length object for each component of the path
+func (bfs *BlobFileIO) MkdirAll(path string) error {
Review Comment:
There was a fair bit of discussion in #1096 regarding how to implement MkdirAll in a way that makes sense for blob stores that don't have a notion of a directory. Worth reviewing this function a bit closer to ensure that even if it passes all tests, it performs as desired.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(catalog/hadoop): Add support for hadoop catalog fs interface methods to blob filesystem [iceberg-go]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on code in PR #1350:
URL: https://github.com/apache/iceberg-go/pull/1350#discussion_r3539706592
##########
io/gocloud/blob.go:
##########
@@ -368,10 +429,194 @@ func (bfs *blobFileIO) DeleteFiles(ctx context.Context, paths []string) ([]strin
return deleted, errs
}
+// MkdirAll mimics creating a directory by creating a zero-length object for each component of the path
+func (bfs *BlobFileIO) MkdirAll(path string) error {
+ key, err := bfs.preprocess(path)
+ if err != nil {
+ return &fs.PathError{Op: "mkdir", Path: path, Err: err}
+ }
+
+ key = strings.Trim(key, "/")
+ if key == "" || key == "." {
+ return nil
+ }
+
+ parts := strings.Split(key, "/")
+ for idx := range parts {
Review Comment:
This writes a marker for every component including the leaf, so an empty namespace ends up with a lone `warehouse/ns/` object. After the tables are dropped that marker is all that's left, and `DropNamespace`'s emptiness walk sees it as a file entry named `ns` and returns `ErrNamespaceNotEmpty` — so an empty blob namespace can never be dropped.
I'd either skip writing the leaf marker here, or have the drop walk skip the namespace's own marker. A CreateNamespace→CreateTable→DropTable→DropNamespace test would have caught it and nothing currently exercises that path. wdyt?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(catalog/hadoop): Add support for hadoop catalog fs interface methods to blob filesystem [iceberg-go]
Posted by "C-Loftus (via GitHub)" <gi...@apache.org>.
C-Loftus commented on code in PR #1350:
URL: https://github.com/apache/iceberg-go/pull/1350#discussion_r3529397482
##########
io/gocloud/blob.go:
##########
@@ -256,10 +317,178 @@ func (bfs *blobFileIO) DeleteFiles(ctx context.Context, paths []string) ([]strin
return deleted, errs
}
+// MkdirAll mimics creating a directory by creating a zero-length object for each component of the path
+func (bfs *BlobFileIO) MkdirAll(path string) error {
+ key, err := bfs.preprocess(path)
+ if err != nil {
+ return &fs.PathError{Op: "mkdir", Path: path, Err: err}
+ }
+
+ key = strings.Trim(key, "/")
+ if key == "" || key == "." {
+ return nil
+ }
+
+ parts := strings.Split(key, "/")
+ for idx := range parts {
+ marker := strings.Join(parts[:idx+1], "/") + "/"
+ if err := bfs.WriteAll(bfs.ctx, marker, nil, nil); err != nil {
+ return &fs.PathError{Op: "mkdir", Path: path, Err: err}
+ }
+ }
+
+ return nil
+}
+
+// ReadFile reads the contents of the file at the given path and returns it as a byte slice.
+func (bfs *BlobFileIO) ReadFile(path string) ([]byte, error) {
+ key, err := bfs.preprocess(path)
+ if err != nil {
+ return nil, &fs.PathError{Op: "ReadFile", Path: path, Err: err}
+ }
+
+ data, err := bfs.ReadAll(bfs.ctx, key)
+ if err != nil {
+ return nil, blobErrToFsErr("ReadFile", path, err)
+ }
+
+ return data, nil
+}
+
+// Stat interprets the input path as a directory or file and returns the corresponding FileInfo.
+// If the path does not exist, it returns fs.ErrNotExist
+func (bfs *BlobFileIO) Stat(path string) (fs.FileInfo, error) {
Review Comment:
Addressed this in https://github.com/apache/iceberg-go/pull/1350/commits/58c4d690d17e332d8b8185fcb5e4a1d50c312d9f Let me know if that seems ok, thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(catalog/hadoop): Add support for hadoop catalog fs interface methods to blob filesystem [iceberg-go]
Posted by "C-Loftus (via GitHub)" <gi...@apache.org>.
C-Loftus commented on PR #1350:
URL: https://github.com/apache/iceberg-go/pull/1350#issuecomment-4893589803
Thanks for reviews all, @laskoviymishka / @zeroshade this should be ready for review again at your convenience
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(catalog/hadoop): Add support for hadoop catalog fs interface methods to blob filesystem [iceberg-go]
Posted by "C-Loftus (via GitHub)" <gi...@apache.org>.
C-Loftus commented on PR #1350:
URL: https://github.com/apache/iceberg-go/pull/1350#issuecomment-4910010703
@laskoviymishka I believe I should have addressed your feedback. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(catalog/hadoop): Add support for hadoop catalog fs interface methods to blob filesystem [iceberg-go]
Posted by "zeroshade (via GitHub)" <gi...@apache.org>.
zeroshade commented on code in PR #1350:
URL: https://github.com/apache/iceberg-go/pull/1350#discussion_r3521446876
##########
io/io.go:
##########
@@ -167,6 +167,54 @@ type ReadDirFile interface {
ReadDir(n int) ([]fs.DirEntry, error)
}
+// StatIO is an extension of IO interface that includes the Stat
+// method for retrieving file or directory information
+type StatIO interface {
+ IO
+
+ // 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
+// method for renaming (moving) files or directories; this must be
+// atomic and can be used for committing metadata updates
+type RenameIO interface {
+ IO
+
+ Rename(oldpath, newpath string) error
+}
+
+// RenameNoReplaceIO is an extension of IO interface that includes the
+// RenameNoReplace method for atomically moving files only when the destination
Review Comment:
[MAJOR] The RenameIO/RenameNoReplaceIO docs here promise atomic commit/no-replace semantics, but BlobFileIO.RenameNoReplace at io/gocloud/blob.go:441 is check-then-copy-then-delete (Stat(newpath)->Copy->Delete). Concurrent commits can both pass the Stat and silently last-writer-wins, so Hadoop commitMetadataFile will not reliably get fs.ErrExist for conflict detection. Please either implement provider-level conditional create/copy for no-replace, or soften these IO docs and the Hadoop allow-unsafe-commits error to state blob-backed commits are non-atomic and can lose concurrent updates.
##########
io/gocloud/blob.go:
##########
@@ -256,10 +317,178 @@ func (bfs *blobFileIO) DeleteFiles(ctx context.Context, paths []string) ([]strin
return deleted, errs
}
+// MkdirAll mimics creating a directory by creating a zero-length object for each component of the path
+func (bfs *BlobFileIO) MkdirAll(path string) error {
+ key, err := bfs.preprocess(path)
+ if err != nil {
+ return &fs.PathError{Op: "mkdir", Path: path, Err: err}
+ }
+
+ key = strings.Trim(key, "/")
+ if key == "" || key == "." {
+ return nil
+ }
+
+ parts := strings.Split(key, "/")
+ for idx := range parts {
+ marker := strings.Join(parts[:idx+1], "/") + "/"
+ if err := bfs.WriteAll(bfs.ctx, marker, nil, nil); err != nil {
+ return &fs.PathError{Op: "mkdir", Path: path, Err: err}
+ }
+ }
+
+ return nil
+}
+
+// ReadFile reads the contents of the file at the given path and returns it as a byte slice.
+func (bfs *BlobFileIO) ReadFile(path string) ([]byte, error) {
+ key, err := bfs.preprocess(path)
+ if err != nil {
+ return nil, &fs.PathError{Op: "ReadFile", Path: path, Err: err}
+ }
+
+ data, err := bfs.ReadAll(bfs.ctx, key)
+ if err != nil {
+ return nil, blobErrToFsErr("ReadFile", path, err)
+ }
+
+ return data, nil
+}
+
+// Stat interprets the input path as a directory or file and returns the corresponding FileInfo.
+// If the path does not exist, it returns fs.ErrNotExist
+func (bfs *BlobFileIO) Stat(path string) (fs.FileInfo, error) {
+ key, err := bfs.preprocess(path)
+ if err != nil {
+ return nil, &fs.PathError{Op: "Stat", Path: path, Err: err}
+ }
+
+ attrs, err := bfs.Attributes(bfs.ctx, key)
+ // if there is no error and we have attributes, we can return a FileInfo for object
+ if err == nil {
+ return blobFileInfo{
+ name: pathpkg.Base(key),
+ size: attrs.Size,
+ mode: fs.ModeIrregular,
+ modTime: attrs.ModTime,
+ sys: attrs,
+ }, nil
+ }
+
+ if gcerrors.Code(err) != gcerrors.NotFound {
+ return nil, blobErrToFsErr("Stat", path, err)
+ }
+
+ marker := directoryMarker(key)
+ // if the marker exists, we can return a FileInfo for the directory
+ if marker != "" {
+ if attrs, markerErr := bfs.Attributes(bfs.ctx, marker); markerErr == nil {
+ return blobFileInfo{
+ name: directoryName(key),
+ mode: fs.ModeDir,
+ modTime: attrs.ModTime,
+ sys: attrs,
+ }, nil
+ } else if gcerrors.Code(markerErr) != gcerrors.NotFound {
+ return nil, blobErrToFsErr("Stat", path, markerErr)
+ }
+ }
+
+ prefix := marker
+ if prefix == "" {
+ prefix = key
+ }
+
+ iter := bfs.List(&blob.ListOptions{Prefix: prefix})
+ if obj, listErr := iter.Next(bfs.ctx); listErr == nil {
+ return blobFileInfo{
+ name: directoryName(key),
+ mode: fs.ModeDir,
+ sys: obj,
+ }, nil
+ } else if listErr != io.EOF {
+ return nil, blobErrToFsErr("Stat", path, listErr)
+ }
+
+ return nil, &fs.PathError{Op: "Stat", Path: path, Err: fs.ErrNotExist}
+}
+
+// Rename renames one file from oldpath to newpath, replacing newpath if it already exists.
+func (bfs *BlobFileIO) Rename(oldpath, newpath string) error {
+ oldKey, err := bfs.preprocess(oldpath)
+ if err != nil {
+ return &fs.PathError{Op: "Rename", Path: oldpath, Err: err}
+ }
+
+ newKey, err := bfs.preprocess(newpath)
+ if err != nil {
+ return &fs.PathError{Op: "Rename", Path: newpath, Err: err}
+ }
+
+ if err := bfs.Copy(bfs.ctx, newKey, oldKey, nil); err != nil {
+ return blobErrToFsErr("Rename", oldpath, err)
+ }
+
+ if err := bfs.Delete(bfs.ctx, oldKey); err != nil && gcerrors.Code(err) != gcerrors.NotFound {
+ return blobErrToFsErr("Rename", oldpath, err)
+ }
+
+ return nil
+}
+
+// RenameNoReplace renames a file or directory from oldpath to newpath, returning an error if newpath already exists.
+func (bfs *BlobFileIO) RenameNoReplace(oldpath, newpath string) error {
+ if _, err := bfs.Stat(newpath); err == nil {
Review Comment:
[NIT] The comment says RenameNoReplace renames "a file or directory", but it delegates to single-object Rename and is not recursive. Please reword this to "one file/object" or document the non-recursive behavior.
##########
io/gocloud/blob.go:
##########
@@ -256,10 +317,178 @@ func (bfs *blobFileIO) DeleteFiles(ctx context.Context, paths []string) ([]strin
return deleted, errs
}
+// MkdirAll mimics creating a directory by creating a zero-length object for each component of the path
+func (bfs *BlobFileIO) MkdirAll(path string) error {
+ key, err := bfs.preprocess(path)
+ if err != nil {
+ return &fs.PathError{Op: "mkdir", Path: path, Err: err}
+ }
+
+ key = strings.Trim(key, "/")
+ if key == "" || key == "." {
+ return nil
+ }
+
+ parts := strings.Split(key, "/")
+ for idx := range parts {
+ marker := strings.Join(parts[:idx+1], "/") + "/"
+ if err := bfs.WriteAll(bfs.ctx, marker, nil, nil); err != nil {
+ return &fs.PathError{Op: "mkdir", Path: path, Err: err}
+ }
+ }
+
+ return nil
+}
+
+// ReadFile reads the contents of the file at the given path and returns it as a byte slice.
+func (bfs *BlobFileIO) ReadFile(path string) ([]byte, error) {
+ key, err := bfs.preprocess(path)
+ if err != nil {
+ return nil, &fs.PathError{Op: "ReadFile", Path: path, Err: err}
+ }
+
+ data, err := bfs.ReadAll(bfs.ctx, key)
+ if err != nil {
+ return nil, blobErrToFsErr("ReadFile", path, err)
+ }
+
+ return data, nil
+}
+
+// Stat interprets the input path as a directory or file and returns the corresponding FileInfo.
+// If the path does not exist, it returns fs.ErrNotExist
+func (bfs *BlobFileIO) Stat(path string) (fs.FileInfo, error) {
Review Comment:
[MINOR] This Stat implementation checks Attributes(key) before interpreting directory markers, so Stat(".../a/") on a marker created by MkdirAll returns IsDir()==false instead of matching local-FS directory behavior. Please TrimRight(key, "/") for stat/dir purposes, or detect keys ending in "/" and return directory info for marker objects.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(catalog/hadoop): Add support for hadoop catalog fs interface methods to blob filesystem [iceberg-go]
Posted by "C-Loftus (via GitHub)" <gi...@apache.org>.
C-Loftus commented on PR #1350:
URL: https://github.com/apache/iceberg-go/pull/1350#issuecomment-4857094061
Marking this as draft until I have confirmation all the integration tests pass
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(catalog/hadoop): Add support for hadoop catalog fs interface methods to blob filesystem [iceberg-go]
Posted by "C-Loftus (via GitHub)" <gi...@apache.org>.
C-Loftus commented on PR #1350:
URL: https://github.com/apache/iceberg-go/pull/1350#issuecomment-4928193221
Thanks for catching that on the comments! Should be fixed now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org