Repository: apache/iceberg-go
Pull request #452 (2025-06-04 22:09:42)

refactor: pass context into actual IO
related https://github.com/apache/iceberg-go/issues/437

All I/O operations must respect the context passed to PlanFiles(ctx) and simmilar methods.

To support that, we stop passing pre-constructed I/O objects and instead construct them ad hoc within PlanFiles, using the provided context


Patch:
diff --git a/README.md b/README.md
index b1e2cc2..63d277c 100644
--- a/README.md
+++ b/README.md
@@ -66,8 +66,8 @@ $ cd iceberg-go/cmd/iceberg && go build .
 | List Tables                 |  X   |      |          |  X   |  X  |
 | Create Table                |  X   |      |          |  X   |  X  |
 | Register Table              |  X   |      |          |  X   |     |
-| Update Current Snapshot     |  X   |      |          |      |  X  |
-| Create New Snapshot         |  X   |      |          |      |  X  |
+| Update Current Snapshot     |  X   |      |          |  X   |  X  |
+| Create New Snapshot         |  X   |      |          |  X   |  X  |
 | Rename Table                |  X   |      |          |  X   |  X  |
 | Drop Table                  |  X   |      |          |  X   |  X  |
 | Alter Table                 |  X   |      |          |  X   |  X  |
diff --git a/catalog/glue/glue.go b/catalog/glue/glue.go
index f3d14c5..65b3899 100644
--- a/catalog/glue/glue.go
+++ b/catalog/glue/glue.go
@@ -223,13 +223,14 @@ func (c *Catalog) LoadTable(ctx context.Context, identifier table.Identifier, pr
 	}
 
 	ctx = utils.WithAwsConfig(ctx, c.awsCfg)
-	// TODO: consider providing a way to directly access the S3 iofs to enable testing of the catalog.
-	iofs, err := io.LoadFS(ctx, props, location)
-	if err != nil {
-		return nil, fmt.Errorf("failed to load table %s.%s: %w", database, tableName, err)
-	}
 
-	icebergTable, err := table.NewFromLocation(identifier, location, iofs, c)
+	icebergTable, err := table.NewFromLocation(
+		ctx,
+		identifier,
+		location,
+		io.LoadFSFunc(props, location),
+		c,
+	)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create table from location %s.%s: %w", database, tableName, err)
 	}
@@ -255,7 +256,11 @@ func (c *Catalog) CreateTable(ctx context.Context, identifier table.Identifier,
 		return nil, err
 	}
 
-	wfs, ok := staged.FS().(io.WriteFileIO)
+	afs, err := staged.FS(ctx)
+	if err != nil {
+		return nil, err
+	}
+	wfs, ok := afs.(io.WriteFileIO)
 	if !ok {
 		return nil, errors.New("loaded filesystem IO does not support writing")
 	}
@@ -317,12 +322,14 @@ func (c *Catalog) RegisterTable(ctx context.Context, identifier table.Identifier
 	}
 	// Load the metadata file to get table properties
 	ctx = utils.WithAwsConfig(ctx, c.awsCfg)
-	iofs, err := io.LoadFS(ctx, nil, metadataLocation)
-	if err != nil {
-		return nil, fmt.Errorf("failed to load metadata file at %s: %w", metadataLocation, err)
-	}
 	// Read the metadata file
-	metadata, err := table.NewFromLocation([]string{tableName}, metadataLocation, iofs, c)
+	metadata, err := table.NewFromLocation(
+		ctx,
+		[]string{tableName},
+		metadataLocation,
+		io.LoadFSFunc(nil, metadataLocation),
+		c,
+	)
 	if err != nil {
 		return nil, fmt.Errorf("failed to read table metadata from %s: %w", metadataLocation, err)
 	}
diff --git a/catalog/internal/utils.go b/catalog/internal/utils.go
index 7e4c8ca..81b8cc3 100644
--- a/catalog/internal/utils.go
+++ b/catalog/internal/utils.go
@@ -136,13 +136,15 @@ func CreateStagedTable(ctx context.Context, catprops iceberg.Properties, nsprops
 
 	ioProps := maps.Clone(catprops)
 	maps.Copy(ioProps, cfg.Properties)
-	fs, err := io.LoadFS(ctx, ioProps, metadataLoc)
-	if err != nil {
-		return table.StagedTable{}, err
-	}
 
 	return table.StagedTable{
-		Table: table.New(ident, metadata, metadataLoc, fs, nil),
+		Table: table.New(
+			ident,
+			metadata,
+			metadataLoc,
+			io.LoadFSFunc(ioProps, metadataLoc),
+			nil,
+		),
 	}, nil
 }
 
@@ -239,10 +241,13 @@ func UpdateAndStageTable(ctx context.Context, current *table.Table, ident table.
 		return nil, err
 	}
 
-	fs, err := io.LoadFS(ctx, updated.Properties(), newLocation)
-	if err != nil {
-		return nil, err
-	}
-
-	return &table.StagedTable{Table: table.New(ident, updated, newLocation, fs, cat)}, nil
+	return &table.StagedTable{
+		Table: table.New(
+			ident,
+			updated,
+			newLocation,
+			io.LoadFSFunc(updated.Properties(), newLocation),
+			cat,
+		),
+	}, nil
 }
diff --git a/catalog/rest/rest.go b/catalog/rest/rest.go
index 9491649..93c88db 100644
--- a/catalog/rest/rest.go
+++ b/catalog/rest/rest.go
@@ -653,12 +653,13 @@ func checkValidNamespace(ident table.Identifier) error {
 }
 
 func (r *Catalog) tableFromResponse(ctx context.Context, identifier []string, metadata table.Metadata, loc string, config iceberg.Properties) (*table.Table, error) {
-	iofs, err := iceio.LoadFS(ctx, config, loc)
-	if err != nil {
-		return nil, err
-	}
-
-	return table.New(identifier, metadata, loc, iofs, r), nil
+	return table.New(
+		identifier,
+		metadata,
+		loc,
+		iceio.LoadFSFunc(config, loc),
+		r,
+	), nil
 }
 
 func (r *Catalog) ListTables(ctx context.Context, namespace table.Identifier) iter.Seq2[table.Identifier, error] {
diff --git a/catalog/sql/sql.go b/catalog/sql/sql.go
index bb8a083..762a643 100644
--- a/catalog/sql/sql.go
+++ b/catalog/sql/sql.go
@@ -283,7 +283,11 @@ func (c *Catalog) CreateTable(ctx context.Context, ident table.Identifier, sc *i
 		return nil, fmt.Errorf("%w: %s", catalog.ErrNoSuchNamespace, ns)
 	}
 
-	wfs, ok := staged.FS().(io.WriteFileIO)
+	afs, err := staged.FS(ctx)
+	if err != nil {
+		return nil, err
+	}
+	wfs, ok := afs.(io.WriteFileIO)
 	if !ok {
 		return nil, errors.New("loaded filesystem IO does not support writing")
 	}
@@ -416,12 +420,13 @@ func (c *Catalog) LoadTable(ctx context.Context, identifier table.Identifier, pr
 	tblProps := maps.Clone(c.props)
 	maps.Copy(props, tblProps)
 
-	iofs, err := io.LoadFS(ctx, tblProps, result.MetadataLocation.String)
-	if err != nil {
-		return nil, err
-	}
-
-	return table.NewFromLocation(identifier, result.MetadataLocation.String, iofs, c)
+	return table.NewFromLocation(
+		ctx,
+		identifier,
+		result.MetadataLocation.String,
+		io.LoadFSFunc(tblProps, result.MetadataLocation.String),
+		c,
+	)
 }
 
 func (c *Catalog) DropTable(ctx context.Context, identifier table.Identifier) error {
diff --git a/cmd/iceberg/output.go b/cmd/iceberg/output.go
index cc31b6d..695ba28 100644
--- a/cmd/iceberg/output.go
+++ b/cmd/iceberg/output.go
@@ -18,6 +18,7 @@
 package main
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"log"
@@ -134,7 +135,12 @@ func (t textOutput) Files(tbl *table.Table, history bool) {
 				snap.SnapshotID, *snap.SchemaID, manifest),
 		})
 
-		manifestList, err := snap.Manifests(tbl.FS())
+		afs, err := tbl.FS(context.TODO())
+		if err != nil {
+			t.Error(err)
+			os.Exit(1)
+		}
+		manifestList, err := snap.Manifests(afs)
 		if err != nil {
 			t.Error(err)
 			os.Exit(1)
@@ -144,7 +150,7 @@ func (t textOutput) Files(tbl *table.Table, history bool) {
 			snapshotTree = append(snapshotTree, pterm.LeveledListItem{
 				Level: 1, Text: "Manifest: " + m.FilePath(),
 			})
-			datafiles, err := m.FetchEntries(tbl.FS(), false)
+			datafiles, err := m.FetchEntries(afs, false)
 			if err != nil {
 				t.Error(err)
 				os.Exit(1)
diff --git a/dev/release/release.sh b/dev/release/release.sh
index 46c4755..bb81b87 100755
--- a/dev/release/release.sh
+++ b/dev/release/release.sh
@@ -40,16 +40,30 @@ fi
 tag="v${version}"
 rc_tag="${tag}-rc${rc}"
 echo "Tagging for release: ${tag}"
-git tag "${tag}" "${rc_tag}"
+git tag "${tag}" "${rc_tag}^{}" -m "Release ${tag}"
 git push origin "${tag}"
 
+release_id="apache-iceberg-go-${version}"
 dist_url="https://dist.apache.org/repos/dist/release/iceberg"
 dist_dev_url="https://dist.apache.org/repos/dist/dev/iceberg"
 
 svn \
-  mv "${dist_dev_url}/apache-iceberg-go-${version}-rc${rc}/" \
-  "${dist_url}/apache-iceberg-go-${version}/" \
-  -m "Apache Iceberg-go ${version}"
+  mv "${dist_dev_url}/${release_id}-rc${rc}/" \
+  "${dist_url}/${release_id}" \
+  -m "Apache Iceberg Go ${version}"
+
+svn co "${dist_url}/${release_id}"
+pushd "${release_id}"
+gh release create "${tag}" \
+  --title "Apache Iceberg Go ${version}" \
+  --generate-notes \
+  --verify-tag \
+  ${release_id}.tar.gz \
+  ${release_id}.tar.gz.asc \
+  ${release_id}.tar.gz.sha512
+popd
+
+rm -rf "${release_id}"
 
 echo "Keep only the latest versions"
 old_releases=$(
diff --git a/dev/release/release_rc.sh b/dev/release/release_rc.sh
index 38b9e40..8a27ad6 100755
--- a/dev/release/release_rc.sh
+++ b/dev/release/release_rc.sh
@@ -98,15 +98,25 @@ if [ "${RELEASE_SIGN}" -gt 0 ]; then
     --repo "${repository}" \
     --skip-existing
 
-  echo "Signing tar.gz and creating checksums"
+  echo "Signing tar.gz"
   cd "${id}"
   gpg --armor --output "${tar_gz}.asc" --detach-sig "${tar_gz}"
-  sha512sum "${tar_gz}" > "${tar_gz}.sha512"
+  echo "Add signature to GitHub release"
+  gh release upload "${rc_tag}" \
+    --clobber \
+    --repo "${repository}" \
+    "${tar_gz}".asc
   cd ..
 fi
 
 if [ "${RELEASE_UPLOAD}" -gt 0 ]; then
   echo "Uploading to ASF dist/dev..."
+  # rename files to remove -rc${rc} suffix before uploading
+  pushd "${id}"
+  for fname in "./*"; do
+    mv "${fname}" "${fname//-rc${rc}/}"
+  done
+  popd
   svn import "${id}" "https://dist.apache.org/repos/dist/dev/iceberg/${id}" -m "Apache Iceberg Go ${version} RC${rc}"
 fi
 
diff --git a/go.mod b/go.mod
index 9224779..bfcc23e 100644
--- a/go.mod
+++ b/go.mod
@@ -28,7 +28,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/config v1.29.14
 	github.com/aws/aws-sdk-go-v2/credentials v1.17.67
 	github.com/aws/aws-sdk-go-v2/service/glue v1.113.0
-	github.com/aws/aws-sdk-go-v2/service/s3 v1.79.4
+	github.com/aws/aws-sdk-go-v2/service/s3 v1.80.0
 	github.com/aws/smithy-go v1.22.3
 	github.com/awsdocs/aws-doc-sdk-examples/gov2/testtools v0.0.0-20250407191926-092f3e54b837
 	github.com/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815
@@ -49,7 +49,7 @@ require (
 	github.com/uptrace/bun/extra/bundebug v1.2.11
 	gocloud.dev v0.41.0
 	golang.org/x/sync v0.14.0
-	google.golang.org/api v0.234.0
+	google.golang.org/api v0.235.0
 	gopkg.in/yaml.v3 v3.0.1
 )
 
diff --git a/go.sum b/go.sum
index 47b8016..34c9673 100644
--- a/go.sum
+++ b/go.sum
@@ -144,8 +144,8 @@ github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.15 h1:dM9/92u2
 github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.15/go.mod h1:SwFBy2vjtA0vZbjjaFtfN045boopadnoVPhu4Fv66vY=
 github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.18.15 h1:moLQUoVq91LiqT1nbvzDukyqAlCv89ZmwaHw/ZFlFZg=
 github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.18.15/go.mod h1:ZH34PJUc8ApjBIfgQCFvkWcUDBtl/WTD+uiYHjd8igA=
-github.com/aws/aws-sdk-go-v2/service/s3 v1.79.4 h1:4yxno6bNHkekkfqG/a1nz/gC2gBwhJSojV1+oTE7K+4=
-github.com/aws/aws-sdk-go-v2/service/s3 v1.79.4/go.mod h1:qbn305Je/IofWBJ4bJz/Q7pDEtnnoInw/dGt71v6rHE=
+github.com/aws/aws-sdk-go-v2/service/s3 v1.80.0 h1:fV4XIU5sn/x8gjRouoJpDVHj+ExJaUk4prYF+eb6qTs=
+github.com/aws/aws-sdk-go-v2/service/s3 v1.80.0/go.mod h1:qbn305Je/IofWBJ4bJz/Q7pDEtnnoInw/dGt71v6rHE=
 github.com/aws/aws-sdk-go-v2/service/sso v1.25.3 h1:1Gw+9ajCV1jogloEv1RRnvfRFia2cL6c9cuKV2Ps+G8=
 github.com/aws/aws-sdk-go-v2/service/sso v1.25.3/go.mod h1:qs4a9T5EMLl/Cajiw2TcbNt2UNo/Hqlyp+GiuG4CFDI=
 github.com/aws/aws-sdk-go-v2/service/ssooidc v1.30.1 h1:hXmVKytPfTy5axZ+fYbR5d0cFmC3JvwLm5kM83luako=
@@ -971,8 +971,8 @@ golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da h1:noIWHXmPHxILtqtCOPIhS
 golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da/go.mod h1:NDW/Ps6MPRej6fsCIbMTohpP40sJ/P/vI1MoTEGwX90=
 gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk=
 gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E=
-google.golang.org/api v0.234.0 h1:d3sAmYq3E9gdr2mpmiWGbm9pHsA/KJmyiLkwKfHBqU4=
-google.golang.org/api v0.234.0/go.mod h1:QpeJkemzkFKe5VCE/PMv7GsUfn9ZF+u+q1Q7w6ckxTg=
+google.golang.org/api v0.235.0 h1:C3MkpQSRxS1Jy6AkzTGKKrpSCOd2WOGrezZ+icKSkKo=
+google.golang.org/api v0.235.0/go.mod h1:QpeJkemzkFKe5VCE/PMv7GsUfn9ZF+u+q1Q7w6ckxTg=
 google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
 google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
 google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc=
diff --git a/io/io.go b/io/io.go
index 523f7c5..248dfed 100644
--- a/io/io.go
+++ b/io/io.go
@@ -294,3 +294,15 @@ func LoadFS(ctx context.Context, props map[string]string, location string) (IO,
 
 	return iofs, nil
 }
+
+// LoadFSFunc is a helper function to create IO Func factory for later usage
+func LoadFSFunc(props map[string]string, location string) func(ctx context.Context) (IO, error) {
+	return func(ctx context.Context) (IO, error) {
+		iofs, err := LoadFS(ctx, props, location)
+		if err != nil {
+			return nil, fmt.Errorf("failed to load metadata file at %s: %w", location, err)
+		}
+
+		return iofs, nil
+	}
+}
diff --git a/table/evaluators.go b/table/evaluators.go
index acee1c3..c7d012f 100644
--- a/table/evaluators.go
+++ b/table/evaluators.go
@@ -751,11 +751,18 @@ func (m *inclusiveMetricsEval) Eval(file iceberg.DataFile) (bool, error) {
 		return rowsCannotMatch, nil
 	}
 
-	m.valueCounts, m.nullCounts = file.ValueCounts(), file.NullValueCounts()
-	m.nanCounts = file.NaNValueCounts()
-	m.lowerBounds, m.upperBounds = file.LowerBoundValues(), file.UpperBoundValues()
+	// avoid race condition while maintaining existing state
+	ev := inclusiveMetricsEval{
+		st:                m.st,
+		includeEmptyFiles: m.includeEmptyFiles,
+		expr:              m.expr,
+	}
 
-	return iceberg.VisitExpr(m.expr, m)
+	ev.valueCounts, ev.nullCounts = file.ValueCounts(), file.NullValueCounts()
+	ev.nanCounts = file.NaNValueCounts()
+	ev.lowerBounds, ev.upperBounds = file.LowerBoundValues(), file.UpperBoundValues()
+
+	return iceberg.VisitExpr(m.expr, &ev)
 }
 
 func (m *inclusiveMetricsEval) mayContainNull(fieldID int) bool {
diff --git a/table/scanner.go b/table/scanner.go
index 14a3811..93dd052 100644
--- a/table/scanner.go
+++ b/table/scanner.go
@@ -154,7 +154,7 @@ func openManifest(io io.IO, manifest iceberg.ManifestFile,
 
 type Scan struct {
 	metadata       Metadata
-	io             io.IO
+	ioF            FSysF
 	rowFilter      iceberg.BooleanExpression
 	selectedFields []string
 	caseSensitive  bool
@@ -299,14 +299,18 @@ func matchDeletesToData(entry iceberg.ManifestEntry, positionalDeletes []iceberg
 
 // fetchPartitionSpecFilteredManifests retrieves the table's current snapshot,
 // fetches its manifest files, and applies partition-spec filters to remove irrelevant manifests.
-func (scan *Scan) fetchPartitionSpecFilteredManifests() ([]iceberg.ManifestFile, error) {
+func (scan *Scan) fetchPartitionSpecFilteredManifests(ctx context.Context) ([]iceberg.ManifestFile, error) {
 	snap := scan.Snapshot()
 	if snap == nil {
 		return nil, nil
 	}
 
+	afs, err := scan.ioF(ctx)
+	if err != nil {
+		return nil, err
+	}
 	// Fetch all manifests for the current snapshot.
-	manifestList, err := snap.Manifests(scan.io)
+	manifestList, err := snap.Manifests(afs)
 	if err != nil {
 		return nil, err
 	}
@@ -354,8 +358,12 @@ func (scan *Scan) collectManifestEntries(
 		}
 
 		g.Go(func() error {
+			fs, err := scan.ioF(ctx)
+			if err != nil {
+				return err
+			}
 			partEval := partitionEvaluators.Get(int(mf.PartitionSpecID()))
-			manifestEntries, err := openManifest(scan.io, mf, partEval, metricsEval)
+			manifestEntries, err := openManifest(fs, mf, partEval, metricsEval)
 			if err != nil {
 				return err
 			}
@@ -390,7 +398,7 @@ func (scan *Scan) collectManifestEntries(
 // building a list of FileScanTasks that match the current Scan criteria.
 func (scan *Scan) PlanFiles(ctx context.Context) ([]FileScanTask, error) {
 	// Step 1: Retrieve filtered manifests based on snapshot and partition specs.
-	manifestList, err := scan.fetchPartitionSpecFilteredManifests()
+	manifestList, err := scan.fetchPartitionSpecFilteredManifests(ctx)
 	if err != nil || len(manifestList) == 0 {
 		return nil, err
 	}
@@ -456,9 +464,14 @@ func (scan *Scan) ToArrowRecords(ctx context.Context) (*arrow.Schema, iter.Seq2[
 		return nil, nil, err
 	}
 
+	fs, err := scan.ioF(ctx)
+	if err != nil {
+		return nil, nil, err
+	}
+
 	return (&arrowScan{
 		metadata:        scan.metadata,
-		fs:              scan.io,
+		fs:              fs,
 		projectedSchema: schema,
 		boundRowFilter:  boundFilter,
 		caseSensitive:   scan.caseSensitive,
diff --git a/table/table.go b/table/table.go
index c91fdfd..b49b240 100644
--- a/table/table.go
+++ b/table/table.go
@@ -33,6 +33,8 @@ import (
 	"golang.org/x/sync/errgroup"
 )
 
+type FSysF func(ctx context.Context) (io.IO, error)
+
 type Identifier = []string
 
 type CatalogIO interface {
@@ -44,8 +46,8 @@ type Table struct {
 	identifier       Identifier
 	metadata         Metadata
 	metadataLocation string
-	fs               io.IO
 	cat              CatalogIO
+	fsF              FSysF
 }
 
 func (t Table) Equals(other Table) bool {
@@ -54,19 +56,19 @@ func (t Table) Equals(other Table) bool {
 		t.metadata.Equals(other.metadata)
 }
 
-func (t Table) Identifier() Identifier               { return t.identifier }
-func (t Table) Metadata() Metadata                   { return t.metadata }
-func (t Table) MetadataLocation() string             { return t.metadataLocation }
-func (t Table) FS() io.IO                            { return t.fs }
-func (t Table) Schema() *iceberg.Schema              { return t.metadata.CurrentSchema() }
-func (t Table) Spec() iceberg.PartitionSpec          { return t.metadata.PartitionSpec() }
-func (t Table) SortOrder() SortOrder                 { return t.metadata.SortOrder() }
-func (t Table) Properties() iceberg.Properties       { return t.metadata.Properties() }
-func (t Table) NameMapping() iceberg.NameMapping     { return t.metadata.NameMapping() }
-func (t Table) Location() string                     { return t.metadata.Location() }
-func (t Table) CurrentSnapshot() *Snapshot           { return t.metadata.CurrentSnapshot() }
-func (t Table) SnapshotByID(id int64) *Snapshot      { return t.metadata.SnapshotByID(id) }
-func (t Table) SnapshotByName(name string) *Snapshot { return t.metadata.SnapshotByName(name) }
+func (t Table) Identifier() Identifier                { return t.identifier }
+func (t Table) Metadata() Metadata                    { return t.metadata }
+func (t Table) MetadataLocation() string              { return t.metadataLocation }
+func (t Table) FS(ctx context.Context) (io.IO, error) { return t.fsF(ctx) }
+func (t Table) Schema() *iceberg.Schema               { return t.metadata.CurrentSchema() }
+func (t Table) Spec() iceberg.PartitionSpec           { return t.metadata.PartitionSpec() }
+func (t Table) SortOrder() SortOrder                  { return t.metadata.SortOrder() }
+func (t Table) Properties() iceberg.Properties        { return t.metadata.Properties() }
+func (t Table) NameMapping() iceberg.NameMapping      { return t.metadata.NameMapping() }
+func (t Table) Location() string                      { return t.metadata.Location() }
+func (t Table) CurrentSnapshot() *Snapshot            { return t.metadata.CurrentSnapshot() }
+func (t Table) SnapshotByID(id int64) *Snapshot       { return t.metadata.SnapshotByID(id) }
+func (t Table) SnapshotByName(name string) *Snapshot  { return t.metadata.SnapshotByName(name) }
 func (t Table) Schemas() map[int]*iceberg.Schema {
 	m := make(map[int]*iceberg.Schema)
 	for _, s := range t.metadata.Schemas() {
@@ -110,15 +112,23 @@ func (t Table) Append(ctx context.Context, rdr array.RecordReader, snapshotProps
 	return txn.Commit(ctx)
 }
 
-func (t Table) AllManifests() iter.Seq2[iceberg.ManifestFile, error] {
+func (t Table) AllManifests(ctx context.Context) iter.Seq2[iceberg.ManifestFile, error] {
+	fs, err := t.fsF(ctx)
+	if err != nil {
+		return func(yield func(iceberg.ManifestFile, error) bool) {
+			yield(nil, err)
+		}
+	}
+
 	type list = tblutils.Enumerated[[]iceberg.ManifestFile]
 	g := errgroup.Group{}
 
 	n := len(t.metadata.Snapshots())
 	ch := make(chan list, n)
+
 	for i, sn := range t.metadata.Snapshots() {
 		g.Go(func() error {
-			manifests, err := sn.Manifests(t.fs)
+			manifests, err := sn.Manifests(fs)
 			if err != nil {
 				return err
 			}
@@ -195,10 +205,13 @@ func (t Table) doCommit(ctx context.Context, updates []Update, reqs []Requiremen
 	if err != nil {
 		return nil, err
 	}
+	fs, err := t.fsF(ctx)
+	if err != nil {
+		return nil, err
+	}
+	deleteOldMetadata(fs, t.metadata, newMeta)
 
-	deleteOldMetadata(t.fs, t.metadata, newMeta)
-
-	return New(t.identifier, newMeta, newLoc, t.fs, t.cat), nil
+	return New(t.identifier, newMeta, newLoc, t.fsF, t.cat), nil
 }
 
 func getFiles(it iter.Seq[MetadataLogEntry]) iter.Seq[string] {
@@ -310,7 +323,7 @@ func WithOptions(opts iceberg.Properties) ScanOption {
 func (t Table) Scan(opts ...ScanOption) *Scan {
 	s := &Scan{
 		metadata:       t.metadata,
-		io:             t.fs,
+		ioF:            t.fsF,
 		rowFilter:      iceberg.AlwaysTrue{},
 		selectedFields: []string{"*"},
 		caseSensitive:  true,
@@ -327,19 +340,29 @@ func (t Table) Scan(opts ...ScanOption) *Scan {
 	return s
 }
 
-func New(ident Identifier, meta Metadata, metadataLocation string, fs io.IO, cat CatalogIO) *Table {
+func New(ident Identifier, meta Metadata, metadataLocation string, fsF FSysF, cat CatalogIO) *Table {
 	return &Table{
 		identifier:       ident,
 		metadata:         meta,
 		metadataLocation: metadataLocation,
-		fs:               fs,
+		fsF:              fsF,
 		cat:              cat,
 	}
 }
 
-func NewFromLocation(ident Identifier, metalocation string, fsys io.IO, cat CatalogIO) (*Table, error) {
+func NewFromLocation(
+	ctx context.Context,
+	ident Identifier,
+	metalocation string,
+	fsysF FSysF,
+	cat CatalogIO,
+) (*Table, error) {
 	var meta Metadata
 
+	fsys, err := fsysF(ctx)
+	if err != nil {
+		return nil, err
+	}
 	if rf, ok := fsys.(io.ReadFileIO); ok {
 		data, err := rf.ReadFile(metalocation)
 		if err != nil {
@@ -361,5 +384,5 @@ func NewFromLocation(ident Identifier, metalocation string, fsys io.IO, cat Cata
 		}
 	}
 
-	return New(ident, meta, metalocation, fsys, cat), nil
+	return New(ident, meta, metalocation, fsysF, cat), nil
 }
diff --git a/table/transaction.go b/table/transaction.go
index 26fa844..6379089 100644
--- a/table/transaction.go
+++ b/table/transaction.go
@@ -116,9 +116,9 @@ func (t *Transaction) apply(updates []Update, reqs []Requirement) error {
 	return nil
 }
 
-func (t *Transaction) appendSnapshotProducer(props iceberg.Properties) *snapshotProducer {
+func (t *Transaction) appendSnapshotProducer(afs io.IO, props iceberg.Properties) *snapshotProducer {
 	manifestMerge := t.meta.props.GetBool(ManifestMergeEnabledKey, ManifestMergeEnabledDefault)
-	updateSnapshot := t.updateSnapshot(props)
+	updateSnapshot := t.updateSnapshot(afs, props)
 	if manifestMerge {
 		return updateSnapshot.mergeAppend()
 	}
@@ -126,10 +126,10 @@ func (t *Transaction) appendSnapshotProducer(props iceberg.Properties) *snapshot
 	return updateSnapshot.fastAppend()
 }
 
-func (t *Transaction) updateSnapshot(props iceberg.Properties) snapshotUpdate {
+func (t *Transaction) updateSnapshot(fs io.IO, props iceberg.Properties) snapshotUpdate {
 	return snapshotUpdate{
 		txn:           t,
-		io:            t.tbl.fs.(io.WriteFileIO),
+		io:            fs.(io.WriteFileIO),
 		snapshotProps: props,
 	}
 }
@@ -150,12 +150,15 @@ func (t *Transaction) AppendTable(ctx context.Context, tbl arrow.Table, batchSiz
 }
 
 func (t *Transaction) Append(ctx context.Context, rdr array.RecordReader, snapshotProps iceberg.Properties) error {
-	appendFiles := t.appendSnapshotProducer(snapshotProps)
-
+	fs, err := t.tbl.fsF(ctx)
+	if err != nil {
+		return err
+	}
+	appendFiles := t.appendSnapshotProducer(fs, snapshotProps)
 	itr := recordsToDataFiles(ctx, t.tbl.Location(), t.meta, recordWritingArgs{
 		sc:        rdr.Schema(),
 		itr:       array.IterFromReader(rdr),
-		fs:        t.tbl.fs.(io.WriteFileIO),
+		fs:        fs.(io.WriteFileIO),
 		writeUUID: &appendFiles.commitUuid,
 	})
 
@@ -215,8 +218,12 @@ func (t *Transaction) ReplaceDataFiles(ctx context.Context, filesToDelete, files
 		return fmt.Errorf("%w: cannot replace files in a table without an existing snapshot", ErrInvalidOperation)
 	}
 
+	fs, err := t.tbl.fsF(ctx)
+	if err != nil {
+		return err
+	}
 	markedForDeletion := make([]iceberg.DataFile, 0, len(setToDelete))
-	for df, err := range s.dataFiles(t.tbl.fs, nil) {
+	for df, err := range s.dataFiles(fs, nil) {
 		if err != nil {
 			return err
 		}
@@ -247,13 +254,13 @@ func (t *Transaction) ReplaceDataFiles(ctx context.Context, filesToDelete, files
 	}
 
 	commitUUID := uuid.New()
-	updater := t.updateSnapshot(snapshotProps).mergeOverwrite(&commitUUID)
+	updater := t.updateSnapshot(fs, snapshotProps).mergeOverwrite(&commitUUID)
 
 	for _, df := range markedForDeletion {
 		updater.deleteDataFile(df)
 	}
 
-	dataFiles := filesToDataFiles(ctx, t.tbl.fs, t.meta, slices.Values(filesToAdd))
+	dataFiles := filesToDataFiles(ctx, fs, t.meta, slices.Values(filesToAdd))
 	for df, err := range dataFiles {
 		if err != nil {
 			return err
@@ -282,7 +289,11 @@ func (t *Transaction) AddFiles(ctx context.Context, files []string, snapshotProp
 	if !ignoreDuplicates {
 		if s := t.meta.currentSnapshot(); s != nil {
 			referenced := make([]string, 0)
-			for df, err := range s.dataFiles(t.tbl.fs, nil) {
+			fs, err := t.tbl.fsF(ctx)
+			if err != nil {
+				return err
+			}
+			for df, err := range s.dataFiles(fs, nil) {
 				if err != nil {
 					return err
 				}
@@ -309,9 +320,14 @@ func (t *Transaction) AddFiles(ctx context.Context, files []string, snapshotProp
 		}
 	}
 
-	updater := t.updateSnapshot(snapshotProps).fastAppend()
+	fs, err := t.tbl.fsF(ctx)
+	if err != nil {
+		return err
+	}
+
+	updater := t.updateSnapshot(fs, snapshotProps).fastAppend()
 
-	dataFiles := filesToDataFiles(ctx, t.tbl.fs, t.meta, slices.Values(files))
+	dataFiles := filesToDataFiles(ctx, fs, t.meta, slices.Values(files))
 	for df, err := range dataFiles {
 		if err != nil {
 			return err
@@ -335,7 +351,7 @@ func (t *Transaction) Scan(opts ...ScanOption) (*Scan, error) {
 
 	s := &Scan{
 		metadata:       updatedMeta,
-		io:             t.tbl.fs,
+		ioF:            t.tbl.fsF,
 		rowFilter:      iceberg.AlwaysTrue{},
 		selectedFields: []string{"*"},
 		caseSensitive:  true,
@@ -358,8 +374,15 @@ func (t *Transaction) StagedTable() (*StagedTable, error) {
 		return nil, err
 	}
 
-	return &StagedTable{Table: New(t.tbl.identifier, updatedMeta,
-		updatedMeta.Location(), t.tbl.fs, t.tbl.cat)}, nil
+	return &StagedTable{
+		Table: New(
+			t.tbl.identifier,
+			updatedMeta,
+			updatedMeta.Location(),
+			t.tbl.fsF,
+			t.tbl.cat,
+		),
+	}, nil
 }
 
 func (t *Transaction) Commit(ctx context.Context) (*Table, error) {
diff --git a/transforms.go b/transforms.go
index fb37e6c..4cd02f5 100644
--- a/transforms.go
+++ b/transforms.go
@@ -720,7 +720,7 @@ func (t DayTransform) MarshalText() ([]byte, error) {
 
 func (DayTransform) String() string { return "day" }
 
-func (DayTransform) ResultType(Type) Type { return PrimitiveTypes.Date }
+func (DayTransform) ResultType(Type) Type { return PrimitiveTypes.Int32 }
 func (DayTransform) PreservesOrder() bool { return true }
 
 func (DayTransform) Equals(other Transform) bool {


Test patch:
diff --git a/catalog/glue/glue_test.go b/catalog/glue/glue_test.go
index 9cf7efd..2a685e1 100644
--- a/catalog/glue/glue_test.go
+++ b/catalog/glue/glue_test.go
@@ -864,31 +864,7 @@ func TestGlueCreateTableSuccessIntegration(t *testing.T) {
 		catalog.WithLocation(metadataLocation),
 	}
 	newTable, err := ctlg.CreateTable(context.TODO(), TableIdentifier(dbName, newTableName), sourceTable.Schema(), createOpts...)
-	defer func() {
-		cleanupErr := ctlg.DropTable(context.TODO(), TableIdentifier(dbName, newTableName))
-		if cleanupErr != nil {
-			t.Logf("Warning: Failed to clean up table %s.%s: %v", dbName, newTableName, cleanupErr)
-		}
-		// Clean up the newly created metadata file in S3
-		if newTable != nil {
-			s3Client := s3.NewFromConfig(awsCfg)
-			metadataLoc := newTable.MetadataLocation()
-			if strings.HasPrefix(metadataLoc, "s3://") {
-				parts := strings.SplitN(strings.TrimPrefix(metadataLoc, "s3://"), "/", 2)
-				if len(parts) == 2 {
-					bucket := parts[0]
-					key := parts[1]
-					_, deleteErr := s3Client.DeleteObject(context.TODO(), &s3.DeleteObjectInput{
-						Bucket: aws.String(bucket),
-						Key:    aws.String(key),
-					})
-					if deleteErr != nil {
-						t.Logf("Warning: Failed to delete metadata file at %s: %v", metadataLoc, deleteErr)
-					}
-				}
-			}
-		}
-	}()
+	defer cleanupTable(t, ctlg, TableIdentifier(dbName, newTableName), awsCfg)
 	assert.NoError(err)
 	assert.Equal([]string{dbName, newTableName}, newTable.Identifier())
 
@@ -1067,31 +1043,7 @@ func TestAlterTableIntegration(t *testing.T) {
 	assert.True(schema.Equals(testTable.Schema()))
 
 	// Clean up table and table location after tests
-	defer func() {
-		cleanupErr := ctlg.DropTable(context.TODO(), TableIdentifier(dbName, tbName))
-		if cleanupErr != nil {
-			t.Logf("Warning: Failed to clean up table %s.%s: %v", dbName, tbName, cleanupErr)
-		}
-		// Clean up the newly created metadata file in S3
-		if testTable != nil {
-			s3Client := s3.NewFromConfig(awsCfg)
-			metadataLoc := testTable.MetadataLocation()
-			if strings.HasPrefix(metadataLoc, "s3://") {
-				parts := strings.SplitN(strings.TrimPrefix(metadataLoc, "s3://"), "/", 2)
-				if len(parts) == 2 {
-					bucket := parts[0]
-					key := parts[1]
-					_, deleteErr := s3Client.DeleteObject(context.TODO(), &s3.DeleteObjectInput{
-						Bucket: aws.String(bucket),
-						Key:    aws.String(key),
-					})
-					if deleteErr != nil {
-						t.Logf("Warning: Failed to delete metadata file at %s: %v", metadataLoc, deleteErr)
-					}
-				}
-			}
-		}
-	}()
+	defer cleanupTable(t, ctlg, tbIdent, awsCfg)
 
 	// Test set table properties
 	updateProps := table.NewSetPropertiesUpdate(map[string]string{
@@ -1159,6 +1111,83 @@ func TestAlterTableIntegration(t *testing.T) {
 	assert.Equal(newFields, testTable.Schema().Fields())
 }
 
+func TestSnapshotManagementIntegration(t *testing.T) {
+	if os.Getenv("TEST_DATABASE_NAME") == "" || os.Getenv("TEST_TABLE_LOCATION") == "" {
+		t.Skip()
+	}
+
+	assert := require.New(t)
+	dbName := os.Getenv("TEST_DATABASE_NAME")
+	tbLocation := os.Getenv("TEST_TABLE_LOCATION")
+	tbName := fmt.Sprintf("table_%d", time.Now().UnixNano())
+	tbIdent := TableIdentifier(dbName, tbName)
+
+	awsCfg, err := config.LoadDefaultConfig(context.TODO(), config.WithClientLogMode(aws.LogRequest|aws.LogResponse))
+	assert.NoError(err)
+	ctlg := NewCatalog(WithAwsConfig(awsCfg))
+
+	// clean up table after test
+	defer cleanupTable(t, ctlg, tbIdent, awsCfg)
+
+	createOpts := []catalog.CreateTableOpt{
+		catalog.WithLocation(tbLocation),
+	}
+	_, err = ctlg.CreateTable(context.TODO(), tbIdent, testSchema, createOpts...)
+	assert.NoError(err)
+
+	testTable, err := ctlg.LoadTable(context.TODO(), tbIdent, nil)
+	assert.NoError(err)
+
+	// Test add new snapshot
+	manifest, schemaid := "s3:/a/b/c.avro", 3
+	newSnap := table.Snapshot{
+		SnapshotID:     25,
+		SequenceNumber: 200,
+		TimestampMs:    1602638573590,
+		ManifestList:   manifest,
+		SchemaID:       &schemaid,
+		Summary: &table.Summary{
+			Operation: table.OpAppend,
+		},
+	}
+
+	_, _, err = ctlg.CommitTable(context.TODO(), testTable, nil, []table.Update{
+		table.NewAddSnapshotUpdate(&newSnap),
+	})
+	assert.NoError(err)
+
+	testTable, err = ctlg.LoadTable(context.TODO(), tbIdent, nil)
+	assert.NoError(err)
+
+	actualSnap := testTable.SnapshotByID(25)
+	assert.Equal(newSnap.SnapshotID, actualSnap.SnapshotID)
+	assert.Equal(newSnap.ParentSnapshotID, actualSnap.ParentSnapshotID)
+	assert.Equal(newSnap.SequenceNumber, actualSnap.SequenceNumber)
+	assert.Equal(newSnap.ManifestList, actualSnap.ManifestList)
+	assert.Equal(newSnap.TimestampMs, actualSnap.TimestampMs)
+	assert.Equal(*newSnap.SchemaID, *actualSnap.SchemaID)
+	assert.Equal(newSnap.Summary.Operation, actualSnap.Summary.Operation)
+
+	// Test update current snapshot
+	_, _, err = ctlg.CommitTable(context.TODO(), testTable, nil, []table.Update{
+		table.NewSetSnapshotRefUpdate(table.MainBranch, 25, table.BranchRef,
+			-1, -1, -1),
+	})
+	assert.NoError(err)
+
+	testTable, err = ctlg.LoadTable(context.TODO(), tbIdent, nil)
+	assert.NoError(err)
+
+	currSnap := testTable.CurrentSnapshot()
+	assert.Equal(newSnap.SnapshotID, currSnap.SnapshotID)
+	assert.Equal(newSnap.ParentSnapshotID, actualSnap.ParentSnapshotID)
+	assert.Equal(newSnap.SequenceNumber, currSnap.SequenceNumber)
+	assert.Equal(newSnap.ManifestList, currSnap.ManifestList)
+	assert.Equal(newSnap.TimestampMs, currSnap.TimestampMs)
+	assert.Equal(*newSnap.SchemaID, *currSnap.SchemaID)
+	assert.Equal(newSnap.Summary.Operation, currSnap.Summary.Operation)
+}
+
 func TestGlueCheckTableExists(t *testing.T) {
 	assert := require.New(t)
 	mockGlueSvc := &mockGlueClient{}
@@ -1192,3 +1221,36 @@ func TestGlueCheckTableNotExists(t *testing.T) {
 	assert.Nil(err)
 	assert.False(exists)
 }
+
+func cleanupTable(t *testing.T, ctlg catalog.Catalog, tbIdent table.Identifier, awsCfg aws.Config) {
+	t.Helper()
+
+	testTable, err := ctlg.LoadTable(context.TODO(), tbIdent, nil)
+	if err != nil {
+		t.Logf("Warning: Failed to load table %s: %v", tbIdent, err)
+	}
+
+	cleanupErr := ctlg.DropTable(context.TODO(), tbIdent)
+	if cleanupErr != nil {
+		t.Logf("Warning: Failed to clean up table %s: %v", tbIdent, cleanupErr)
+	}
+
+	if testTable != nil {
+		s3Client := s3.NewFromConfig(awsCfg)
+		metadataLoc := testTable.MetadataLocation()
+		if strings.HasPrefix(metadataLoc, "s3://") {
+			parts := strings.SplitN(strings.TrimPrefix(metadataLoc, "s3://"), "/", 2)
+			if len(parts) == 2 {
+				bucket := parts[0]
+				key := parts[1]
+				_, deleteErr := s3Client.DeleteObject(context.TODO(), &s3.DeleteObjectInput{
+					Bucket: aws.String(bucket),
+					Key:    aws.String(key),
+				})
+				if deleteErr != nil {
+					t.Logf("Warning: Failed to delete metadata file at %s: %v", metadataLoc, deleteErr)
+				}
+			}
+		}
+	}
+}
diff --git a/catalog/rest/rest_integration_test.go b/catalog/rest/rest_integration_test.go
index edba0d5..4bd2807 100644
--- a/catalog/rest/rest_integration_test.go
+++ b/catalog/rest/rest_integration_test.go
@@ -33,6 +33,7 @@ import (
 	"github.com/apache/iceberg-go/catalog/rest"
 	"github.com/apache/iceberg-go/io"
 	"github.com/apache/iceberg-go/table"
+	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 )
 
@@ -254,11 +255,11 @@ func (s *RestIntegrationSuite) TestWriteCommitTable() {
 	pqfile, err := url.JoinPath(location, "data", "test_commit_table_data", "test.parquet")
 	s.Require().NoError(err)
 
-	fw, err := tbl.FS().(io.WriteFileIO).Create(pqfile)
+	fw, err := mustFS(s.T(), tbl).(io.WriteFileIO).Create(pqfile)
 	s.Require().NoError(err)
 	s.Require().NoError(pqarrow.WriteTable(table, fw, table.NumRows(),
 		nil, pqarrow.DefaultWriterProps()))
-	defer tbl.FS().Remove(pqfile)
+	defer mustFS(s.T(), tbl).Remove(pqfile)
 
 	txn := tbl.NewTransaction()
 	s.Require().NoError(txn.AddFiles(s.ctx, []string{pqfile}, nil, false))
@@ -266,7 +267,7 @@ func (s *RestIntegrationSuite) TestWriteCommitTable() {
 	s.Require().NoError(err)
 
 	mf := []iceberg.ManifestFile{}
-	for m, err := range updated.AllManifests() {
+	for m, err := range updated.AllManifests(s.ctx) {
 		s.Require().NoError(err)
 		s.Require().NotNil(m)
 		mf = append(mf, m)
@@ -274,13 +275,19 @@ func (s *RestIntegrationSuite) TestWriteCommitTable() {
 
 	s.Len(mf, 1)
 	s.EqualValues(1, mf[0].AddedDataFiles())
-	entries, err := mf[0].FetchEntries(updated.FS(), false)
+	entries, err := mf[0].FetchEntries(mustFS(s.T(), updated), false)
 	s.Require().NoError(err)
 
 	s.Len(entries, 1)
 	s.Equal(pqfile, entries[0].DataFile().FilePath())
 }
 
+func mustFS(t *testing.T, tbl *table.Table) io.IO {
+	r, err := tbl.FS(context.Background())
+	require.NoError(t, err)
+	return r
+}
+
 func TestRestIntegration(t *testing.T) {
 	suite.Run(t, new(RestIntegrationSuite))
 }
diff --git a/table/table_test.go b/table/table_test.go
index 0b00a5a..cef39a5 100644
--- a/table/table_test.go
+++ b/table/table_test.go
@@ -59,6 +59,13 @@ func TestTable(t *testing.T) {
 	suite.Run(t, new(TableTestSuite))
 }
 
+func mustFS(t *testing.T, tbl *table.Table) iceio.IO {
+	r, err := tbl.FS(context.Background())
+	require.NoError(t, err)
+
+	return r
+}
+
 func (t *TableTestSuite) SetupSuite() {
 	var mockfs internal.MockFS
 	mockfs.Test(t.T())
@@ -66,13 +73,21 @@ func (t *TableTestSuite) SetupSuite() {
 		Return(&internal.MockFile{Contents: bytes.NewReader([]byte(table.ExampleTableMetadataV2))}, nil)
 	defer mockfs.AssertExpectations(t.T())
 
-	tbl, err := table.NewFromLocation([]string{"foo"}, "s3://bucket/test/location/uuid.metadata.json", &mockfs, nil)
+	tbl, err := table.NewFromLocation(
+		context.Background(),
+		[]string{"foo"},
+		"s3://bucket/test/location/uuid.metadata.json",
+		func(ctx context.Context) (iceio.IO, error) {
+			return &mockfs, nil
+		},
+		nil,
+	)
 	t.Require().NoError(err)
 	t.Require().NotNil(tbl)
 
 	t.Equal([]string{"foo"}, tbl.Identifier())
 	t.Equal("s3://bucket/test/location/uuid.metadata.json", tbl.MetadataLocation())
-	t.Equal(&mockfs, tbl.FS())
+	t.Equal(&mockfs, mustFS(t.T(), tbl))
 
 	t.tbl = tbl
 }
@@ -84,7 +99,15 @@ func (t *TableTestSuite) TestNewTableFromReadFile() {
 		Return([]byte(table.ExampleTableMetadataV2), nil)
 	defer mockfsReadFile.AssertExpectations(t.T())
 
-	tbl2, err := table.NewFromLocation([]string{"foo"}, "s3://bucket/test/location/uuid.metadata.json", &mockfsReadFile, nil)
+	tbl2, err := table.NewFromLocation(
+		t.T().Context(),
+		[]string{"foo"},
+		"s3://bucket/test/location/uuid.metadata.json",
+		func(ctx context.Context) (iceio.IO, error) {
+			return &mockfsReadFile, nil
+		},
+		nil,
+	)
 	t.Require().NoError(err)
 	t.Require().NotNil(tbl2)
 
@@ -298,7 +321,15 @@ func (t *TableWritingTestSuite) createTable(identifier table.Identifier, formatV
 		t.location, iceberg.Properties{"format-version": strconv.Itoa(formatVersion)})
 	t.Require().NoError(err)
 
-	return table.New(identifier, meta, t.getMetadataLoc(), iceio.LocalFS{}, nil)
+	return table.New(
+		identifier,
+		meta,
+		t.getMetadataLoc(),
+		func(ctx context.Context) (iceio.IO, error) {
+			return iceio.LocalFS{}, nil
+		},
+		nil,
+	)
 }
 
 func (t *TableWritingTestSuite) TestAddFilesUnpartitioned() {
@@ -311,7 +342,7 @@ func (t *TableWritingTestSuite) TestAddFilesUnpartitioned() {
 	files := make([]string, 0)
 	for i := range 5 {
 		filePath := fmt.Sprintf("%s/unpartitioned/test-%d.parquet", t.location, i)
-		t.writeParquet(tbl.FS().(iceio.WriteFileIO), filePath, t.arrTbl)
+		t.writeParquet(mustFS(t.T(), tbl).(iceio.WriteFileIO), filePath, t.arrTbl)
 		files = append(files, filePath)
 	}
 
@@ -358,7 +389,7 @@ func (t *TableWritingTestSuite) TestAddFilesFileNotFound() {
 	files := make([]string, 0)
 	for i := range 5 {
 		filePath := fmt.Sprintf("%s/unpartitioned_file_not_found/test-%d.parquet", t.location, i)
-		t.writeParquet(tbl.FS().(iceio.WriteFileIO), filePath, t.arrTbl)
+		t.writeParquet(mustFS(t.T(), tbl).(iceio.WriteFileIO), filePath, t.arrTbl)
 		files = append(files, filePath)
 	}
 
@@ -379,7 +410,7 @@ func (t *TableWritingTestSuite) TestAddFilesUnpartitionedHasFieldIDs() {
 	files := make([]string, 0)
 	for i := range 5 {
 		filePath := fmt.Sprintf("%s/unpartitioned_with_ids/test-%d.parquet", t.location, i)
-		t.writeParquet(tbl.FS().(iceio.WriteFileIO), filePath, t.arrTblWithIDs)
+		t.writeParquet(mustFS(t.T(), tbl).(iceio.WriteFileIO), filePath, t.arrTblWithIDs)
 		files = append(files, filePath)
 	}
 
@@ -408,7 +439,7 @@ func (t *TableWritingTestSuite) TestAddFilesFailsSchemaMismatch() {
 		  {"foo": false, "bar": "bar_string", "baz": "456", "qux": "2024-03-07"}]`,
 	})
 	t.Require().NoError(err)
-	t.writeParquet(tbl.FS().(iceio.WriteFileIO), filePath, mismatchTable)
+	t.writeParquet(mustFS(t.T(), tbl).(iceio.WriteFileIO), filePath, mismatchTable)
 
 	files := []string{filePath}
 
@@ -451,7 +482,7 @@ func (t *TableWritingTestSuite) TestAddFilesPartitionedTable() {
 		t.Require().NoError(err)
 		defer table.Release()
 
-		t.writeParquet(tbl.FS().(iceio.WriteFileIO), filePath, table)
+		t.writeParquet(mustFS(t.T(), tbl).(iceio.WriteFileIO), filePath, table)
 		files = append(files, filePath)
 	}
 
@@ -479,11 +510,11 @@ func (t *TableWritingTestSuite) TestAddFilesPartitionedTable() {
 			},
 		})
 
-	m, err := stagedTbl.CurrentSnapshot().Manifests(tbl.FS())
+	m, err := stagedTbl.CurrentSnapshot().Manifests(mustFS(t.T(), tbl))
 	t.Require().NoError(err)
 
 	for _, manifest := range m {
-		entries, err := manifest.FetchEntries(tbl.FS(), false)
+		entries, err := manifest.FetchEntries(mustFS(t.T(), tbl), false)
 		t.Require().NoError(err)
 
 		for _, e := range entries {
@@ -509,7 +540,7 @@ func (t *TableWritingTestSuite) TestAddFilesToBucketPartitionedTableFails() {
 		t.Require().NoError(err)
 		defer table.Release()
 
-		t.writeParquet(tbl.FS().(iceio.WriteFileIO), filePath, table)
+		t.writeParquet(mustFS(t.T(), tbl).(iceio.WriteFileIO), filePath, table)
 		files = append(files, filePath)
 	}
 
@@ -537,7 +568,7 @@ func (t *TableWritingTestSuite) TestAddFilesToPartitionedTableFailsLowerAndUpper
 		t.Require().NoError(err)
 		defer table.Release()
 
-		t.writeParquet(tbl.FS().(iceio.WriteFileIO), filePath, table)
+		t.writeParquet(mustFS(t.T(), tbl).(iceio.WriteFileIO), filePath, table)
 		files = append(files, filePath)
 	}
 
@@ -576,8 +607,8 @@ func (t *TableWritingTestSuite) TestAddFilesWithLargeAndRegular() {
 	t.Require().NoError(err)
 	defer arrTableLarge.Release()
 
-	t.writeParquet(tbl.FS().(iceio.WriteFileIO), filePath, arrTable)
-	t.writeParquet(tbl.FS().(iceio.WriteFileIO), filePathLarge, arrTableLarge)
+	t.writeParquet(mustFS(t.T(), tbl).(iceio.WriteFileIO), filePath, arrTable)
+	t.writeParquet(mustFS(t.T(), tbl).(iceio.WriteFileIO), filePathLarge, arrTableLarge)
 
 	tx := tbl.NewTransaction()
 	t.Require().NoError(tx.AddFiles(t.ctx, []string{filePath, filePathLarge}, nil, false))
@@ -600,7 +631,7 @@ func (t *TableWritingTestSuite) TestAddFilesValidUpcast() {
 	tbl := t.createTable(ident, t.formatVersion, *iceberg.UnpartitionedSpec, t.tableSchemaPromotedTypes)
 
 	filePath := fmt.Sprintf("%s/test_table_with_valid_upcast_v%d/test.parquet", t.location, t.formatVersion)
-	t.writeParquet(tbl.FS().(iceio.WriteFileIO), filePath, t.arrTablePromotedTypes)
+	t.writeParquet(mustFS(t.T(), tbl).(iceio.WriteFileIO), filePath, t.arrTablePromotedTypes)
 
 	tx := tbl.NewTransaction()
 	t.Require().NoError(tx.AddFiles(t.ctx, []string{filePath}, nil, false))
@@ -648,7 +679,7 @@ func (t *TableWritingTestSuite) TestAddFilesSubsetOfSchema() {
 	filePath := fmt.Sprintf("%s/test_table_with_subset_of_schema_v%d/test.parquet", t.location, t.formatVersion)
 	withoutCol := dropColFromTable(0, t.arrTbl)
 	defer withoutCol.Release()
-	t.writeParquet(tbl.FS().(iceio.WriteFileIO), filePath, withoutCol)
+	t.writeParquet(mustFS(t.T(), tbl).(iceio.WriteFileIO), filePath, withoutCol)
 
 	tx := tbl.NewTransaction()
 	t.Require().NoError(tx.AddFiles(t.ctx, []string{filePath}, nil, false))
@@ -683,7 +714,7 @@ func (t *TableWritingTestSuite) TestAddFilesDuplicateFilesInFilePaths() {
 	tbl := t.createTable(ident, t.formatVersion, *iceberg.UnpartitionedSpec, t.tableSchema)
 
 	filePath := fmt.Sprintf("%s/test_table_with_duplicate_files_v%d/test.parquet", t.location, t.formatVersion)
-	t.writeParquet(tbl.FS().(iceio.WriteFileIO), filePath, t.arrTbl)
+	t.writeParquet(mustFS(t.T(), tbl).(iceio.WriteFileIO), filePath, t.arrTbl)
 
 	tx := tbl.NewTransaction()
 	err := tx.AddFiles(t.ctx, []string{filePath, filePath}, nil, false)
@@ -701,7 +732,7 @@ func (t *TableWritingTestSuite) TestAddFilesReferencedByCurrentSnapshot() {
 	files := make([]string, 0)
 	for i := range 5 {
 		filePath := fmt.Sprintf("%s/add_files_referenced/test-%d.parquet", t.location, i)
-		t.writeParquet(tbl.FS().(iceio.WriteFileIO), filePath, t.arrTbl)
+		t.writeParquet(mustFS(t.T(), tbl).(iceio.WriteFileIO), filePath, t.arrTbl)
 		files = append(files, filePath)
 	}
 
@@ -723,7 +754,7 @@ func (t *TableWritingTestSuite) TestAddFilesReferencedCurrentSnapshotIgnoreDupli
 	files := make([]string, 0)
 	for i := range 5 {
 		filePath := fmt.Sprintf("%s/add_files_referenced/test-%d.parquet", t.location, i)
-		t.writeParquet(tbl.FS().(iceio.WriteFileIO), filePath, t.arrTbl)
+		t.writeParquet(mustFS(t.T(), tbl).(iceio.WriteFileIO), filePath, t.arrTbl)
 		files = append(files, filePath)
 	}
 
@@ -735,7 +766,7 @@ func (t *TableWritingTestSuite) TestAddFilesReferencedCurrentSnapshotIgnoreDupli
 	t.Require().NoError(err)
 
 	added, existing, deleted := []int32{}, []int32{}, []int32{}
-	for m, err := range staged.AllManifests() {
+	for m, err := range staged.AllManifests(t.T().Context()) {
 		t.Require().NoError(err)
 		added = append(added, m.AddedDataFiles())
 		existing = append(existing, m.ExistingDataFiles())
@@ -790,7 +821,15 @@ func (t *TableWritingTestSuite) TestReplaceDataFiles() {
 
 	ctx := context.Background()
 
-	tbl := table.New(ident, meta, t.getMetadataLoc(), fs, &mockedCatalog{})
+	tbl := table.New(
+		ident,
+		meta,
+		t.getMetadataLoc(),
+		func(ctx context.Context) (iceio.IO, error) {
+			return fs, nil
+		},
+		&mockedCatalog{},
+	)
 	for i := range 5 {
 		tx := tbl.NewTransaction()
 		t.Require().NoError(tx.AddFiles(ctx, files[i:i+1], nil, false))
@@ -798,7 +837,7 @@ func (t *TableWritingTestSuite) TestReplaceDataFiles() {
 		t.Require().NoError(err)
 	}
 
-	mflist, err := tbl.CurrentSnapshot().Manifests(tbl.FS())
+	mflist, err := tbl.CurrentSnapshot().Manifests(mustFS(t.T(), tbl))
 	t.Require().NoError(err)
 	t.Len(mflist, 5)
 
@@ -1105,12 +1144,12 @@ func (t *TableWritingTestSuite) TestMergeManifests() {
 	tblC, err = tblC.AppendTable(t.ctx, arrTable, 1, nil)
 	t.Require().NoError(err)
 
-	manifestList, err := tblA.CurrentSnapshot().Manifests(tblA.FS())
+	manifestList, err := tblA.CurrentSnapshot().Manifests(mustFS(t.T(), tblA))
 	t.Require().NoError(err)
 	t.Len(manifestList, 1)
-	t.validateManifestFileLength(tblA.FS(), manifestList[0])
+	t.validateManifestFileLength(mustFS(t.T(), tblA), manifestList[0])
 
-	entries, err := manifestList[0].FetchEntries(tblA.FS(), false)
+	entries, err := manifestList[0].FetchEntries(mustFS(t.T(), tblA), false)
 	t.Require().NoError(err)
 	t.Len(entries, 3)
 
@@ -1124,20 +1163,20 @@ func (t *TableWritingTestSuite) TestMergeManifests() {
 		}
 	}
 
-	manifestList, err = tblB.CurrentSnapshot().Manifests(tblB.FS())
+	manifestList, err = tblB.CurrentSnapshot().Manifests(mustFS(t.T(), tblB))
 	t.Require().NoError(err)
 	t.Len(manifestList, 3)
 
 	for _, m := range manifestList {
-		t.validateManifestFileLength(tblB.FS(), m)
+		t.validateManifestFileLength(mustFS(t.T(), tblB), m)
 	}
 
-	manifestList, err = tblC.CurrentSnapshot().Manifests(tblC.FS())
+	manifestList, err = tblC.CurrentSnapshot().Manifests(mustFS(t.T(), tblC))
 	t.Require().NoError(err)
 	t.Len(manifestList, 3)
 
 	for _, m := range manifestList {
-		t.validateManifestFileLength(tblC.FS(), m)
+		t.validateManifestFileLength(mustFS(t.T(), tblC), m)
 	}
 
 	resultA, err := tblA.Scan().ToArrowTable(t.ctx)
@@ -1204,9 +1243,8 @@ func TestNullableStructRequiredField(t *testing.T) {
 	sc, err := table.ArrowSchemaToIcebergWithFreshIDs(arrowSchema, false)
 	require.NoError(t, err)
 
-	ctx := context.TODO()
-	require.NoError(t, cat.CreateNamespace(ctx, table.Identifier{"testing"}, nil))
-	tbl, err := cat.CreateTable(ctx, table.Identifier{"testing", "nullable_struct_required_field"}, sc,
+	require.NoError(t, cat.CreateNamespace(t.Context(), table.Identifier{"testing"}, nil))
+	tbl, err := cat.CreateTable(t.Context(), table.Identifier{"testing", "nullable_struct_required_field"}, sc,
 		catalog.WithProperties(iceberg.Properties{"format-version": "2"}),
 		catalog.WithLocation("file://"+loc))
 	require.NoError(t, err)
@@ -1226,7 +1264,7 @@ func TestNullableStructRequiredField(t *testing.T) {
 	defer arrTable.Release()
 
 	tx := tbl.NewTransaction()
-	require.NoError(t, tx.AppendTable(ctx, arrTable, N, nil))
+	require.NoError(t, tx.AppendTable(t.Context(), arrTable, N, nil))
 	stagedTbl, err := tx.StagedTable()
 	require.NoError(t, err)
 	require.NotNil(t, stagedTbl)
@@ -1303,7 +1341,9 @@ func (t *TableWritingTestSuite) TestDeleteOldMetadataLogsErrorOnFileNotFound() {
 		table.UnsortedSortOrder, t.location, iceberg.Properties{"format-version": strconv.Itoa(t.formatVersion), "write.metadata.delete-after-commit.enabled": "true"})
 	t.Require().NoError(err)
 
-	tbl := table.New(ident, meta, t.getMetadataLoc(), fs, &DeleteOldMetadataMockedCatalog{})
+	tbl := table.New(ident, meta, t.getMetadataLoc(), func(ctx context.Context) (iceio.IO, error) {
+		return fs, nil
+	}, &DeleteOldMetadataMockedCatalog{})
 	ctx := context.Background()
 
 	// transaction 1 to create metadata file
@@ -1347,7 +1387,15 @@ func (t *TableWritingTestSuite) TestDeleteOldMetadataNoErrorLogsOnFileFound() {
 	meta, err := table.NewMetadata(t.tableSchemaPromotedTypes, iceberg.UnpartitionedSpec, table.UnsortedSortOrder, t.location, iceberg.Properties{"format-version": strconv.Itoa(t.formatVersion), "write.metadata.delete-after-commit.enabled": "true"})
 	t.Require().NoError(err)
 
-	tbl := table.New(ident, meta, t.getMetadataLoc(), fs, &DeleteOldMetadataMockedCatalog{})
+	tbl := table.New(
+		ident,
+		meta,
+		t.getMetadataLoc(),
+		func(ctx context.Context) (iceio.IO, error) {
+			return fs, nil
+		},
+		&DeleteOldMetadataMockedCatalog{},
+	)
 
 	ctx := context.Background()
 
diff --git a/table/transaction_test.go b/table/transaction_test.go
index bef954e..40f015c 100644
--- a/table/transaction_test.go
+++ b/table/transaction_test.go
@@ -97,7 +97,7 @@ func (s *SparkIntegrationTestSuite) TestAddFile() {
 	rec := bldr.NewRecord()
 	defer rec.Release()
 
-	fw, err := tbl.FS().(iceio.WriteFileIO).Create(filename)
+	fw, err := mustFS(s.T(), tbl).(iceio.WriteFileIO).Create(filename)
 	if err != nil {
 		panic(err)
 	}
diff --git a/transforms_test.go b/transforms_test.go
index 3e7a6d5..4abebeb 100644
--- a/transforms_test.go
+++ b/transforms_test.go
@@ -18,8 +18,11 @@
 package iceberg_test
 
 import (
+	"bytes"
+	"reflect"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/apache/arrow-go/v18/arrow/decimal"
 	"github.com/apache/iceberg-go"
@@ -139,3 +142,98 @@ func TestToHumanString(t *testing.T) {
 		})
 	}
 }
+
+func TestManifestPartitionVals(t *testing.T) {
+	// Sanity checks that the source and result types of the transform are
+	// compatible with their use to generate partition data in manifests.
+	ts := time.Date(1971, 2, 10, 10, 20, 30, 4_000_000, time.UTC)
+	tests := []struct {
+		transform    iceberg.Transform
+		input        iceberg.Literal
+		expectResult iceberg.Literal
+	}{
+		{
+			transform:    iceberg.HourTransform{},
+			input:        iceberg.TimestampLiteral(ts.UnixMicro()),
+			expectResult: iceberg.Int32Literal((365+40)*24 + 10),
+		},
+		{
+			transform:    iceberg.DayTransform{},
+			input:        iceberg.TimestampLiteral(ts.UnixMicro()),
+			expectResult: iceberg.Int32Literal(365 + 40),
+		},
+		{
+			transform:    iceberg.MonthTransform{},
+			input:        iceberg.TimestampLiteral(ts.UnixMicro()),
+			expectResult: iceberg.Int32Literal(13),
+		},
+		{
+			transform:    iceberg.YearTransform{},
+			input:        iceberg.TimestampLiteral(ts.UnixMicro()),
+			expectResult: iceberg.Int32Literal(1),
+		},
+		{
+			transform:    iceberg.TruncateTransform{Width: 100},
+			input:        iceberg.Int64Literal(123456789),
+			expectResult: iceberg.Int64Literal(123456700),
+		},
+		{
+			transform:    iceberg.IdentityTransform{},
+			input:        iceberg.StringLiteral("foobar"),
+			expectResult: iceberg.StringLiteral("foobar"),
+		},
+		{
+			transform:    iceberg.BucketTransform{NumBuckets: 128},
+			input:        iceberg.StringLiteral("foobar"),
+			expectResult: iceberg.Int32Literal(61),
+		},
+	}
+	for _, tt := range tests {
+		t.Run(reflect.TypeOf(tt.transform).String(), func(t *testing.T) {
+			result := tt.transform.Apply(iceberg.Optional[iceberg.Literal]{Val: tt.input, Valid: true})
+			require.True(t, result.Valid)
+			assert.Equal(t, tt.expectResult, result.Val)
+
+			schema := iceberg.NewSchema(0, iceberg.NestedField{
+				Name: "abc",
+				ID:   1,
+				Type: tt.input.Type(),
+			})
+			partitionSpec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+				Name:      "transformed_abc",
+				SourceID:  1,
+				FieldID:   1000,
+				Transform: tt.transform,
+			})
+			dataFile, err := iceberg.NewDataFileBuilder(
+				partitionSpec, iceberg.EntryContentData,
+				"1234.parquet", iceberg.ParquetFile,
+				map[int]any{1000: result.Val.Any()},
+				100, 100_000,
+			)
+			require.NoError(t, err)
+			snapshotID := int64(123)
+			seqNum := int64(456)
+			manifestEntry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapshotID, &seqNum, &seqNum, dataFile.Build())
+			var buf bytes.Buffer
+			manifestFile, err := iceberg.WriteManifest(
+				"abc.avro", &buf,
+				2, partitionSpec, schema, 123,
+				[]iceberg.ManifestEntry{manifestEntry},
+			)
+			require.NoError(t, err)
+
+			// Now make sure we can deserialize and re-serialize the manifest, too.
+			entries, err := iceberg.ReadManifest(manifestFile, &buf, false)
+			require.NoError(t, err)
+
+			buf.Reset()
+			_, err = iceberg.WriteManifest(
+				"abc.avro", &buf,
+				2, partitionSpec, schema, 123,
+				entries,
+			)
+			require.NoError(t, err)
+		})
+	}
+}