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