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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
