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: [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]
