Repository: apache/iceberg-go
Pull request #330 (2025-03-11 20:13:06)

feat(table): Basic Transaction and AddFiles
And here we go! Snapshot producers, basic Transaction object, and a basic implementation of adding files to a table!

Complete with unit tests and a general TableWritingTestSuite.



zeroshade: We're in the home stretch @Fokko @kevinjqliu!! Thanks so much for the quick reviews and feedback on all of these. 
zeroshade: > There is quite a bit code here, but if I understand it correctly. We first write Parquet files, and then add them to the table using `AddFiles`. I think this is wrong, since `AddFiles` is meant as a migration feature. This method will fetch the footer, check the schema etc. Also, the writers that come with Iceberg-go, should always produce the Field-IDs.

This *only* implements the migration feature of `AddFiles`. This does not yet implement any writing of parquet files by `iceberg-go` outside of the unit tests writing parquet files without field ids for testing the `AddFiles` function.

> Instead, when we write Parquet, we directly want to convert that into a DataFile, and then we can add them to the table using add_file

Yup, i plan on doing that as future work. It was an easier code path to put together the `AddFiles` migration functions than to implement all of the work for splitting up records and tables for determining when and how to write out the parquet files. 

When I changed the `Append` to just return a NotImplemented error a whole bunch of code for the manifest merging was now unused, so I've removed that from this PR to simplify it more. Is that sufficient to reduce this to make it easier to review? Otherwise I can look into splitting it more along what you've suggested.
zeroshade: @Fokko The difficulty with splitting out the AddFiles portion to a separate PR is that all of the table writing unit tests that I added rely on AddFiles being implemented for actually testing the whole way through. I can't figure out a good way to easily test the internal manifest generation and handling without doing a full end-to-end test by actually adding files to the table.

What do you think?
zeroshade: The easiest part to excise as a separate PR is likely the stats aggregation implementation pieces. I can remove that from this PR (most of the tests should still work fine for adding files, just without the bounds existing in the result) and then I can create a PR for just the stats aggregations. Would that shrink this enough?
zeroshade: @Fokko I've added the test cases and handling of decimals. Do you still need me to split this PR into smaller chunks or anything else? Or are we good here and ready for the next one? :smile:

Patch:
diff --git a/go.mod b/go.mod
index 72aa91a..35062d1 100644
--- a/go.mod
+++ b/go.mod
@@ -22,12 +22,12 @@ go 1.23.0
 toolchain go1.23.6
 
 require (
-	github.com/apache/arrow-go/v18 v18.1.1-0.20250226170053-efecae3596e6
+	github.com/apache/arrow-go/v18 v18.2.0
 	github.com/aws/aws-sdk-go-v2 v1.36.3
 	github.com/aws/aws-sdk-go-v2/config v1.29.9
 	github.com/aws/aws-sdk-go-v2/credentials v1.17.62
 	github.com/aws/aws-sdk-go-v2/service/glue v1.106.1
-	github.com/aws/aws-sdk-go-v2/service/s3 v1.78.1
+	github.com/aws/aws-sdk-go-v2/service/s3 v1.78.2
 	github.com/aws/smithy-go v1.22.3
 	github.com/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815
 	github.com/google/uuid v1.6.0
@@ -46,7 +46,7 @@ require (
 	github.com/uptrace/bun/extra/bundebug v1.2.11
 	gocloud.dev v0.40.0
 	golang.org/x/sync v0.12.0
-	google.golang.org/api v0.224.0
+	google.golang.org/api v0.226.0
 	gopkg.in/yaml.v3 v3.0.1
 )
 
@@ -54,48 +54,56 @@ require (
 	atomicgo.dev/cursor v0.2.0 // indirect
 	atomicgo.dev/keyboard v0.2.9 // indirect
 	atomicgo.dev/schedule v0.1.0 // indirect
-	cloud.google.com/go v0.118.0 // indirect
+	cel.dev/expr v0.19.2 // indirect
+	cloud.google.com/go v0.119.0 // indirect
 	cloud.google.com/go/auth v0.15.0 // indirect
 	cloud.google.com/go/auth/oauth2adapt v0.2.7 // indirect
 	cloud.google.com/go/compute/metadata v0.6.0 // indirect
-	cloud.google.com/go/iam v1.2.2 // indirect
-	cloud.google.com/go/storage v1.43.0 // indirect
-	github.com/JohnCGriffin/overflow v0.0.0-20211019200055-46fa312c352c // indirect
+	cloud.google.com/go/iam v1.4.2 // indirect
+	cloud.google.com/go/monitoring v1.24.0 // indirect
+	cloud.google.com/go/storage v1.51.0 // indirect
+	github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.25.0 // indirect
+	github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.51.0 // indirect
+	github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.51.0 // indirect
 	github.com/andybalholm/brotli v1.1.1 // indirect
 	github.com/antlr4-go/antlr/v4 v4.13.1 // indirect
 	github.com/apache/thrift v0.21.0 // indirect
-	github.com/aws/aws-sdk-go v1.55.5 // indirect
+	github.com/aws/aws-sdk-go v1.55.6 // indirect
 	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.10 // indirect
 	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.30 // indirect
-	github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.17.11 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.17.66 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.34 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.34 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.34 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.3 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.6.2 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.7.0 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.15 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.18.15 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sso v1.25.1 // indirect
 	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.29.1 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sts v1.33.17 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	github.com/cncf/xds/go v0.0.0-20250121191232-2f005788dc42 // indirect
 	github.com/cockroachdb/apd/v3 v3.2.1 // indirect
-	github.com/containerd/console v1.0.3 // indirect
+	github.com/containerd/console v1.0.4 // indirect
 	github.com/creasty/defaults v1.8.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/dustin/go-humanize v1.0.1 // indirect
+	github.com/envoyproxy/go-control-plane/envoy v1.32.4 // indirect
+	github.com/envoyproxy/protoc-gen-validate v1.2.1 // indirect
 	github.com/fatih/color v1.18.0 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/goccy/go-json v0.10.5 // indirect
-	github.com/goccy/go-yaml v1.11.0 // indirect
-	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
-	github.com/golang/snappy v0.0.4 // indirect
+	github.com/goccy/go-yaml v1.16.0 // indirect
+	github.com/golang/groupcache v0.0.0-20241129210726-2c02b8208cf8 // indirect
+	github.com/golang/snappy v1.0.0 // indirect
 	github.com/google/flatbuffers v25.2.10+incompatible // indirect
 	github.com/google/s2a-go v0.1.9 // indirect
 	github.com/google/wire v0.6.0 // indirect
-	github.com/googleapis/enterprise-certificate-proxy v0.3.5 // indirect
+	github.com/googleapis/enterprise-certificate-proxy v0.3.6 // indirect
 	github.com/googleapis/gax-go/v2 v2.14.1 // indirect
 	github.com/gookit/color v1.5.4 // indirect
 	github.com/jinzhu/inflection v1.0.0 // indirect
@@ -103,7 +111,7 @@ require (
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/klauspost/asmfmt v1.3.2 // indirect
 	github.com/klauspost/compress v1.18.0 // indirect
-	github.com/klauspost/cpuid/v2 v2.2.9 // indirect
+	github.com/klauspost/cpuid/v2 v2.2.10 // indirect
 	github.com/lithammer/fuzzysearch v1.1.8 // indirect
 	github.com/mattn/go-colorable v0.1.14 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
@@ -116,12 +124,13 @@ require (
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/ncruces/go-strftime v0.1.9 // indirect
 	github.com/pierrec/lz4/v4 v4.1.22 // indirect
+	github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/puzpuzpuz/xsync/v3 v3.5.1 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
 	github.com/rivo/uniseg v0.4.7 // indirect
 	github.com/stretchr/objx v0.5.2 // indirect
-	github.com/substrait-io/substrait v0.66.1-0.20250205013839-a30b3e2d7ec6 // indirect
+	github.com/substrait-io/substrait v0.69.0 // indirect
 	github.com/tmthrgd/go-hex v0.0.0-20190904060850-447a3041c3bc // indirect
 	github.com/vmihailenco/msgpack/v5 v5.4.1 // indirect
 	github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
@@ -129,29 +138,32 @@ require (
 	github.com/zeebo/xxh3 v1.0.2 // indirect
 	go.opencensus.io v0.24.0 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
-	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.59.0 // indirect
-	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.59.0 // indirect
-	go.opentelemetry.io/otel v1.34.0 // indirect
-	go.opentelemetry.io/otel/metric v1.34.0 // indirect
-	go.opentelemetry.io/otel/trace v1.34.0 // indirect
-	golang.org/x/crypto v0.35.0 // indirect
-	golang.org/x/exp v0.0.0-20250228200357-dead58393ab7 // indirect
-	golang.org/x/mod v0.23.0 // indirect
-	golang.org/x/net v0.36.0 // indirect
-	golang.org/x/oauth2 v0.27.0 // indirect
-	golang.org/x/sys v0.30.0 // indirect
-	golang.org/x/term v0.29.0 // indirect
-	golang.org/x/text v0.22.0 // indirect
-	golang.org/x/time v0.10.0 // indirect
-	golang.org/x/tools v0.30.0 // indirect
+	go.opentelemetry.io/contrib/detectors/gcp v1.34.0 // indirect
+	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.60.0 // indirect
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0 // indirect
+	go.opentelemetry.io/otel v1.35.0 // indirect
+	go.opentelemetry.io/otel/metric v1.35.0 // indirect
+	go.opentelemetry.io/otel/sdk v1.35.0 // indirect
+	go.opentelemetry.io/otel/sdk/metric v1.35.0 // indirect
+	go.opentelemetry.io/otel/trace v1.35.0 // indirect
+	golang.org/x/crypto v0.36.0 // indirect
+	golang.org/x/exp v0.0.0-20250305212735-054e65f0b394 // indirect
+	golang.org/x/mod v0.24.0 // indirect
+	golang.org/x/net v0.37.0 // indirect
+	golang.org/x/oauth2 v0.28.0 // indirect
+	golang.org/x/sys v0.31.0 // indirect
+	golang.org/x/term v0.30.0 // indirect
+	golang.org/x/text v0.23.0 // indirect
+	golang.org/x/time v0.11.0 // indirect
+	golang.org/x/tools v0.31.0 // indirect
 	golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da // indirect
-	google.golang.org/genproto v0.0.0-20241118233622-e639e219e697 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20241209162323-e6fa225c2576 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20250227231956-55c901821b1e // indirect
-	google.golang.org/grpc v1.70.0 // indirect
+	google.golang.org/genproto v0.0.0-20250313205543-e70fdf4c4cb4 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20250313205543-e70fdf4c4cb4 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20250313205543-e70fdf4c4cb4 // indirect
+	google.golang.org/grpc v1.71.0 // indirect
 	google.golang.org/protobuf v1.36.5 // indirect
 	modernc.org/libc v1.61.13 // indirect
 	modernc.org/mathutil v1.7.1 // indirect
 	modernc.org/memory v1.8.2 // indirect
-	modernc.org/sqlite v1.36.0 // indirect
+	modernc.org/sqlite v1.36.1 // indirect
 )
diff --git a/go.sum b/go.sum
index 3d97872..bea6919 100644
--- a/go.sum
+++ b/go.sum
@@ -6,24 +6,38 @@ atomicgo.dev/keyboard v0.2.9 h1:tOsIid3nlPLZ3lwgG8KZMp/SFmr7P0ssEN5JUsm78K8=
 atomicgo.dev/keyboard v0.2.9/go.mod h1:BC4w9g00XkxH/f1HXhW2sXmJFOCWbKn9xrOunSFtExQ=
 atomicgo.dev/schedule v0.1.0 h1:nTthAbhZS5YZmgYbb2+DH8uQIZcTlIrd4eYr3UQxEjs=
 atomicgo.dev/schedule v0.1.0/go.mod h1:xeUa3oAkiuHYh8bKiQBRojqAMq3PXXbJujjb0hw8pEU=
+cel.dev/expr v0.19.2 h1:V354PbqIXr9IQdwy4SYA4xa0HXaWq1BUPAGzugBY5V4=
+cel.dev/expr v0.19.2/go.mod h1:MrpN08Q+lEBs+bGYdLxxHkZoUSsCp0nSKTs0nTymJgw=
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
-cloud.google.com/go v0.118.0 h1:tvZe1mgqRxpiVa3XlIGMiPcEUbP1gNXELgD4y/IXmeQ=
-cloud.google.com/go v0.118.0/go.mod h1:zIt2pkedt/mo+DQjcT4/L3NDxzHPR29j5HcclNH+9PM=
+cloud.google.com/go v0.119.0 h1:tw7OjErMzJKbbjaEHkrt60KQrK5Wus/boCZ7tm5/RNE=
+cloud.google.com/go v0.119.0/go.mod h1:fwB8QLzTcNevxqi8dcpR+hoMIs3jBherGS9VUBDAW08=
 cloud.google.com/go/auth v0.15.0 h1:Ly0u4aA5vG/fsSsxu98qCQBemXtAtJf+95z9HK+cxps=
 cloud.google.com/go/auth v0.15.0/go.mod h1:WJDGqZ1o9E9wKIL+IwStfyn/+s59zl4Bi+1KQNVXLZ8=
 cloud.google.com/go/auth/oauth2adapt v0.2.7 h1:/Lc7xODdqcEw8IrZ9SvwnlLX6j9FHQM74z6cBk9Rw6M=
 cloud.google.com/go/auth/oauth2adapt v0.2.7/go.mod h1:NTbTTzfvPl1Y3V1nPpOgl2w6d/FjO7NNUQaWSox6ZMc=
 cloud.google.com/go/compute/metadata v0.6.0 h1:A6hENjEsCDtC1k8byVsgwvVcioamEHvZ4j01OwKxG9I=
 cloud.google.com/go/compute/metadata v0.6.0/go.mod h1:FjyFAW1MW0C203CEOMDTu3Dk1FlqW3Rga40jzHL4hfg=
-cloud.google.com/go/iam v1.2.2 h1:ozUSofHUGf/F4tCNy/mu9tHLTaxZFLOUiKzjcgWHGIA=
-cloud.google.com/go/iam v1.2.2/go.mod h1:0Ys8ccaZHdI1dEUilwzqng/6ps2YB6vRsjIe00/+6JY=
-cloud.google.com/go/longrunning v0.6.2 h1:xjDfh1pQcWPEvnfjZmwjKQEcHnpz6lHjfy7Fo0MK+hc=
-cloud.google.com/go/longrunning v0.6.2/go.mod h1:k/vIs83RN4bE3YCswdXC5PFfWVILjm3hpEUlSko4PiI=
-cloud.google.com/go/storage v1.43.0 h1:CcxnSohZwizt4LCzQHWvBf1/kvtHUn7gk9QERXPyXFs=
-cloud.google.com/go/storage v1.43.0/go.mod h1:ajvxEa7WmZS1PxvKRq4bq0tFT3vMd502JwstCcYv0Q0=
+cloud.google.com/go/iam v1.4.2 h1:4AckGYAYsowXeHzsn/LCKWIwSWLkdb0eGjH8wWkd27Q=
+cloud.google.com/go/iam v1.4.2/go.mod h1:REGlrt8vSlh4dfCJfSEcNjLGq75wW75c5aU3FLOYq34=
+cloud.google.com/go/logging v1.13.0 h1:7j0HgAp0B94o1YRDqiqm26w4q1rDMH7XNRU34lJXHYc=
+cloud.google.com/go/logging v1.13.0/go.mod h1:36CoKh6KA/M0PbhPKMq6/qety2DCAErbhXT62TuXALA=
+cloud.google.com/go/longrunning v0.6.5 h1:sD+t8DO8j4HKW4QfouCklg7ZC1qC4uzVZt8iz3uTW+Q=
+cloud.google.com/go/longrunning v0.6.5/go.mod h1:Et04XK+0TTLKa5IPYryKf5DkpwImy6TluQ1QTLwlKmY=
+cloud.google.com/go/monitoring v1.24.0 h1:csSKiCJ+WVRgNkRzzz3BPoGjFhjPY23ZTcaenToJxMM=
+cloud.google.com/go/monitoring v1.24.0/go.mod h1:Bd1PRK5bmQBQNnuGwHBfUamAV1ys9049oEPHnn4pcsc=
+cloud.google.com/go/storage v1.51.0 h1:ZVZ11zCiD7b3k+cH5lQs/qcNaoSz3U9I0jgwVzqDlCw=
+cloud.google.com/go/storage v1.51.0/go.mod h1:YEJfu/Ki3i5oHC/7jyTgsGZwdQ8P9hqMqvpi5kRKGgc=
+cloud.google.com/go/trace v1.11.3 h1:c+I4YFjxRQjvAhRmSsmjpASUKq88chOX854ied0K/pE=
+cloud.google.com/go/trace v1.11.3/go.mod h1:pt7zCYiDSQjC9Y2oqCsh9jF4GStB/hmjrYLsxRR27q8=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
-github.com/JohnCGriffin/overflow v0.0.0-20211019200055-46fa312c352c h1:RGWPOewvKIROun94nF7v2cua9qP+thov/7M50KEoeSU=
-github.com/JohnCGriffin/overflow v0.0.0-20211019200055-46fa312c352c/go.mod h1:X0CRv0ky0k6m906ixxpzmDRLvX58TFUKS2eePweuyxk=
+github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.25.0 h1:3c8yed4lgqTt+oTQ+JNMDo+F4xprBf+O/il4ZC0nRLw=
+github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.25.0/go.mod h1:obipzmGjfSjam60XLwGfqUkJsfiheAl+TUjG+4yzyPM=
+github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.51.0 h1:fYE9p3esPxA/C0rQ0AHhP0drtPXDRhaWiwg1DPqO7IU=
+github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.51.0/go.mod h1:BnBReJLvVYx2CS/UHOgVz2BXKXD9wsQPxZug20nZhd0=
+github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0.51.0 h1:OqVGm6Ei3x5+yZmSJG1Mh2NwHvpVmZ08CB5qJhT9Nuk=
+github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0.51.0/go.mod h1:SZiPHWGOOk3bl8tkevxkoiwPgsIl6CwrWcbwjfHZpdM=
+github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.51.0 h1:6/0iUd0xrnX7qt+mLNRwg5c0PGv8wpE8K90ryANQwMI=
+github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.51.0/go.mod h1:otE2jQekW/PqXk1Awf5lmfokJx4uwuqcj1ab5SpGeW0=
 github.com/MarvinJWendt/testza v0.1.0/go.mod h1:7AxNvlfeHP7Z/hDQ5JtE3OKYT3XFUeLCDE2DQninSqs=
 github.com/MarvinJWendt/testza v0.2.1/go.mod h1:God7bhG8n6uQxwdScay+gjm9/LnO4D3kkcZX4hv9Rp8=
 github.com/MarvinJWendt/testza v0.2.8/go.mod h1:nwIcjmr0Zz+Rcwfh3/4UhBp7ePKVhuBExvZqnKYWlII=
@@ -37,13 +51,13 @@ github.com/andybalholm/brotli v1.1.1 h1:PR2pgnyFznKEugtsUo0xLdDop5SKXd5Qf5ysW+7X
 github.com/andybalholm/brotli v1.1.1/go.mod h1:05ib4cKhjx3OQYUY22hTVd34Bc8upXjOLL2rKwwZBoA=
 github.com/antlr4-go/antlr/v4 v4.13.1 h1:SqQKkuVZ+zWkMMNkjy5FZe5mr5WURWnlpmOuzYWrPrQ=
 github.com/antlr4-go/antlr/v4 v4.13.1/go.mod h1:GKmUxMtwp6ZgGwZSva4eWPC5mS6vUAmOABFgjdkM7Nw=
-github.com/apache/arrow-go/v18 v18.1.1-0.20250226170053-efecae3596e6 h1:RWgwATcu+8ZR65fmb/esaQkQqoCBhiRnB6WPMfAxJOs=
-github.com/apache/arrow-go/v18 v18.1.1-0.20250226170053-efecae3596e6/go.mod h1:ev6qDk3+bf7eh/goQbjA4ZViRDhu+9pmad8YTLvupIg=
+github.com/apache/arrow-go/v18 v18.2.0 h1:QhWqpgZMKfWOniGPhbUxrHohWnooGURqL2R2Gg4SO1Q=
+github.com/apache/arrow-go/v18 v18.2.0/go.mod h1:Ic/01WSwGJWRrdAZcxjBZ5hbApNJ28K96jGYaxzzGUc=
 github.com/apache/thrift v0.21.0 h1:tdPmh/ptjE1IJnhbhrcl2++TauVjy242rkV/UzJChnE=
 github.com/apache/thrift v0.21.0/go.mod h1:W1H8aR/QRtYNvrPeFXBtobyRkd0/YVhTc6i07XIAgDw=
 github.com/atomicgo/cursor v0.0.1/go.mod h1:cBON2QmmrysudxNBFthvMtN32r3jxVRIvzkUiF/RuIk=
-github.com/aws/aws-sdk-go v1.55.5 h1:KKUZBfBoyqy5d3swXyiC7Q76ic40rYcbqH7qjh59kzU=
-github.com/aws/aws-sdk-go v1.55.5/go.mod h1:eRwEWoyTWFMVYVQzKMNHWP5/RV4xIUGMQfXQHfHkpNU=
+github.com/aws/aws-sdk-go v1.55.6 h1:cSg4pvZ3m8dgYcgqB97MrcdjUmZ1BeMYKUxMMB89IPk=
+github.com/aws/aws-sdk-go v1.55.6/go.mod h1:eRwEWoyTWFMVYVQzKMNHWP5/RV4xIUGMQfXQHfHkpNU=
 github.com/aws/aws-sdk-go-v2 v1.36.3 h1:mJoei2CxPutQVxaATCzDUjcZEjVRdpsiiXi2o38yqWM=
 github.com/aws/aws-sdk-go-v2 v1.36.3/go.mod h1:LLXuLpgzEbD766Z5ECcRmi8AzSwfZItDtmABVkRLGzg=
 github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.10 h1:zAybnyUQXIZ5mok5Jqwlf58/TFE7uvd3IAsa1aF9cXs=
@@ -54,8 +68,8 @@ github.com/aws/aws-sdk-go-v2/credentials v1.17.62 h1:fvtQY3zFzYJ9CfixuAQ96IxDrBa
 github.com/aws/aws-sdk-go-v2/credentials v1.17.62/go.mod h1:ElETBxIQqcxej++Cs8GyPBbgMys5DgQPTwo7cUPDKt8=
 github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.30 h1:x793wxmUWVDhshP8WW2mlnXuFrO4cOd3HLBroh1paFw=
 github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.30/go.mod h1:Jpne2tDnYiFascUEs2AWHJL9Yp7A5ZVy3TNyxaAjD6M=
-github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.17.11 h1:FEDZD/Axt5tKSkPAs967KZ++MkvYdBqr0a+cetRbjLM=
-github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.17.11/go.mod h1:dvlsbA32KfvCzqwTiX7maABgFek2RyUuYEJ3kyn/PmQ=
+github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.17.66 h1:MTLivtC3s89de7Fe3P8rzML/8XPNRfuyJhlRTsCEt0k=
+github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.17.66/go.mod h1:NAuQ2s6gaFEsuTIb2+P5t6amB1w5MhvJFxppoezGWH0=
 github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.34 h1:ZK5jHhnrioRkUNOc+hOgQKlUL5JeC3S6JgLxtQ+Rm0Q=
 github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.34/go.mod h1:p4VfIceZokChbA9FzMbRGz5OV+lekcVtHlPKEO0gSZY=
 github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.34 h1:SZwFm17ZUNNg5Np0ioo/gq8Mn6u9w19Mri8DnJ15Jf0=
@@ -68,14 +82,14 @@ github.com/aws/aws-sdk-go-v2/service/glue v1.106.1 h1:OYx855o5eYaymW30LgbhVPFjB6
 github.com/aws/aws-sdk-go-v2/service/glue v1.106.1/go.mod h1:6FqWCqW0Py6VOvY42NQyf9e7N+sNVnDEiHFklCCCoQc=
 github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.3 h1:eAh2A4b5IzM/lum78bZ590jy36+d/aFLgKF/4Vd1xPE=
 github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.3/go.mod h1:0yKJC/kb8sAnmlYa6Zs3QVYqaC8ug2AbnNChv5Ox3uA=
-github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.6.2 h1:t/gZFyrijKuSU0elA5kRngP/oU3mc0I+Dvp8HwRE4c0=
-github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.6.2/go.mod h1:iu6FSzgt+M2/x3Dk8zhycdIcHjEFb36IS8HVUVFoMg0=
+github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.7.0 h1:lguz0bmOoGzozP9XfRJR1QIayEYo+2vP/No3OfLF0pU=
+github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.7.0/go.mod h1:iu6FSzgt+M2/x3Dk8zhycdIcHjEFb36IS8HVUVFoMg0=
 github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.15 h1:dM9/92u2F1JbDaGooxTq18wmmFzbJRfXfVfy96/1CXM=
 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.78.1 h1:1M0gSbyP6q06gl3384wpoKPaH9G16NPqZFieEhLboSU=
-github.com/aws/aws-sdk-go-v2/service/s3 v1.78.1/go.mod h1:4qzsZSzB/KiX2EzDjs9D7A8rI/WGJxZceVJIHqtJjIU=
+github.com/aws/aws-sdk-go-v2/service/s3 v1.78.2 h1:jIiopHEV22b4yQP2q36Y0OmwLbsxNWdWwfZRR5QRRO4=
+github.com/aws/aws-sdk-go-v2/service/s3 v1.78.2/go.mod h1:U5SNqwhXB3Xe6F47kXvWihPl/ilGaEDe8HD/50Z9wxc=
 github.com/aws/aws-sdk-go-v2/service/sso v1.25.1 h1:8JdC7Gr9NROg1Rusk25IcZeTO59zLxsKgE0gkh5O6h0=
 github.com/aws/aws-sdk-go-v2/service/sso v1.25.1/go.mod h1:qs4a9T5EMLl/Cajiw2TcbNt2UNo/Hqlyp+GiuG4CFDI=
 github.com/aws/aws-sdk-go-v2/service/ssooidc v1.29.1 h1:KwuLovgQPcdjNMfFt9OhUd9a2OwcOKhxfvF4glTzLuA=
@@ -85,12 +99,17 @@ github.com/aws/aws-sdk-go-v2/service/sts v1.33.17/go.mod h1:cQnB8CUnxbMU82JvlqjK
 github.com/aws/smithy-go v1.22.3 h1:Z//5NuZCSW6R4PhQ93hShNbyBbn8BWCmCVCt+Q8Io5k=
 github.com/aws/smithy-go v1.22.3/go.mod h1:t1ufH5HMublsJYulve2RKmHDC15xu1f26kHCp/HgceI=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
+github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
+github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
 github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
+github.com/cncf/xds/go v0.0.0-20250121191232-2f005788dc42 h1:Om6kYQYDUk5wWbT0t0q6pvyM49i9XZAv9dDrkDA7gjk=
+github.com/cncf/xds/go v0.0.0-20250121191232-2f005788dc42/go.mod h1:W+zGtBO5Y1IgJhy4+A9GOqVhqLpfZi+vwmdNXUehLA8=
 github.com/cockroachdb/apd/v3 v3.2.1 h1:U+8j7t0axsIgvQUqthuNm82HIrYXodOV2iWLWtEaIwg=
 github.com/cockroachdb/apd/v3 v3.2.1/go.mod h1:klXJcjp+FffLTHlhIG69tezTDvdP065naDsHzKhYSqc=
-github.com/containerd/console v1.0.3 h1:lIr7SlA5PxZyMV30bDW0MGbiOPXwc63yRuCP0ARubLw=
 github.com/containerd/console v1.0.3/go.mod h1:7LqA/THxQ86k76b8c/EMSiaJ3h1eZkMkXar0TQ1gf3U=
+github.com/containerd/console v1.0.4 h1:F2g4+oChYvBTsASRTz8NP6iIAi97J3TtSAsLbIFn4ro=
+github.com/containerd/console v1.0.4/go.mod h1:YynlIjWYF8myEu6sdkwKIvGQq+cOckRm6So2avqoYAk=
 github.com/creasty/defaults v1.8.0 h1:z27FJxCAa0JKt3utc0sCImAEb+spPucmKoOdLHvHYKk=
 github.com/creasty/defaults v1.8.0/go.mod h1:iGzKe6pbEHnpMPtfDXZEr0NVxWnPTjb1bbDy08fPzYM=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -103,7 +122,15 @@ github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+m
 github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=
+github.com/envoyproxy/go-control-plane v0.13.4 h1:zEqyPVyku6IvWCFwux4x9RxkLOMUL+1vC9xUFv5l2/M=
+github.com/envoyproxy/go-control-plane v0.13.4/go.mod h1:kDfuBlDVsSj2MjrLEtRWtHlsWIFcGyB2RMO44Dc5GZA=
+github.com/envoyproxy/go-control-plane/envoy v1.32.4 h1:jb83lalDRZSpPWW2Z7Mck/8kXZ5CQAFYVjQcdVIr83A=
+github.com/envoyproxy/go-control-plane/envoy v1.32.4/go.mod h1:Gzjc5k8JcJswLjAx1Zm+wSYE20UrLtt7JZMWiWQXQEw=
+github.com/envoyproxy/go-control-plane/ratelimit v0.1.0 h1:/G9QYbddjL25KvtKTv3an9lx6VBE2cnb8wp1vEGNYGI=
+github.com/envoyproxy/go-control-plane/ratelimit v0.1.0/go.mod h1:Wk+tMFAFbCXaJPzVVHnPgRKdUdwW/KdbRt94AzgRee4=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
+github.com/envoyproxy/protoc-gen-validate v1.2.1 h1:DEo3O99U8j4hBFwbJfrz9VtgcDfUKS7KJ7spH3d86P8=
+github.com/envoyproxy/protoc-gen-validate v1.2.1/go.mod h1:d/C80l/jxXLdfEIhX1W2TmLfsJ31lvEjwamM4DxlWXU=
 github.com/fatih/color v1.18.0 h1:S8gINlzdQ840/4pfAwic/ZE0djQEH3wM94VfqLTZcOM=
 github.com/fatih/color v1.18.0/go.mod h1:4FelSpRwEGDpQ12mAdzqdOukCy4u8WUtOY6lkT/6HfU=
 github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg=
@@ -113,20 +140,14 @@ github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY=
 github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
 github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
 github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
-github.com/go-playground/locales v0.13.0 h1:HyWk6mgj5qFqCT5fjGBuRArbVDfE4hi8+e8ceBS/t7Q=
-github.com/go-playground/locales v0.13.0/go.mod h1:taPMhCMXrRLJO55olJkUXHZBHCxTMfnGwq/HNwmWNS8=
-github.com/go-playground/universal-translator v0.17.0 h1:icxd5fm+REJzpZx7ZfpaD876Lmtgy7VtROAbHHXk8no=
-github.com/go-playground/universal-translator v0.17.0/go.mod h1:UkSxE5sNxxRwHyU+Scu5vgOQjsIJAF8j9muTVoKLVtA=
-github.com/go-playground/validator/v10 v10.11.1 h1:prmOlTVv+YjZjmRmNSF3VmspqJIxJWXmqUsHwfTRRkQ=
-github.com/go-playground/validator/v10 v10.11.1/go.mod h1:i+3WkQ1FvaUjjxh1kSvIA4dMGDBiPU55YFDl0WbKdWU=
 github.com/goccy/go-json v0.10.5 h1:Fq85nIqj+gXn/S5ahsiTlK3TmC85qgirsdTP/+DeaC4=
 github.com/goccy/go-json v0.10.5/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M=
-github.com/goccy/go-yaml v1.11.0 h1:n7Z+zx8S9f9KgzG6KtQKf+kwqXZlLNR2F6018Dgau54=
-github.com/goccy/go-yaml v1.11.0/go.mod h1:H+mJrWtjPTJAHvRbV09MCK9xYwODM+wRTVFFTWckfng=
+github.com/goccy/go-yaml v1.16.0 h1:d7m1G7A0t+logajVtklHfDYJs2Et9g3gHwdBNNFou0w=
+github.com/goccy/go-yaml v1.16.0/go.mod h1:XBurs7gK8ATbW4ZPGKgcbrY1Br56PdM69F7LkFRi1kA=
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
 github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
-github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da h1:oI5xCqsCo564l8iNU+DwB5epxmsaqB+rhGL0m5jtYqE=
-github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
+github.com/golang/groupcache v0.0.0-20241129210726-2c02b8208cf8 h1:f+oWsMOmNPc8JmEHVZIycC7hBoQxHH9pNKQORJNozsQ=
+github.com/golang/groupcache v0.0.0-20241129210726-2c02b8208cf8/go.mod h1:wcDNUvekVysuuOpQKo3191zZyTpiI6se1N1ULghS0sw=
 github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
@@ -139,8 +160,8 @@ github.com/golang/protobuf v1.4.1/go.mod h1:U8fpvMrcmy5pZrNK1lt4xCsGvpyWQ/VVv6QD
 github.com/golang/protobuf v1.4.3/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
 github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
 github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
-github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM=
-github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/golang/snappy v1.0.0 h1:Oy607GVXHs7RtbggtPBnr2RmDArIsAefDwvrdWvRhGs=
+github.com/golang/snappy v1.0.0/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/google/flatbuffers v25.2.10+incompatible h1:F3vclr7C3HpB1k9mxCGRMXq6FdUalZ6H/pNX4FP1v0Q=
 github.com/google/flatbuffers v25.2.10+incompatible/go.mod h1:1AeVuKshWv4vARoZatz6mlQ0JxURH0Kv5+zNeJKJCa8=
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
@@ -168,8 +189,8 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/wire v0.6.0 h1:HBkoIh4BdSxoyo9PveV8giw7ZsaBOvzWKfcg/6MrVwI=
 github.com/google/wire v0.6.0/go.mod h1:F4QhpQ9EDIdJ1Mbop/NZBRB+5yrR6qg3BnctaoUk6NA=
-github.com/googleapis/enterprise-certificate-proxy v0.3.5 h1:VgzTY2jogw3xt39CusEnFJWm7rlsq5yL5q9XdLOuP5g=
-github.com/googleapis/enterprise-certificate-proxy v0.3.5/go.mod h1:MkHOF77EYAE7qfSuSS9PU6g4Nt4e11cnsDUowfwewLA=
+github.com/googleapis/enterprise-certificate-proxy v0.3.6 h1:GW/XbdyBFQ8Qe+YAmFU9uHLo7OnF5tL52HFAgMmyrf4=
+github.com/googleapis/enterprise-certificate-proxy v0.3.6/go.mod h1:MkHOF77EYAE7qfSuSS9PU6g4Nt4e11cnsDUowfwewLA=
 github.com/googleapis/gax-go/v2 v2.14.1 h1:hb0FFeiPaQskmvakKu5EbCbpntQn48jyHuvrkurSS/Q=
 github.com/googleapis/gax-go/v2 v2.14.1/go.mod h1:Hb/NubMaVM88SrNkvl8X/o8XWwDJEPqouaLeN2IUxoA=
 github.com/gookit/color v1.4.2/go.mod h1:fqRyamkC1W8uxl+lxCQxOT09l/vYfZ+QeiX3rKQHCoQ=
@@ -193,8 +214,8 @@ github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYW
 github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/klauspost/cpuid/v2 v2.0.10/go.mod h1:g2LTdtYhdyuGPqyWyv7qRAmj1WBqxuObKfj5c0PQa7c=
 github.com/klauspost/cpuid/v2 v2.0.12/go.mod h1:g2LTdtYhdyuGPqyWyv7qRAmj1WBqxuObKfj5c0PQa7c=
-github.com/klauspost/cpuid/v2 v2.2.9 h1:66ze0taIn2H33fBvCkXuv9BmCwDfafmiIVpKV9kKGuY=
-github.com/klauspost/cpuid/v2 v2.2.9/go.mod h1:rqkxqrZ1EhYM9G+hXH7YdowN5R5RGN6NK4QwQ3WMXF8=
+github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE=
+github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
@@ -202,8 +223,6 @@ github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
-github.com/leodido/go-urn v1.2.0 h1:hpXL4XnriNwQ/ABnpepYM/1vCLWNDfUNts8dX3xTG6Y=
-github.com/leodido/go-urn v1.2.0/go.mod h1:+8+nEpDfqqsY+g338gtMEUOtuK+4dEMhiQEgxpxOKII=
 github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
 github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
 github.com/lithammer/fuzzysearch v1.1.8 h1:/HIuJnjHuXS8bKaiTMeeDlW2/AyIWk2brx1V8LFgLN4=
@@ -232,6 +251,8 @@ github.com/ncruces/go-strftime v0.1.9 h1:bY0MQC28UADQmHmaF5dgpLmImcShSi2kHU9XLdh
 github.com/ncruces/go-strftime v0.1.9/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls=
 github.com/pierrec/lz4/v4 v4.1.22 h1:cKFw6uJDK+/gfw5BcDL0JL5aBsAFdsIT18eRtLj7VIU=
 github.com/pierrec/lz4/v4 v4.1.22/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
+github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 h1:GFCKgmp0tecUJ0sJuv4pzYCqS9+RGSn52M3FUwPs+uo=
+github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10/go.mod h1:t/avpk3KcrXxUnYOhZhMXJlSEyie6gQbtLq5NM3loB8=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
@@ -269,8 +290,8 @@ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO
 github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
 github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
 github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
-github.com/substrait-io/substrait v0.66.1-0.20250205013839-a30b3e2d7ec6 h1:XqtxwYFCjS4L0o1QD4ipGHCuFG94U0f6BeldbilGQjU=
-github.com/substrait-io/substrait v0.66.1-0.20250205013839-a30b3e2d7ec6/go.mod h1:MPFNw6sToJgpD5Z2rj0rQrdP/Oq8HG7Z2t3CAEHtkHw=
+github.com/substrait-io/substrait v0.69.0 h1:qfwUe1qKa3PsCclMpubQOF6nqIqS14geUuvzJ1P7gsM=
+github.com/substrait-io/substrait v0.69.0/go.mod h1:MPFNw6sToJgpD5Z2rj0rQrdP/Oq8HG7Z2t3CAEHtkHw=
 github.com/substrait-io/substrait-go/v3 v3.9.0 h1:sRJf0ID9q2TPxJ9eH+oAniepMqt9fYW0Hy32CScT2cI=
 github.com/substrait-io/substrait-go/v3 v3.9.0/go.mod h1:VG7jCqtUm28bSngHwq86FywtU74knJ25LNX63SZ53+E=
 github.com/tmthrgd/go-hex v0.0.0-20190904060850-447a3041c3bc h1:9lRDQMhESg+zvGYmW5DyG0UqvY96Bu5QYsTLvCHdrgo=
@@ -311,20 +332,24 @@ go.opencensus.io v0.24.0 h1:y73uSU6J157QMP2kn2r30vwW1A2W2WFwSCGnAVxeaD0=
 go.opencensus.io v0.24.0/go.mod h1:vNK8G9p7aAivkbmorf4v+7Hgx+Zs0yY+0fOtgBfjQKo=
 go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA=
 go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A=
-go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.59.0 h1:rgMkmiGfix9vFJDcDi1PK8WEQP4FLQwLDfhp5ZLpFeE=
-go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.59.0/go.mod h1:ijPqXp5P6IRRByFVVg9DY8P5HkxkHE5ARIa+86aXPf4=
-go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.59.0 h1:CV7UdSGJt/Ao6Gp4CXckLxVRRsRgDHoI8XjbL3PDl8s=
-go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.59.0/go.mod h1:FRmFuRJfag1IZ2dPkHnEoSFVgTVPUd2qf5Vi69hLb8I=
-go.opentelemetry.io/otel v1.34.0 h1:zRLXxLCgL1WyKsPVrgbSdMN4c0FMkDAskSTQP+0hdUY=
-go.opentelemetry.io/otel v1.34.0/go.mod h1:OWFPOQ+h4G8xpyjgqo4SxJYdDQ/qmRH+wivy7zzx9oI=
-go.opentelemetry.io/otel/metric v1.34.0 h1:+eTR3U0MyfWjRDhmFMxe2SsW64QrZ84AOhvqS7Y+PoQ=
-go.opentelemetry.io/otel/metric v1.34.0/go.mod h1:CEDrp0fy2D0MvkXE+dPV7cMi8tWZwX3dmaIhwPOaqHE=
-go.opentelemetry.io/otel/sdk v1.34.0 h1:95zS4k/2GOy069d321O8jWgYsW3MzVV+KuSPKp7Wr1A=
-go.opentelemetry.io/otel/sdk v1.34.0/go.mod h1:0e/pNiaMAqaykJGKbi+tSjWfNNHMTxoC9qANsCzbyxU=
-go.opentelemetry.io/otel/sdk/metric v1.32.0 h1:rZvFnvmvawYb0alrYkjraqJq0Z4ZUJAiyYCU9snn1CU=
-go.opentelemetry.io/otel/sdk/metric v1.32.0/go.mod h1:PWeZlq0zt9YkYAp3gjKZ0eicRYvOh1Gd+X99x6GHpCQ=
-go.opentelemetry.io/otel/trace v1.34.0 h1:+ouXS2V8Rd4hp4580a8q23bg0azF2nI8cqLYnC8mh/k=
-go.opentelemetry.io/otel/trace v1.34.0/go.mod h1:Svm7lSjQD7kG7KJ/MUHPVXSDGz2OX4h0M2jHBhmSfRE=
+go.opentelemetry.io/contrib/detectors/gcp v1.34.0 h1:JRxssobiPg23otYU5SbWtQC//snGVIM3Tx6QRzlQBao=
+go.opentelemetry.io/contrib/detectors/gcp v1.34.0/go.mod h1:cV4BMFcscUR/ckqLkbfQmF0PRsq8w/lMGzdbCSveBHo=
+go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.60.0 h1:x7wzEgXfnzJcHDwStJT+mxOz4etr2EcexjqhBvmoakw=
+go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.60.0/go.mod h1:rg+RlpR5dKwaS95IyyZqj5Wd4E13lk/msnTS0Xl9lJM=
+go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0 h1:sbiXRNDSWJOTobXh5HyQKjq6wUC5tNybqjIqDpAY4CU=
+go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0/go.mod h1:69uWxva0WgAA/4bu2Yy70SLDBwZXuQ6PbBpbsa5iZrQ=
+go.opentelemetry.io/otel v1.35.0 h1:xKWKPxrxB6OtMCbmMY021CqC45J+3Onta9MqjhnusiQ=
+go.opentelemetry.io/otel v1.35.0/go.mod h1:UEqy8Zp11hpkUrL73gSlELM0DupHoiq72dR+Zqel/+Y=
+go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.29.0 h1:WDdP9acbMYjbKIyJUhTvtzj601sVJOqgWdUxSdR/Ysc=
+go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.29.0/go.mod h1:BLbf7zbNIONBLPwvFnwNHGj4zge8uTCM/UPIVW1Mq2I=
+go.opentelemetry.io/otel/metric v1.35.0 h1:0znxYu2SNyuMSQT4Y9WDWej0VpcsxkuklLa4/siN90M=
+go.opentelemetry.io/otel/metric v1.35.0/go.mod h1:nKVFgxBZ2fReX6IlyW28MgZojkoAkJGaE8CpgeAU3oE=
+go.opentelemetry.io/otel/sdk v1.35.0 h1:iPctf8iprVySXSKJffSS79eOjl9pvxV9ZqOWT0QejKY=
+go.opentelemetry.io/otel/sdk v1.35.0/go.mod h1:+ga1bZliga3DxJ3CQGg3updiaAJoNECOgJREo9KHGQg=
+go.opentelemetry.io/otel/sdk/metric v1.35.0 h1:1RriWBmCKgkeHEhM7a2uMjMUfP7MsOF5JpUCaEqEI9o=
+go.opentelemetry.io/otel/sdk/metric v1.35.0/go.mod h1:is6XYCUMpcKi+ZsOvfluY5YstFnhW0BidkR+gL+qN+w=
+go.opentelemetry.io/otel/trace v1.35.0 h1:dPpEfJu1sDIqruz7BHFG3c7528f6ddfSWfFDVt/xgMs=
+go.opentelemetry.io/otel/trace v1.35.0/go.mod h1:WUk7DtFp1Aw2MkvqGdwiXYDZZNvA/1J8o6xRXLrIkyc=
 gocloud.dev v0.40.0 h1:f8LgP+4WDqOG/RXoUcyLpeIAGOcAbZrZbDQCUee10ng=
 gocloud.dev v0.40.0/go.mod h1:drz+VyYNBvrMTW0KZiBAYEdl8lbNZx+OQ7oQvdrFmSQ=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
@@ -332,11 +357,11 @@ golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPh
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.13.0/go.mod h1:y6Z2r+Rw4iayiXXAIxJIDAJ1zMW4yaTpebo8fPOliYc=
 golang.org/x/crypto v0.18.0/go.mod h1:R0j02AL6hcrfOiy9T4ZYp/rcWeMxM3L6QYxlOuEG1mg=
-golang.org/x/crypto v0.35.0 h1:b15kiHdrGCHrP6LvwaQ3c03kgNhhiMgvlhxHQhmg2Xs=
-golang.org/x/crypto v0.35.0/go.mod h1:dy7dXNW32cAb/6/PRuTNsix8T+vJAqvuIy5Bli/x0YQ=
+golang.org/x/crypto v0.36.0 h1:AnAEvhDddvBdpY+uR+MyHmuZzzNqXSe/GvuDeob5L34=
+golang.org/x/crypto v0.36.0/go.mod h1:Y4J0ReaxCR1IMaabaSMugxJES1EpwhBHhv2bDHklZvc=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
-golang.org/x/exp v0.0.0-20250228200357-dead58393ab7 h1:aWwlzYV971S4BXRS9AmqwDLAD85ouC6X+pocatKY58c=
-golang.org/x/exp v0.0.0-20250228200357-dead58393ab7/go.mod h1:BHOTPb3L19zxehTsLoJXVaTktb06DFgmdW6Wb9s8jqk=
+golang.org/x/exp v0.0.0-20250305212735-054e65f0b394 h1:nDVHiLt8aIbd/VzvPWN6kSOPE7+F/fNFDSXLVYkE/Iw=
+golang.org/x/exp v0.0.0-20250305212735-054e65f0b394/go.mod h1:sIifuuw/Yco/y6yb6+bDNfyeQ/MdPUy/hKEMYQV17cM=
 golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
 golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU=
 golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
@@ -344,8 +369,8 @@ golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91
 golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
 golang.org/x/mod v0.12.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
 golang.org/x/mod v0.14.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
-golang.org/x/mod v0.23.0 h1:Zb7khfcRGKk+kqfxFaP5tZqCnDZMjC5VtUBs87Hr6QM=
-golang.org/x/mod v0.23.0/go.mod h1:6SkKJ3Xj0I0BrPOZoBy3bdMptDDU9oJrpohJ3eWZ1fY=
+golang.org/x/mod v0.24.0 h1:ZfthKaKaT4NrhGVZHO1/WDTwGES4De8KtWO0SIbNJMU=
+golang.org/x/mod v0.24.0/go.mod h1:IXM97Txy2VM4PJ3gI61r1YEk/gAj6zAHN3AdZt6S9Ww=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190213061140-3a22650c66bd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -359,11 +384,11 @@ golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
 golang.org/x/net v0.15.0/go.mod h1:idbUs1IY1+zTqbi8yxTbhexhEEk5ur9LInksu6HrEpk=
 golang.org/x/net v0.20.0/go.mod h1:z8BVo6PvndSri0LbOE3hAn0apkU+1YvI6E70E9jsnvY=
-golang.org/x/net v0.36.0 h1:vWF2fRbw4qslQsQzgFqZff+BItCvGFQqKzKIzx1rmoA=
-golang.org/x/net v0.36.0/go.mod h1:bFmbeoIPfrw4sMHNhb4J9f6+tPziuGjq7Jk/38fxi1I=
+golang.org/x/net v0.37.0 h1:1zLorHbz+LYj7MQlSf1+2tPIIgibq2eL5xkrGk6f+2c=
+golang.org/x/net v0.37.0/go.mod h1:ivrbrMbzFq5J41QOQh0siUuly180yBYtLp+CKbEaFx8=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
-golang.org/x/oauth2 v0.27.0 h1:da9Vo7/tDv5RH/7nZDz1eMGS/q1Vv1N/7FCrBhI9I3M=
-golang.org/x/oauth2 v0.27.0/go.mod h1:onh5ek6nERTohokkhCD/y2cV4Do3fxFHFuAejCkRWT8=
+golang.org/x/oauth2 v0.28.0 h1:CrgCKl8PPAVtLnU3c+EDw6x11699EWlsDeWNWKdIOkc=
+golang.org/x/oauth2 v0.28.0/go.mod h1:onh5ek6nERTohokkhCD/y2cV4Do3fxFHFuAejCkRWT8=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -385,13 +410,14 @@ golang.org/x/sys v0.0.0-20211013075003-97ac67df715c/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220319134239-a9b59b0215f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
-golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc=
-golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.31.0 h1:ioabZlmFYtWhL+TRYpcnNlLwhyxaM9kWTDEmfnprqik=
+golang.org/x/sys v0.31.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210220032956-6a3ed077a48d/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210615171337-6886f2dfbf5b/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
@@ -400,8 +426,8 @@ golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
 golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo=
 golang.org/x/term v0.12.0/go.mod h1:owVbMEjm3cBLCHdkQu9b1opXd4ETQWc3BhuQGKgXgvU=
 golang.org/x/term v0.16.0/go.mod h1:yn7UURbUtPyrVJPGPq404EukNFxcm/foM+bV/bfcDsY=
-golang.org/x/term v0.29.0 h1:L6pJp37ocefwRRtYPKSWOWzOtWSxVajvz2ldH/xi3iU=
-golang.org/x/term v0.29.0/go.mod h1:6bl4lRlvVuDgSf3179VpIxBF0o10JUpXWOnI7nErv7s=
+golang.org/x/term v0.30.0 h1:PQ39fJZ+mfadBm0y5WlL4vlM7Sx1Hgf13sMIY2+QS9Y=
+golang.org/x/term v0.30.0/go.mod h1:NYYFdzHoI5wRh/h5tDMdMqCqPJZEuNqVR5xJLd/n67g=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
@@ -409,10 +435,10 @@ golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
 golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
 golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
-golang.org/x/text v0.22.0 h1:bofq7m3/HAFvbF51jz3Q9wLg3jkvSPuiZu/pD1XwgtM=
-golang.org/x/text v0.22.0/go.mod h1:YRoo4H8PVmsu+E3Ou7cqLVH8oXWIHVoX0jqUWALQhfY=
-golang.org/x/time v0.10.0 h1:3usCWA8tQn0L8+hFJQNgzpWbd89begxN66o1Ojdn5L4=
-golang.org/x/time v0.10.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
+golang.org/x/text v0.23.0 h1:D71I7dUrlY+VX0gQShAThNGHFxZ13dGLBHQLVl1mJlY=
+golang.org/x/text v0.23.0/go.mod h1:/BLNzu4aZCJ1+kcD0DNRotWKage4q2rGVAg4o22unh4=
+golang.org/x/time v0.11.0 h1:/bpjEDfN9tkoN/ryeYHnv5hcMlc8ncjMcM4XBk5NWV0=
+golang.org/x/time v0.11.0/go.mod h1:CDIdPxbZBQxdj6cxyCIdrNogrJKMJ7pr37NYpMcMDSg=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY=
@@ -423,34 +449,34 @@ golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc
 golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=
 golang.org/x/tools v0.13.0/go.mod h1:HvlwmtVNQAhOuCjW7xxvovg8wbNq7LwfXh/k7wXUl58=
 golang.org/x/tools v0.17.0/go.mod h1:xsh6VxdV005rRVaS6SSAf9oiAqljS7UZUacMZ8Bnsps=
-golang.org/x/tools v0.30.0 h1:BgcpHewrV5AUp2G9MebG4XPFI1E2W41zU1SaqVA9vJY=
-golang.org/x/tools v0.30.0/go.mod h1:c347cR/OJfw5TI+GfX7RUPNMdDRRbjvYTS0jPyvsVtY=
+golang.org/x/tools v0.31.0 h1:0EedkvKDbh+qistFTd0Bcwe/YLh4vHwWEkiI0toFIBU=
+golang.org/x/tools v0.31.0/go.mod h1:naFTU+Cev749tSJRXJlna0T3WxKvb1kWEx15xA4SdmQ=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da h1:noIWHXmPHxILtqtCOPIhSt0ABwskkZKjD3bXGnZGpNY=
 golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da/go.mod h1:NDW/Ps6MPRej6fsCIbMTohpP40sJ/P/vI1MoTEGwX90=
 gonum.org/v1/gonum v0.15.1 h1:FNy7N6OUZVUaWG9pTiD+jlhdQ3lMP+/LcTpJ6+a8sQ0=
 gonum.org/v1/gonum v0.15.1/go.mod h1:eZTZuRFrzu5pcyjN5wJhcIhnUdNijYxX1T2IcrOGY0o=
-google.golang.org/api v0.224.0 h1:Ir4UPtDsNiwIOHdExr3fAj4xZ42QjK7uQte3lORLJwU=
-google.golang.org/api v0.224.0/go.mod h1:3V39my2xAGkodXy0vEqcEtkqgw2GtrFL5WuBZlCTCOQ=
+google.golang.org/api v0.226.0 h1:9A29y1XUD+YRXfnHkO66KggxHBZWg9LsTGqm7TkUvtQ=
+google.golang.org/api v0.226.0/go.mod h1:WP/0Xm4LVvMOCldfvOISnWquSRWbG2kArDZcg+W2DbY=
 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=
 google.golang.org/genproto v0.0.0-20190819201941-24fa4b261c55/go.mod h1:DMBHOl98Agz4BDEuKkezgsaosCRResVns1a3J2ZsMNc=
 google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013/go.mod h1:NbSheEEYHJ7i3ixzK3sjbqSGDJWnxyFXZblF3eUsNvo=
-google.golang.org/genproto v0.0.0-20241118233622-e639e219e697 h1:ToEetK57OidYuqD4Q5w+vfEnPvPpuTwedCNVohYJfNk=
-google.golang.org/genproto v0.0.0-20241118233622-e639e219e697/go.mod h1:JJrvXBWRZaFMxBufik1a4RpFw4HhgVtBBWQeQgUj2cc=
-google.golang.org/genproto/googleapis/api v0.0.0-20241209162323-e6fa225c2576 h1:CkkIfIt50+lT6NHAVoRYEyAvQGFM7xEwXUUywFvEb3Q=
-google.golang.org/genproto/googleapis/api v0.0.0-20241209162323-e6fa225c2576/go.mod h1:1R3kvZ1dtP3+4p4d3G8uJ8rFk/fWlScl38vanWACI08=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20250227231956-55c901821b1e h1:YA5lmSs3zc/5w+xsRcHqpETkaYyK63ivEPzNTcUUlSA=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20250227231956-55c901821b1e/go.mod h1:LuRYeWDFV6WOn90g357N17oMCaxpgCnbi/44qJvDn2I=
+google.golang.org/genproto v0.0.0-20250313205543-e70fdf4c4cb4 h1:kCjWYliqPA8g5z87mbjnf/cdgQqMzBfp9xYre5qKu2A=
+google.golang.org/genproto v0.0.0-20250313205543-e70fdf4c4cb4/go.mod h1:SqIx1NV9hcvqdLHo7uNZDS5lrUJybQ3evo3+z/WBfA0=
+google.golang.org/genproto/googleapis/api v0.0.0-20250313205543-e70fdf4c4cb4 h1:IFnXJq3UPB3oBREOodn1v1aGQeZYQclEmvWRMN0PSsY=
+google.golang.org/genproto/googleapis/api v0.0.0-20250313205543-e70fdf4c4cb4/go.mod h1:c8q6Z6OCqnfVIqUFJkCzKcrj8eCvUrz+K4KRzSTuANg=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20250313205543-e70fdf4c4cb4 h1:iK2jbkWL86DXjEx0qiHcRE9dE4/Ahua5k6V8OWFb//c=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20250313205543-e70fdf4c4cb4/go.mod h1:LuRYeWDFV6WOn90g357N17oMCaxpgCnbi/44qJvDn2I=
 google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
 google.golang.org/grpc v1.23.0/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyacEbxg=
 google.golang.org/grpc v1.25.1/go.mod h1:c3i+UQWmh7LiEpx4sFZnkU36qjEYZ0imhYfXVyQciAY=
 google.golang.org/grpc v1.27.0/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8abTk=
 google.golang.org/grpc v1.33.2/go.mod h1:JMHMWHQWaTccqQQlmk3MJZS+GWXOdAesneDmEnv2fbc=
-google.golang.org/grpc v1.70.0 h1:pWFv03aZoHzlRKHWicjsZytKAiYCtNS0dHbXnIdq7jQ=
-google.golang.org/grpc v1.70.0/go.mod h1:ofIJqVKDXx/JiXrwr2IG4/zwdH9txy3IlF40RmcJSQw=
+google.golang.org/grpc v1.71.0 h1:kF77BGdPTQ4/JZWMlb9VpJ5pa25aqvVqogsxNHHdeBg=
+google.golang.org/grpc v1.71.0/go.mod h1:H0GRtasmQOh9LkFoCPDu3ZrwUtD1YGE+b2vYBYd/8Ec=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
 google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
 google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM=
@@ -494,8 +520,8 @@ modernc.org/opt v0.1.4 h1:2kNGMRiUjrp4LcaPuLY2PzUfqM/w9N23quVwhKt5Qm8=
 modernc.org/opt v0.1.4/go.mod h1:03fq9lsNfvkYSfxrfUhZCWPk1lm4cq4N+Bh//bEtgns=
 modernc.org/sortutil v1.2.1 h1:+xyoGf15mM3NMlPDnFqrteY07klSFxLElE2PVuWIJ7w=
 modernc.org/sortutil v1.2.1/go.mod h1:7ZI3a3REbai7gzCLcotuw9AC4VZVpYMjDzETGsSMqJE=
-modernc.org/sqlite v1.36.0 h1:EQXNRn4nIS+gfsKeUTymHIz1waxuv5BzU7558dHSfH8=
-modernc.org/sqlite v1.36.0/go.mod h1:7MPwH7Z6bREicF9ZVUR78P1IKuxfZ8mRIDHD0iD+8TU=
+modernc.org/sqlite v1.36.1 h1:bDa8BJUH4lg6EGkLbahKe/8QqoF8p9gArSc6fTqYhyQ=
+modernc.org/sqlite v1.36.1/go.mod h1:7MPwH7Z6bREicF9ZVUR78P1IKuxfZ8mRIDHD0iD+8TU=
 modernc.org/strutil v1.2.1 h1:UneZBkQA+DX2Rp35KcM69cSsNES9ly8mQWD71HKlOA0=
 modernc.org/strutil v1.2.1/go.mod h1:EHkiggD70koQxjVdSBM3JKM7k6L0FbGE5eymy9i3B9A=
 modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y=
diff --git a/literals.go b/literals.go
index 06ac8de..0671e77 100644
--- a/literals.go
+++ b/literals.go
@@ -1215,8 +1215,9 @@ func (d DecimalLiteral) MarshalBinary() (data []byte, err error) {
 	// stored as unscaled value in two's compliment big-endian values
 	// using the minimum number of bytes for the values
 	n := decimal128.Num(d.Val).BigInt()
+	minBytes := (n.BitLen() + 8) / 8
 	// bytes gives absolute value as big-endian bytes
-	data = n.Bytes()
+	data = n.FillBytes(make([]byte, minBytes))
 	if n.Sign() < 0 {
 		// convert to 2's complement for negative value
 		for i, v := range data {
diff --git a/table/arrow_utils.go b/table/arrow_utils.go
index 752f677..f987bb4 100644
--- a/table/arrow_utils.go
+++ b/table/arrow_utils.go
@@ -19,21 +19,31 @@ package table
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"iter"
 	"maps"
 	"regexp"
 	"slices"
 	"strconv"
 	"strings"
+	"unsafe"
 
 	"github.com/apache/arrow-go/v18/arrow"
 	"github.com/apache/arrow-go/v18/arrow/array"
 	"github.com/apache/arrow-go/v18/arrow/compute"
+	"github.com/apache/arrow-go/v18/arrow/decimal128"
 	"github.com/apache/arrow-go/v18/arrow/extensions"
 	"github.com/apache/arrow-go/v18/arrow/memory"
+	"github.com/apache/arrow-go/v18/parquet/file"
 	"github.com/apache/arrow-go/v18/parquet/metadata"
+	"github.com/apache/arrow-go/v18/parquet/pqarrow"
 	"github.com/apache/iceberg-go"
 	"github.com/apache/iceberg-go/internal"
+	iceio "github.com/apache/iceberg-go/io"
+	tblutils "github.com/apache/iceberg-go/table/internal"
+	"github.com/google/uuid"
+	"github.com/pterm/pterm"
 )
 
 // constants to look for as Keys in Arrow field metadata
@@ -900,6 +910,139 @@ func ToRequestedSchema(ctx context.Context, requested, fileSchema *iceberg.Schem
 	return out, nil
 }
 
+type schemaCompatVisitor struct {
+	provided *iceberg.Schema
+
+	errorData pterm.TableData
+}
+
+func checkSchemaCompat(requested, provided *iceberg.Schema) error {
+	sc := &schemaCompatVisitor{
+		provided:  provided,
+		errorData: pterm.TableData{{"", "Table Field", "Requested Field"}},
+	}
+
+	_, compat := iceberg.PreOrderVisit(requested, sc)
+
+	return compat
+}
+
+func checkArrowSchemaCompat(requested *iceberg.Schema, provided *arrow.Schema, downcastNanoToMicro bool) error {
+	mapping := requested.NameMapping()
+	providedSchema, err := ArrowSchemaToIceberg(provided, downcastNanoToMicro, mapping)
+	if err != nil {
+		return err
+	}
+
+	return checkSchemaCompat(requested, providedSchema)
+}
+
+func (sc *schemaCompatVisitor) isFieldCompat(lhs iceberg.NestedField) bool {
+	rhs, ok := sc.provided.FindFieldByID(lhs.ID)
+	if !ok {
+		if lhs.Required {
+			sc.errorData = append(sc.errorData,
+				[]string{"❌", lhs.String(), "missing"})
+
+			return false
+		}
+		sc.errorData = append(sc.errorData,
+			[]string{"✅", lhs.String(), "missing"})
+
+		return true
+	}
+
+	if lhs.Required && !rhs.Required {
+		sc.errorData = append(sc.errorData,
+			[]string{"❌", lhs.String(), rhs.String()})
+
+		return false
+	}
+
+	if lhs.Type.Equals(rhs.Type) {
+		sc.errorData = append(sc.errorData,
+			[]string{"✅", lhs.String(), rhs.String()})
+
+		return true
+	}
+
+	// we only check that parent node is also of the same type
+	// we check the type of the child nodes as we traverse them later
+	switch lhs.Type.(type) {
+	case *iceberg.StructType:
+		if rhs, ok := rhs.Type.(*iceberg.StructType); ok {
+			sc.errorData = append(sc.errorData,
+				[]string{"✅", lhs.String(), rhs.String()})
+
+			return true
+		}
+	case *iceberg.ListType:
+		if rhs, ok := rhs.Type.(*iceberg.ListType); ok {
+			sc.errorData = append(sc.errorData,
+				[]string{"✅", lhs.String(), rhs.String()})
+
+			return true
+		}
+	case *iceberg.MapType:
+		if rhs, ok := rhs.Type.(*iceberg.MapType); ok {
+			sc.errorData = append(sc.errorData,
+				[]string{"✅", lhs.String(), rhs.String()})
+
+			return true
+		}
+	}
+
+	if _, err := iceberg.PromoteType(rhs.Type, lhs.Type); err != nil {
+		sc.errorData = append(sc.errorData,
+			[]string{"❌", lhs.String(), rhs.String()})
+
+		return false
+	}
+
+	sc.errorData = append(sc.errorData,
+		[]string{"✅", lhs.String(), rhs.String()})
+
+	return true
+}
+
+func (sc *schemaCompatVisitor) Schema(s *iceberg.Schema, v func() bool) bool {
+	if !v() {
+		pterm.DisableColor()
+		tbl := pterm.DefaultTable.WithHasHeader(true).WithData(sc.errorData)
+		tbl.Render()
+		txt, _ := tbl.Srender()
+		pterm.EnableColor()
+		panic("mismatch in fields:\n" + txt)
+	}
+
+	return true
+}
+
+func (sc *schemaCompatVisitor) Struct(st iceberg.StructType, v []func() bool) bool {
+	out := true
+	for _, res := range v {
+		out = res() && out
+	}
+
+	return out
+}
+
+func (sc *schemaCompatVisitor) Field(n iceberg.NestedField, v func() bool) bool {
+	return sc.isFieldCompat(n) && v()
+}
+
+func (sc *schemaCompatVisitor) List(l iceberg.ListType, v func() bool) bool {
+	return sc.isFieldCompat(l.ElementField()) && v()
+}
+
+func (sc *schemaCompatVisitor) Map(m iceberg.MapType, vk, vv func() bool) bool {
+	return sc.isFieldCompat(m.KeyField()) && sc.isFieldCompat(m.ValueField()) && vk() && vv()
+}
+
+func (sc *schemaCompatVisitor) Primitive(p iceberg.PrimitiveType) bool {
+	return true
+}
+
 func must[T any](v T, err error) T {
 	if err != nil {
 		panic(err)
@@ -908,6 +1051,344 @@ func must[T any](v T, err error) T {
 	return v
 }
 
+func primitiveToPhysicalType(typ iceberg.Type) string {
+	switch t := typ.(type) {
+	case iceberg.BooleanType:
+		return "BOOLEAN"
+	case iceberg.Int32Type:
+		return "INT32"
+	case iceberg.Int64Type:
+		return "INT64"
+	case iceberg.Float32Type:
+		return "FLOAT"
+	case iceberg.Float64Type:
+		return "DOUBLE"
+	case iceberg.DateType:
+		return "INT32"
+	case iceberg.TimeType:
+		return "INT64"
+	case iceberg.TimestampType:
+		return "INT64"
+	case iceberg.TimestampTzType:
+		return "INT64"
+	case iceberg.StringType:
+		return "BYTE_ARRAY"
+	case iceberg.UUIDType:
+		return "FIXED_LEN_BYTE_ARRAY"
+	case iceberg.FixedType:
+		return "FIXED_LEN_BYTE_ARRAY"
+	case iceberg.BinaryType:
+		return "BYTE_ARRAY"
+	case iceberg.DecimalType:
+		switch {
+		case t.Precision() <= 9:
+			return "INT32"
+		case t.Precision() <= 18:
+			return "INT64"
+		default:
+			return "FIXED_LEN_BYTE_ARRAY"
+		}
+	default:
+		panic(fmt.Errorf("expected primitive type, got: %s", typ))
+	}
+}
+
+type typedStat[T iceberg.LiteralType] interface {
+	Min() T
+	Max() T
+}
+
+type wrappedBinaryStats struct {
+	*metadata.ByteArrayStatistics
+}
+
+func (w wrappedBinaryStats) Min() []byte {
+	return w.ByteArrayStatistics.Min()
+}
+
+func (w wrappedBinaryStats) Max() []byte {
+	return w.ByteArrayStatistics.Max()
+}
+
+type wrappedStringStats struct {
+	*metadata.ByteArrayStatistics
+}
+
+func (w wrappedStringStats) Min() string {
+	data := w.ByteArrayStatistics.Min()
+
+	return unsafe.String(unsafe.SliceData(data), len(data))
+}
+
+func (w wrappedStringStats) Max() string {
+	data := w.ByteArrayStatistics.Max()
+
+	return unsafe.String(unsafe.SliceData(data), len(data))
+}
+
+type wrappedUUIDStats struct {
+	*metadata.FixedLenByteArrayStatistics
+}
+
+func (w wrappedUUIDStats) Min() uuid.UUID {
+	uid, err := uuid.FromBytes(w.FixedLenByteArrayStatistics.Min())
+	if err != nil {
+		panic(err)
+	}
+
+	return uid
+}
+
+func (w wrappedUUIDStats) Max() uuid.UUID {
+	uid, err := uuid.FromBytes(w.FixedLenByteArrayStatistics.Max())
+	if err != nil {
+		panic(err)
+	}
+
+	return uid
+}
+
+type wrappedFLBAStats struct {
+	*metadata.FixedLenByteArrayStatistics
+}
+
+func (w wrappedFLBAStats) Min() []byte {
+	return w.FixedLenByteArrayStatistics.Min()
+}
+
+func (w wrappedFLBAStats) Max() []byte {
+	return w.FixedLenByteArrayStatistics.Max()
+}
+
+type wrappedDecStats struct {
+	*metadata.FixedLenByteArrayStatistics
+	scale int
+}
+
+func (w wrappedDecStats) Min() iceberg.Decimal {
+	dec, err := tblutils.BigEndianToDecimal(w.FixedLenByteArrayStatistics.Min())
+	if err != nil {
+		panic(err)
+	}
+
+	return iceberg.Decimal{Val: dec, Scale: w.scale}
+}
+
+func (w wrappedDecStats) Max() iceberg.Decimal {
+	dec, err := tblutils.BigEndianToDecimal(w.FixedLenByteArrayStatistics.Max())
+	if err != nil {
+		panic(err)
+	}
+
+	return iceberg.Decimal{Val: dec, Scale: w.scale}
+}
+
+type statsAgg interface {
+	update(stats metadata.TypedStatistics)
+	minAsBytes() ([]byte, error)
+	maxAsBytes() ([]byte, error)
+}
+
+type statsAggregator[T iceberg.LiteralType] struct {
+	curMin iceberg.TypedLiteral[T]
+	curMax iceberg.TypedLiteral[T]
+
+	cmp           iceberg.Comparator[T]
+	primitiveType iceberg.PrimitiveType
+	truncLen      int
+}
+
+func newStatAgg[T iceberg.LiteralType](typ iceberg.PrimitiveType, trunc int) statsAgg {
+	var z T
+
+	return &statsAggregator[T]{
+		primitiveType: typ,
+		truncLen:      trunc,
+		cmp:           iceberg.NewLiteral(z).(iceberg.TypedLiteral[T]).Comparator(),
+	}
+}
+
+func createStatsAgg(typ iceberg.PrimitiveType, physicalTypeStr string, truncLen int) (statsAgg, error) {
+	expectedPhysical := primitiveToPhysicalType(typ)
+	if physicalTypeStr != expectedPhysical {
+		switch {
+		case physicalTypeStr == "INT32" && expectedPhysical == "INT64":
+		case physicalTypeStr == "FLOAT" && expectedPhysical == "DOUBLE":
+		default:
+			return nil, fmt.Errorf("unexpected physical type %s for %s, expected %s",
+				physicalTypeStr, typ, expectedPhysical)
+		}
+	}
+
+	switch physicalTypeStr {
+	case "BOOLEAN":
+		return newStatAgg[bool](typ, truncLen), nil
+	case "INT32":
+		switch typ.(type) {
+		case iceberg.DecimalType:
+			return &decAsIntAgg[int32]{
+				newStatAgg[int32](typ, truncLen).(*statsAggregator[int32]),
+			}, nil
+		}
+
+		return newStatAgg[int32](typ, truncLen), nil
+	case "INT64":
+		switch typ.(type) {
+		case iceberg.DecimalType:
+			return &decAsIntAgg[int64]{
+				newStatAgg[int64](typ, truncLen).(*statsAggregator[int64]),
+			}, nil
+		}
+
+		return newStatAgg[int64](typ, truncLen), nil
+	case "FLOAT":
+		return newStatAgg[float32](typ, truncLen), nil
+	case "DOUBLE":
+		return newStatAgg[float64](typ, truncLen), nil
+	case "FIXED_LEN_BYTE_ARRAY":
+		switch typ.(type) {
+		case iceberg.UUIDType:
+			return newStatAgg[uuid.UUID](typ, truncLen), nil
+		case iceberg.DecimalType:
+			return newStatAgg[iceberg.Decimal](typ, truncLen), nil
+		default:
+			return newStatAgg[[]byte](typ, truncLen), nil
+		}
+	case "BYTE_ARRAY":
+		if typ.Equals(iceberg.PrimitiveTypes.String) {
+			return newStatAgg[string](typ, truncLen), nil
+		}
+
+		return newStatAgg[[]byte](typ, truncLen), nil
+	default:
+		return nil, fmt.Errorf("unsupported physical type: %s", physicalTypeStr)
+	}
+}
+
+func (s *statsAggregator[T]) update(stats metadata.TypedStatistics) {
+	st := stats.(typedStat[T])
+	s.updateMin(st.Min())
+	s.updateMax(st.Max())
+}
+
+func (s *statsAggregator[T]) toBytes(val iceberg.Literal) ([]byte, error) {
+	v, err := val.To(s.primitiveType)
+	if err != nil {
+		return nil, err
+	}
+
+	return v.MarshalBinary()
+}
+
+func (s *statsAggregator[T]) updateMin(val T) {
+	if s.curMin == nil {
+		s.curMin = iceberg.NewLiteral(val).(iceberg.TypedLiteral[T])
+	} else {
+		if s.cmp(val, s.curMin.Value()) < 0 {
+			s.curMin = iceberg.NewLiteral(val).(iceberg.TypedLiteral[T])
+		}
+	}
+}
+
+func (s *statsAggregator[T]) updateMax(val T) {
+	if s.curMax == nil {
+		s.curMax = iceberg.NewLiteral(val).(iceberg.TypedLiteral[T])
+	} else {
+		if s.cmp(val, s.curMax.Value()) > 0 {
+			s.curMax = iceberg.NewLiteral(val).(iceberg.TypedLiteral[T])
+		}
+	}
+}
+
+func (s *statsAggregator[T]) minAsBytes() ([]byte, error) {
+	if s.curMin == nil {
+		return nil, nil
+	}
+
+	if s.truncLen > 0 {
+		return s.toBytes((&iceberg.TruncateTransform{Width: s.truncLen}).
+			Apply(iceberg.Optional[iceberg.Literal]{Valid: true, Val: s.curMin}).Val)
+	}
+
+	return s.toBytes(s.curMin)
+}
+
+func (s *statsAggregator[T]) maxAsBytes() ([]byte, error) {
+	if s.curMax == nil {
+		return nil, nil
+	}
+
+	if s.truncLen <= 0 {
+		return s.toBytes(s.curMax)
+	}
+
+	switch s.primitiveType.(type) {
+	case iceberg.StringType:
+		if !s.curMax.Type().Equals(s.primitiveType) {
+			return nil, errors.New("expected current max to be a string")
+		}
+
+		curMax := any(s.curMax.Value()).(string)
+		result := truncateUpperBoundText(curMax, s.truncLen)
+		if result != "" {
+			return s.toBytes(iceberg.StringLiteral(result))
+		}
+
+		return nil, nil
+	case iceberg.BinaryType:
+		if !s.curMax.Type().Equals(s.primitiveType) {
+			return nil, errors.New("expected current max to be a binary")
+		}
+
+		curMax := any(s.curMax.Value()).([]byte)
+		result := truncateUpperBoundBinary(curMax, s.truncLen)
+		if len(result) > 0 {
+			return s.toBytes(iceberg.BinaryLiteral(result))
+		}
+
+		return nil, nil
+	default:
+		return nil, fmt.Errorf("%s cannot be truncated for upper bound", s.primitiveType)
+	}
+}
+
+type decAsIntAgg[T int32 | int64] struct {
+	*statsAggregator[T]
+}
+
+func (s *decAsIntAgg[T]) minAsBytes() ([]byte, error) {
+	if s.curMin == nil {
+		return nil, nil
+	}
+
+	lit := iceberg.DecimalLiteral(iceberg.Decimal{
+		Val:   decimal128.FromI64(int64(s.curMin.Value())),
+		Scale: s.primitiveType.(iceberg.DecimalType).Scale(),
+	})
+	if s.truncLen > 0 {
+		return s.toBytes((&iceberg.TruncateTransform{Width: s.truncLen}).
+			Apply(iceberg.Optional[iceberg.Literal]{Valid: true, Val: lit}).Val)
+	}
+
+	return s.toBytes(lit)
+}
+
+func (s *decAsIntAgg[T]) maxAsBytes() ([]byte, error) {
+	if s.curMax == nil {
+		return nil, nil
+	}
+
+	lit := iceberg.DecimalLiteral(iceberg.Decimal{
+		Val:   decimal128.FromI64(int64(s.curMax.Value())),
+		Scale: s.primitiveType.(iceberg.DecimalType).Scale(),
+	})
+	if s.truncLen <= 0 {
+		return s.toBytes(lit)
+	}
+
+	return nil, fmt.Errorf("%s cannot be truncated for upper bound", s.primitiveType)
+}
+
 type metricModeType string
 
 const (
@@ -1146,17 +1627,45 @@ type dataFileStatistics struct {
 	valueCounts     map[int]int64
 	nullValueCounts map[int]int64
 	nanValueCounts  map[int]int64
-	// column aggregates
-	splitOffsets []int64
+	colAggs         map[int]statsAgg
+	splitOffsets    []int64
 }
 
 func (d *dataFileStatistics) toDataFile(spec iceberg.PartitionSpec, path string, format iceberg.FileFormat, filesize int64) (iceberg.DataFile, error) {
 	bldr, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentData,
-		path, iceberg.ParquetFile, nil, d.recordCount, filesize)
+		path, format, nil, d.recordCount, filesize)
 	if err != nil {
 		return nil, err
 	}
 
+	lowerBounds := make(map[int][]byte)
+	upperBounds := make(map[int][]byte)
+
+	for fieldID, agg := range d.colAggs {
+		min, err := agg.minAsBytes()
+		if err != nil {
+			return nil, err
+		}
+
+		max, err := agg.maxAsBytes()
+		if err != nil {
+			return nil, err
+		}
+
+		if len(min) > 0 {
+			lowerBounds[fieldID] = min
+		}
+		if len(max) > 0 {
+			upperBounds[fieldID] = max
+		}
+	}
+
+	if len(lowerBounds) > 0 {
+		bldr.LowerBoundValues(lowerBounds)
+	}
+	if len(upperBounds) > 0 {
+		bldr.UpperBoundValues(upperBounds)
+	}
 	bldr.ColumnSizes(d.colSizes)
 	bldr.ValueCounts(d.valueCounts)
 	bldr.NullValueCounts(d.nullValueCounts)
@@ -1174,6 +1683,7 @@ func dataFileStatsFromParquetMetadata(pqmeta *metadata.FileMetaData, statsCols m
 		nullValueCounts = make(map[int]int64)
 		nanValueCounts  = make(map[int]int64)
 		invalidateCol   = make(map[int]struct{})
+		colAggs         = make(map[int]statsAgg)
 	)
 
 	for rg := range pqmeta.NumRowGroups() {
@@ -1227,11 +1737,36 @@ func dataFileStatsFromParquetMetadata(pqmeta *metadata.FileMetaData, statsCols m
 				nullValueCounts[fieldID] += stats.NullCount()
 			}
 
-			if statsCol.mode.typ == metricModeCounts {
+			if statsCol.mode.typ == metricModeCounts || !stats.HasMinMax() {
 				continue
 			}
 
-			// TODO stats aggregator!
+			agg, ok := colAggs[fieldID]
+			if !ok {
+				agg, err = createStatsAgg(statsCol.icebergTyp, stats.Type().String(), statsCol.mode.len)
+				if err != nil {
+					panic(err)
+				}
+
+				colAggs[fieldID] = agg
+			}
+
+			switch typ := statsCol.icebergTyp.(type) {
+			case iceberg.BinaryType:
+				stats = wrappedBinaryStats{stats.(*metadata.ByteArrayStatistics)}
+			case iceberg.UUIDType:
+				stats = wrappedUUIDStats{stats.(*metadata.FixedLenByteArrayStatistics)}
+			case iceberg.StringType:
+				stats = wrappedStringStats{stats.(*metadata.ByteArrayStatistics)}
+			case iceberg.FixedType:
+				stats = wrappedFLBAStats{stats.(*metadata.FixedLenByteArrayStatistics)}
+			case iceberg.DecimalType:
+				if typ.Precision() > 18 {
+					stats = wrappedDecStats{stats.(*metadata.FixedLenByteArrayStatistics), typ.Scale()}
+				}
+			}
+
+			agg.update(stats)
 		}
 
 	}
@@ -1242,6 +1777,11 @@ func dataFileStatsFromParquetMetadata(pqmeta *metadata.FileMetaData, statsCols m
 
 		return ok
 	})
+	maps.DeleteFunc(colAggs, func(fieldID int, _ statsAgg) bool {
+		_, ok := invalidateCol[fieldID]
+
+		return ok
+	})
 
 	return &dataFileStatistics{
 		recordCount:     pqmeta.GetNumRows(),
@@ -1250,5 +1790,56 @@ func dataFileStatsFromParquetMetadata(pqmeta *metadata.FileMetaData, statsCols m
 		nullValueCounts: nullValueCounts,
 		nanValueCounts:  nanValueCounts,
 		splitOffsets:    splitOffsets,
+		colAggs:         colAggs,
+	}
+}
+
+func parquetFilesToDataFiles(fileIO iceio.IO, meta *MetadataBuilder, paths iter.Seq[string]) iter.Seq2[iceberg.DataFile, error] {
+	return func(yield func(iceberg.DataFile, error) bool) {
+		defer func() {
+			if r := recover(); r != nil {
+				switch e := r.(type) {
+				case string:
+					yield(nil, fmt.Errorf("error encountered during parquet file conversion: %s", e))
+				case error:
+					yield(nil, fmt.Errorf("error encountered during parquet file conversion: %w", e))
+				}
+			}
+		}()
+
+		for filePath := range paths {
+			inputFile := must(fileIO.Open(filePath))
+			defer inputFile.Close()
+
+			rdr := must(file.NewParquetReader(inputFile))
+			defer rdr.Close()
+
+			arrRdr := must(pqarrow.NewFileReader(rdr, pqarrow.ArrowReadProperties{}, memory.DefaultAllocator))
+			arrSchema := must(arrRdr.Schema())
+
+			if hasIDs := must(VisitArrowSchema(arrSchema, hasIDs{})); hasIDs {
+				yield(nil, fmt.Errorf("%w: cannot add file %s because it has field-ids. add-files only supports the addition of files without field_ids",
+					iceberg.ErrNotImplemented, filePath))
+
+				return
+			}
+
+			if err := checkArrowSchemaCompat(meta.CurrentSchema(), arrSchema, false); err != nil {
+				panic(err)
+			}
+
+			statistics := dataFileStatsFromParquetMetadata(rdr.MetaData(),
+				must(computeStatsPlan(meta.CurrentSchema(), meta.props)),
+				must(parquetPathToIDMapping(meta.CurrentSchema())))
+
+			df, err := statistics.toDataFile(meta.CurrentSpec(), filePath, iceberg.ParquetFile, rdr.MetaData().GetSourceFileSize())
+			if err != nil {
+				panic(err)
+			}
+
+			if !yield(df, nil) {
+				return
+			}
+		}
 	}
 }
diff --git a/table/internal/utils.go b/table/internal/utils.go
index 7b5f7d8..cab0d73 100644
--- a/table/internal/utils.go
+++ b/table/internal/utils.go
@@ -19,6 +19,11 @@ package internal
 
 import (
 	"container/heap"
+	"encoding/binary"
+	"fmt"
+
+	"github.com/apache/arrow-go/v18/arrow/decimal"
+	"github.com/apache/arrow-go/v18/arrow/decimal128"
 )
 
 // Enumerated is a quick way to represent a sequenced value that can
@@ -79,3 +84,58 @@ func MakeSequencedChan[T any](bufferSize uint, source <-chan T, comesAfter, isNe
 
 	return out
 }
+
+func u64FromBigEndianShifted(buf []byte) uint64 {
+	var bytes [8]byte
+	copy(bytes[8-len(buf):], buf)
+
+	return binary.BigEndian.Uint64(bytes[:])
+}
+
+func BigEndianToDecimal(buf []byte) (decimal.Decimal128, error) {
+	const (
+		minDecBytes = 1
+		maxDecBytes = 16
+	)
+
+	if len(buf) < minDecBytes || len(buf) > maxDecBytes {
+		return decimal.Decimal128{},
+			fmt.Errorf("invalid length for conversion to decimal: %d, must be between %d and %d",
+				len(buf), minDecBytes, maxDecBytes)
+	}
+
+	// big endian, so first byte is the MSB and host the sign bit
+	isNeg := int8(buf[0]) < 0
+
+	var hi, lo int64
+	// step 1. extract high bits
+	highBitsOffset := max(0, len(buf)-8)
+	highBits := u64FromBigEndianShifted(buf[:highBitsOffset])
+
+	if highBitsOffset == 8 {
+		hi = int64(highBits)
+	} else {
+		if isNeg && len(buf) < maxDecBytes {
+			hi = -1
+		}
+
+		hi = int64(uint64(hi) << (uint64(highBitsOffset) * 8))
+		hi |= int64(highBits)
+	}
+
+	// step 2. extract low bits
+	lowBitsOffset := min(len(buf), 8)
+	lowBits := u64FromBigEndianShifted(buf[highBitsOffset:])
+	if lowBitsOffset == 8 {
+		lo = int64(lowBits)
+	} else {
+		if isNeg && len(buf) < 8 {
+			lo = -1
+		}
+
+		lo = int64(uint64(lo) << (uint64(lowBitsOffset) * 8))
+		lo |= int64(lowBits)
+	}
+
+	return decimal128.New(hi, uint64(lo)), nil
+}
diff --git a/table/metadata.go b/table/metadata.go
index 6828979..454320e 100644
--- a/table/metadata.go
+++ b/table/metadata.go
@@ -18,6 +18,7 @@
 package table
 
 import (
+	"encoding/binary"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -43,6 +44,25 @@ const (
 	addSortOrderAction   = "add-sort-order"
 )
 
+func generateSnapshotID() int64 {
+	var (
+		rndUUID = uuid.New()
+		out     [8]byte
+	)
+
+	for i := range 8 {
+		lhs, rhs := rndUUID[i], rndUUID[i+8]
+		out[i] = lhs ^ rhs
+	}
+
+	snapshotID := int64(binary.LittleEndian.Uint64(out[:]))
+	if snapshotID < 0 {
+		snapshotID = -snapshotID
+	}
+
+	return snapshotID
+}
+
 // Metadata for an iceberg table as specified in the Iceberg spec
 //
 // https://iceberg.apache.org/spec/#iceberg-table-spec
@@ -214,6 +234,37 @@ func (b *MetadataBuilder) CurrentSchema() *iceberg.Schema {
 	return s
 }
 
+func (b *MetadataBuilder) nextSequenceNumber() int64 {
+	if b.formatVersion > 1 {
+		if b.lastSequenceNumber == nil {
+			return 0
+		}
+
+		return *b.lastSequenceNumber + 1
+	}
+
+	return 0
+}
+
+func (b *MetadataBuilder) newSnapshotID() int64 {
+	snapshotID := generateSnapshotID()
+	for slices.ContainsFunc(b.snapshotList, func(s Snapshot) bool { return s.SnapshotID == snapshotID }) {
+		snapshotID = generateSnapshotID()
+	}
+
+	return snapshotID
+}
+
+func (b *MetadataBuilder) currentSnapshot() *Snapshot {
+	if b.currentSnapshotID == nil {
+		return nil
+	}
+
+	s, _ := b.SnapshotByID(*b.currentSnapshotID)
+
+	return s
+}
+
 func (b *MetadataBuilder) AddSchema(schema *iceberg.Schema, newLastColumnID int, initial bool) (*MetadataBuilder, error) {
 	if newLastColumnID < b.lastColumnId {
 		return nil, fmt.Errorf("%w: newLastColumnID %d, must be >= %d", iceberg.ErrInvalidArgument, newLastColumnID, b.lastColumnId)
diff --git a/table/properties.go b/table/properties.go
index e2f9e8c..0005049 100644
--- a/table/properties.go
+++ b/table/properties.go
@@ -27,8 +27,22 @@ const (
 
 	DefaultNameMappingKey = "schema.name-mapping.default"
 
-	MetricsModeColumnConfPrefix = "write.metadata.metrics.column"
-
+	MetricsModeColumnConfPrefix    = "write.metadata.metrics.column"
 	DefaultWriteMetricsModeKey     = "write.metadata.metrics.default"
 	DefaultWriteMetricsModeDefault = "truncate(16)"
+
+	ManifestMergeEnabledKey     = "commit.manifest-merge.enabled"
+	ManifestMergeEnabledDefault = false
+
+	ManifestTargetSizeBytesKey     = "commit.manifest.target-size-bytes"
+	ManifestTargetSizeBytesDefault = 8 * 1024 * 1024 // 8 MB
+
+	ManifestMinMergeCountKey     = "commit.manifest.min-count-to-merge"
+	ManifestMinMergeCountDefault = 100
+
+	WritePartitionSummaryLimitKey     = "write.summary.partition-limit"
+	WritePartitionSummaryLimitDefault = 0
+
+	MetadataDeleteAfterCommitEnabledKey     = "write.metadata.delete-after-commit.enabled"
+	MetadataDeleteAfterCommitEnabledDefault = false
 )
diff --git a/table/snapshot_producers.go b/table/snapshot_producers.go
new file mode 100644
index 0000000..298a2f5
--- /dev/null
+++ b/table/snapshot_producers.go
@@ -0,0 +1,384 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"maps"
+	"slices"
+	"sync/atomic"
+	"unicode/utf8"
+
+	"github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/internal"
+	iceio "github.com/apache/iceberg-go/io"
+	"github.com/google/uuid"
+	"golang.org/x/sync/errgroup"
+)
+
+type producerImpl interface {
+	processManifests(manifests []iceberg.ManifestFile) ([]iceberg.ManifestFile, error)
+	existingManifests() ([]iceberg.ManifestFile, error)
+	deletedEntries() ([]iceberg.ManifestEntry, error)
+}
+
+func newManifestFileName(num int, commit uuid.UUID) string {
+	return fmt.Sprintf("%s-m%d.avro", commit, num)
+}
+
+func newManifestListFileName(snapshotID int64, attempt int, commit uuid.UUID) string {
+	// mimics behavior of java
+	// https://github.com/apache/iceberg/blob/c862b9177af8e2d83122220764a056f3b96fd00c/core/src/main/java/org/apache/iceberg/SnapshotProducer.java#L491
+	return fmt.Sprintf("snap-%d-%d-%s.avro", snapshotID, attempt, commit)
+}
+
+func newFastAppendFilesProducer(op Operation, txn *Transaction, fs iceio.WriteFileIO, commitUUID *uuid.UUID, snapshotProps iceberg.Properties) *snapshotProducer {
+	prod := createSnapshotProducer(op, txn, fs, commitUUID, snapshotProps)
+	prod.producerImpl = &fastAppendFiles{base: prod}
+
+	return prod
+}
+
+type fastAppendFiles struct {
+	base *snapshotProducer
+}
+
+func (fa *fastAppendFiles) processManifests(manifests []iceberg.ManifestFile) ([]iceberg.ManifestFile, error) {
+	return manifests, nil
+}
+
+func (fa *fastAppendFiles) existingManifests() ([]iceberg.ManifestFile, error) {
+	existing := make([]iceberg.ManifestFile, 0)
+	if fa.base.parentSnapshotID > 0 {
+		previous, err := fa.base.txn.meta.SnapshotByID(fa.base.parentSnapshotID)
+		if err != nil {
+			return nil, fmt.Errorf("could not find parent snapshot %d", fa.base.parentSnapshotID)
+		}
+
+		manifests, err := previous.Manifests(fa.base.io)
+		if err != nil {
+			return nil, err
+		}
+
+		for _, m := range manifests {
+			if m.HasAddedFiles() || m.HasExistingFiles() || m.SnapshotID() == fa.base.snapshotID {
+				existing = append(existing, m)
+			}
+		}
+	}
+
+	return existing, nil
+}
+
+func (fa *fastAppendFiles) deletedEntries() ([]iceberg.ManifestEntry, error) {
+	return nil, nil
+}
+
+type snapshotProducer struct {
+	producerImpl
+
+	commitUuid       uuid.UUID
+	io               iceio.WriteFileIO
+	txn              *Transaction
+	op               Operation
+	snapshotID       int64
+	parentSnapshotID int64
+	addedFiles       []iceberg.DataFile
+	manifestCount    atomic.Int32
+	deletedFiles     map[string]iceberg.DataFile
+	snapshotProps    iceberg.Properties
+}
+
+func createSnapshotProducer(op Operation, txn *Transaction, fs iceio.WriteFileIO, commitUUID *uuid.UUID, snapshotProps iceberg.Properties) *snapshotProducer {
+	var (
+		commit         uuid.UUID
+		parentSnapshot int64 = -1
+	)
+
+	if commitUUID == nil {
+		commit = uuid.New()
+	} else {
+		commit = *commitUUID
+	}
+
+	if snap := txn.meta.currentSnapshot(); snap != nil {
+		parentSnapshot = snap.SnapshotID
+	}
+
+	return &snapshotProducer{
+		commitUuid:       commit,
+		io:               fs,
+		txn:              txn,
+		op:               op,
+		snapshotID:       txn.meta.newSnapshotID(),
+		parentSnapshotID: parentSnapshot,
+		addedFiles:       []iceberg.DataFile{},
+		deletedFiles:     make(map[string]iceberg.DataFile),
+		snapshotProps:    snapshotProps,
+	}
+}
+
+func (sp *snapshotProducer) spec(id int) iceberg.PartitionSpec {
+	if spec, _ := sp.txn.meta.GetSpecByID(id); spec != nil {
+		return *spec
+	}
+
+	return iceberg.NewPartitionSpec()
+}
+
+func (sp *snapshotProducer) appendDataFile(df iceberg.DataFile) *snapshotProducer {
+	sp.addedFiles = append(sp.addedFiles, df)
+
+	return sp
+}
+
+func (sp *snapshotProducer) newManifestOutput() (io.WriteCloser, string, error) {
+	provider, err := sp.txn.tbl.LocationProvider()
+	if err != nil {
+		return nil, "", err
+	}
+	fname := newManifestFileName(int(sp.manifestCount.Add(1)), sp.commitUuid)
+	filepath := provider.NewMetadataLocation(fname)
+	f, err := sp.io.Create(filepath)
+	if err != nil {
+		return nil, "", fmt.Errorf("could not create manifest file: %w", err)
+	}
+
+	return f, filepath, nil
+}
+
+func (sp *snapshotProducer) manifests() ([]iceberg.ManifestFile, error) {
+	var g errgroup.Group
+
+	results := [...][]iceberg.ManifestFile{nil, nil, nil}
+
+	if len(sp.addedFiles) > 0 {
+		g.Go(func() error {
+			out, path, err := sp.newManifestOutput()
+			if err != nil {
+				return err
+			}
+			defer out.Close()
+
+			counter := &internal.CountingWriter{W: out}
+
+			wr, err := iceberg.NewManifestWriter(sp.txn.meta.formatVersion, counter,
+				sp.txn.meta.CurrentSpec(), sp.txn.meta.CurrentSchema(),
+				sp.snapshotID)
+			if err != nil {
+				return err
+			}
+
+			for _, df := range sp.addedFiles {
+				err := wr.Add(iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &sp.snapshotID,
+					nil, nil, df))
+				if err != nil {
+					return err
+				}
+			}
+
+			mf, err := wr.ToManifestFile(path, counter.Count)
+			if err == nil {
+				results[0] = append(results[0], mf)
+			}
+
+			return err
+		})
+	}
+
+	deleted, err := sp.deletedEntries()
+	if err != nil {
+		return nil, err
+	}
+
+	if len(deleted) > 0 {
+		g.Go(func() error {
+			partitionGroups := map[int][]iceberg.ManifestEntry{}
+			for _, entry := range deleted {
+				specid := int(entry.DataFile().SpecID())
+
+				group := partitionGroups[specid]
+				partitionGroups[specid] = append(group, entry)
+			}
+
+			for specid, entries := range partitionGroups {
+				out, path, err := sp.newManifestOutput()
+				if err != nil {
+					return err
+				}
+				defer out.Close()
+
+				mf, err := iceberg.WriteManifest(path, out, sp.txn.meta.formatVersion,
+					sp.spec(specid), sp.txn.meta.CurrentSchema(), sp.snapshotID, entries)
+				if err != nil {
+					return err
+				}
+				results[1] = append(results[1], mf)
+			}
+
+			return nil
+		})
+	}
+
+	g.Go(func() error {
+		m, err := sp.existingManifests()
+		if err != nil {
+			return err
+		}
+		results[2] = m
+
+		return nil
+	})
+
+	if err := g.Wait(); err != nil {
+		return nil, err
+	}
+
+	manifests := slices.Concat(results[0], results[1], results[2])
+
+	return sp.processManifests(manifests)
+}
+
+func (sp *snapshotProducer) summary(props iceberg.Properties) (Summary, error) {
+	var ssc SnapshotSummaryCollector
+	partitionSummaryLimit := sp.txn.meta.props.
+		GetInt(WritePartitionSummaryLimitKey, WritePartitionSummaryLimitDefault)
+	ssc.setPartitionSummaryLimit(partitionSummaryLimit)
+
+	currentSchema := sp.txn.meta.CurrentSchema()
+	partitionSpec := sp.txn.meta.CurrentSpec()
+	for _, df := range sp.addedFiles {
+		ssc.addFile(df, currentSchema, partitionSpec)
+	}
+
+	if len(sp.deletedFiles) > 0 {
+		specs := sp.txn.meta.specs
+		for _, df := range sp.deletedFiles {
+			ssc.removeFile(df, currentSchema, specs[df.SpecID()])
+		}
+	}
+
+	var previousSnapshot *Snapshot
+	if sp.parentSnapshotID > 0 {
+		previousSnapshot, _ = sp.txn.meta.SnapshotByID(sp.parentSnapshotID)
+	}
+
+	var previousSummary iceberg.Properties
+	if previousSnapshot != nil {
+		previousSummary = previousSnapshot.Summary.Properties
+	}
+
+	summaryProps := ssc.build()
+	maps.Copy(summaryProps, props)
+
+	return updateSnapshotSummaries(Summary{
+		Operation:  sp.op,
+		Properties: summaryProps,
+	}, previousSummary, sp.op == OpOverwrite)
+}
+
+func (sp *snapshotProducer) commit() ([]Update, []Requirement, error) {
+	newManifests, err := sp.manifests()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	nextSequence := sp.txn.meta.nextSequenceNumber()
+	summary, err := sp.summary(sp.snapshotProps)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	fname := newManifestListFileName(sp.snapshotID, 0, sp.commitUuid)
+	locProvider, err := sp.txn.tbl.LocationProvider()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	manifestListFilePath := locProvider.NewMetadataLocation(fname)
+
+	var parentSnapshot *int64
+	if sp.parentSnapshotID > 0 {
+		parentSnapshot = &sp.parentSnapshotID
+	}
+
+	out, err := sp.io.Create(manifestListFilePath)
+	if err != nil {
+		return nil, nil, err
+	}
+	defer out.Close()
+
+	err = iceberg.WriteManifestList(sp.txn.meta.formatVersion, out,
+		sp.snapshotID, parentSnapshot, &nextSequence, newManifests)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	snapshot := Snapshot{
+		SnapshotID:       sp.snapshotID,
+		ParentSnapshotID: parentSnapshot,
+		SequenceNumber:   nextSequence,
+		ManifestList:     manifestListFilePath,
+		Summary:          &summary,
+		SchemaID:         &sp.txn.meta.currentSchemaID,
+	}
+
+	return []Update{
+			NewAddSnapshotUpdate(&snapshot),
+			NewSetSnapshotRefUpdate("main", sp.snapshotID, BranchRef, -1, -1, -1),
+		}, []Requirement{
+			AssertRefSnapshotID("main", sp.txn.meta.currentSnapshotID),
+		}, nil
+}
+
+func truncateUpperBoundText(s string, trunc int) string {
+	if trunc == utf8.RuneCountInString(s) {
+		return s
+	}
+
+	result := []rune(s)[:trunc]
+	for i := len(result) - 1; i >= 0; i-- {
+		next := result[i] + 1
+		if utf8.ValidRune(next) {
+			result[i] = next
+
+			return string(result)
+		}
+	}
+
+	return ""
+}
+
+func truncateUpperBoundBinary(val []byte, trunc int) []byte {
+	result := val[:trunc]
+	if bytes.Equal(result, val) {
+		return result
+	}
+
+	for i := len(result) - 1; i >= 0; i-- {
+		if result[i] < 255 {
+			result[i]++
+
+			return result
+		}
+	}
+
+	return nil
+}
diff --git a/table/table.go b/table/table.go
index b6b7786..2584b72 100644
--- a/table/table.go
+++ b/table/table.go
@@ -19,10 +19,13 @@ package table
 
 import (
 	"context"
+	"fmt"
+	"iter"
 	"runtime"
 	"slices"
 
 	"github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/internal"
 	"github.com/apache/iceberg-go/io"
 )
 
@@ -55,6 +58,7 @@ func (t Table) Schema() *iceberg.Schema              { return t.metadata.Current
 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) }
@@ -72,6 +76,64 @@ func (t Table) LocationProvider() (LocationProvider, error) {
 	return LoadLocationProvider(t.metadataLocation, t.metadata.Properties())
 }
 
+func (t Table) NewTransaction() *Transaction {
+	meta, _ := MetadataBuilderFromBase(t.metadata)
+
+	return &Transaction{
+		tbl:  &t,
+		meta: meta,
+		reqs: []Requirement{},
+	}
+}
+
+func (t Table) doCommit(ctx context.Context, updates []Update, reqs []Requirement) (*Table, error) {
+	newMeta, newLoc, err := t.cat.CommitTable(ctx, &t, reqs, updates)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := deleteOldMetadata(t.fs, t.metadata, newMeta); err != nil {
+		return nil, err
+	}
+
+	return New(t.identifier, newMeta, newLoc, t.fs, t.cat), nil
+}
+
+func getFiles(it iter.Seq[MetadataLogEntry]) iter.Seq[string] {
+	return func(yield func(string) bool) {
+		next, stop := iter.Pull(it)
+		defer stop()
+		for {
+			entry, ok := next()
+			if !ok {
+				return
+			}
+			if !yield(entry.MetadataFile) {
+				return
+			}
+		}
+	}
+}
+
+func deleteOldMetadata(fs io.IO, baseMeta, newMeta Metadata) error {
+	deleteAfterCommit := newMeta.Properties().GetBool(MetadataDeleteAfterCommitEnabledKey,
+		MetadataDeleteAfterCommitEnabledDefault)
+
+	if deleteAfterCommit {
+		removedPrevious := slices.Collect(getFiles(baseMeta.PreviousFiles()))
+		currentMetadata := slices.Collect(getFiles(newMeta.PreviousFiles()))
+		toRemove := internal.Difference(removedPrevious, currentMetadata)
+
+		for _, file := range toRemove {
+			if err := fs.Remove(file); err != nil {
+				return fmt.Errorf("failed to delete old metadata file %s: %w", file, err)
+			}
+		}
+	}
+
+	return nil
+}
+
 type ScanOption func(*Scan)
 
 func noopOption(*Scan) {}
diff --git a/table/transaction.go b/table/transaction.go
new file mode 100644
index 0000000..15785b5
--- /dev/null
+++ b/table/transaction.go
@@ -0,0 +1,193 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"slices"
+	"sync"
+	"time"
+
+	"github.com/apache/arrow-go/v18/arrow/array"
+	"github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/io"
+)
+
+type snapshotUpdate struct {
+	txn           *Transaction
+	io            io.WriteFileIO
+	snapshotProps iceberg.Properties
+}
+
+func (s snapshotUpdate) fastAppend() *snapshotProducer {
+	return newFastAppendFilesProducer(OpAppend, s.txn, s.io, nil, s.snapshotProps)
+}
+
+type Transaction struct {
+	tbl  *Table
+	meta *MetadataBuilder
+
+	reqs []Requirement
+
+	mx        sync.Mutex
+	committed bool
+}
+
+func (t *Transaction) apply(updates []Update, reqs []Requirement) error {
+	t.mx.Lock()
+	defer t.mx.Unlock()
+
+	if t.committed {
+		return errors.New("transaction has already been committed")
+	}
+
+	current, err := t.meta.Build()
+	if err != nil {
+		return err
+	}
+
+	for _, r := range reqs {
+		if err := r.Validate(current); err != nil {
+			return err
+		}
+	}
+
+	existing := map[string]struct{}{}
+	for _, r := range t.reqs {
+		existing[r.GetType()] = struct{}{}
+	}
+
+	for _, r := range reqs {
+		if _, ok := existing[r.GetType()]; !ok {
+			t.reqs = append(t.reqs, r)
+		}
+	}
+
+	prevUpdates, prevLastUpdated := len(t.meta.updates), t.meta.lastUpdatedMS
+	for _, u := range updates {
+		if err := u.Apply(t.meta); err != nil {
+			return err
+		}
+	}
+
+	// u.Apply will add updates to t.meta.updates if they are not no-ops
+	// and actually perform changes. So let's check if we actually had any
+	// changes added and thus need to update the lastupdated value.
+	if prevUpdates < len(t.meta.updates) {
+		if prevLastUpdated == t.meta.lastUpdatedMS {
+			t.meta.lastUpdatedMS = time.Now().UnixMilli()
+		}
+	}
+
+	return nil
+}
+
+func (t *Transaction) updateSnapshot(props iceberg.Properties) snapshotUpdate {
+	return snapshotUpdate{
+		txn:           t,
+		io:            t.tbl.fs.(io.WriteFileIO),
+		snapshotProps: props,
+	}
+}
+
+func (t *Transaction) SetProperties(props iceberg.Properties) error {
+	if len(props) > 0 {
+		return t.apply([]Update{NewSetPropertiesUpdate(props)}, nil)
+	}
+
+	return nil
+}
+
+func (t *Transaction) Append(rdr array.RecordReader, snapshotProps iceberg.Properties) error {
+	return iceberg.ErrNotImplemented
+}
+
+func (t *Transaction) AddFiles(files []string, snapshotProps iceberg.Properties, checkDuplicates bool) error {
+	set := make(map[string]string)
+	for _, f := range files {
+		set[f] = f
+	}
+
+	if len(set) != len(files) {
+		return errors.New("file paths must be unique for AppendDataFiles")
+	}
+
+	if checkDuplicates {
+		return iceberg.ErrNotImplemented
+	}
+
+	if t.meta.NameMapping() == nil {
+		nameMapping := t.meta.CurrentSchema().NameMapping()
+		mappingJson, err := json.Marshal(nameMapping)
+		if err != nil {
+			return err
+		}
+		err = t.SetProperties(iceberg.Properties{DefaultNameMappingKey: string(mappingJson)})
+		if err != nil {
+			return err
+		}
+	}
+
+	updater := t.updateSnapshot(snapshotProps).fastAppend()
+
+	dataFiles := parquetFilesToDataFiles(t.tbl.fs, t.meta, slices.Values(files))
+	for df, err := range dataFiles {
+		if err != nil {
+			return err
+		}
+		updater.appendDataFile(df)
+	}
+
+	updates, reqs, err := updater.commit()
+	if err != nil {
+		return err
+	}
+
+	return t.apply(updates, reqs)
+}
+
+func (t *Transaction) StagedTable() (*Table, error) {
+	updatedMeta, err := t.meta.Build()
+	if err != nil {
+		return nil, err
+	}
+
+	return New(t.tbl.identifier, updatedMeta,
+		updatedMeta.Location(), t.tbl.fs, t.tbl.cat), nil
+}
+
+func (t *Transaction) Commit(ctx context.Context) (*Table, error) {
+	t.mx.Lock()
+	defer t.mx.Unlock()
+
+	if t.committed {
+		return nil, errors.New("transaction has already been committed")
+	}
+
+	t.committed = true
+
+	if len(t.meta.updates) > 0 {
+		t.reqs = append(t.reqs, AssertTableUUID(t.meta.uuid))
+
+		return t.tbl.doCommit(ctx, t.meta.updates, t.reqs)
+	}
+
+	return t.tbl, nil
+}
diff --git a/types.go b/types.go
index ea93007..e3fea7a 100644
--- a/types.go
+++ b/types.go
@@ -26,7 +26,7 @@ import (
 	"strings"
 	"time"
 
-	"github.com/apache/arrow-go/v18/arrow/decimal128"
+	"github.com/apache/arrow-go/v18/arrow/decimal"
 )
 
 var (
@@ -484,7 +484,7 @@ func (d DecimalType) Scale() int     { return d.scale }
 func (DecimalType) primitive()       {}
 
 type Decimal struct {
-	Val   decimal128.Num
+	Val   decimal.Decimal128
 	Scale int
 }
 


Test patch:
diff --git a/literals_test.go b/literals_test.go
index 96c6606..20e21e6 100644
--- a/literals_test.go
+++ b/literals_test.go
@@ -840,6 +840,11 @@ func TestUnmarshalBinary(t *testing.T) {
 			[]byte{0xff, 0xed, 0x29, 0x79},
 			iceberg.DecimalLiteral{Scale: 4, Val: decimal128.FromI64(-1234567)},
 		},
+		{
+			iceberg.DecimalTypeOf(8, 2),
+			[]byte{0x00, 0xbc, 0x61, 0x4e},
+			iceberg.DecimalLiteral{Scale: 2, Val: decimal128.FromI64(12345678)},
+		},
 	}
 
 	for _, tt := range tests {
diff --git a/table/arrow_utils_internal_test.go b/table/arrow_utils_internal_test.go
index 85730d1..5584d50 100644
--- a/table/arrow_utils_internal_test.go
+++ b/table/arrow_utils_internal_test.go
@@ -21,22 +21,27 @@ import (
 	"bytes"
 	"cmp"
 	"math"
+	"math/big"
 	"slices"
+	"strings"
 	"testing"
+	"time"
 
 	"github.com/apache/arrow-go/v18/arrow/array"
+	"github.com/apache/arrow-go/v18/arrow/decimal128"
 	"github.com/apache/arrow-go/v18/arrow/memory"
 	"github.com/apache/arrow-go/v18/parquet"
 	"github.com/apache/arrow-go/v18/parquet/file"
 	"github.com/apache/arrow-go/v18/parquet/metadata"
 	"github.com/apache/arrow-go/v18/parquet/pqarrow"
 	"github.com/apache/iceberg-go"
+	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 )
 
-func constructTestTable(t *testing.T, writeStats bool) (*metadata.FileMetaData, Metadata) {
+func constructTestTable(t *testing.T, writeStats []string) (*metadata.FileMetaData, Metadata) {
 	tableMeta, err := ParseMetadataString(`{
 		"format-version": 2,
         "location": "s3://bucket/test/location",
@@ -141,9 +146,19 @@ func constructTestTable(t *testing.T, writeStats bool) (*metadata.FileMetaData,
 	rec := bldr.NewRecord()
 	defer rec.Release()
 
+	var opts []parquet.WriterProperty
+	if len(writeStats) == 0 {
+		opts = append(opts, parquet.WithStats(true))
+	} else {
+		opts = append(opts, parquet.WithStats(false))
+		for _, stat := range writeStats {
+			opts = append(opts, parquet.WithStatsFor(stat, true))
+		}
+	}
+
 	var buf bytes.Buffer
 	wr, err := pqarrow.NewFileWriter(arrowSchema, &buf,
-		parquet.NewWriterProperties(parquet.WithStats(writeStats)),
+		parquet.NewWriterProperties(opts...),
 		pqarrow.DefaultWriterProps())
 	require.NoError(t, err)
 
@@ -161,8 +176,8 @@ type FileStatsMetricsSuite struct {
 	suite.Suite
 }
 
-func (suite *FileStatsMetricsSuite) getDataFile(meta iceberg.Properties) iceberg.DataFile {
-	fileMeta, tableMeta := constructTestTable(suite.T(), true)
+func (suite *FileStatsMetricsSuite) getDataFile(meta iceberg.Properties, writeStats []string) iceberg.DataFile {
+	fileMeta, tableMeta := constructTestTable(suite.T(), writeStats)
 	require.NotNil(suite.T(), tableMeta)
 	require.NotNil(suite.T(), fileMeta)
 
@@ -190,12 +205,12 @@ func (suite *FileStatsMetricsSuite) getDataFile(meta iceberg.Properties) iceberg
 }
 
 func (suite *FileStatsMetricsSuite) TestRecordCount() {
-	df := suite.getDataFile(nil)
+	df := suite.getDataFile(nil, nil)
 	suite.EqualValues(int64(4), df.Count())
 }
 
 func (suite *FileStatsMetricsSuite) TestValueCounts() {
-	df := suite.getDataFile(nil)
+	df := suite.getDataFile(nil, nil)
 	suite.Len(df.ValueCounts(), 7)
 	suite.EqualValues(4, df.ValueCounts()[1])
 	suite.EqualValues(4, df.ValueCounts()[2])
@@ -207,7 +222,7 @@ func (suite *FileStatsMetricsSuite) TestValueCounts() {
 }
 
 func (suite *FileStatsMetricsSuite) TestColumnSizes() {
-	df := suite.getDataFile(nil)
+	df := suite.getDataFile(nil, nil)
 	suite.Len(df.ColumnSizes(), 7)
 	suite.Greater(df.ColumnSizes()[1], int64(0))
 	suite.Greater(df.ColumnSizes()[2], int64(0))
@@ -217,13 +232,13 @@ func (suite *FileStatsMetricsSuite) TestColumnSizes() {
 }
 
 func (suite *FileStatsMetricsSuite) TestOffsets() {
-	df := suite.getDataFile(nil)
+	df := suite.getDataFile(nil, nil)
 	suite.Len(df.SplitOffsets(), 1)
 	suite.EqualValues(4, df.SplitOffsets()[0])
 }
 
 func (suite *FileStatsMetricsSuite) TestNullValueCounts() {
-	df := suite.getDataFile(nil)
+	df := suite.getDataFile(nil, nil)
 	suite.Len(df.NullValueCounts(), 7)
 	suite.EqualValues(1, df.NullValueCounts()[1])
 	suite.EqualValues(0, df.NullValueCounts()[2])
@@ -236,8 +251,23 @@ func (suite *FileStatsMetricsSuite) TestNullValueCounts() {
 	// pqarrow doesn't currently write the NaN counts
 }
 
+func (suite *FileStatsMetricsSuite) TestBounds() {
+	df := suite.getDataFile(nil, nil)
+	suite.Len(df.LowerBoundValues(), 2)
+	suite.Equal([]byte("aaaaaaaaaaaaaaaa"), df.LowerBoundValues()[1])
+	lb, err := iceberg.LiteralFromBytes(iceberg.PrimitiveTypes.Float32, df.LowerBoundValues()[2])
+	suite.Require().NoError(err)
+	suite.Equal(float32(1.69), lb.(iceberg.Float32Literal).Value())
+
+	suite.Len(df.UpperBoundValues(), 2)
+	suite.Equal([]byte("zzzzzzzzzzzzzzz{"), df.UpperBoundValues()[1])
+	ub, err := iceberg.LiteralFromBytes(iceberg.PrimitiveTypes.Float32, df.UpperBoundValues()[2])
+	suite.Require().NoError(err)
+	suite.Equal(float32(100), ub.(iceberg.Float32Literal).Value())
+}
+
 func (suite *FileStatsMetricsSuite) TestMetricsModeNone() {
-	df := suite.getDataFile(iceberg.Properties{"write.metadata.metrics.default": "none"})
+	df := suite.getDataFile(iceberg.Properties{"write.metadata.metrics.default": "none"}, nil)
 	suite.Len(df.ValueCounts(), 0)
 	suite.Len(df.ColumnSizes(), 0)
 	suite.Len(df.NullValueCounts(), 0)
@@ -247,7 +277,7 @@ func (suite *FileStatsMetricsSuite) TestMetricsModeNone() {
 }
 
 func (suite *FileStatsMetricsSuite) TestMetricsModeCounts() {
-	df := suite.getDataFile(iceberg.Properties{"write.metadata.metrics.default": "counts"})
+	df := suite.getDataFile(iceberg.Properties{"write.metadata.metrics.default": "counts"}, nil)
 	suite.Len(df.ValueCounts(), 7)
 	suite.Len(df.NullValueCounts(), 7)
 	suite.Len(df.NaNValueCounts(), 0)
@@ -256,31 +286,73 @@ func (suite *FileStatsMetricsSuite) TestMetricsModeCounts() {
 }
 
 func (suite *FileStatsMetricsSuite) TestMetricsModeFull() {
-	df := suite.getDataFile(iceberg.Properties{"write.metadata.metrics.default": "full"})
+	df := suite.getDataFile(iceberg.Properties{"write.metadata.metrics.default": "full"}, nil)
 	suite.Len(df.ValueCounts(), 7)
 	suite.Len(df.NullValueCounts(), 7)
 	suite.Len(df.NaNValueCounts(), 0)
-	// bound values not yet implemented
+	suite.Len(df.LowerBoundValues(), 2)
+	suite.Equal([]byte("aaaaaaaaaaaaaaaaaaaa"), df.LowerBoundValues()[1])
+	lb, err := iceberg.LiteralFromBytes(iceberg.PrimitiveTypes.Float32, df.LowerBoundValues()[2])
+	suite.Require().NoError(err)
+	suite.Equal(float32(1.69), lb.(iceberg.Float32Literal).Value())
+
+	suite.Len(df.UpperBoundValues(), 2)
+	suite.Equal([]byte("zzzzzzzzzzzzzzzzzzzz"), df.UpperBoundValues()[1])
+	ub, err := iceberg.LiteralFromBytes(iceberg.PrimitiveTypes.Float32, df.UpperBoundValues()[2])
+	suite.Require().NoError(err)
+	suite.Equal(float32(100), ub.(iceberg.Float32Literal).Value())
 }
 
 func (suite *FileStatsMetricsSuite) TestMetricsModeNonDefaultTrunc() {
-	df := suite.getDataFile(iceberg.Properties{"write.metadata.metrics.default": "truncate(2)"})
+	df := suite.getDataFile(iceberg.Properties{"write.metadata.metrics.default": "truncate(2)"}, nil)
 	suite.Len(df.ValueCounts(), 7)
 	suite.Len(df.NullValueCounts(), 7)
 	suite.Len(df.NaNValueCounts(), 0)
-	// bound values not yet implemented
+	suite.Len(df.LowerBoundValues(), 2)
+	suite.Equal([]byte("aa"), df.LowerBoundValues()[1])
+	lb, err := iceberg.LiteralFromBytes(iceberg.PrimitiveTypes.Float32, df.LowerBoundValues()[2])
+	suite.Require().NoError(err)
+	suite.Equal(float32(1.69), lb.(iceberg.Float32Literal).Value())
+
+	suite.Len(df.UpperBoundValues(), 2)
+	suite.Equal([]byte("z{"), df.UpperBoundValues()[1])
+	ub, err := iceberg.LiteralFromBytes(iceberg.PrimitiveTypes.Float32, df.UpperBoundValues()[2])
+	suite.Require().NoError(err)
+	suite.Equal(float32(100), ub.(iceberg.Float32Literal).Value())
 }
 
 func (suite *FileStatsMetricsSuite) TestColumnMetricsMode() {
 	df := suite.getDataFile(iceberg.Properties{
 		"write.metadata.metrics.default":        "truncate(2)",
 		"write.metadata.metrics.column.strings": "none",
-	})
+	}, nil)
 
 	suite.Len(df.ValueCounts(), 6)
 	suite.Len(df.NullValueCounts(), 6)
 	suite.Len(df.NaNValueCounts(), 0)
-	// bound values not yet implemented
+
+	suite.Len(df.LowerBoundValues(), 1)
+	lb, err := iceberg.LiteralFromBytes(iceberg.PrimitiveTypes.Float32, df.LowerBoundValues()[2])
+	suite.Require().NoError(err)
+	suite.Equal(float32(1.69), lb.(iceberg.Float32Literal).Value())
+
+	suite.Len(df.UpperBoundValues(), 1)
+	ub, err := iceberg.LiteralFromBytes(iceberg.PrimitiveTypes.Float32, df.UpperBoundValues()[2])
+	suite.Require().NoError(err)
+	suite.Equal(float32(100), ub.(iceberg.Float32Literal).Value())
+}
+
+func (suite *FileStatsMetricsSuite) TestReadMissingStats() {
+	df := suite.getDataFile(nil, []string{"strings"})
+
+	stringColIdx := 1
+	suite.Len(df.LowerBoundValues(), 1)
+	suite.Equal([]byte("aaaaaaaaaaaaaaaa"), df.LowerBoundValues()[stringColIdx])
+	suite.Len(df.UpperBoundValues(), 1)
+	suite.Equal([]byte("zzzzzzzzzzzzzzz{"), df.UpperBoundValues()[stringColIdx])
+
+	suite.Len(df.NullValueCounts(), 1)
+	suite.EqualValues(1, df.NullValueCounts()[stringColIdx])
 }
 
 func TestFileMetrics(t *testing.T) {
@@ -408,3 +480,194 @@ func TestStatsTypes(t *testing.T) {
 		iceberg.PrimitiveTypes.Int32,
 	}, actual)
 }
+
+func constructTestTablePrimitiveTypes(t *testing.T) (*metadata.FileMetaData, Metadata) {
+	tableMeta, err := ParseMetadataString(`{
+        "format-version": 2,
+        "location": "s3://bucket/test/location",
+        "last-column-id": 7,
+        "current-schema-id": 0,
+        "schemas": [
+            {
+                "type": "struct",
+                "schema-id": 0,
+                "fields": [
+                    {"id": 1, "name": "booleans", "required": false, "type": "boolean"},
+                    {"id": 2, "name": "ints", "required": false, "type": "int"},
+                    {"id": 3, "name": "longs", "required": false, "type": "long"},
+                    {"id": 4, "name": "floats", "required": false, "type": "float"},
+                    {"id": 5, "name": "doubles", "required": false, "type": "double"},
+                    {"id": 6, "name": "dates", "required": false, "type": "date"},
+                    {"id": 7, "name": "times", "required": false, "type": "time"},
+                    {"id": 8, "name": "timestamps", "required": false, "type": "timestamp"},
+                    {"id": 9, "name": "timestamptzs", "required": false, "type": "timestamptz"},
+                    {"id": 10, "name": "strings", "required": false, "type": "string"},
+                    {"id": 11, "name": "uuids", "required": false, "type": "uuid"},
+                    {"id": 12, "name": "binaries", "required": false, "type": "binary"},
+					{"id": 13, "name": "small_dec", "required": false, "type": "decimal(8, 2)"},
+					{"id": 14, "name": "med_dec", "required": false, "type": "decimal(16, 2)"},
+					{"id": 15, "name": "large_dec", "required": false, "type": "decimal(24, 2)"}
+                ]
+            }
+        ],
+		"last-updated-ms": -1,
+        "default-spec-id": 0,
+        "partition-specs": [{"spec-id": 0, "fields": []}],
+        "properties": {}
+	}`)
+	require.NoError(t, err)
+
+	arrowSchema, err := SchemaToArrowSchema(tableMeta.Schemas()[0], nil, true, false)
+	require.NoError(t, err)
+
+	rec, _, err := array.RecordFromJSON(memory.DefaultAllocator, arrowSchema, strings.NewReader(`[
+		{
+			"booleans": true, 
+			"ints": 23,
+			"longs": 54,
+			"floats": 454.1223,
+			"doubles": 8542.12,
+			"dates": "2022-01-02",
+			"times": "17:30:34",
+			"timestamps": "2022-01-02T17:30:34.399",
+			"timestamptzs": "2022-01-02T17:30:34.399",
+			"strings": "hello",
+			"uuids": "`+uuid.NewMD5(uuid.NameSpaceDNS, []byte("foo")).String()+`",
+			"binaries": "aGVsbG8=",
+			"small_dec": "123456.78",
+			"med_dec": "12345678901234.56",
+			"large_dec": "1234567890123456789012.34"
+		},
+		{
+			"booleans": false,
+			"ints": 89,
+			"longs": 2,
+			"floats": 24342.29,
+			"doubles": -43.9,
+			"dates": "2023-02-04",
+			"times": "13:21:04",
+			"timestamps": "2023-02-04T13:21:04.354",
+			"timestamptzs": "2023-02-04T13:21:04.354",
+			"strings": "world",
+			"uuids": "`+uuid.NewMD5(uuid.NameSpaceDNS, []byte("bar")).String()+`",
+			"binaries": "d29ybGQ=",
+			"small_dec": "876543.21",
+			"med_dec": "65432109876543.21",
+			"large_dec": "4321098765432109876543.21"
+		}
+	]`))
+	require.NoError(t, err)
+	defer rec.Release()
+
+	var buf bytes.Buffer
+	wr, err := pqarrow.NewFileWriter(arrowSchema, &buf,
+		parquet.NewWriterProperties(parquet.WithStats(true),
+			parquet.WithStoreDecimalAsInteger(true)),
+		pqarrow.DefaultWriterProps())
+	require.NoError(t, err)
+
+	require.NoError(t, wr.Write(rec))
+	require.NoError(t, wr.Close())
+
+	rdr, err := file.NewParquetReader(bytes.NewReader(buf.Bytes()))
+	require.NoError(t, err)
+	defer rdr.Close()
+
+	return rdr.MetaData(), tableMeta
+}
+
+func assertBounds[T iceberg.LiteralType](t *testing.T, bound []byte, typ iceberg.Type, expected T) {
+	lit, err := iceberg.LiteralFromBytes(typ, bound)
+	require.NoError(t, err)
+	assert.Equal(t, expected, lit.(iceberg.TypedLiteral[T]).Value())
+}
+
+func toDate(tm time.Time) iceberg.Date {
+	return iceberg.Date(tm.Truncate(24*time.Hour).Unix() / int64((time.Hour * 24).Seconds()))
+}
+
+func TestMetricsPrimitiveTypes(t *testing.T) {
+	meta, tblMeta := constructTestTablePrimitiveTypes(t)
+	require.NotNil(t, tblMeta)
+	require.NotNil(t, meta)
+
+	tblMeta.Properties()["write.metadata.metrics.default"] = "truncate(2)"
+
+	collector, err := computeStatsPlan(tblMeta.CurrentSchema(), tblMeta.Properties())
+	require.NoError(t, err)
+	mapping, err := parquetPathToIDMapping(tblMeta.CurrentSchema())
+	require.NoError(t, err)
+
+	stats := dataFileStatsFromParquetMetadata(meta, collector, mapping)
+	df, err := stats.toDataFile(tblMeta.PartitionSpec(), "fake-path.parquet",
+		iceberg.ParquetFile, meta.GetSourceFileSize())
+	require.NoError(t, err)
+
+	assert.Len(t, df.ValueCounts(), 15)
+	assert.Len(t, df.NullValueCounts(), 15)
+	assert.Len(t, df.NaNValueCounts(), 0)
+
+	assert.Len(t, df.LowerBoundValues(), 15)
+	assertBounds(t, df.LowerBoundValues()[1], iceberg.PrimitiveTypes.Bool, false)
+	assertBounds(t, df.LowerBoundValues()[2], iceberg.PrimitiveTypes.Int32, int32(23))
+	assertBounds(t, df.LowerBoundValues()[3], iceberg.PrimitiveTypes.Int64, int64(2))
+	assertBounds(t, df.LowerBoundValues()[4], iceberg.PrimitiveTypes.Float32, float32(454.1223))
+	assertBounds(t, df.LowerBoundValues()[5], iceberg.PrimitiveTypes.Float64, -43.9)
+	assertBounds(t, df.LowerBoundValues()[6], iceberg.PrimitiveTypes.Date,
+		toDate(time.Date(2022, time.January, 2, 0, 0, 0, 0, time.UTC)))
+	assertBounds(t, df.LowerBoundValues()[7], iceberg.PrimitiveTypes.Time,
+		iceberg.Time(time.Duration(13*time.Hour+21*time.Minute+4*time.Second).Microseconds()))
+	assertBounds(t, df.LowerBoundValues()[8], iceberg.PrimitiveTypes.Timestamp,
+		iceberg.Timestamp(time.Date(2022, time.January, 2, 17, 30, 34, 399000000, time.UTC).UnixMicro()))
+	assertBounds(t, df.LowerBoundValues()[9], iceberg.PrimitiveTypes.TimestampTz,
+		iceberg.Timestamp(time.Date(2022, time.January, 2, 17, 30, 34, 399000000, time.UTC).UnixMicro()))
+	assertBounds(t, df.LowerBoundValues()[10], iceberg.PrimitiveTypes.String, "he")
+	assertBounds(t, df.LowerBoundValues()[11], iceberg.PrimitiveTypes.UUID,
+		uuid.NewMD5(uuid.NameSpaceDNS, []byte("foo")))
+	assertBounds(t, df.LowerBoundValues()[12], iceberg.PrimitiveTypes.Binary, []byte("he"))
+	assertBounds(t, df.LowerBoundValues()[13], iceberg.DecimalTypeOf(8, 2), iceberg.Decimal{
+		Val:   decimal128.FromI64(12345678),
+		Scale: 2,
+	})
+	assertBounds(t, df.LowerBoundValues()[14], iceberg.DecimalTypeOf(16, 2), iceberg.Decimal{
+		Val:   decimal128.FromI64(1234567890123456),
+		Scale: 2,
+	})
+	expected, _ := (&big.Int{}).SetString("123456789012345678901234", 10)
+	assertBounds(t, df.LowerBoundValues()[15], iceberg.DecimalTypeOf(24, 2), iceberg.Decimal{
+		Val:   decimal128.FromBigInt(expected),
+		Scale: 2,
+	})
+
+	assert.Len(t, df.UpperBoundValues(), 15)
+	assertBounds(t, df.UpperBoundValues()[1], iceberg.PrimitiveTypes.Bool, true)
+	assertBounds(t, df.UpperBoundValues()[2], iceberg.PrimitiveTypes.Int32, int32(89))
+	assertBounds(t, df.UpperBoundValues()[3], iceberg.PrimitiveTypes.Int64, int64(54))
+	assertBounds(t, df.UpperBoundValues()[4], iceberg.PrimitiveTypes.Float32, float32(24342.29))
+	assertBounds(t, df.UpperBoundValues()[5], iceberg.PrimitiveTypes.Float64, 8542.12)
+	assertBounds(t, df.UpperBoundValues()[6], iceberg.PrimitiveTypes.Date,
+		toDate(time.Date(2023, time.February, 4, 0, 0, 0, 0, time.UTC)))
+	assertBounds(t, df.UpperBoundValues()[7], iceberg.PrimitiveTypes.Time,
+		iceberg.Time(time.Duration(17*time.Hour+30*time.Minute+34*time.Second).Microseconds()))
+	assertBounds(t, df.UpperBoundValues()[8], iceberg.PrimitiveTypes.Timestamp,
+		iceberg.Timestamp(time.Date(2023, time.February, 4, 13, 21, 4, 354000000, time.UTC).UnixMicro()))
+	assertBounds(t, df.UpperBoundValues()[9], iceberg.PrimitiveTypes.TimestampTz,
+		iceberg.Timestamp(time.Date(2023, time.February, 4, 13, 21, 4, 354000000, time.UTC).UnixMicro()))
+	assertBounds(t, df.UpperBoundValues()[10], iceberg.PrimitiveTypes.String, "wp")
+	assertBounds(t, df.UpperBoundValues()[11], iceberg.PrimitiveTypes.UUID,
+		uuid.NewMD5(uuid.NameSpaceDNS, []byte("bar")))
+	assertBounds(t, df.UpperBoundValues()[12], iceberg.PrimitiveTypes.Binary, []byte("wp"))
+	assertBounds(t, df.UpperBoundValues()[13], iceberg.DecimalTypeOf(8, 2), iceberg.Decimal{
+		Val:   decimal128.FromI64(87654321),
+		Scale: 2,
+	})
+	assertBounds(t, df.UpperBoundValues()[14], iceberg.DecimalTypeOf(16, 2), iceberg.Decimal{
+		Val:   decimal128.FromI64(6543210987654321),
+		Scale: 2,
+	})
+	expectedUpper, _ := (&big.Int{}).SetString("432109876543210987654321", 10)
+	assertBounds(t, df.UpperBoundValues()[15], iceberg.DecimalTypeOf(24, 2), iceberg.Decimal{
+		Val:   decimal128.FromBigInt(expectedUpper),
+		Scale: 2,
+	})
+}
diff --git a/table/snapshots_internal_test.go b/table/snapshots_internal_test.go
index c374178..5c8afe0 100644
--- a/table/snapshots_internal_test.go
+++ b/table/snapshots_internal_test.go
@@ -194,3 +194,14 @@ func TestInvalidOperation(t *testing.T) {
 	_, err := updateSnapshotSummaries(Summary{Operation: OpReplace}, nil, false)
 	assert.ErrorIs(t, err, iceberg.ErrNotImplemented)
 }
+
+func TestTruncateUpperBoundString(t *testing.T) {
+	assert.Equal(t, "ab", truncateUpperBoundText("aaaa", 2))
+	// \u10FFFF\u10FFFF\x00
+	assert.Equal(t, "", truncateUpperBoundText("\xf4\x8f\xbf\xbf\xf4\x8f\xbf\xbf\x00", 2))
+}
+
+func TestTruncateUpperBoundBinary(t *testing.T) {
+	assert.Equal(t, []byte{0x01, 0x03}, truncateUpperBoundBinary([]byte{0x01, 0x02, 0x03}, 2))
+	assert.Nil(t, truncateUpperBoundBinary([]byte{0xff, 0xff, 0x00}, 2))
+}
diff --git a/table/table_test.go b/table/table_test.go
index f139ae2..8ae66d7 100644
--- a/table/table_test.go
+++ b/table/table_test.go
@@ -19,11 +19,21 @@ package table_test
 
 import (
 	"bytes"
+	"context"
+	"fmt"
+	"strconv"
 	"testing"
 
+	"github.com/apache/arrow-go/v18/arrow"
+	"github.com/apache/arrow-go/v18/arrow/array"
+	"github.com/apache/arrow-go/v18/arrow/memory"
+	"github.com/apache/arrow-go/v18/parquet/pqarrow"
 	"github.com/apache/iceberg-go"
 	"github.com/apache/iceberg-go/internal"
+	iceio "github.com/apache/iceberg-go/io"
 	"github.com/apache/iceberg-go/table"
+	"github.com/google/uuid"
+	"github.com/pterm/pterm"
 	"github.com/stretchr/testify/suite"
 )
 
@@ -128,3 +138,236 @@ func (t *TableTestSuite) TestSnapshotByName() {
 
 	t.True(testSnapshot.Equals(*t.tbl.SnapshotByName("test")))
 }
+
+type TableWritingTestSuite struct {
+	suite.Suite
+
+	tableSchema      *iceberg.Schema
+	arrSchema        *arrow.Schema
+	arrTbl           arrow.Table
+	arrSchemaWithIDs *arrow.Schema
+	arrTblWithIDs    arrow.Table
+	arrSchemaUpdated *arrow.Schema
+	arrTblUpdated    arrow.Table
+
+	location      string
+	formatVersion int
+}
+
+func (t *TableWritingTestSuite) SetupSuite() {
+	mem := memory.DefaultAllocator
+
+	t.tableSchema = iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Name: "foo", Type: iceberg.PrimitiveTypes.Bool},
+		iceberg.NestedField{ID: 2, Name: "bar", Type: iceberg.PrimitiveTypes.String},
+		iceberg.NestedField{ID: 4, Name: "baz", Type: iceberg.PrimitiveTypes.Int32},
+		iceberg.NestedField{ID: 10, Name: "qux", Type: iceberg.PrimitiveTypes.Date})
+
+	t.arrSchema = arrow.NewSchema([]arrow.Field{
+		{Name: "foo", Type: arrow.FixedWidthTypes.Boolean, Nullable: true},
+		{Name: "bar", Type: arrow.BinaryTypes.String, Nullable: true},
+		{Name: "baz", Type: arrow.PrimitiveTypes.Int32, Nullable: true},
+		{Name: "qux", Type: arrow.PrimitiveTypes.Date32, Nullable: true},
+	}, nil)
+
+	var err error
+	t.arrTbl, err = array.TableFromJSON(mem, t.arrSchema, []string{
+		`[{"foo": true, "bar": "bar_string", "baz": 123, "qux": "2024-03-07"}]`,
+	})
+	t.Require().NoError(err)
+
+	t.arrSchemaWithIDs = arrow.NewSchema([]arrow.Field{
+		{
+			Name: "foo", Type: arrow.FixedWidthTypes.Boolean,
+			Metadata: arrow.MetadataFrom(map[string]string{"PARQUET:field_id": "1"}),
+		},
+		{
+			Name: "bar", Type: arrow.BinaryTypes.String,
+			Metadata: arrow.MetadataFrom(map[string]string{"PARQUET:field_id": "2"}),
+		},
+		{
+			Name: "baz", Type: arrow.PrimitiveTypes.Int32,
+			Metadata: arrow.MetadataFrom(map[string]string{"PARQUET:field_id": "3"}),
+		},
+		{
+			Name: "qux", Type: arrow.PrimitiveTypes.Date32,
+			Metadata: arrow.MetadataFrom(map[string]string{"PARQUET:field_id": "4"}),
+		},
+	}, nil)
+
+	t.arrTblWithIDs, err = array.TableFromJSON(mem, t.arrSchemaWithIDs, []string{
+		`[{"foo": true, "bar": "bar_string", "baz": 123, "qux": "2024-03-07"}]`,
+	})
+	t.Require().NoError(err)
+
+	t.arrSchemaUpdated = arrow.NewSchema([]arrow.Field{
+		{Name: "foo", Type: arrow.FixedWidthTypes.Boolean, Nullable: true},
+		{Name: "baz", Type: arrow.PrimitiveTypes.Int32, Nullable: true},
+		{Name: "qux", Type: arrow.PrimitiveTypes.Date32, Nullable: true},
+		{Name: "quux", Type: arrow.PrimitiveTypes.Int32, Nullable: true},
+	}, nil)
+
+	t.arrTblUpdated, err = array.TableFromJSON(mem, t.arrSchemaUpdated, []string{
+		`[{"foo": true, "baz": 123, "qux": "2024-03-07", "quux": 234}]`,
+	})
+	t.Require().NoError(err)
+}
+
+func (t *TableWritingTestSuite) SetupTest() {
+	t.location = t.T().TempDir()
+}
+
+func (t *TableWritingTestSuite) TearDownSuite() {
+	t.arrTbl.Release()
+	t.arrTblUpdated.Release()
+	t.arrTblWithIDs.Release()
+}
+
+func (t *TableWritingTestSuite) getMetadataLoc() string {
+	return fmt.Sprintf("%s/metadata/%05d-%s.metadata.json",
+		t.location, 1, uuid.New().String())
+}
+
+func (t *TableWritingTestSuite) writeParquet(fio iceio.WriteFileIO, filePath string, arrTbl arrow.Table) {
+	fo, err := fio.Create(filePath)
+	t.Require().NoError(err)
+
+	t.Require().NoError(pqarrow.WriteTable(arrTbl, fo, arrTbl.NumRows(),
+		nil, pqarrow.DefaultWriterProps()))
+}
+
+func (t *TableWritingTestSuite) createTable(identifier table.Identifier, formatVersion int, spec iceberg.PartitionSpec, sc *iceberg.Schema) *table.Table {
+	meta, err := table.NewMetadata(sc, &spec, table.UnsortedSortOrder,
+		t.getMetadataLoc(), iceberg.Properties{"format-version": strconv.Itoa(formatVersion)})
+	t.Require().NoError(err)
+
+	return table.New(identifier, meta, t.location, iceio.LocalFS{}, nil)
+}
+
+func (t *TableWritingTestSuite) TestAddFilesUnpartitioned() {
+	ident := table.Identifier{"default", "unpartitioned_table_v" + strconv.Itoa(t.formatVersion)}
+	tbl := t.createTable(ident, t.formatVersion,
+		*iceberg.UnpartitionedSpec, t.tableSchema)
+
+	t.NotNil(tbl)
+
+	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)
+		files = append(files, filePath)
+	}
+
+	tx := tbl.NewTransaction()
+	t.Require().NoError(tx.AddFiles(files, nil, false))
+
+	stagedTbl, err := tx.StagedTable()
+	t.Require().NoError(err)
+	t.NotNil(stagedTbl.NameMapping())
+
+	t.Equal(stagedTbl.CurrentSnapshot().Summary,
+		&table.Summary{
+			Operation: table.OpAppend,
+			Properties: iceberg.Properties{
+				"added-data-files":       "5",
+				"added-files-size":       "3660",
+				"added-records":          "5",
+				"total-data-files":       "5",
+				"total-delete-files":     "0",
+				"total-equality-deletes": "0",
+				"total-files-size":       "3660",
+				"total-position-deletes": "0",
+				"total-records":          "5",
+			},
+		})
+
+	contents, err := stagedTbl.Scan().ToArrowTable(context.Background())
+	t.Require().NoError(err)
+	defer contents.Release()
+
+	t.EqualValues(5, contents.NumRows())
+}
+
+func (t *TableWritingTestSuite) TestAddFilesFileNotFound() {
+	ident := table.Identifier{"default", "unpartitioned_table_file_not_found_v" + strconv.Itoa(t.formatVersion)}
+	tbl := t.createTable(ident, t.formatVersion,
+		*iceberg.UnpartitionedSpec, t.tableSchema)
+
+	t.NotNil(tbl)
+
+	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)
+		files = append(files, filePath)
+	}
+
+	files = append(files, t.location+"/unpartitioned_file_not_found/unknown.parquet")
+	tx := tbl.NewTransaction()
+	err := tx.AddFiles(files, nil, false)
+	t.Error(err)
+	t.ErrorContains(err, "no such file or directory")
+}
+
+func (t *TableWritingTestSuite) TestAddFilesUnpartitionedHasFieldIDs() {
+	ident := table.Identifier{"default", "unpartitioned_table_with_ids_v" + strconv.Itoa(t.formatVersion)}
+	tbl := t.createTable(ident, t.formatVersion,
+		*iceberg.UnpartitionedSpec, t.tableSchema)
+
+	t.NotNil(tbl)
+
+	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)
+		files = append(files, filePath)
+	}
+
+	tx := tbl.NewTransaction()
+	err := tx.AddFiles(files, nil, false)
+	t.Error(err)
+	t.ErrorIs(err, iceberg.ErrNotImplemented)
+}
+
+func (t *TableWritingTestSuite) TestAddFilesFailsSchemaMismatch() {
+	ident := table.Identifier{"default", "unpartitioned_table_schema_mismatch_v" + strconv.Itoa(t.formatVersion)}
+	tbl := t.createTable(ident, t.formatVersion,
+		*iceberg.UnpartitionedSpec, t.tableSchema)
+	t.Require().NotNil(tbl)
+
+	wrongSchema := arrow.NewSchema([]arrow.Field{
+		{Name: "foo", Type: arrow.FixedWidthTypes.Boolean, Nullable: true},
+		{Name: "bar", Type: arrow.BinaryTypes.String, Nullable: true},
+		{Name: "baz", Type: arrow.BinaryTypes.String, Nullable: true}, // should be int32
+		{Name: "qux", Type: arrow.PrimitiveTypes.Date32, Nullable: true},
+	}, nil)
+
+	filePath := fmt.Sprintf("%s/unpartitioned_schema_mismatch_v%d/test.parquet", t.location, t.formatVersion)
+	mismatchTable, err := array.TableFromJSON(memory.DefaultAllocator, wrongSchema, []string{
+		`[{"foo": true, "bar": "bar_string", "baz": "123", "qux": "2024-03-07"},
+		  {"foo": false, "bar": "bar_string", "baz": "456", "qux": "2024-03-07"}]`,
+	})
+	t.Require().NoError(err)
+	t.writeParquet(tbl.FS().(iceio.WriteFileIO), filePath, mismatchTable)
+
+	files := []string{filePath}
+
+	pterm.DisableOutput() // disable error to console
+	defer pterm.EnableOutput()
+
+	tx := tbl.NewTransaction()
+	err = tx.AddFiles(files, nil, false)
+	t.Error(err)
+	t.EqualError(err, `error encountered during parquet file conversion: error encountered during schema visitor: mismatch in fields:
+   | Table Field              | Requested Field         
+✅ | 1: foo: optional boolean | 1: foo: optional boolean
+✅ | 2: bar: optional string  | 2: bar: optional string 
+❌ | 3: baz: optional int     | 3: baz: optional string 
+✅ | 4: qux: optional date    | 4: qux: optional date   
+`)
+}
+
+func TestTableWriting(t *testing.T) {
+	suite.Run(t, &TableWritingTestSuite{formatVersion: 1})
+	suite.Run(t, &TableWritingTestSuite{formatVersion: 2})
+}