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