zeroshade commented on code in PR #2392:
URL: https://github.com/apache/iceberg-go/pull/2392#discussion_r4469929656
##########
io/gocloud/blobfs/blob.go:
##########
@@ -300,12 +407,17 @@ func (bfs *FileIO) Open(_ context.Context, path string)
(icebergio.File, error)
key, name := path, pathpkg.Base(path)
- r, err := bfs.NewReader(bfs.ctx, key, nil)
+ // Read only the attributes: a reader opened here would start a GET of
the
+ // whole object, which ReadAt, the way Parquet files are read, never
uses.
+ attrs, err := bfs.Attributes(bfs.ctx, key)
Review Comment:
`Attributes` is a HEAD request. For files read with `Open` + `Read`
(manifest lists, manifests), `main` costs 1 GET because `Read` consumes the GET
that `Open` starts; this costs HEAD + GET. Could the size come from the GET
response (`blob.Reader.Size()`) instead, with that response reused by `Read`
and by the first small-object `ReadAt`?
##########
io/gocloud/blobfs/blob.go:
##########
@@ -37,26 +37,131 @@ import (
"gocloud.dev/gcerrors"
)
+// wholeObjectReadMaxSize is the largest object ReadAt downloads whole on its
+// first call. Reading a Parquet file issues a ReadAt for its footer and for
+// every column chunk; for a small file one GET of the object is far cheaper
+// than one GET per column. The downloaded object is held until Close.
+const wholeObjectReadMaxSize = 8 << 20
+
// blobOpenFile describes a single open blob as a File.
// It implements the iceberg-go/io.File interface.
-// It is based on gocloud.dev/blob.iofsOpenFile which:
-// - Doesn't support the `io.ReaderAt` interface
-// - Is not externally accessible, so copied here
+//
+// Opening a blobOpenFile reads only the object's attributes. Read starts a
+// streaming GET from the current offset on first use, and ReadAt issues a
+// ranged GET per call, or downloads an object of at most
+// wholeObjectReadMaxSize once and serves every ReadAt from memory.
type blobOpenFile struct {
- *blob.Reader
-
name, key string
+ size int64
+ modTime time.Time
b *FileIO
ctx context.Context
+
+ mu sync.Mutex // guards r and pos, used by Read and Seek
+ r io.ReadCloser
+ pos int64
+
+ whole sync.Once // downloads data for ReadAt of a small object
+ data []byte
+ wholeErr error
}
-func (f *blobOpenFile) ReadAt(p []byte, off int64) (n int, err error) {
- var rdr io.ReadCloser
+func (f *blobOpenFile) rangeReader(off, length int64) (io.ReadCloser, error) {
if f.b.newRangeReader != nil {
- rdr, err = f.b.newRangeReader(f.ctx, f.key, off, int64(len(p)))
- } else {
- rdr, err = f.b.NewRangeReader(f.ctx, f.key, off, int64(len(p)),
nil)
+ return f.b.newRangeReader(f.ctx, f.key, off, length)
+ }
+
+ return f.b.NewRangeReader(f.ctx, f.key, off, length, nil)
+}
+
+func (f *blobOpenFile) Read(p []byte) (int, error) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+
+ if f.pos >= f.size {
+ return 0, io.EOF
}
+ if f.r == nil {
+ r, err := f.rangeReader(f.pos, -1)
+ if err != nil {
+ return 0, err
+ }
+ f.r = r
+ }
+ n, err := f.r.Read(p)
+ f.pos += int64(n)
+
+ return n, err
+}
+
+func (f *blobOpenFile) Seek(offset int64, whence int) (int64, error) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+
+ switch whence {
+ case io.SeekCurrent:
+ offset += f.pos
+ case io.SeekEnd:
+ offset += f.size
+ case io.SeekStart:
+ default:
+ return 0, fmt.Errorf("blobfs: invalid whence %d", whence)
+ }
+ if offset < 0 {
+ return 0, fmt.Errorf("blobfs: negative position %d", offset)
+ }
+ if offset != f.pos && f.r != nil {
+ err := f.r.Close()
+ f.r = nil
+ if err != nil {
+ return 0, err
+ }
+ }
+ f.pos = offset
+
+ return offset, nil
+}
+
+func (f *blobOpenFile) Close() error {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+
+ if f.r == nil {
+ return nil
+ }
+ err := f.r.Close()
+ f.r = nil
+
+ return err
+}
+
+func (f *blobOpenFile) ReadAt(p []byte, off int64) (int, error) {
+ if f.size > wholeObjectReadMaxSize {
+ return f.readRange(p, off)
+ }
+
+ f.whole.Do(func() {
+ data := make([]byte, f.size)
+ _, f.wholeErr = f.readRange(data, 0)
Review Comment:
`wholeErr` is stored by the `sync.Once`, so one transient GET failure makes
every later `ReadAt` on this file fail; on `main` each `ReadAt` was
independent. A zero-byte object also sends a zero-length GET here.
##########
io/gocloud/blobfs/blob.go:
##########
@@ -37,26 +37,131 @@ import (
"gocloud.dev/gcerrors"
)
+// wholeObjectReadMaxSize is the largest object ReadAt downloads whole on its
+// first call. Reading a Parquet file issues a ReadAt for its footer and for
+// every column chunk; for a small file one GET of the object is far cheaper
+// than one GET per column. The downloaded object is held until Close.
+const wholeObjectReadMaxSize = 8 << 20
Review Comment:
What is 8 MiB based on? Reading only the footer of a 7.5 MiB file goes from
about 2.3 KB to 7.5 MiB, and reading 2 of 20 columns from about 786 KiB to 7.5
MiB. Footer-only reads are what `AddDataFilesByPath` does.
##########
io/gocloud/blobfs/blob_test.go:
##########
@@ -1169,3 +1170,94 @@ func TestBlobFileIORemoveFallsBackToDirectoryMarker(t
*testing.T) {
require.NoError(t, err)
assert.False(t, exists)
}
+
+// countingFileIO returns a FileIO over bucket that counts the GETs it issues.
+func countingFileIO(ctx context.Context, bucket *blob.Bucket, gets *int)
*FileIO {
+ return &FileIO{
+ Bucket: bucket,
+ extractObject: identityObjectLocation,
+ ctx: ctx,
+ newRangeReader: func(ctx context.Context, key string, offset,
length int64) (io.ReadCloser, error) {
+ *gets++
+
+ return bucket.NewRangeReader(ctx, key, offset, length,
nil)
+ },
+ }
+}
+
+func TestOpenReadAtGets(t *testing.T) {
+ ctx := context.Background()
+ bucket := memblob.OpenBucket(nil)
+ defer bucket.Close()
+
+ small := []byte("0123456789")
+ large := bytes.Repeat([]byte{7}, wholeObjectReadMaxSize+1)
+ require.NoError(t, bucket.WriteAll(ctx, "small", small, nil))
+ require.NoError(t, bucket.WriteAll(ctx, "large", large, nil))
+
+ tests := []struct {
+ key string
+ content []byte
+ wantGets int
+ }{
+ {"small", small, 1},
+ {"large", large, 3},
+ }
+ for _, tt := range tests {
+ t.Run(tt.key, func(t *testing.T) {
+ var gets int
+ f, err := countingFileIO(ctx, bucket, &gets).Open(ctx,
tt.key)
+ require.NoError(t, err)
+ defer f.Close()
+
+ info, err := f.Stat()
+ require.NoError(t, err)
+ assert.Equal(t, int64(len(tt.content)), info.Size())
+ assert.Zero(t, gets, "Open must not read the object")
Review Comment:
`countingFileIO` only counts `newRangeReader`, so this passes even though
`Open` now sends a HEAD. Counting `Attributes` too would make the test cover
the real request count.
--
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]
