zeroshade commented on code in PR #1857:
URL: https://github.com/apache/iceberg-go/pull/1857#discussion_r3865585680
##########
catalog/rest/vended_creds.go:
##########
@@ -230,3 +271,181 @@ func (v *vendedCredentialRefresher) close() error {
return nil
}
+
+// prefixScopedIO selects a plan credential using the actual object location
+// passed to Open/Remove. A single plan may cover metadata, data, and delete
+// files in different storage prefixes, so resolving credentials once at plan
+// creation would be incorrect.
+type prefixScopedIO struct {
+ ctx context.Context
+ baseProps iceberg.Properties
+ credentials []StorageCredential
+
+ mu sync.Mutex
+ filesystems map[string]iceio.IO
+ closed bool
+ nowFunc func() time.Time
+}
+
+func newPrefixScopedIO(ctx context.Context, baseProps iceberg.Properties,
credentials []StorageCredential) *prefixScopedIO {
+ return &prefixScopedIO{
+ ctx: ctx,
+ baseProps: maps.Clone(baseProps),
+ credentials: slices.Clone(credentials),
+ filesystems: make(map[string]iceio.IO),
+ }
+}
+
+func (p *prefixScopedIO) Open(name string) (iceio.File, error) {
+ fs, err := p.filesystemFor(name)
+ if err != nil {
+ return nil, err
+ }
+
+ return fs.Open(name)
+}
+
+func (p *prefixScopedIO) Remove(name string) error {
+ fs, err := p.filesystemFor(name)
+ if err != nil {
+ return err
+ }
+
+ return fs.Remove(name)
+}
+
+func (p *prefixScopedIO) filesystemFor(name string) (iceio.IO, error) {
+ credentialIndex := matchingStorageCredentialIndex(p.credentials, name)
+ key := scopedFilesystemKey(credentialIndex, name)
+
+ p.mu.Lock()
+ if p.closed {
+ p.mu.Unlock()
+
+ return nil, errors.New("prefix-scoped IO is closed")
+ }
+ if credentialIndex >= 0 {
+ if expiresAt, ok :=
parseCredentialExpiry(p.credentials[credentialIndex].Config); ok &&
+ p.now().After(expiresAt) {
+ p.mu.Unlock()
+
+ return nil, fmt.Errorf("%w: %s expired at %s",
+ ErrVendedCredentialsExpired, name,
expiresAt.Format(time.RFC3339))
+ }
+ }
+ if fs, ok := p.filesystems[key]; ok {
+ p.mu.Unlock()
+
+ return fs, nil
+ }
+ p.mu.Unlock()
+
+ props := p.propertiesForLocation(name)
+
+ fs, err := iceio.LoadFS(p.ctx, props, name)
+ if err != nil {
+ return nil, err
+ }
+
+ p.mu.Lock()
+ if p.closed {
+ p.mu.Unlock()
+
+ return nil, errors.New("prefix-scoped IO is closed")
+ }
+ if existing, ok := p.filesystems[key]; ok {
Review Comment:
Loading outside the mutex avoids serialization, but this double-check
discards the newly created `fs` without closing it when another goroutine wins
the cache race. The `p.closed` branch above has the same leak. Cloud
implementations can own closable clients—for example, `BlobFileIO` promotes
`blob.Bucket.Close`.
I reproduced this 20/20 times with two concurrent blocked `LoadFS` calls:
expected the losing filesystem to be closed once; the close count remained
zero. Please close the newly loaded filesystem before returning from either
discard branch and add regressions for both concurrent insertion and
close-during-load.
--
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]
