This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new b8e4b2eeb refactor(io)!: add context to StatIO.Stat (#2448)
b8e4b2eeb is described below
commit b8e4b2eeb6e113e72c408506f41123e6eab041b6
Author: Minh Vu <[email protected]>
AuthorDate: Mon Nov 9 18:45:33 2026 +0100
refactor(io)!: add context to StatIO.Stat (#2448)
* refactor(io)!: add context to StatIO.Stat
* fix(io): preserve stat operation context and propagate errors
---
catalog/hadoop/hadoop_context_test.go | 17 +++++-
catalog/hadoop/hadoop_s3_integration_test.go | 6 +-
catalog/hadoop/hadoop_test.go | 2 +-
catalog/hadoop/namespaces.go | 29 +++++----
catalog/hadoop/tables.go | 6 +-
io/gocloud/blobfs/blob.go | 12 ++--
io/gocloud/blobfs/blob_test.go | 91 ++++++++++++++++++++++++++--
io/io.go | 8 ++-
io/local.go | 2 +-
io/local_test.go | 13 ++++
table/conflict_validation.go | 8 +--
table/conflict_validation_test.go | 18 +++++-
table/snapshots.go | 2 +-
table/snapshots_test.go | 23 +++++++
website/src/releases.md | 1 +
15 files changed, 195 insertions(+), 43 deletions(-)
diff --git a/catalog/hadoop/hadoop_context_test.go
b/catalog/hadoop/hadoop_context_test.go
index 6f6297303..476c7c773 100644
--- a/catalog/hadoop/hadoop_context_test.go
+++ b/catalog/hadoop/hadoop_context_test.go
@@ -53,12 +53,12 @@ func (f *contextBoundHadoopFS) localPath(name string)
string {
return filepath.Join(f.root, parsed.Host,
filepath.FromSlash(strings.TrimPrefix(parsed.Path, "/")))
}
-func (f *contextBoundHadoopFS) Stat(name string) (fs.FileInfo, error) {
- if err := f.ctx.Err(); err != nil {
+func (f *contextBoundHadoopFS) Stat(ctx context.Context, name string)
(fs.FileInfo, error) {
+ if err := ctx.Err(); err != nil {
return nil, err
}
- return f.LocalFS.Stat(f.localPath(name))
+ return f.LocalFS.Stat(ctx, f.localPath(name))
}
func (f *contextBoundHadoopFS) MkdirAll(name string) error {
@@ -137,6 +137,17 @@ func TestNewCatalogFilesystemContextLifetime(t *testing.T)
{
require.ErrorIs(t, ctx.Err(), context.Canceled)
require.NoError(t, factoryContext.Err())
require.NoError(t,
cat.CreateNamespace(context.Background(), table.Identifier{"ns"}, nil))
+ canceledOperationCtx, cancelOperation :=
context.WithCancel(context.Background())
+ cancelOperation()
+ _, err = cat.CheckNamespaceExists(canceledOperationCtx,
table.Identifier{"ns"})
+ require.ErrorIs(t, err, context.Canceled)
+ _, err = cat.ListNamespaces(canceledOperationCtx,
table.Identifier{"ns"})
+ require.ErrorIs(t, err, context.Canceled)
+ err = cat.CreateNamespace(canceledOperationCtx,
table.Identifier{"canceled"}, nil)
+ require.ErrorIs(t, err, context.Canceled)
+ exists, err :=
cat.CheckNamespaceExists(context.Background(), table.Identifier{"canceled"})
+ require.NoError(t, err)
+ require.False(t, exists)
fs := cat.(*Catalog).filesystem
location := warehouse + "/ns/after-cancel.txt"
diff --git a/catalog/hadoop/hadoop_s3_integration_test.go
b/catalog/hadoop/hadoop_s3_integration_test.go
index 4bc8135fd..78de42e65 100644
--- a/catalog/hadoop/hadoop_s3_integration_test.go
+++ b/catalog/hadoop/hadoop_s3_integration_test.go
@@ -197,11 +197,11 @@ func (s *HadoopS3IntegrationSuite)
TestPurgeTableRemovesTableRoot() {
s.Require().NoError(s.cat.PurgeTable(s.ctx, ident))
- _, err = statFS.Stat(tbl.MetadataLocation())
+ _, err = statFS.Stat(s.ctx, tbl.MetadataLocation())
s.ErrorIs(err, fs.ErrNotExist)
- _, err = statFS.Stat(dataPath)
+ _, err = statFS.Stat(s.ctx, dataPath)
s.ErrorIs(err, fs.ErrNotExist)
- _, err = statFS.Stat(tbl.Location())
+ _, err = statFS.Stat(s.ctx, tbl.Location())
s.ErrorIs(err, fs.ErrNotExist)
exists, err := s.cat.CheckTableExists(s.ctx, ident)
diff --git a/catalog/hadoop/hadoop_test.go b/catalog/hadoop/hadoop_test.go
index 571379503..f0d1c4fe1 100644
--- a/catalog/hadoop/hadoop_test.go
+++ b/catalog/hadoop/hadoop_test.go
@@ -973,7 +973,7 @@ type statFailingFS struct {
statCalls int
}
-func (f *statFailingFS) Stat(string) (fs.FileInfo, error) {
+func (f *statFailingFS) Stat(context.Context, string) (fs.FileInfo, error) {
f.statCalls++
return nil, errors.New("stat should not be called")
diff --git a/catalog/hadoop/namespaces.go b/catalog/hadoop/namespaces.go
index 375107ea5..13ce9c889 100644
--- a/catalog/hadoop/namespaces.go
+++ b/catalog/hadoop/namespaces.go
@@ -32,7 +32,7 @@ import (
"github.com/apache/iceberg-go/table"
)
-func (c *Catalog) CreateNamespace(_ context.Context, ns table.Identifier,
props iceberg.Properties) error {
+func (c *Catalog) CreateNamespace(ctx context.Context, ns table.Identifier,
props iceberg.Properties) error {
if err := validateIdentifier(ns); err != nil {
return err
}
@@ -42,7 +42,7 @@ func (c *Catalog) CreateNamespace(_ context.Context, ns
table.Identifier, props
}
// Raise an error if the namespace already exists
- if err := c.checkedMkdirAll(ns); err != nil {
+ if err := c.checkedMkdirAll(ctx, ns); err != nil {
if errors.Is(err, fs.ErrExist) {
return fmt.Errorf("%w: %s",
catalog.ErrNamespaceAlreadyExists, strings.Join(ns, "."))
}
@@ -103,14 +103,14 @@ func (c *Catalog) DropNamespace(ctx context.Context, ns
table.Identifier) error
return c.filesystem.Remove(ctx, path)
}
-func (c *Catalog) CheckNamespaceExists(_ context.Context, ns table.Identifier)
(bool, error) {
+func (c *Catalog) CheckNamespaceExists(ctx context.Context, ns
table.Identifier) (bool, error) {
if err := validateIdentifier(ns); err != nil {
return false, err
}
path := c.namespaceToPath(ns)
- info, err := c.filesystem.Stat(path)
+ info, err := c.filesystem.Stat(ctx, path)
if errors.Is(err, fs.ErrNotExist) {
return false, nil
}
@@ -136,10 +136,13 @@ func (c *Catalog) ListNamespaces(ctx context.Context,
parent table.Identifier) (
} else {
path = c.namespaceToPath(parent)
- info, err := c.filesystem.Stat(path)
+ info, err := c.filesystem.Stat(ctx, path)
if errors.Is(err, fs.ErrNotExist) || (err == nil &&
!info.IsDir()) {
return nil, fmt.Errorf("%w: %s",
catalog.ErrNoSuchNamespace, strings.Join(parent, "."))
}
+ if err != nil {
+ return nil, fmt.Errorf("hadoop catalog: failed to stat
namespace: %w", err)
+ }
}
result := []table.Identifier{}
@@ -184,14 +187,14 @@ func (c *Catalog) ListNamespaces(ctx context.Context,
parent table.Identifier) (
// LoadNamespaceProperties returns a synthetic "location" property for the
// namespace. This is a Go-only convenience — the Hadoop catalog does not
// persist user-defined namespace properties.
-func (c *Catalog) LoadNamespaceProperties(_ context.Context, ns
table.Identifier) (iceberg.Properties, error) {
+func (c *Catalog) LoadNamespaceProperties(ctx context.Context, ns
table.Identifier) (iceberg.Properties, error) {
if err := validateIdentifier(ns); err != nil {
return nil, err
}
path := c.namespaceToPath(ns)
- info, err := c.filesystem.Stat(path)
+ info, err := c.filesystem.Stat(ctx, path)
if errors.Is(err, fs.ErrNotExist) || (err == nil && !info.IsDir()) {
return nil, fmt.Errorf("%w: %s", catalog.ErrNoSuchNamespace,
strings.Join(ns, "."))
}
@@ -221,22 +224,24 @@ func (c *Catalog) UpdateNamespaceProperties(_
context.Context, _ table.Identifie
// all subdirectories of a given identifier path exist before creating the
full path.
// This function is not atomic and may not return ErrNamespaceAlreadyExists if
// called concurrently with other calls that change the same identifier
-func (c *Catalog) checkedMkdirAll(id table.Identifier) error {
+func (c *Catalog) checkedMkdirAll(ctx context.Context, id table.Identifier)
error {
path := c.namespaceToPath(id)
// Start at index 1 to skip the root warehouse directory
for pathIndex := 1; pathIndex < len(id); pathIndex++ {
subPath := id[:pathIndex]
parentPath := c.namespaceToPath(subPath)
- if _, err := c.filesystem.Stat(parentPath); err != nil {
+ if _, err := c.filesystem.Stat(ctx, parentPath); err != nil {
return err
}
}
// Check the final element in the path
- if _, err := c.filesystem.Stat(path); err == nil {
- // If there is no error and stat returns successfully,
- // it means that it must already exist
+ _, err := c.filesystem.Stat(ctx, path)
+ if err == nil {
return fs.ErrExist
}
+ if !errors.Is(err, fs.ErrNotExist) {
+ return err
+ }
return c.filesystem.MkdirAll(path)
}
diff --git a/catalog/hadoop/tables.go b/catalog/hadoop/tables.go
index 783fc5005..0bc34e1c1 100644
--- a/catalog/hadoop/tables.go
+++ b/catalog/hadoop/tables.go
@@ -58,7 +58,7 @@ func (c *Catalog) CreateTable(ctx context.Context, ident
table.Identifier, sc *i
ns := catalog.NamespaceFromIdent(ident)
nsPath := c.namespaceToPath(ns)
- info, err := c.filesystem.Stat(nsPath)
+ info, err := c.filesystem.Stat(ctx, nsPath)
if errors.Is(err, fs.ErrNotExist) || (err == nil && !info.IsDir()) {
return nil, fmt.Errorf("%w: %s", catalog.ErrNoSuchNamespace,
strings.Join(ns, "."))
}
@@ -287,7 +287,7 @@ func (c *Catalog) commitMetadataFile(ctx context.Context,
ident table.Identifier
conflictErr, strings.Join(ident, "."),
existingPath)
}
- claimInfo, err := c.filesystem.Stat(claimPath)
+ claimInfo, err := c.filesystem.Stat(ctx, claimPath)
if err != nil {
if errors.Is(err, fs.ErrNotExist) {
continue
@@ -371,7 +371,7 @@ func (c *Catalog) ListTables(ctx context.Context, ns
table.Identifier) iter.Seq2
nsPath := c.namespaceToPath(ns)
- info, err := c.filesystem.Stat(nsPath)
+ info, err := c.filesystem.Stat(ctx, nsPath)
if errors.Is(err, fs.ErrNotExist) || (err == nil &&
!info.IsDir()) {
yield(nil, fmt.Errorf("%w: %s",
catalog.ErrNoSuchNamespace, strings.Join(ns, ".")))
diff --git a/io/gocloud/blobfs/blob.go b/io/gocloud/blobfs/blob.go
index beee3912a..07c38ea8c 100644
--- a/io/gocloud/blobfs/blob.go
+++ b/io/gocloud/blobfs/blob.go
@@ -686,7 +686,7 @@ func (bfs *FileIO) ReadFile(path string) ([]byte, error) {
// 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 *FileIO) Stat(path string) (fs.FileInfo, error) {
+func (bfs *FileIO) Stat(ctx context.Context, path string) (fs.FileInfo, error)
{
key, err := bfs.preprocess(path)
if err != nil {
return nil, &fs.PathError{Op: "Stat", Path: path, Err: err}
@@ -696,7 +696,7 @@ func (bfs *FileIO) Stat(path string) (fs.FileInfo, error) {
// 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 {
+ if attrs, markerErr := bfs.Attributes(ctx, key); markerErr ==
nil {
return blobFileInfo{
name: directoryName(key),
mode: fs.ModeDir,
@@ -708,7 +708,7 @@ func (bfs *FileIO) Stat(path string) (fs.FileInfo, error) {
}
}
- attrs, err := bfs.Attributes(bfs.ctx, key)
+ attrs, err := bfs.Attributes(ctx, key)
// if there is no error and we have attributes, we can return a
FileInfo for object
if err == nil {
return blobFileInfo{
@@ -727,7 +727,7 @@ func (bfs *FileIO) Stat(path string) (fs.FileInfo, error) {
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 {
+ if attrs, markerErr := bfs.Attributes(ctx, marker); markerErr
== nil {
return blobFileInfo{
name: directoryName(key),
mode: fs.ModeDir,
@@ -745,7 +745,7 @@ func (bfs *FileIO) Stat(path string) (fs.FileInfo, error) {
}
iter := bfs.List(&blob.ListOptions{Prefix: prefix})
- if obj, listErr := iter.Next(bfs.ctx); listErr == nil {
+ if obj, listErr := iter.Next(ctx); listErr == nil {
return blobFileInfo{
name: directoryName(key),
mode: fs.ModeDir,
@@ -783,7 +783,7 @@ func (bfs *FileIO) Rename(oldpath, newpath string) error {
// RenameNoReplace renames one file/object (non-recursive) from oldpath to
newpath, returning an error if newpath already exists.
func (bfs *FileIO) RenameNoReplace(oldpath, newpath string) error {
- if _, err := bfs.Stat(newpath); err == nil {
+ if _, err := bfs.Stat(bfs.ctx, newpath); err == nil {
return &fs.PathError{Op: "RenameNoReplace", Path: newpath, Err:
fs.ErrExist}
// if the error is just that the file doesn't exist, we can
continue with the rename
// since we will be creating the new file. If the error is
something else, we should return it.
diff --git a/io/gocloud/blobfs/blob_test.go b/io/gocloud/blobfs/blob_test.go
index ce3edaab2..68976854b 100644
--- a/io/gocloud/blobfs/blob_test.go
+++ b/io/gocloud/blobfs/blob_test.go
@@ -228,6 +228,87 @@ func TestBlobFileIOWriteOperationsUseOperationContext(t
*testing.T) {
}
}
+type statContextKey struct{}
+
+type contextAwareAttributesBucket struct {
+ driver.Bucket
+ attributesCtx chan context.Context
+}
+
+func (b *contextAwareAttributesBucket) ErrorCode(error) gcerrors.ErrorCode {
return gcerrors.Unknown }
+
+func (b *contextAwareAttributesBucket) Attributes(ctx context.Context, _
string) (*driver.Attributes, error) {
+ b.attributesCtx <- ctx
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+
+ return &driver.Attributes{Size: 1}, nil
+}
+
+func (b *contextAwareAttributesBucket) Close() error { return nil }
+
+func TestBlobFileIOStatUsesOperationContext(t *testing.T) {
+ driverBucket := &contextAwareAttributesBucket{attributesCtx: make(chan
context.Context, 1)}
+ bucket := blob.NewBucket(driverBucket)
+ t.Cleanup(func() { require.NoError(t, bucket.Close()) })
+
+ lifetimeCtx := context.WithValue(context.Background(),
statContextKey{}, "lifetime")
+ fileIO := New(lifetimeCtx, bucket,
DefaultObjectLocationExtractor("my-bucket"))
+ ctx, cancel :=
context.WithCancel(context.WithValue(context.Background(), statContextKey{},
"operation"))
+ cancel()
+
+ _, err := fileIO.Stat(ctx, "s3://my-bucket/data/file.parquet")
+ assert.ErrorIs(t, err, context.Canceled)
+
+ gotCtx := <-driverBucket.attributesCtx
+ assert.Equal(t, "operation", gotCtx.Value(statContextKey{}))
+}
+
+type statListContextBucket struct {
+ driver.Bucket
+ listCtx context.Context
+}
+
+func (*statListContextBucket) ErrorCode(err error) gcerrors.ErrorCode {
+ if errors.Is(err, fs.ErrNotExist) {
+ return gcerrors.NotFound
+ }
+
+ return gcerrors.Unknown
+}
+
+func (*statListContextBucket) Attributes(context.Context, string)
(*driver.Attributes, error) {
+ return nil, fs.ErrNotExist
+}
+
+func (b *statListContextBucket) ListPaged(ctx context.Context, _
*driver.ListOptions) (*driver.ListPage, error) {
+ b.listCtx = ctx
+
+ return &driver.ListPage{
+ Objects: []*driver.ListObject{{Key: "data/child.parquet"}},
+ }, nil
+}
+
+func (*statListContextBucket) Close() error { return nil }
+
+func TestBlobFileIOStatDirectoryListingUsesOperationContext(t *testing.T) {
+ driverBucket := &statListContextBucket{}
+ bucket := blob.NewBucket(driverBucket)
+ t.Cleanup(func() { require.NoError(t, bucket.Close()) })
+
+ lifetimeCtx, cancelLifetime :=
context.WithCancel(context.WithValue(context.Background(), statContextKey{},
"lifetime"))
+ fileIO := New(lifetimeCtx, bucket,
DefaultObjectLocationExtractor("my-bucket"))
+ cancelLifetime()
+ operationCtx := context.WithValue(context.Background(),
statContextKey{}, "operation")
+
+ info, err := fileIO.Stat(operationCtx, "s3://my-bucket/data")
+ require.NoError(t, err)
+ require.True(t, info.IsDir())
+ require.NotNil(t, driverBucket.listCtx)
+ assert.Equal(t, "operation",
driverBucket.listCtx.Value(statContextKey{}))
+}
+
func TestBlobFileIOWriteFileHonorsOperationCancellation(t *testing.T) {
t.Parallel()
@@ -948,22 +1029,22 @@ func TestBlobFileIOStat(t *testing.T) {
bfs := New(ctx, bucket, DefaultObjectLocationExtractor("test-bucket"))
- fileInfo, err := bfs.Stat("s3://test-bucket/data/file.parquet")
+ fileInfo, err := bfs.Stat(ctx, "s3://test-bucket/data/file.parquet")
require.NoError(t, err)
assert.Equal(t, "file.parquet", fileInfo.Name())
assert.Equal(t, int64(len("content")), fileInfo.Size())
assert.False(t, fileInfo.IsDir())
- dirInfo, err := bfs.Stat("s3://test-bucket/data")
+ dirInfo, err := bfs.Stat(ctx, "s3://test-bucket/data")
require.NoError(t, err)
assert.Equal(t, "data", dirInfo.Name())
assert.True(t, dirInfo.IsDir())
require.NoError(t, bfs.MkdirAll("s3://test-bucket/foo"))
- markerInfoWithSlash, err := bfs.Stat("s3://test-bucket/foo/")
+ markerInfoWithSlash, err := bfs.Stat(ctx, "s3://test-bucket/foo/")
require.NoError(t, err)
assert.Equal(t, "foo", markerInfoWithSlash.Name())
- markerInfoWithoutSlash, err := bfs.Stat("s3://test-bucket/foo")
+ markerInfoWithoutSlash, err := bfs.Stat(ctx, "s3://test-bucket/foo")
require.NoError(t, err)
assert.Equal(t, "foo", markerInfoWithoutSlash.Name())
// The trailing slash in the path should not affect the
@@ -971,7 +1052,7 @@ func TestBlobFileIOStat(t *testing.T) {
assert.True(t, markerInfoWithSlash.IsDir())
assert.True(t, markerInfoWithoutSlash.IsDir())
- _, err = bfs.Stat("s3://test-bucket/missing")
+ _, err = bfs.Stat(ctx, "s3://test-bucket/missing")
require.ErrorIs(t, err, fs.ErrNotExist)
}
diff --git a/io/io.go b/io/io.go
index 9ad825b62..bafa70825 100644
--- a/io/io.go
+++ b/io/io.go
@@ -181,9 +181,11 @@ type ReadDirFile interface {
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)
+ // Stat returns a FileInfo describing the named file or directory, or
an error
+ // satisfying errors.Is(err, fs.ErrNotExist) if the file does not
exist. The
+ // context bounds remote stat operations; local filesystem
implementations may
+ // ignore it.
+ Stat(ctx context.Context, path string) (fs.FileInfo, error)
}
// RenameIO is an extension of IO interface that includes the Rename
diff --git a/io/local.go b/io/local.go
index 43efbceed..108514b5a 100644
--- a/io/local.go
+++ b/io/local.go
@@ -183,7 +183,7 @@ func (LocalFS) Mkdir(name string) error {
return os.Mkdir(path, 0o755)
}
-func (LocalFS) Stat(name string) (fs.FileInfo, error) {
+func (LocalFS) Stat(_ context.Context, name string) (fs.FileInfo, error) {
path, err := localPath(name)
if err != nil {
return nil, err
diff --git a/io/local_test.go b/io/local_test.go
index b45c1ec37..7f1b21351 100644
--- a/io/local_test.go
+++ b/io/local_test.go
@@ -159,6 +159,19 @@ func TestLocalFSParsesFileURIs(t *testing.T) {
}
}
+func TestLocalFSStatIgnoresContext(t *testing.T) {
+ t.Parallel()
+
+ path := filepath.Join(t.TempDir(), "metadata.json")
+ require.NoError(t, os.WriteFile(path, []byte("metadata"), 0o600))
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+
+ info, err := (LocalFS{}).Stat(ctx, path)
+ require.NoError(t, err)
+ assert.Equal(t, int64(len("metadata")), info.Size())
+}
+
func TestLocalFSParsesEscapedFileURIsConsistently(t *testing.T) {
t.Parallel()
diff --git a/table/conflict_validation.go b/table/conflict_validation.go
index 8850f8095..aa1bc0318 100644
--- a/table/conflict_validation.go
+++ b/table/conflict_validation.go
@@ -249,7 +249,7 @@ func (c *conflictManifestIO) Open(ctx context.Context, name
string) (iceio.File,
info, statErr := f.Stat()
if statErr != nil || info == nil || info.Size() < 0 {
if statIO, ok := c.base.(iceio.StatIO); ok {
- info, statErr = statIO.Stat(name)
+ info, statErr = statIO.Stat(ctx, name)
}
}
if statErr != nil || info == nil || info.Size() < 0 {
@@ -271,16 +271,16 @@ func (c *conflictManifestIO) Remove(context.Context,
string) error {
return errConflictManifestIOReadOnly
}
-func (c *conflictManifestIO) Stat(name string) (fs.FileInfo, error) {
+func (c *conflictManifestIO) Stat(ctx context.Context, name string)
(fs.FileInfo, error) {
if data, ok := c.files[name]; ok {
return conflictManifestFileInfo{name: name, size:
int64(len(data))}, nil
}
if statIO, ok := c.base.(iceio.StatIO); ok {
- return statIO.Stat(name)
+ return statIO.Stat(ctx, name)
}
- f, err := c.Open(context.Background(), name)
+ f, err := c.Open(ctx, name)
if err != nil {
return nil, err
}
diff --git a/table/conflict_validation_test.go
b/table/conflict_validation_test.go
index 07604e298..855010959 100644
--- a/table/conflict_validation_test.go
+++ b/table/conflict_validation_test.go
@@ -868,9 +868,11 @@ func
TestValidateNoConflictingDataFiles_SnapshotIsolationIsNoOp(t *testing.T) {
type conflictValidationStatIO struct {
*trackingCallsIO
+ statCtx context.Context
}
-func (c *conflictValidationStatIO) Stat(name string) (fs.FileInfo, error) {
+func (c *conflictValidationStatIO) Stat(ctx context.Context, name string)
(fs.FileInfo, error) {
+ c.statCtx = ctx
data, ok := c.files[name]
if !ok {
return nil, fs.ErrNotExist
@@ -879,6 +881,20 @@ func (c *conflictValidationStatIO) Stat(name string)
(fs.FileInfo, error) {
return conflictManifestFileInfo{name: name, size: int64(len(data))}, nil
}
+func TestConflictManifestIOStatPassesContext(t *testing.T) {
+ type key struct{}
+ ctx := context.WithValue(t.Context(), key{}, "caller")
+ tio := newTrackingCallsIO()
+ tio.files["manifest.avro"] = []byte("manifest")
+ base := &conflictValidationStatIO{trackingCallsIO: tio}
+ cache := newConflictManifestIO(base)
+
+ _, err := cache.Stat(ctx, "manifest.avro")
+ require.NoError(t, err)
+ require.NotNil(t, base.statCtx)
+ assert.Equal(t, "caller", base.statCtx.Value(key{}))
+}
+
func TestConflictValidationReusesSharedManifestReads(t *testing.T) {
tio := newTrackingCallsIO()
dir := filepath.ToSlash(t.TempDir())
diff --git a/table/snapshots.go b/table/snapshots.go
index b5a2d30b1..188516d4d 100644
--- a/table/snapshots.go
+++ b/table/snapshots.go
@@ -469,7 +469,7 @@ func embeddedManifestLength(ctx context.Context, fio
iceio.IO, path string) (_ i
}
if statIO, ok := fio.(iceio.StatIO); ok {
- info, err := statIO.Stat(path)
+ info, err := statIO.Stat(ctx, path)
if err != nil {
return 0, fmt.Errorf("could not stat embedded manifest
%q: %w", path, err)
}
diff --git a/table/snapshots_test.go b/table/snapshots_test.go
index d6188671e..c3d2c4bb5 100644
--- a/table/snapshots_test.go
+++ b/table/snapshots_test.go
@@ -21,6 +21,7 @@ import (
"bytes"
"context"
"encoding/json"
+ "io/fs"
"os"
"testing"
@@ -600,3 +601,25 @@ func TestSnapshotManifestsPassContextToOpen(t *testing.T) {
})
}
}
+
+type contextStatProbe struct {
+ iomock.ContextProbe
+}
+
+func (p *contextStatProbe) Stat(ctx context.Context, _ string) (fs.FileInfo,
error) {
+ p.Ctx = ctx
+
+ return nil, iomock.ErrContextProbe
+}
+
+func TestSnapshotEmbeddedManifestStatPassesContext(t *testing.T) {
+ type key struct{}
+ ctx := context.WithValue(t.Context(), key{}, "caller")
+ probe := &contextStatProbe{}
+ snapshot := table.Snapshot{SnapshotID: 1, ManifestLocations:
[]string{"s3://bucket/m1.avro"}}
+
+ _, err := snapshot.Manifests(ctx, probe)
+ require.ErrorIs(t, err, iomock.ErrContextProbe)
+ require.NotNil(t, probe.Ctx)
+ assert.Equal(t, "caller", probe.Ctx.Value(key{}))
+}
diff --git a/website/src/releases.md b/website/src/releases.md
index f4885948d..098f8a25a 100644
--- a/website/src/releases.md
+++ b/website/src/releases.md
@@ -42,6 +42,7 @@ Per-release notes (highlights, breaking changes,
contributors) are published on
### Unreleased compatibility notes
+- **`io.StatIO.Stat` accepts a context.** The method now takes a
`context.Context` before the path, so custom `StatIO` implementations and
callers must pass the operation context.
- **Schemas store binary and fixed defaults as hex strings.** `NewSchema`,
`NewSchemaWithIdentifiers` and `UpdateSchema.AddColumn` (through `Apply()`)
convert `[]byte`, `BinaryLiteral` and `FixedLiteral` defaults of binary and
fixed fields, including fields nested in structs, lists and maps, to the
lowercase hex string iceberg-go writes to metadata JSON, so a field read back
from a schema holds a `string` such as `"0aff"`. Code that type-asserts these
defaults to `[]byte` must decode the [...]
- **Predicates reject NaN literals when bound.** `EqualTo`, `NotEqualTo`,
`LessThan`, `LessThanEqual`, `GreaterThan`, `GreaterThanEqual`, `IsIn`, `NotIn`
and `LiteralPredicate`/`SetPredicate` with a NaN float literal now fail in
`Bind` with `ErrInvalidArgument`, as Java does. That includes the string
`"NaN"` on a float column and a NaN on a decimal column, which used to bind as
zero. Scans, `ExpressionEvaluator` and the metrics evaluators all bind first,
so they return the error too. Use [...]
- **SQL view APIs align with `catalog.ViewCatalog`.** `CreateView` now takes a
`*view.Version`, schema and options, and returns `*view.View`. `LoadView` now
returns `*view.View`; use `Metadata()` to access its metadata. If the caller
leaves `DefaultCatalog` empty, SQL still sets it to the catalog name.
