Repository: apache/iceberg-go
Pull request #310 (2025-02-19 17:11:07)

ci(catalog/rest): initial framework for rest catalog integration tests
Brought up by #291 we need to add some simple integration tests for the REST catalog to ensure that our requests and handling are correct according to the spec and not just according to our mocked request/responses.

This creates an initial framework for adding REST integration tests (more need to be added) which already discovered an issue in the delete requests that is also fixed by this PR.


Patch:
diff --git a/.github/workflows/go-integration.yml b/.github/workflows/go-integration.yml
index 0060737..86daf89 100644
--- a/.github/workflows/go-integration.yml
+++ b/.github/workflows/go-integration.yml
@@ -66,6 +66,7 @@ jobs:
         run: |          
           go test -tags integration -v -run="^TestScanner" ./table
           go test -tags integration -v ./io
+          go test -tags integration -v -run="^TestRestIntegration$" ./catalog/rest
 
       - name: Show debug logs
         if: ${{ failure() }}
diff --git a/README.md b/README.md
index 90a3b20..4a5fe13 100644
--- a/README.md
+++ b/README.md
@@ -72,7 +72,7 @@ $ cd iceberg-go/cmd/iceberg && go build .
 | Alter Table              |      |      |          |      |  X  |
 | Set Table Properties     |  X   |      |          |      |  X  |
 | Create Namespace         |  X   |      |          |  X   |  X  |
-| Check Namespace Exists   |  X   |      |          |      |  X  |
+| Check Namespace Exists   |  X   |      |          |  X   |  X  |
 | Drop Namespace           |  X   |      |          |  X   |  X  |
 | Set Namespace Properties |  X   |      |          |  X   |  X  |
 | List View                |  X   |      |          |      |     |
diff --git a/catalog/catalog.go b/catalog/catalog.go
index 6e9a98b..bc04a6d 100644
--- a/catalog/catalog.go
+++ b/catalog/catalog.go
@@ -31,6 +31,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"iter"
 	"maps"
 	"strings"
 
@@ -79,7 +80,7 @@ type Catalog interface {
 	CommitTable(context.Context, *table.Table, []table.Requirement, []table.Update) (table.Metadata, string, error)
 	// ListTables returns a list of table identifiers in the catalog, with the returned
 	// identifiers containing the information required to load the table via that catalog.
-	ListTables(ctx context.Context, namespace table.Identifier) ([]table.Identifier, error)
+	ListTables(ctx context.Context, namespace table.Identifier) iter.Seq2[table.Identifier, error]
 	// LoadTable loads a table from the catalog and returns a Table with the metadata.
 	LoadTable(ctx context.Context, identifier table.Identifier, props iceberg.Properties) (*table.Table, error)
 	// DropTable tells the catalog to drop the table entirely.
diff --git a/catalog/glue/glue.go b/catalog/glue/glue.go
index 41f61c8..5b9cb0e 100644
--- a/catalog/glue/glue.go
+++ b/catalog/glue/glue.go
@@ -21,6 +21,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"iter"
 	"strconv"
 	_ "unsafe"
 
@@ -155,33 +156,34 @@ func NewCatalog(opts ...Option) *Catalog {
 // ListTables returns a list of Iceberg tables in the given Glue database.
 //
 // The namespace should just contain the Glue database name.
-func (c *Catalog) ListTables(ctx context.Context, namespace table.Identifier) ([]table.Identifier, error) {
-	database, err := identifierToGlueDatabase(namespace)
-	if err != nil {
-		return nil, err
-	}
-
-	params := &glue.GetTablesInput{CatalogId: c.catalogId, DatabaseName: aws.String(database)}
-
-	var icebergTables []table.Identifier
-
-	for {
-		tblsRes, err := c.glueSvc.GetTables(ctx, params)
+func (c *Catalog) ListTables(ctx context.Context, namespace table.Identifier) iter.Seq2[table.Identifier, error] {
+	return func(yield func(table.Identifier, error) bool) {
+		database, err := identifierToGlueDatabase(namespace)
 		if err != nil {
-			return nil, fmt.Errorf("failed to list tables in namespace %s: %w", database, err)
+			yield(table.Identifier{}, err)
+			return
 		}
 
-		icebergTables = append(icebergTables,
-			filterTableListByType(database, tblsRes.TableList, glueTypeIceberg)...)
+		paginator := glue.NewGetTablesPaginator(c.glueSvc, &glue.GetTablesInput{
+			CatalogId:    c.catalogId,
+			DatabaseName: aws.String(database),
+		})
 
-		if tblsRes.NextToken == nil {
-			break
-		}
+		for paginator.HasMorePages() {
+			page, err := paginator.NextPage(ctx)
+			if err != nil {
+				yield(table.Identifier{}, fmt.Errorf("failed to list tables in namespace %s: %w", database, err))
+				return
+			}
 
-		params.NextToken = tblsRes.NextToken
+			icebergTables := filterTableListByType(database, page.TableList, glueTypeIceberg)
+			for _, tbl := range icebergTables {
+				if !yield(tbl, nil) {
+					return
+				}
+			}
+		}
 	}
-
-	return icebergTables, nil
 }
 
 // LoadTable loads a table from the catalog table details.
@@ -363,6 +365,19 @@ func (c *Catalog) CreateNamespace(ctx context.Context, namespace table.Identifie
 	return nil
 }
 
+func (c *Catalog) CheckNamespaceExists(ctx context.Context, namespace table.Identifier) (bool, error) {
+	databaseName, err := identifierToGlueDatabase(namespace)
+	if err != nil {
+		return false, err
+	}
+
+	_, err = c.getDatabase(ctx, databaseName)
+	if err != nil {
+		return false, err
+	}
+	return true, nil
+}
+
 // DropNamespace deletes an Iceberg namespace from the Glue catalog.
 func (c *Catalog) DropNamespace(ctx context.Context, namespace table.Identifier) error {
 	databaseName, err := identifierToGlueDatabase(namespace)
@@ -543,14 +558,12 @@ func DatabaseIdentifier(database string) table.Identifier {
 
 func filterTableListByType(database string, tableList []types.Table, tableType string) []table.Identifier {
 	var filtered []table.Identifier
-
 	for _, tbl := range tableList {
 		if tbl.Parameters[tableTypePropsKey] != tableType {
 			continue
 		}
 		filtered = append(filtered, TableIdentifier(database, aws.ToString(tbl.Name)))
 	}
-
 	return filtered
 }
 
diff --git a/catalog/rest/rest.go b/catalog/rest/rest.go
index d097c9a..55950d4 100644
--- a/catalog/rest/rest.go
+++ b/catalog/rest/rest.go
@@ -155,7 +155,7 @@ type createTableRequest struct {
 	Location      string                 `json:"location,omitempty"`
 	PartitionSpec *iceberg.PartitionSpec `json:"partition-spec,omitempty"`
 	WriteOrder    *table.SortOrder       `json:"write-order,omitempty"`
-	StageCreate   bool                   `json:"stage-create,omitempty"`
+	StageCreate   bool                   `json:"stage-create"`
 	Props         iceberg.Properties     `json:"properties,omitempty"`
 }
 
@@ -268,7 +268,7 @@ func do[T any](ctx context.Context, method string, baseURI *url.URL, path []stri
 		return ret, handleNon200(rsp, override)
 	}
 
-	if method == http.MethodHead {
+	if method == http.MethodHead || method == http.MethodDelete {
 		return
 	}
 
@@ -649,27 +649,60 @@ func (r *Catalog) tableFromResponse(ctx context.Context, identifier []string, me
 	return table.New(id, metadata, loc, iofs), nil
 }
 
-func (r *Catalog) ListTables(ctx context.Context, namespace table.Identifier) ([]table.Identifier, error) {
-	if err := checkValidNamespace(namespace); err != nil {
-		return nil, err
+func (r *Catalog) ListTables(ctx context.Context, namespace table.Identifier) iter.Seq2[table.Identifier, error] {
+	return func(yield func(table.Identifier, error) bool) {
+		pageSize := r.getPageSize(ctx)
+		var pageToken string
+
+		for {
+			tables, nextPageToken, err := r.listTablesPage(ctx, namespace, pageToken, pageSize)
+			if err != nil {
+				yield(table.Identifier{}, err)
+				return
+			}
+			for _, tbl := range tables {
+				if !yield(tbl, nil) {
+					return
+				}
+			}
+			if nextPageToken == "" {
+				return
+			}
+			pageToken = nextPageToken
+		}
 	}
+}
 
+func (r *Catalog) listTablesPage(ctx context.Context, namespace table.Identifier, pageToken string, pageSize int) ([]table.Identifier, string, error) {
+	if err := checkValidNamespace(namespace); err != nil {
+		return nil, "", err
+	}
 	ns := strings.Join(namespace, namespaceSeparator)
-	path := []string{"namespaces", ns, "tables"}
+	uri := r.baseURI.JoinPath("namespaces", ns, "tables")
+
+	v := url.Values{}
+	if pageSize >= 0 {
+		v.Set("page-size", strconv.Itoa(pageSize))
+	}
+	if pageToken != "" {
+		v.Set("page-token", pageToken)
+	}
 
+	uri.RawQuery = v.Encode()
 	type resp struct {
-		Identifiers []identifier `json:"identifiers"`
+		Identifiers   []identifier `json:"identifiers"`
+		NextPageToken string       `json:"next-page-token,omitempty"`
 	}
-	rsp, err := doGet[resp](ctx, r.baseURI, path, r.cl, map[int]error{http.StatusNotFound: catalog.ErrNoSuchNamespace})
+	rsp, err := doGet[resp](ctx, uri, []string{}, r.cl, map[int]error{http.StatusNotFound: catalog.ErrNoSuchNamespace})
 	if err != nil {
-		return nil, err
+		return nil, "", err
 	}
-
 	out := make([]table.Identifier, len(rsp.Identifiers))
 	for i, id := range rsp.Identifiers {
 		out[i] = append(id.Namespace, id.Name)
 	}
-	return out, nil
+	return out, rsp.NextPageToken, nil
+
 }
 
 func splitIdentForPath(ident table.Identifier) (string, string, error) {
@@ -866,19 +899,19 @@ func (r *Catalog) PurgeTable(ctx context.Context, identifier table.Identifier) e
 
 func (r *Catalog) RenameTable(ctx context.Context, from, to table.Identifier) (*table.Table, error) {
 	type payload struct {
-		From identifier `json:"from"`
-		To   identifier `json:"to"`
+		Source      identifier `json:"source"`
+		Destination identifier `json:"destination"`
 	}
-	f := identifier{
+	src := identifier{
 		Namespace: catalog.NamespaceFromIdent(from),
 		Name:      catalog.TableNameFromIdent(from),
 	}
-	t := identifier{
+	dst := identifier{
 		Namespace: catalog.NamespaceFromIdent(to),
 		Name:      catalog.TableNameFromIdent(to),
 	}
 
-	_, err := doPost[payload, any](ctx, r.baseURI, []string{"tables", "rename"}, payload{From: f, To: t}, r.cl,
+	_, err := doPost[payload, any](ctx, r.baseURI, []string{"tables", "rename"}, payload{Source: src, Destination: dst}, r.cl,
 		map[int]error{http.StatusNotFound: catalog.ErrNoSuchTable})
 	if err != nil {
 		return nil, err
diff --git a/catalog/sql/sql.go b/catalog/sql/sql.go
index 22edee1..657576b 100644
--- a/catalog/sql/sql.go
+++ b/catalog/sql/sql.go
@@ -22,6 +22,7 @@ import (
 	"database/sql"
 	"errors"
 	"fmt"
+	"iter"
 	"maps"
 	"path"
 	"slices"
@@ -524,9 +525,15 @@ func (c *Catalog) DropNamespace(ctx context.Context, namespace table.Identifier)
 		return fmt.Errorf("%w: %s", catalog.ErrNoSuchNamespace, nsToDelete)
 	}
 
-	tbls, err := c.ListTables(ctx, namespace)
-	if err != nil {
-		return err
+	tbls := make([]table.Identifier, 0)
+	iter := c.ListTables(ctx, namespace)
+
+	for tbl, err := range iter {
+		tbls = append(tbls, tbl)
+		if err != nil {
+			return err
+		}
+		break // there is already at least a table
 	}
 
 	if len(tbls) > 0 {
@@ -576,7 +583,23 @@ func (c *Catalog) LoadNamespaceProperties(ctx context.Context, namespace table.I
 	})
 }
 
-func (c *Catalog) ListTables(ctx context.Context, namespace table.Identifier) ([]table.Identifier, error) {
+func (c *Catalog) ListTables(ctx context.Context, namespace table.Identifier) iter.Seq2[table.Identifier, error] {
+	tables, err := c.listTablesAll(ctx, namespace)
+	if err != nil {
+		return func(yield func(table.Identifier, error) bool) {
+			yield(table.Identifier{}, err)
+		}
+	}
+	return func(yield func(table.Identifier, error) bool) {
+		for _, t := range tables {
+			if !yield(t, nil) {
+				return
+			}
+		}
+	}
+}
+
+func (c *Catalog) listTablesAll(ctx context.Context, namespace table.Identifier) ([]table.Identifier, error) {
 	if len(namespace) > 0 {
 		exists, err := c.namespaceExists(ctx, strings.Join(namespace, "."))
 		if err != nil {
diff --git a/cmd/iceberg/main.go b/cmd/iceberg/main.go
index da74601..31845ee 100644
--- a/cmd/iceberg/main.go
+++ b/cmd/iceberg/main.go
@@ -273,10 +273,12 @@ func list(ctx context.Context, output Output, cat catalog.Catalog, parent string
 	}
 
 	if len(ids) == 0 && parent != "" {
-		ids, err = cat.ListTables(ctx, prnt)
-		if err != nil {
-			output.Error(err)
-			os.Exit(1)
+		iter := cat.ListTables(ctx, prnt)
+		for id, err := range iter {
+			ids = append(ids, id)
+			if err != nil {
+				output.Error(err)
+			}
 		}
 	}
 	output.Identifiers(ids)
diff --git a/go.mod b/go.mod
index 526d57c..4ae6456 100644
--- a/go.mod
+++ b/go.mod
@@ -23,12 +23,12 @@ toolchain go1.23.6
 
 require (
 	github.com/apache/arrow-go/v18 v18.1.1-0.20250218215100-460f5004b92d
-	github.com/aws/aws-sdk-go-v2 v1.36.1
-	github.com/aws/aws-sdk-go-v2/config v1.29.6
-	github.com/aws/aws-sdk-go-v2/credentials v1.17.59
-	github.com/aws/aws-sdk-go-v2/service/glue v1.105.9
-	github.com/aws/aws-sdk-go-v2/service/s3 v1.77.0
-	github.com/aws/smithy-go v1.22.2
+	github.com/aws/aws-sdk-go-v2 v1.36.2
+	github.com/aws/aws-sdk-go-v2/config v1.29.7
+	github.com/aws/aws-sdk-go-v2/credentials v1.17.60
+	github.com/aws/aws-sdk-go-v2/service/glue v1.105.10
+	github.com/aws/aws-sdk-go-v2/service/s3 v1.77.1
+	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
 	github.com/hamba/avro/v2 v2.28.0
@@ -36,18 +36,18 @@ require (
 	github.com/stretchr/testify v1.10.0
 	github.com/substrait-io/substrait-go/v3 v3.7.0
 	github.com/twmb/murmur3 v1.1.8
-	github.com/uptrace/bun v1.2.9
-	github.com/uptrace/bun/dialect/mssqldialect v1.2.9
-	github.com/uptrace/bun/dialect/mysqldialect v1.2.9
-	github.com/uptrace/bun/dialect/oracledialect v1.2.9
-	github.com/uptrace/bun/dialect/pgdialect v1.2.9
-	github.com/uptrace/bun/dialect/sqlitedialect v1.2.9
-	github.com/uptrace/bun/driver/sqliteshim v1.2.9
-	github.com/uptrace/bun/extra/bundebug v1.2.9
+	github.com/uptrace/bun v1.2.10
+	github.com/uptrace/bun/dialect/mssqldialect v1.2.10
+	github.com/uptrace/bun/dialect/mysqldialect v1.2.10
+	github.com/uptrace/bun/dialect/oracledialect v1.2.10
+	github.com/uptrace/bun/dialect/pgdialect v1.2.10
+	github.com/uptrace/bun/dialect/sqlitedialect v1.2.10
+	github.com/uptrace/bun/driver/sqliteshim v1.2.10
+	github.com/uptrace/bun/extra/bundebug v1.2.10
 	gocloud.dev v0.40.0
-	golang.org/x/exp v0.0.0-20250106191152-7588d65b2ba8
+	golang.org/x/exp v0.0.0-20250215185904-eff6e970281f
 	golang.org/x/sync v0.11.0
-	google.golang.org/api v0.221.0
+	google.golang.org/api v0.222.0
 	gopkg.in/yaml.v3 v3.0.1
 )
 
@@ -66,20 +66,20 @@ require (
 	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-v2/aws/protocol/eventstream v1.6.9 // indirect
-	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.28 // 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.29 // indirect
 	github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.17.11 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.32 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.32 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.2 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.32 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.2 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.6.0 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.13 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.18.13 // indirect
-	github.com/aws/aws-sdk-go-v2/service/sso v1.24.15 // indirect
-	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.28.14 // indirect
-	github.com/aws/aws-sdk-go-v2/service/sts v1.33.14 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.33 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.33 // 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.33 // 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.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.14 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.18.14 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.24.16 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.28.15 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sts v1.33.15 // indirect
 	github.com/cockroachdb/apd/v3 v3.2.1 // indirect
 	github.com/containerd/console v1.0.3 // indirect
 	github.com/creasty/defaults v1.8.0 // indirect
@@ -118,7 +118,7 @@ require (
 	github.com/ncruces/go-strftime v0.1.9 // indirect
 	github.com/pierrec/lz4/v4 v4.1.22 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	github.com/puzpuzpuz/xsync/v3 v3.5.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
@@ -147,11 +147,11 @@ require (
 	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-20250207221924-e9438ea467c6 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20250212204824-5a70512c5d8b // indirect
 	google.golang.org/grpc v1.70.0 // indirect
 	google.golang.org/protobuf v1.36.5 // indirect
-	modernc.org/libc v1.61.9 // 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.34.5 // indirect
+	modernc.org/sqlite v1.35.0 // indirect
 )
diff --git a/go.sum b/go.sum
index 08f8ad3..8ec5666 100644
--- a/go.sum
+++ b/go.sum
@@ -44,46 +44,46 @@ github.com/apache/thrift v0.21.0/go.mod h1:W1H8aR/QRtYNvrPeFXBtobyRkd0/YVhTc6i07
 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-v2 v1.36.1 h1:iTDl5U6oAhkNPba0e1t1hrwAo02ZMqbrGq4k5JBWM5E=
-github.com/aws/aws-sdk-go-v2 v1.36.1/go.mod h1:5PMILGVKiW32oDzjj6RU52yrNrDPUHcbZQYr1sM7qmM=
-github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.9 h1:VZPDrbzdsU1ZxhyWrvROqLY0nxFWgMCAzhn/nYz3X48=
-github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.9/go.mod h1:3XkePX5dSaxveLAYY7nsbsZZrKxCyEuE5pM4ziFxyGg=
-github.com/aws/aws-sdk-go-v2/config v1.29.6 h1:fqgqEKK5HaZVWLQoLiC9Q+xDlSp+1LYidp6ybGE2OGg=
-github.com/aws/aws-sdk-go-v2/config v1.29.6/go.mod h1:Ft+WLODzDQmCTHDvqAH1JfC2xxbZ0MxpZAcJqmE1LTQ=
-github.com/aws/aws-sdk-go-v2/credentials v1.17.59 h1:9btwmrt//Q6JcSdgJOLI98sdr5p7tssS9yAsGe8aKP4=
-github.com/aws/aws-sdk-go-v2/credentials v1.17.59/go.mod h1:NM8fM6ovI3zak23UISdWidyZuI1ghNe2xjzUZAyT+08=
-github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.28 h1:KwsodFKVQTlI5EyhRSugALzsV6mG/SGrdjlMXSZSdso=
-github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.28/go.mod h1:EY3APf9MzygVhKuPXAc5H+MkGb8k/DOSQjWS0LgkKqI=
+github.com/aws/aws-sdk-go-v2 v1.36.2 h1:Ub6I4lq/71+tPb/atswvToaLGVMxKZvjYDVOWEExOcU=
+github.com/aws/aws-sdk-go-v2 v1.36.2/go.mod h1:LLXuLpgzEbD766Z5ECcRmi8AzSwfZItDtmABVkRLGzg=
+github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.10 h1:zAybnyUQXIZ5mok5Jqwlf58/TFE7uvd3IAsa1aF9cXs=
+github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.10/go.mod h1:qqvMj6gHLR/EXWZw4ZbqlPbQUyenf4h82UQUlKc+l14=
+github.com/aws/aws-sdk-go-v2/config v1.29.7 h1:71nqi6gUbAUiEQkypHQcNVSFJVUFANpSeUNShiwWX2M=
+github.com/aws/aws-sdk-go-v2/config v1.29.7/go.mod h1:yqJQ3nh2HWw/uxd56bicyvmDW4KSc+4wN6lL8pYjynU=
+github.com/aws/aws-sdk-go-v2/credentials v1.17.60 h1:1dq+ELaT5ogfmqtV1eocq8SpOK1NRsuUfmhQtD/XAh4=
+github.com/aws/aws-sdk-go-v2/credentials v1.17.60/go.mod h1:HDes+fn/xo9VeszXqjBVkxOo/aUy8Mc6QqKvZk32GlE=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.29 h1:JO8pydejFKmGcUNiiwt75dzLHRWthkwApIvPoyUtXEg=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.29/go.mod h1:adxZ9i9DRmB8zAT0pO0yGnsmu0geomp5a3uq5XpgOJ8=
 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/internal/configsources v1.3.32 h1:BjUcr3X3K0wZPGFg2bxOWW3VPN8rkE3/61zhP+IHviA=
-github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.32/go.mod h1:80+OGC/bgzzFFTUmcuwD0lb4YutwQeKLFpmt6hoWapU=
-github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.32 h1:m1GeXHVMJsRsUAqG6HjZWx9dj7F5TR+cF1bjyfYyBd4=
-github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.32/go.mod h1:IitoQxGfaKdVLNg0hD8/DXmAqNy0H4K2H2Sf91ti8sI=
-github.com/aws/aws-sdk-go-v2/internal/ini v1.8.2 h1:Pg9URiobXy85kgFev3og2CuOZ8JZUBENF+dcgWBaYNk=
-github.com/aws/aws-sdk-go-v2/internal/ini v1.8.2/go.mod h1:FbtygfRFze9usAadmnGJNc8KsP346kEe+y2/oyhGAGc=
-github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.32 h1:OIHj/nAhVzIXGzbAE+4XmZ8FPvro3THr6NlqErJc3wY=
-github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.32/go.mod h1:LiBEsDo34OJXqdDlRGsilhlIiXR7DL+6Cx2f4p1EgzI=
-github.com/aws/aws-sdk-go-v2/service/glue v1.105.9 h1:GzlbilXxPKYpGiPgy7T9CVj0OkllW7LA+aptBRYFFo8=
-github.com/aws/aws-sdk-go-v2/service/glue v1.105.9/go.mod h1:W6FO+0G0oJ30CUrwKBrT/UTCD1hm4tZNbiaGF/FGTZI=
-github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.2 h1:D4oz8/CzT9bAEYtVhSBmFj2dNOtaHOtMKc2vHBwYizA=
-github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.2/go.mod h1:Za3IHqTQ+yNcRHxu1OFucBh0ACZT4j4VQFF0BqpZcLY=
-github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.6.0 h1:kT2WeWcFySdYpPgyqJMSUE7781Qucjtn6wBvrgm9P+M=
-github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.6.0/go.mod h1:WYH1ABybY7JK9TITPnk6ZlP7gQB8psI4c9qDmMsnLSA=
-github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.13 h1:SYVGSFQHlchIcy6e7x12bsrxClCXSP5et8cqVhL8cuw=
-github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.13/go.mod h1:kizuDaLX37bG5WZaoxGPQR/LNFXpxp0vsUnqfkWXfNE=
-github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.18.13 h1:OBsrtam3rk8NfBEq7OLOMm5HtQ9Yyw32X4UQMya/wjw=
-github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.18.13/go.mod h1:3U4gFA5pmoCOja7aq4nSaIAGbaOHv2Yl2ug018cmC+Q=
-github.com/aws/aws-sdk-go-v2/service/s3 v1.77.0 h1:RCOi1rDmLqOICym/6UeS2cqKED4T4m966w2rl1HfL+g=
-github.com/aws/aws-sdk-go-v2/service/s3 v1.77.0/go.mod h1:VC4EKSHqT3nzOcU955VWHMGsQ+w67wfAUBSjC8NOo8U=
-github.com/aws/aws-sdk-go-v2/service/sso v1.24.15 h1:/eE3DogBjYlvlbhd2ssWyeuovWunHLxfgw3s/OJa4GQ=
-github.com/aws/aws-sdk-go-v2/service/sso v1.24.15/go.mod h1:2PCJYpi7EKeA5SkStAmZlF6fi0uUABuhtF8ILHjGc3Y=
-github.com/aws/aws-sdk-go-v2/service/ssooidc v1.28.14 h1:M/zwXiL2iXUrHputuXgmO94TVNmcenPHxgLXLutodKE=
-github.com/aws/aws-sdk-go-v2/service/ssooidc v1.28.14/go.mod h1:RVwIw3y/IqxC2YEXSIkAzRDdEU1iRabDPaYjpGCbCGQ=
-github.com/aws/aws-sdk-go-v2/service/sts v1.33.14 h1:TzeR06UCMUq+KA3bDkujxK1GVGy+G8qQN/QVYzGLkQE=
-github.com/aws/aws-sdk-go-v2/service/sts v1.33.14/go.mod h1:dspXf/oYWGWo6DEvj98wpaTeqt5+DMidZD0A9BYTizc=
-github.com/aws/smithy-go v1.22.2 h1:6D9hW43xKFrRx/tXXfAlIZc4JI+yQe6snnWcQyxSyLQ=
-github.com/aws/smithy-go v1.22.2/go.mod h1:irrKGvNn1InZwb2d7fkIRNucdfwR8R+Ts3wxYa/cJHg=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.33 h1:knLyPMw3r3JsU8MFHWctE4/e2qWbPaxDYLlohPvnY8c=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.33/go.mod h1:EBp2HQ3f+XCB+5J+IoEbGhoV7CpJbnrsd4asNXmTL0A=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.33 h1:K0+Ne08zqti8J9jwENxZ5NoUyBnaFDTu3apwQJWrwwA=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.33/go.mod h1:K97stwwzaWzmqxO8yLGHhClbVW1tC6VT1pDLk1pGrq4=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3 h1:bIqFDwgGXXN1Kpp99pDOdKMTTb5d2KyU5X/BZxjOkRo=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3/go.mod h1:H5O/EsxDWyU+LP/V8i5sm8cxoZgc2fdNR9bxlOFrQTo=
+github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.33 h1:/frG8aV09yhCVSOEC2pzktflJJO48NwY3xntHBwxHiA=
+github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.33/go.mod h1:8vwASlAcV366M+qxZnjNzCjeastk1Rt1bpSRaGZanGU=
+github.com/aws/aws-sdk-go-v2/service/glue v1.105.10 h1:ycWzmgAhHyFpa5EzFrYldClRa6cQvk4PVY6aE9SUlLk=
+github.com/aws/aws-sdk-go-v2/service/glue v1.105.10/go.mod h1:Vl8+9gVeL4/C8dinktlSYkDt32MVRggaCRNVE7c/6Iw=
+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.1 h1:7SuukGpyIgF5EiAbf1dZRxP+xSnY1WjiHBjL08fjJeE=
+github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.6.1/go.mod h1:k+Vce/8R28tSozjdWphkrNhK8zLmdS9RgiDNZl6p8Rw=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.14 h1:2scbY6//jy/s8+5vGrk7l1+UtHl0h9A4MjOO2k/TM2E=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.14/go.mod h1:bRpZPHZpSe5YRHmPfK3h1M7UBFCn2szHzyx0rw04zro=
+github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.18.14 h1:fgdkfsxTehqPcIQa24G/Omwv9RocTq2UcONNX/OnrZI=
+github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.18.14/go.mod h1:wMxQ3OE8fiM8z2YRAeb2J8DLTTWMvRyYYuQOs26AbTQ=
+github.com/aws/aws-sdk-go-v2/service/s3 v1.77.1 h1:5bI9tJL2Z0FGFtp/LPDv0eyliFBHCn7LAhqpQuL+7kk=
+github.com/aws/aws-sdk-go-v2/service/s3 v1.77.1/go.mod h1:njj3tSJONkfdLt4y6X8pyqeM6sJLNZxmzctKKV+n1GM=
+github.com/aws/aws-sdk-go-v2/service/sso v1.24.16 h1:YV6xIKDJp6U7YB2bxfud9IENO1LRpGhe2Tv/OKtPrOQ=
+github.com/aws/aws-sdk-go-v2/service/sso v1.24.16/go.mod h1:DvbmMKgtpA6OihFJK13gHMZOZrCHttz8wPHGKXqU+3o=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.28.15 h1:kMyK3aKotq1aTBsj1eS8ERJLjqYRRRcsmP33ozlCvlk=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.28.15/go.mod h1:5uPZU7vSNzb8Y0dm75xTikinegPYK3uJmIHQZFq5Aqo=
+github.com/aws/aws-sdk-go-v2/service/sts v1.33.15 h1:ht1jVmeeo2anR7zDiYJLSnRYnO/9NILXXu42FP3rJg0=
+github.com/aws/aws-sdk-go-v2/service/sts v1.33.15/go.mod h1:xWZ5cOiFe3czngChE4LhCBqUxNwgfwndEF7XlYP/yD8=
+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/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=
@@ -244,8 +244,8 @@ github.com/pterm/pterm v0.12.36/go.mod h1:NjiL09hFhT/vWjQHSj1athJpx6H8cjpHXNAK5b
 github.com/pterm/pterm v0.12.40/go.mod h1:ffwPLwlbXxP+rxT0GsgDTzS3y3rmpAO1NMjUkGTYf8s=
 github.com/pterm/pterm v0.12.80 h1:mM55B+GnKUnLMUSqhdINe4s6tOuVQIetQ3my8JGyAIg=
 github.com/pterm/pterm v0.12.80/go.mod h1:c6DeF9bSnOSeFPZlfs4ZRAFcf5SCoTwvwQ5xaKGQlHo=
-github.com/puzpuzpuz/xsync/v3 v3.5.0 h1:i+cMcpEDY1BkNm7lPDkCtE4oElsYLn+EKF8kAu2vXT4=
-github.com/puzpuzpuz/xsync/v3 v3.5.0/go.mod h1:VjzYrABPabuM4KyBh1Ftq6u8nhwY5tBPKP9jpmh0nnA=
+github.com/puzpuzpuz/xsync/v3 v3.5.1 h1:GJYJZwO6IdxN/IKbneznS6yPkVC+c3zyY/j19c++5Fg=
+github.com/puzpuzpuz/xsync/v3 v3.5.1/go.mod h1:VjzYrABPabuM4KyBh1Ftq6u8nhwY5tBPKP9jpmh0nnA=
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
 github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
@@ -277,22 +277,22 @@ github.com/tmthrgd/go-hex v0.0.0-20190904060850-447a3041c3bc h1:9lRDQMhESg+zvGYm
 github.com/tmthrgd/go-hex v0.0.0-20190904060850-447a3041c3bc/go.mod h1:bciPuU6GHm1iF1pBvUfxfsH0Wmnc2VbpgvbI9ZWuIRs=
 github.com/twmb/murmur3 v1.1.8 h1:8Yt9taO/WN3l08xErzjeschgZU2QSrwm1kclYq+0aRg=
 github.com/twmb/murmur3 v1.1.8/go.mod h1:Qq/R7NUyOfr65zD+6Q5IHKsJLwP7exErjN6lyyq3OSQ=
-github.com/uptrace/bun v1.2.9 h1:OOt2DlIcRUMSZPr6iXDFg/LaQd59kOxbAjpIVHddKRs=
-github.com/uptrace/bun v1.2.9/go.mod h1:r2ZaaGs9Ru5bpGTr8GQfp8jp+TlCav9grYCPOu2CJSg=
-github.com/uptrace/bun/dialect/mssqldialect v1.2.9 h1:UBEaY6bdbqMcVd5/iWKM4BWCab1QYYnNA8dnW9Dq+LI=
-github.com/uptrace/bun/dialect/mssqldialect v1.2.9/go.mod h1:jFQVfoQYugBD05LVj0i0WUZF1Bw9KMFIrHNH7S1M7oI=
-github.com/uptrace/bun/dialect/mysqldialect v1.2.9 h1:nn3xVxTLhZh9LScwVD81yJX38DQ1LlriH8/WWEdWpdk=
-github.com/uptrace/bun/dialect/mysqldialect v1.2.9/go.mod h1:Qe9QIiyUUxcxmhxskMNV++dEWhR6UQvLV3WYiQhWyKQ=
-github.com/uptrace/bun/dialect/oracledialect v1.2.9 h1:qOHbK/AVwIpwcPwnCWR4KD3E+vdKC7+iQs8V8baCnf0=
-github.com/uptrace/bun/dialect/oracledialect v1.2.9/go.mod h1:4T17uZPy5neikxhYoZOywrBoWCOyb0YPSVyiMDOfYBA=
-github.com/uptrace/bun/dialect/pgdialect v1.2.9 h1:caf5uFbOGiXvadV6pA5gn87k0awFFxL1kuuY3SpxnWk=
-github.com/uptrace/bun/dialect/pgdialect v1.2.9/go.mod h1:m7L9JtOp/Lt8HccET70ULxplMweE/u0S9lNUSxz2duo=
-github.com/uptrace/bun/dialect/sqlitedialect v1.2.9 h1:HLzGWXBh07sT8zhVPy6veYbbGrAtYq0KzyRHXBj+GjA=
-github.com/uptrace/bun/dialect/sqlitedialect v1.2.9/go.mod h1:dUR+ecoCWA0FIa9vhQVRnGtYYPpuCLJoEEtX9E1aiBU=
-github.com/uptrace/bun/driver/sqliteshim v1.2.9 h1:vVBGtXFUNTlaijRQS3PTIH2QoNBr63orDd2aF6j3rBk=
-github.com/uptrace/bun/driver/sqliteshim v1.2.9/go.mod h1:BXAh3Tv9qiGrzV6WAl4fm1WoGya7doYTpsKBJzNk2do=
-github.com/uptrace/bun/extra/bundebug v1.2.9 h1:3SU66p+q76XhfeUUzl9XooVu7hVNueZ/2Q3J8S1uzCU=
-github.com/uptrace/bun/extra/bundebug v1.2.9/go.mod h1:/rp83jYAtwZUQIz+L3KwvREXaSd5GQGPJUusqq+Qtis=
+github.com/uptrace/bun v1.2.10 h1:6TlxUQhGxiiv7MHjzxbV6ZNt/Im0PIQ3S45riAmbnGA=
+github.com/uptrace/bun v1.2.10/go.mod h1:ww5G8h59UrOnCHmZ8O1I/4Djc7M/Z3E+EWFS2KLB6dQ=
+github.com/uptrace/bun/dialect/mssqldialect v1.2.10 h1:6nNUpA7lxWNEG77Pp109uQuHbMYdNreXLulDR+Yq87I=
+github.com/uptrace/bun/dialect/mssqldialect v1.2.10/go.mod h1:fDs/IHcnOLeCGUV3+bDl7l3eYW1yogrxB63PvwxuWIU=
+github.com/uptrace/bun/dialect/mysqldialect v1.2.10 h1:5H6r7nQu9Cs55j8Os391Co6TXt+DGFyS7tj8yVLl6XY=
+github.com/uptrace/bun/dialect/mysqldialect v1.2.10/go.mod h1:bTFIm0aeLP5IoxjdU/ZpMeQZhOMgMsRNW3BG+dLq0AY=
+github.com/uptrace/bun/dialect/oracledialect v1.2.10 h1:LTX9x672qI7Y1Xj5IAfl73Euv09k4UZWavLQsUr51FI=
+github.com/uptrace/bun/dialect/oracledialect v1.2.10/go.mod h1:CJHWypzs2Oe1uxYQiP+23PwV22NtOriXtju3yf4g6oU=
+github.com/uptrace/bun/dialect/pgdialect v1.2.10 h1:+PAGCVyWDoAjMuAgn0+ud7fu3It8+Xvk7HQAJ5wCXMQ=
+github.com/uptrace/bun/dialect/pgdialect v1.2.10/go.mod h1:hv0zsoc3PeW5fl3JeBglZT1vl2FoERY+QwvuvKsKATA=
+github.com/uptrace/bun/dialect/sqlitedialect v1.2.10 h1:/74GDx1hnRrrmIvqpNbbFwD28sW1z+i/QjQSVy6XnnY=
+github.com/uptrace/bun/dialect/sqlitedialect v1.2.10/go.mod h1:xBx+N2q4G4s51tAxZU5vKB3Zu0bFl1uRmKqZwCPBilg=
+github.com/uptrace/bun/driver/sqliteshim v1.2.10 h1:D9CDMuvhbAEgWOTNDgHcXDMMBdO2bxZPB1I7BwmE7VI=
+github.com/uptrace/bun/driver/sqliteshim v1.2.10/go.mod h1:kVsO2rGHv/Ee4XBDNXWlR/kJ3LEy4Imd2wzF4gGq2R8=
+github.com/uptrace/bun/extra/bundebug v1.2.10 h1:9Ot6fJ1vemrc0qBYp0roJCogTl9den1PAFcYygBiKoc=
+github.com/uptrace/bun/extra/bundebug v1.2.10/go.mod h1:xnuXkwPrC0gNalR2bde8PobgjwXGCo4D9nZoV/2ghzQ=
 github.com/vmihailenco/msgpack/v5 v5.4.1 h1:cQriyiUvjTwOHg8QZaPihLWeRAAVoCpE00IUPn0Bjt8=
 github.com/vmihailenco/msgpack/v5 v5.4.1/go.mod h1:GaZTsDaehaPpQVyxrf5mtQlH+pc21PIudVV/E3rRQok=
 github.com/vmihailenco/tagparser/v2 v2.0.0 h1:y09buUbR+b5aycVFQs/g70pqKVZNBmxwAhO7/IwNM9g=
@@ -335,8 +335,8 @@ golang.org/x/crypto v0.18.0/go.mod h1:R0j02AL6hcrfOiy9T4ZYp/rcWeMxM3L6QYxlOuEG1m
 golang.org/x/crypto v0.33.0 h1:IOBPskki6Lysi0lo9qQvbxiQ+FvsCC/YWOecCHAixus=
 golang.org/x/crypto v0.33.0/go.mod h1:bVdXmD7IV/4GdElGPozy6U7lWdRXA4qyRVGJV57uQ5M=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
-golang.org/x/exp v0.0.0-20250106191152-7588d65b2ba8 h1:yqrTHse8TCMW1M1ZCP+VAR/l0kKxwaAIqN/il7x4voA=
-golang.org/x/exp v0.0.0-20250106191152-7588d65b2ba8/go.mod h1:tujkw807nyEEAamNbDrEGzRav+ilXA7PCRAd6xsmwiU=
+golang.org/x/exp v0.0.0-20250215185904-eff6e970281f h1:oFMYAjX0867ZD2jcNiLBrI9BdpmEkvPyi5YrBGXbamg=
+golang.org/x/exp v0.0.0-20250215185904-eff6e970281f/go.mod h1:BHOTPb3L19zxehTsLoJXVaTktb06DFgmdW6Wb9s8jqk=
 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=
@@ -431,8 +431,8 @@ golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da h1:noIWHXmPHxILtqtCOPIhS
 golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da/go.mod h1:NDW/Ps6MPRej6fsCIbMTohpP40sJ/P/vI1MoTEGwX90=
 gonum.org/v1/gonum v0.15.1 h1:FNy7N6OUZVUaWG9pTiD+jlhdQ3lMP+/LcTpJ6+a8sQ0=
 gonum.org/v1/gonum v0.15.1/go.mod h1:eZTZuRFrzu5pcyjN5wJhcIhnUdNijYxX1T2IcrOGY0o=
-google.golang.org/api v0.221.0 h1:qzaJfLhDsbMeFee8zBRdt/Nc+xmOuafD/dbdgGfutOU=
-google.golang.org/api v0.221.0/go.mod h1:7sOU2+TL4TxUTdbi0gWgAIg7tH5qBXxoyhtL+9x3biQ=
+google.golang.org/api v0.222.0 h1:Aiewy7BKLCuq6cUCeOUrsAlzjXPqBkEeQ/iwGHVQa/4=
+google.golang.org/api v0.222.0/go.mod h1:efZia3nXpWELrwMlN5vyQrD4GmJN1Vw0x68Et3r+a9c=
 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=
@@ -442,8 +442,8 @@ google.golang.org/genproto v0.0.0-20241118233622-e639e219e697 h1:ToEetK57OidYuqD
 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-20250207221924-e9438ea467c6 h1:2duwAxN2+k0xLNpjnHTXoMUgnv6VPSp5fiqTuwSxjmI=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20250207221924-e9438ea467c6/go.mod h1:8BS3B93F/U1juMFq9+EDk+qOT5CO1R9IzXxG3PTqiRk=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20250212204824-5a70512c5d8b h1:FQtJ1MxbXoIIrZHZ33M+w5+dAP9o86rgpjoKr/ZmT7k=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20250212204824-5a70512c5d8b/go.mod h1:8BS3B93F/U1juMFq9+EDk+qOT5CO1R9IzXxG3PTqiRk=
 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=
@@ -478,14 +478,14 @@ honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWh
 honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 modernc.org/cc/v4 v4.24.4 h1:TFkx1s6dCkQpd6dKurBNmpo+G8Zl4Sq/ztJ+2+DEsh0=
 modernc.org/cc/v4 v4.24.4/go.mod h1:uVtb5OGqUKpoLWhqwNQo/8LwvoiEBLvZXIQ/SmO6mL0=
-modernc.org/ccgo/v4 v4.23.13 h1:PFiaemQwE/jdwi8XEHyEV+qYWoIuikLP3T4rvDeJb00=
-modernc.org/ccgo/v4 v4.23.13/go.mod h1:vdN4h2WR5aEoNondUx26K7G8X+nuBscYnAEWSRmN2/0=
+modernc.org/ccgo/v4 v4.23.16 h1:Z2N+kk38b7SfySC1ZkpGLN2vthNJP1+ZzGZIlH7uBxo=
+modernc.org/ccgo/v4 v4.23.16/go.mod h1:nNma8goMTY7aQZQNTyN9AIoJfxav4nvTnvKThAeMDdo=
 modernc.org/fileutil v1.3.0 h1:gQ5SIzK3H9kdfai/5x41oQiKValumqNTDXMvKo62HvE=
 modernc.org/fileutil v1.3.0/go.mod h1:XatxS8fZi3pS8/hKG2GH/ArUogfxjpEKs3Ku3aK4JyQ=
-modernc.org/gc/v2 v2.6.1 h1:+Qf6xdG8l7B27TQ8D8lw/iFMUj1RXRBOuMUWziJOsk8=
-modernc.org/gc/v2 v2.6.1/go.mod h1:YgIahr1ypgfe7chRuJi2gD7DBQiKSLMPgBQe9oIiito=
-modernc.org/libc v1.61.9 h1:PLSBXVkifXGELtJ5BOnBUyAHr7lsatNwFU/RRo4kfJM=
-modernc.org/libc v1.61.9/go.mod h1:61xrnzk/aR8gr5bR7Uj/lLFLuXu2/zMpIjcry63Eumk=
+modernc.org/gc/v2 v2.6.3 h1:aJVhcqAte49LF+mGveZ5KPlsp4tdGdAOT4sipJXADjw=
+modernc.org/gc/v2 v2.6.3/go.mod h1:YgIahr1ypgfe7chRuJi2gD7DBQiKSLMPgBQe9oIiito=
+modernc.org/libc v1.61.13 h1:3LRd6ZO1ezsFiX1y+bHd1ipyEHIJKvuprv0sLTBwLW8=
+modernc.org/libc v1.61.13/go.mod h1:8F/uJWL/3nNil0Lgt1Dpz+GgkApWh04N3el3hxJcA6E=
 modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU=
 modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg=
 modernc.org/memory v1.8.2 h1:cL9L4bcoAObu4NkxOlKWBWtNHIsnnACGF/TbqQ6sbcI=
@@ -494,8 +494,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.34.5 h1:Bb6SR13/fjp15jt70CL4f18JIN7p7dnMExd+UFnF15g=
-modernc.org/sqlite v1.34.5/go.mod h1:YLuNmX9NKs8wRNK2ko1LW1NGYcc9FkBO69JOt1AR9JE=
+modernc.org/sqlite v1.35.0 h1:yQps4fegMnZFdphtzlfQTCNBWtS0CZv48pRpW3RFHRw=
+modernc.org/sqlite v1.35.0/go.mod h1:9cr2sicr7jIaWTBKQmAxQLfBv9LL0su4ZTEV+utt3ic=
 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/internal/avro_schemas.go b/internal/avro_schemas.go
index 893b7b2..92365f8 100644
--- a/internal/avro_schemas.go
+++ b/internal/avro_schemas.go
@@ -17,552 +17,407 @@
 
 package internal
 
-import "github.com/hamba/avro/v2"
+import (
+	"fmt"
 
-const (
-	ManifestListV1Key  = "manifest-list-v1"
-	ManifestListV2Key  = "manifest-list-v2"
-	ManifestEntryV1Key = "manifest-entry-v1"
-	ManifestEntryV2Key = "manifest-entry-v2"
+	"github.com/hamba/avro/v2"
 )
 
+func NullableSchema(schema avro.Schema) avro.Schema {
+	return Must(avro.NewUnionSchema([]avro.Schema{
+		NullSchema, schema,
+	}))
+}
+
+var (
+	// for x := range 24 {
+	// 		math.Floor(math.Log10(math.Abs(math.Pow(2, float64(8*x-1)-1))))
+	// }
+	// maxPrecision = [...]int{-1, 2, 4, 6, 9, 11, 14, 16, 18, 21,
+	// 	23, 26, 28, 31, 33, 35, 38, 40, 43, 45, 47, 50, 52, 55}
+	// for p := range 40 {
+	//	  for pos := range 24 {
+	//       if p > maxPrecision[pos] {
+	//          requiredLength[p] = pos-1
+	//          break
+	//       }
+	//    }
+	// }
+	requiredLength = [...]int{1, 1, 1, 2, 2, 3, 3, 4, 4, 4, 5, 5,
+		6, 6, 6, 7, 7, 8, 8, 9, 9, 9, 10, 10, 11, 11, 11, 12, 12,
+		13, 13, 13, 14, 14, 15, 15, 16, 16, 16, 17}
+)
+
+// DecimalRequiredBytes returns the required number of bytes to store a
+// decimal value of the given precision. If the precision is outside of
+// the range (0, 40], this returns -1 as it is invalid.
+func DecimalRequiredBytes(precision int) int {
+	if precision <= 0 || precision >= 40 {
+		return -1
+	}
+
+	return requiredLength[precision]
+}
+
+func DecimalSchema(precision, scale int) avro.Schema {
+	return Must(avro.NewFixedSchema("fixed", "",
+		DecimalRequiredBytes(precision), avro.NewDecimalLogicalSchema(precision, scale)))
+}
+
 var (
+	NullSchema           = avro.NewNullSchema()
+	BoolSchema           = avro.NewPrimitiveSchema(avro.Boolean, nil)
+	NullableBoolSchema   = NullableSchema(BoolSchema)
+	BinarySchema         = avro.NewPrimitiveSchema(avro.Bytes, nil)
+	NullableBinarySchema = NullableSchema(BinarySchema)
+	StringSchema         = avro.NewPrimitiveSchema(avro.String, nil)
+	IntSchema            = avro.NewPrimitiveSchema(avro.Int, nil)
+	NullableIntSchema    = NullableSchema(IntSchema)
+	LongSchema           = avro.NewPrimitiveSchema(avro.Long, nil)
+	NullableLongSchema   = NullableSchema(LongSchema)
+	FloatSchema          = avro.NewPrimitiveSchema(avro.Float, nil)
+	DoubleSchema         = avro.NewPrimitiveSchema(avro.Double, nil)
+	DateSchema           = avro.NewPrimitiveSchema(avro.Int, avro.NewPrimitiveLogicalSchema(avro.Date))
+	TimeSchema           = avro.NewPrimitiveSchema(avro.Long, avro.NewPrimitiveLogicalSchema(avro.TimeMicros))
+	TimestampSchema      = avro.NewPrimitiveSchema(avro.Long, avro.NewPrimitiveLogicalSchema(avro.TimestampMicros),
+		avro.WithProps(map[string]any{"adjust-to-utc": false}))
+	TimestampTzSchema = avro.NewPrimitiveSchema(avro.Long, avro.NewPrimitiveLogicalSchema(avro.TimestampMicros),
+		avro.WithProps(map[string]any{"adjust-to-utc": true}))
+	UUIDSchema = Must(avro.NewFixedSchema("uuid", "", 16, avro.NewPrimitiveLogicalSchema(avro.UUID)))
+
 	AvroSchemaCache avro.SchemaCache
 )
 
+func newMapSchema(name string, keySchema, valueSchema avro.Schema, keyFieldID, valueFieldID int) avro.Schema {
+	return avro.NewArraySchema(
+		Must(avro.NewRecordSchema(name, "", []*avro.Field{
+			Must(avro.NewField("key", keySchema, avro.WithProps(map[string]any{"field-id": keyFieldID}))),
+			Must(avro.NewField("value", valueSchema, avro.WithProps(map[string]any{"field-id": valueFieldID}))),
+		})), avro.WithProps(map[string]any{"logicalType": "map"}))
+}
+
+func WithFieldID(id int) avro.SchemaOption {
+	return avro.WithProps(map[string]any{"field-id": id})
+}
+
 func init() {
-	AvroSchemaCache.Add(ManifestListV1Key, avro.MustParse(`{
-		"type": "record",
-		"name": "manifest_file",
-		"fields": [
-			{"name": "manifest_path", "type": "string", "doc": "Location URI with FS scheme", "field-id": 500},
-			{"name": "manifest_length", "type": "long", "doc": "Total file size in bytes", "field-id": 501},
-			{"name": "partition_spec_id", "type": "int", "doc": "Spec ID used to write", "field-id": 502},
-			{
-				"name": "added_snapshot_id",
-				"type": "long",
-				"doc": "Snapshot ID that added the manifest",
-				"field-id": 503
-			},
-			{
-				"name": "added_data_files_count",
-				"type": ["null", "int"],
-				"doc": "Added entry count",
-				"field-id": 504
-			},
-			{
-				"name": "existing_data_files_count",
-				"type": ["null", "int"],
-				"doc": "Existing entry count",
-				"field-id": 505
-			},
-			{
-				"name": "deleted_data_files_count",
-				"type": ["null", "int"],
-				"doc": "Deleted entry count",
-				"field-id": 506
-			},
-			{
-				"name": "partitions",
-				"type": [
-					"null",
-					{
-						"type": "array",
-						"items": {
-							"type": "record",
-							"name": "r508",
-							"fields": [
-								{
-									"name": "contains_null",
-									"type": "boolean",
-									"doc": "True if any file has a null partition value",
-									"field-id": 509
-								},
-								{
-									"name": "contains_nan",
-									"type": ["null", "boolean"],
-									"doc": "True if any file has a nan partition value",
-									"field-id": 518
-								},
-								{
-									"name": "lower_bound",
-									"type": ["null", "bytes"],
-									"doc": "Partition lower bound for all files",								
-									"field-id": 510
-								},
-								{
-									"name": "upper_bound",
-									"type": ["null", "bytes"],
-									"doc": "Partition upper bound for all files",								
-									"field-id": 511
-								}
-							]
-						},
-						"element-id": 508
-					}
-				],
-				"doc": "Summary for each partition",			
-				"field-id": 507
-			},
-			{"name": "added_rows_count", "type": ["null", "long"], "doc": "Added rows count", "field-id": 512},
-			{
-				"name": "existing_rows_count",
-				"type": ["null", "long"],
-				"doc": "Existing rows count",			
-				"field-id": 513
-			},
-			{
-				"name": "deleted_rows_count",
-				"type": ["null", "long"],
-				"doc": "Deleted rows count",			
-				"field-id": 514
-			}
-		]
-	}`))
+	AvroSchemaCache.Add("field_summary", Must(avro.NewRecordSchema("field_summary", "", []*avro.Field{
+		Must(avro.NewField("contains_null",
+			BoolSchema,
+			avro.WithDoc("true if the field contains null values"),
+			WithFieldID(509))),
+		Must(avro.NewField("contains_nan",
+			NullableBoolSchema,
+			avro.WithDoc("true if the field contains NaN values"),
+			WithFieldID(518))),
+		Must(avro.NewField("lower_bound", NullableBinarySchema,
+			avro.WithDoc("serialized lower bound"),
+			WithFieldID(510))),
+		Must(avro.NewField("upper_bound", NullableBinarySchema,
+			avro.WithDoc("serialized upper bound"),
+			WithFieldID(511))),
+	}, WithFieldID(508))))
+
+	AvroSchemaCache.Add("manifest_list_file_v1", Must(avro.NewRecordSchema("manifest_file", "", []*avro.Field{
+		Must(avro.NewField("manifest_path",
+			StringSchema,
+			avro.WithDoc("Location URI with FS scheme"),
+			WithFieldID(500))),
+		Must(avro.NewField("manifest_length",
+			LongSchema,
+			avro.WithDoc("Total file size in bytes"),
+			WithFieldID(501))),
+		Must(avro.NewField("partition_spec_id",
+			IntSchema,
+			avro.WithDoc("Spec ID used to write"),
+			WithFieldID(502))),
+		Must(avro.NewField("added_snapshot_id",
+			LongSchema,
+			avro.WithDoc("Snapshot ID that added the manifest"),
+			WithFieldID(503))),
+		Must(avro.NewField("added_files_count",
+			NullableIntSchema,
+			avro.WithDoc("Added entry count"),
+			WithFieldID(504))),
+		Must(avro.NewField("existing_files_count",
+			NullableIntSchema,
+			avro.WithDoc("Existing entry count"),
+			WithFieldID(505))),
+		Must(avro.NewField("deleted_files_count",
+			NullableIntSchema,
+			avro.WithDoc("Deleted entry count"),
+			WithFieldID(506))),
+		Must(avro.NewField("partitions",
+			NullableSchema(
+				avro.NewArraySchema(AvroSchemaCache.Get("field_summary"))),
+			avro.WithDoc("Partition field summaries"),
+			WithFieldID(507))),
+		Must(avro.NewField("added_rows_count",
+			NullableLongSchema,
+			avro.WithDoc("Added row count"),
+			WithFieldID(512))),
+		Must(avro.NewField("existing_rows_count",
+			NullableLongSchema,
+			avro.WithDoc("Existing row count"),
+			WithFieldID(513))),
+		Must(avro.NewField("deleted_rows_count",
+			NullableLongSchema,
+			avro.WithDoc("Deleted row count"),
+			WithFieldID(514))),
+		Must(avro.NewField("key_metadata", NullableBinarySchema,
+			avro.WithDoc("Key metadata"),
+			WithFieldID(519))),
+	})))
+
+	AvroSchemaCache.Add("manifest_list_file_v2", Must(avro.NewRecordSchema("manifest_file", "", []*avro.Field{
+		Must(avro.NewField("manifest_path",
+			StringSchema,
+			avro.WithDoc("Location URI with FS scheme"),
+			WithFieldID(500))),
+		Must(avro.NewField("manifest_length",
+			LongSchema,
+			avro.WithDoc("Total file size in bytes"),
+			WithFieldID(501))),
+		Must(avro.NewField("partition_spec_id",
+			IntSchema,
+			avro.WithDoc("Spec ID used to write"),
+			WithFieldID(502))),
+		Must(avro.NewField("content", IntSchema,
+			avro.WithDoc("Content type"),
+			avro.WithDefault(0),
+			WithFieldID(517))),
+		Must(avro.NewField("sequence_number", LongSchema,
+			avro.WithDoc("Sequence number"),
+			avro.WithDefault(int64(0)),
+			WithFieldID(515))),
+		Must(avro.NewField("min_sequence_number", LongSchema,
+			avro.WithDoc("Minimum sequence number"),
+			avro.WithDefault(int64(0)),
+			WithFieldID(516))),
+		Must(avro.NewField("added_snapshot_id",
+			LongSchema,
+			avro.WithDoc("Snapshot ID that added the manifest"),
+			WithFieldID(503))),
+		Must(avro.NewField("added_files_count",
+			IntSchema,
+			avro.WithDoc("Added entry count"),
+			WithFieldID(504))),
+		Must(avro.NewField("existing_files_count",
+			IntSchema,
+			avro.WithDoc("Existing entry count"),
+			WithFieldID(505))),
+		Must(avro.NewField("deleted_files_count",
+			IntSchema,
+			avro.WithDoc("Deleted entry count"),
+			WithFieldID(506))),
+		Must(avro.NewField("partitions",
+			NullableSchema(
+				avro.NewArraySchema(AvroSchemaCache.Get("field_summary"))),
+			avro.WithDoc("Partition field summaries"),
+			WithFieldID(507))),
+		Must(avro.NewField("added_rows_count",
+			LongSchema,
+			avro.WithDoc("Added row count"),
+			WithFieldID(512))),
+		Must(avro.NewField("existing_rows_count",
+			LongSchema,
+			avro.WithDoc("Existing row count"),
+			WithFieldID(513))),
+		Must(avro.NewField("deleted_rows_count",
+			LongSchema,
+			avro.WithDoc("Deleted row count"),
+			WithFieldID(514))),
+		Must(avro.NewField("key_metadata", NullableBinarySchema,
+			avro.WithDoc("Key metadata"),
+			WithFieldID(519))),
+	})))
+
+	AvroSchemaCache.Add("data_file_v1", Must(avro.NewRecordSchema("data_file", "", []*avro.Field{
+		Must(avro.NewField("file_path",
+			StringSchema,
+			avro.WithDoc("Location URI with FS scheme"),
+			WithFieldID(100))),
+		Must(avro.NewField("file_format",
+			StringSchema,
+			avro.WithDoc("File format name: avro, orc, parquet"),
+			WithFieldID(101))),
+		// skip partition field, we'll add that dynamically as needed
+		Must(avro.NewField("record_count",
+			LongSchema,
+			avro.WithDoc("Number of records in the file"),
+			WithFieldID(103))),
+		Must(avro.NewField("file_size_in_bytes",
+			LongSchema,
+			avro.WithDoc("Size of the file in bytes"),
+			WithFieldID(104))),
+		Must(avro.NewField("block_size_in_bytes",
+			LongSchema,
+			avro.WithDoc("Deprecated. Always write default in v1. Do not write in v2."),
+			avro.WithDefault(int64(64*1024*1024)),
+			WithFieldID(105))),
+		Must(avro.NewField("column_sizes",
+			NullableSchema(newMapSchema("k117_v118", IntSchema, LongSchema, 117, 118)),
+			avro.WithDoc("map of column id to total size on disk"),
+			WithFieldID(108))),
+		Must(avro.NewField("value_counts",
+			NullableSchema(newMapSchema("k119_v120", IntSchema, LongSchema, 119, 120)),
+			avro.WithDoc("map of value to count"),
+			WithFieldID(109))),
+		Must(avro.NewField("null_value_counts",
+			NullableSchema(newMapSchema("k121_v122", IntSchema, LongSchema, 121, 122)),
+			avro.WithDoc("map of value to count"),
+			WithFieldID(110))),
+		Must(avro.NewField("nan_value_counts",
+			NullableSchema(newMapSchema("k138_v139", IntSchema, LongSchema, 138, 139)),
+			avro.WithDoc("map of value to count"),
+			WithFieldID(137))),
+		Must(avro.NewField("lower_bounds",
+			NullableSchema(newMapSchema("k126_v127", IntSchema, BinarySchema, 126, 127)),
+			avro.WithDoc("map of column id to lower bound"),
+			WithFieldID(125))),
+		Must(avro.NewField("upper_bounds",
+			NullableSchema(newMapSchema("k129_v130", IntSchema, BinarySchema, 129, 130)),
+			avro.WithDoc("map of column id to upper bound"),
+			WithFieldID(128))),
+		Must(avro.NewField("key_metadata", NullableBinarySchema,
+			avro.WithDoc("Encryption Key Metadata Blob"),
+			WithFieldID(131))),
+		Must(avro.NewField("split_offsets",
+			NullableSchema(avro.NewArraySchema(LongSchema,
+				WithFieldID(133))),
+			avro.WithDoc("splitable offsets"),
+			WithFieldID(132))),
+		Must(avro.NewField("sort_order_id",
+			NullableIntSchema,
+			avro.WithDoc("Sort order ID"),
+			WithFieldID(140))),
+	})))
+
+	AvroSchemaCache.Add("data_file_v2", Must(avro.NewRecordSchema("data_file", "", []*avro.Field{
+		Must(avro.NewField("content", IntSchema,
+			avro.WithDoc("Content type"),
+			avro.WithDefault(0),
+			WithFieldID(134))),
+		Must(avro.NewField("file_path",
+			StringSchema,
+			avro.WithDoc("Location URI with FS scheme"),
+			WithFieldID(100))),
+		Must(avro.NewField("file_format",
+			StringSchema,
+			avro.WithDoc("File format name: avro, orc, parquet"),
+			WithFieldID(101))),
+		// skip partition field, we'll add that dynamically as needed
+		Must(avro.NewField("record_count",
+			LongSchema,
+			avro.WithDoc("Number of records in the file"),
+			WithFieldID(103))),
+		Must(avro.NewField("file_size_in_bytes",
+			LongSchema,
+			avro.WithDoc("Size of the file in bytes"),
+			WithFieldID(104))),
+		Must(avro.NewField("column_sizes",
+			NullableSchema(newMapSchema("k117_v118", IntSchema, LongSchema, 117, 118)),
+			avro.WithDoc("map of column id to total size on disk"),
+			WithFieldID(108))),
+		Must(avro.NewField("value_counts",
+			NullableSchema(newMapSchema("k119_v120", IntSchema, LongSchema, 119, 120)),
+			avro.WithDoc("map of value to count"),
+			WithFieldID(109))),
+		Must(avro.NewField("null_value_counts",
+			NullableSchema(newMapSchema("k121_v122", IntSchema, LongSchema, 121, 122)),
+			avro.WithDoc("map of value to count"),
+			WithFieldID(110))),
+		Must(avro.NewField("nan_value_counts",
+			NullableSchema(newMapSchema("k138_v139", IntSchema, LongSchema, 138, 139)),
+			avro.WithDoc("map of value to count"),
+			WithFieldID(137))),
+		Must(avro.NewField("lower_bounds",
+			NullableSchema(newMapSchema("k126_v127", IntSchema, BinarySchema, 126, 127)),
+			avro.WithDoc("map of column id to lower bound"),
+			WithFieldID(125))),
+		Must(avro.NewField("upper_bounds",
+			NullableSchema(newMapSchema("k129_v130", IntSchema, BinarySchema, 129, 130)),
+			avro.WithDoc("map of column id to upper bound"),
+			WithFieldID(128))),
+		Must(avro.NewField("key_metadata", NullableBinarySchema,
+			avro.WithDoc("Encryption Key Metadata Blob"),
+			WithFieldID(131))),
+		Must(avro.NewField("split_offsets",
+			NullableSchema(avro.NewArraySchema(LongSchema,
+				WithFieldID(133))),
+			avro.WithDoc("splitable offsets"),
+			WithFieldID(132))),
+		Must(avro.NewField("equality_ids",
+			NullableSchema(avro.NewArraySchema(LongSchema,
+				WithFieldID(136))),
+			avro.WithDoc("field ids used to determine row equality in equality delete files"),
+			WithFieldID(135))),
+		Must(avro.NewField("sort_order_id",
+			NullableIntSchema,
+			avro.WithDoc("Sort order ID"),
+			WithFieldID(140))),
+	})))
+
+	AvroSchemaCache.Add("manifest_entry_v1", Must(avro.NewRecordSchema("manifest_entry", "", []*avro.Field{
+		Must(avro.NewField("status", IntSchema, WithFieldID(0))),
+		Must(avro.NewField("snapshot_id", LongSchema, WithFieldID(1))),
+		// leave data_file for dyanmic generation
+	})))
+
+	AvroSchemaCache.Add("manifest_entry_v2", Must(avro.NewRecordSchema("manifest_entry", "", []*avro.Field{
+		Must(avro.NewField("status", IntSchema, WithFieldID(0))),
+		Must(avro.NewField("snapshot_id", NullableLongSchema, WithFieldID(1))),
+		Must(avro.NewField("sequence_number", NullableLongSchema, WithFieldID(3))),
+		Must(avro.NewField("file_sequence_number", NullableLongSchema, WithFieldID(4))),
+		// leave data_file for dynamic generation
+	})))
+}
+
+func newDataFileSchema(partitionType avro.Schema, version int) (avro.Schema, error) {
+	key := fmt.Sprintf("data_file_v%d", version)
+	schema := AvroSchemaCache.Get(key)
+
+	partField, err := avro.NewField("partition",
+		partitionType, WithFieldID(102))
+	if err != nil {
+		return nil, err
+	}
+
+	return avro.NewRecordSchema("data_file", "",
+		append(schema.(*avro.RecordSchema).Fields(), partField))
+}
+
+func NewManifestFileSchema(version int) (avro.Schema, error) {
+	switch version {
+	case 1, 2:
+	default:
+		return nil, fmt.Errorf("unsupported iceberg spec version: %d", version)
+	}
+
+	key := fmt.Sprintf("manifest_list_file_v%d", version)
+	return AvroSchemaCache.Get(key), nil
+}
+
+func NewManifestEntrySchema(partitionType avro.Schema, version int) (avro.Schema, error) {
+	switch version {
+	case 1, 2:
+	default:
+		return nil, fmt.Errorf("unsupported iceberg spec version: %d", version)
+	}
+
+	dfschema, err := newDataFileSchema(partitionType, version)
+	if err != nil {
+		return nil, err
+	}
 
-	AvroSchemaCache.Add(ManifestListV2Key, avro.MustParse(`{
-        "type": "record",
-        "name": "manifest_file",
-        "fields": [
-            {"name": "manifest_path", "type": "string", "doc": "Location URI with FS scheme", "field-id": 500},
-            {"name": "manifest_length", "type": "long", "doc": "Total file size in bytes", "field-id": 501},
-            {"name": "partition_spec_id", "type": "int", "doc": "Spec ID used to write", "field-id": 502},
-            {"name": "content", "type": "int", "doc": "Contents of the manifest: 0=data, 1=deletes", "field-id": 517},
-            {
-                "name": "sequence_number",
-                "type": "long",
-                "doc": "Sequence number when the manifest was added",
-                "field-id": 515
-            },
-            {
-                "name": "min_sequence_number",
-                "type": "long",
-                "doc": "Lowest sequence number in the manifest",
-                "field-id": 516
-            },
-            {"name": "added_snapshot_id", "type": "long", "doc": "Snapshot ID that added the manifest", "field-id": 503},
-            {"name": "added_files_count", "type": "int", "doc": "Added entry count", "field-id": 504},
-            {"name": "existing_files_count", "type": "int", "doc": "Existing entry count", "field-id": 505},
-            {"name": "deleted_files_count", "type": "int", "doc": "Deleted entry count", "field-id": 506},
-            {"name": "added_rows_count", "type": "long", "doc": "Added rows count", "field-id": 512},
-            {"name": "existing_rows_count", "type": "long", "doc": "Existing rows count", "field-id": 513},
-            {"name": "deleted_rows_count", "type": "long", "doc": "Deleted rows count", "field-id": 514},
-            {
-                "name": "partitions",
-                "type": [
-                    "null",
-                    {
-                        "type": "array",
-                        "items": {
-                            "type": "record",
-                            "name": "r508",
-                            "fields": [
-                                {
-                                    "name": "contains_null",
-                                    "type": "boolean",
-                                    "doc": "True if any file has a null partition value",
-                                    "field-id": 509
-                                },
-                                {
-                                    "name": "contains_nan",
-                                    "type": ["null", "boolean"],
-                                    "doc": "True if any file has a nan partition value",                                  
-                                    "field-id": 518
-                                },
-                                {
-                                    "name": "lower_bound",
-                                    "type": ["null", "bytes"],
-                                    "doc": "Partition lower bound for all files",                                    
-                                    "field-id": 510
-                                },
-                                {
-                                    "name": "upper_bound",
-                                    "type": ["null", "bytes"],
-                                    "doc": "Partition upper bound for all files",
-                                    "field-id": 511
-                                }
-                            ]
-                        },
-                        "element-id": 508
-                    }
-                ],
-                "doc": "Summary for each partition",
-                "field-id": 507
-            }
-        ]
-    }`))
+	dfField, err := avro.NewField("data_file", dfschema, WithFieldID(2))
+	if err != nil {
+		return nil, err
+	}
 
-	AvroSchemaCache.Add(ManifestEntryV1Key, avro.MustParse(`{
-        "type": "record",
-        "name": "manifest_entry",
-        "fields": [
-            {"name": "status", "type": "int", "field-id": 0},
-            {"name": "snapshot_id", "type": "long", "field-id": 1},
-            {
-                "name": "data_file",
-                "type": {
-                    "type": "record",
-                    "name": "r2",
-                    "fields": [
-                        {"name": "file_path", "type": "string", "doc": "Location URI with FS scheme", "field-id": 100},
-                        {
-                            "name": "file_format",
-                            "type": "string",
-                            "doc": "File format name: avro, orc, or parquet",
-                            "field-id": 101
-                        },
-                        {
-                            "name": "partition",
-                            "type": {
-                                "type": "record",
-                                "name": "r102",
-                                "fields": [
-                                    {"field-id": 1000, "name": "VendorID", "type": ["null", "int"]},
-                                    {
-                                        "field-id": 1001,                                        
-                                        "name": "tpep_pickup_datetime",
-                                        "type": ["null", {"type": "int", "logicalType": "date"}]
-                                    }
-                                ]
-                            },
-                            "field-id": 102
-                        },
-                        {"name": "record_count", "type": "long", "doc": "Number of records in the file", "field-id": 103},
-                        {"name": "file_size_in_bytes", "type": "long", "doc": "Total file size in bytes", "field-id": 104},
-                        {"name": "block_size_in_bytes", "type": "long", "field-id": 105},
-                        {
-                            "name": "column_sizes",
-                            "type": [
-                                "null",
-                                {
-                                    "type": "array",
-                                    "items": {
-                                        "type": "record",
-                                        "name": "k117_v118",
-                                        "fields": [
-                                            {"name": "key", "type": "int", "field-id": 117},
-                                            {"name": "value", "type": "long", "field-id": 118}
-                                        ]
-                                    },
-                                    "logicalType": "map"
-                                }
-                            ],
-                            "doc": "Map of column id to total size on disk",                            
-                            "field-id": 108
-                        },
-                        {
-                            "name": "value_counts",
-                            "type": [
-                                "null",
-                                {
-                                    "type": "array",
-                                    "items": {
-                                        "type": "record",
-                                        "name": "k119_v120",
-                                        "fields": [
-                                            {"name": "key", "type": "int", "field-id": 119},
-                                            {"name": "value", "type": "long", "field-id": 120}
-                                        ]
-                                    },
-                                    "logicalType": "map"
-                                }
-                            ],
-                            "doc": "Map of column id to total count, including null and NaN",                            
-                            "field-id": 109
-                        },
-                        {
-                            "name": "null_value_counts",
-                            "type": [
-                                "null",
-                                {
-                                    "type": "array",
-                                    "items": {
-                                        "type": "record",
-                                        "name": "k121_v122",
-                                        "fields": [
-                                            {"name": "key", "type": "int", "field-id": 121},
-                                            {"name": "value", "type": "long", "field-id": 122}
-                                        ]
-                                    },
-                                    "logicalType": "map"
-                                }
-                            ],
-                            "doc": "Map of column id to null value count",                            
-                            "field-id": 110
-                        },
-                        {
-                            "name": "nan_value_counts",
-                            "type": [
-                                "null",
-                                {
-                                    "type": "array",
-                                    "items": {
-                                        "type": "record",
-                                        "name": "k138_v139",
-                                        "fields": [
-                                            {"name": "key", "type": "int", "field-id": 138},
-                                            {"name": "value", "type": "long", "field-id": 139}
-                                        ]
-                                    },
-                                    "logicalType": "map"
-                                }
-                            ],
-                            "doc": "Map of column id to number of NaN values in the column",                            
-                            "field-id": 137
-                        },
-                        {
-                            "name": "lower_bounds",
-                            "type": [
-                                "null",
-                                {
-                                    "type": "array",
-                                    "items": {
-                                        "type": "record",
-                                        "name": "k126_v127",
-                                        "fields": [
-                                            {"name": "key", "type": "int", "field-id": 126},
-                                            {"name": "value", "type": "bytes", "field-id": 127}
-                                        ]
-                                    },
-                                    "logicalType": "map"
-                                }
-                            ],
-                            "doc": "Map of column id to lower bound",                            
-                            "field-id": 125
-                        },
-                        {
-                            "name": "upper_bounds",
-                            "type": [
-                                "null",
-                                {
-                                    "type": "array",
-                                    "items": {
-                                        "type": "record",
-                                        "name": "k129_v130",
-                                        "fields": [
-                                            {"name": "key", "type": "int", "field-id": 129},
-                                            {"name": "value", "type": "bytes", "field-id": 130}
-                                        ]
-                                    },
-                                    "logicalType": "map"
-                                }
-                            ],
-                            "doc": "Map of column id to upper bound",                            
-                            "field-id": 128
-                        },
-                        {
-                            "name": "key_metadata",
-                            "type": ["null", "bytes"],
-                            "doc": "Encryption key metadata blob",                            
-                            "field-id": 131
-                        },
-                        {
-                            "name": "split_offsets",
-                            "type": ["null", {"type": "array", "items": "long", "element-id": 133}],
-                            "doc": "Splittable offsets",                            
-                            "field-id": 132
-                        },
-                        {
-                            "name": "sort_order_id",
-                            "type": ["null", "int"],
-                            "doc": "Sort order ID",                            
-                            "field-id": 140
-                        }
-                    ]
-                },
-                "field-id": 2
-            }
-        ]
-    }`))
+	key := fmt.Sprintf("manifest_entry_v%d", version)
+	schema := AvroSchemaCache.Get(key)
 
-	AvroSchemaCache.Add(ManifestEntryV2Key, avro.MustParse(`{
-        "type": "record",
-        "name": "manifest_entry",
-        "fields": [
-            {"name": "status", "type": "int", "field-id": 0},
-            {"name": "snapshot_id", "type": ["null", "long"], "field-id": 1},
-			{"name": "sequence_number", "type": ["null", "long"], "field-id": 3},
-			{"name": "file_sequence_number", "type": ["null", "long"], "field-id": 4},
-            {
-                "name": "data_file",
-                "type": {
-                    "type": "record",
-                    "name": "r2",
-                    "fields": [
-						{"name": "content", "type": "int", "doc": "Type of content stored by the data file", "field-id": 134},
-                        {"name": "file_path", "type": "string", "doc": "Location URI with FS scheme", "field-id": 100},
-                        {
-                            "name": "file_format",
-                            "type": "string",
-                            "doc": "File format name: avro, orc, or parquet",
-                            "field-id": 101
-                        },
-                        {
-                            "name": "partition",
-                            "type": {
-                                "type": "record",
-                                "name": "r102",
-                                "fields": [
-                                    {"field-id": 1000, "name": "VendorID", "type": ["null", "int"]},
-                                    {
-                                        "field-id": 1001,                                        
-                                        "name": "tpep_pickup_datetime",
-                                        "type": ["null", {"type": "int", "logicalType": "date"}]
-                                    }
-                                ]
-                            },
-                            "field-id": 102
-                        },
-                        {"name": "record_count", "type": "long", "doc": "Number of records in the file", "field-id": 103},
-                        {"name": "file_size_in_bytes", "type": "long", "doc": "Total file size in bytes", "field-id": 104},                        
-                        {
-                            "name": "column_sizes",
-                            "type": [
-                                "null",
-                                {
-                                    "type": "array",
-                                    "items": {
-                                        "type": "record",
-                                        "name": "k117_v118",
-                                        "fields": [
-                                            {"name": "key", "type": "int", "field-id": 117},
-                                            {"name": "value", "type": "long", "field-id": 118}
-                                        ]
-                                    },
-                                    "logicalType": "map"
-                                }
-                            ],
-                            "doc": "Map of column id to total size on disk",                            
-                            "field-id": 108
-                        },
-                        {
-                            "name": "value_counts",
-                            "type": [
-                                "null",
-                                {
-                                    "type": "array",
-                                    "items": {
-                                        "type": "record",
-                                        "name": "k119_v120",
-                                        "fields": [
-                                            {"name": "key", "type": "int", "field-id": 119},
-                                            {"name": "value", "type": "long", "field-id": 120}
-                                        ]
-                                    },
-                                    "logicalType": "map"
-                                }
-                            ],
-                            "doc": "Map of column id to total count, including null and NaN",                            
-                            "field-id": 109
-                        },
-                        {
-                            "name": "null_value_counts",
-                            "type": [
-                                "null",
-                                {
-                                    "type": "array",
-                                    "items": {
-                                        "type": "record",
-                                        "name": "k121_v122",
-                                        "fields": [
-                                            {"name": "key", "type": "int", "field-id": 121},
-                                            {"name": "value", "type": "long", "field-id": 122}
-                                        ]
-                                    },
-                                    "logicalType": "map"
-                                }
-                            ],
-                            "doc": "Map of column id to null value count",                            
-                            "field-id": 110
-                        },
-                        {
-                            "name": "nan_value_counts",
-                            "type": [
-                                "null",
-                                {
-                                    "type": "array",
-                                    "items": {
-                                        "type": "record",
-                                        "name": "k138_v139",
-                                        "fields": [
-                                            {"name": "key", "type": "int", "field-id": 138},
-                                            {"name": "value", "type": "long", "field-id": 139}
-                                        ]
-                                    },
-                                    "logicalType": "map"
-                                }
-                            ],
-                            "doc": "Map of column id to number of NaN values in the column",                            
-                            "field-id": 137
-                        },
-                        {
-                            "name": "lower_bounds",
-                            "type": [
-                                "null",
-                                {
-                                    "type": "array",
-                                    "items": {
-                                        "type": "record",
-                                        "name": "k126_v127",
-                                        "fields": [
-                                            {"name": "key", "type": "int", "field-id": 126},
-                                            {"name": "value", "type": "bytes", "field-id": 127}
-                                        ]
-                                    },
-                                    "logicalType": "map"
-                                }
-                            ],
-                            "doc": "Map of column id to lower bound",                            
-                            "field-id": 125
-                        },
-                        {
-                            "name": "upper_bounds",
-                            "type": [
-                                "null",
-                                {
-                                    "type": "array",
-                                    "items": {
-                                        "type": "record",
-                                        "name": "k129_v130",
-                                        "fields": [
-                                            {"name": "key", "type": "int", "field-id": 129},
-                                            {"name": "value", "type": "bytes", "field-id": 130}
-                                        ]
-                                    },
-                                    "logicalType": "map"
-                                }
-                            ],
-                            "doc": "Map of column id to upper bound",                            
-                            "field-id": 128
-                        },
-                        {
-                            "name": "key_metadata",
-                            "type": ["null", "bytes"],
-                            "doc": "Encryption key metadata blob",                            
-                            "field-id": 131
-                        },
-                        {
-                            "name": "split_offsets",
-                            "type": ["null", {"type": "array", "items": "long", "element-id": 133}],
-                            "doc": "Splittable offsets",                            
-                            "field-id": 132
-                        },
-						{
-							"name": "equality_ids",
-							"type": ["null", {"type": "array", "items": "int", "element-id": 136}],
-							"doc": "Field ids used to determine row equality for delete files",
-							"field-id": 135
-						},
-                        {
-                            "name": "sort_order_id",
-                            "type": ["null", "int"],
-                            "doc": "Sort order ID",                            
-                            "field-id": 140
-                        }
-                    ]
-                },
-                "field-id": 2
-            }
-        ]
-    }`))
+	return avro.NewRecordSchema("manifest_entry", "",
+		append(schema.(*avro.RecordSchema).Fields(), dfField))
 }
diff --git a/internal/must.go b/internal/must.go
new file mode 100644
index 0000000..e3d24c6
--- /dev/null
+++ b/internal/must.go
@@ -0,0 +1,26 @@
+// 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 internal
+
+func Must[T any](v T, err error) T {
+	if err != nil {
+		panic(err)
+	}
+
+	return v
+}
diff --git a/literals.go b/literals.go
index be4bcd9..cd89e63 100644
--- a/literals.go
+++ b/literals.go
@@ -108,6 +108,11 @@ func NewLiteral[T LiteralType](val T) Literal {
 	panic("can't happen due to literal type constraint")
 }
 
+func getComparator[T LiteralType]() Comparator[T] {
+	var z T
+	return NewLiteral(z).(TypedLiteral[T]).Comparator()
+}
+
 // LiteralFromBytes uses the defined Iceberg spec for how to serialize a value of
 // a the provided type and returns the appropriate Literal value from it.
 //
diff --git a/manifest.go b/manifest.go
index 208e09a..6eea875 100644
--- a/manifest.go
+++ b/manifest.go
@@ -18,13 +18,20 @@
 package iceberg
 
 import (
+	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
+	"math"
+	"reflect"
+	"slices"
+	"strconv"
 	"sync"
 	"time"
 
 	"github.com/apache/iceberg-go/internal"
 	iceio "github.com/apache/iceberg-go/io"
+	"github.com/google/uuid"
 
 	"github.com/hamba/avro/v2"
 	"github.com/hamba/avro/v2/ocf"
@@ -40,6 +47,17 @@ const (
 	ManifestContentDeletes ManifestContent = 1
 )
 
+func (m ManifestContent) String() string {
+	switch m {
+	case ManifestContentData:
+		return "data"
+	case ManifestContentDeletes:
+		return "deletes"
+	default:
+		return "UNKNOWN"
+	}
+}
+
 type FieldSummary struct {
 	ContainsNull bool    `avro:"contains_null"`
 	ContainsNaN  *bool   `avro:"contains_nan"`
@@ -67,6 +85,11 @@ func NewManifestV1Builder(path string, length int64, partitionSpecID int32, adde
 	}
 }
 
+func (b *ManifestV1Builder) PartitionType(st *StructType) *ManifestV1Builder {
+	b.m.partitionType = st
+	return b
+}
+
 func (b *ManifestV1Builder) AddedFiles(cnt int32) *ManifestV1Builder {
 	b.m.AddedFilesCount = &cnt
 	return b
@@ -112,6 +135,10 @@ func (b *ManifestV1Builder) KeyMetadata(km []byte) *ManifestV1Builder {
 // a pointer to the constructed manifest file. Further calls to the modifier
 // methods after calling build would modify the constructed ManifestFile.
 func (b *ManifestV1Builder) Build() ManifestFile {
+	if b.m.partitionType == nil {
+		b.m.partitionType = UnpartitionedSpec.PartitionType(nil)
+	}
+
 	return b.m
 }
 
@@ -126,13 +153,14 @@ func (f *fallbackManifestFileV1) toManifest() *manifestFileV1 {
 }
 
 type manifestFileV1 struct {
+	partitionType      *StructType
 	Path               string          `avro:"manifest_path"`
 	Len                int64           `avro:"manifest_length"`
 	SpecID             int32           `avro:"partition_spec_id"`
 	AddedSnapshotID    int64           `avro:"added_snapshot_id"`
-	AddedFilesCount    *int32          `avro:"added_data_files_count"`
-	ExistingFilesCount *int32          `avro:"existing_data_files_count"`
-	DeletedFilesCount  *int32          `avro:"deleted_data_files_count"`
+	AddedFilesCount    *int32          `avro:"added_files_count"`
+	ExistingFilesCount *int32          `avro:"existing_files_count"`
+	DeletedFilesCount  *int32          `avro:"deleted_files_count"`
 	AddedRowsCount     *int64          `avro:"added_rows_count"`
 	ExistingRowsCount  *int64          `avro:"existing_rows_count"`
 	DeletedRowsCount   *int64          `avro:"deleted_rows_count"`
@@ -215,10 +243,10 @@ func (m *manifestFileV1) FetchEntries(fs iceio.IO, discardDeleted bool) ([]Manif
 	return fetchManifestEntries(m, fs, discardDeleted)
 }
 
-// WriteEntries writes a list of manifest entries to an avro file.
-func (m *manifestFileV1) WriteEntries(out io.Writer, entries []ManifestEntry) error {
-	return writeManifestEntries(out, entries, m.Version())
-}
+// // WriteEntries writes a list of manifest entries to an avro file.
+// func (m *manifestFileV1) WriteEntries(out io.Writer, entries []ManifestEntry) error {
+// 	return writeManifestEntries(out, m.partitionType, entries, m.Version())
+// }
 
 // ManifestV2Builder is a helper for building a V2 manifest file
 // struct which will conform to the ManifestFile interface.
@@ -242,6 +270,11 @@ func NewManifestV2Builder(path string, length int64, partitionSpecID int32, cont
 	}
 }
 
+func (b *ManifestV2Builder) PartitionSchema(st *StructType) *ManifestV2Builder {
+	b.m.partitionType = st
+	return b
+}
+
 func (b *ManifestV2Builder) SequenceNum(num, minSeqNum int64) *ManifestV2Builder {
 	b.m.SeqNumber, b.m.MinSeqNumber = num, minSeqNum
 	return b
@@ -292,10 +325,15 @@ func (b *ManifestV2Builder) KeyMetadata(km []byte) *ManifestV2Builder {
 // a pointer to the constructed manifest file. Further calls to the modifier
 // methods after calling build would modify the constructed ManifestFile.
 func (b *ManifestV2Builder) Build() ManifestFile {
+	if b.m.partitionType == nil {
+		b.m.partitionType = UnpartitionedSpec.PartitionType(nil)
+	}
+
 	return b.m
 }
 
 type manifestFileV2 struct {
+	partitionType      *StructType
 	Path               string          `avro:"manifest_path"`
 	Len                int64           `avro:"manifest_length"`
 	SpecID             int32           `avro:"partition_spec_id"`
@@ -370,10 +408,10 @@ func (m *manifestFileV2) FetchEntries(fs iceio.IO, discardDeleted bool) ([]Manif
 	return fetchManifestEntries(m, fs, discardDeleted)
 }
 
-// WriteEntries writes a list of manifest entries to an avro file.
-func (m *manifestFileV2) WriteEntries(out io.Writer, entries []ManifestEntry) error {
-	return writeManifestEntries(out, entries, m.Version())
-}
+// // WriteEntries writes a list of manifest entries to an avro file.
+// func (m *manifestFileV2) WriteEntries(out io.Writer, entries []ManifestEntry) error {
+// 	return writeManifestEntries(out, m.partitionType, entries, m.Version())
+// }
 
 func getFieldIDMap(sc avro.Schema) (map[string]int, map[int]avro.LogicalType) {
 	getField := func(rs *avro.RecordSchema, name string) *avro.Field {
@@ -540,10 +578,10 @@ type ManifestFile interface {
 	// If discardDeleted is true, entries for files containing deleted rows
 	// will be skipped.
 	FetchEntries(fs iceio.IO, discardDeleted bool) ([]ManifestEntry, error)
-	// WriteEntries writes a list of manifest entries to a provided
-	// io.Writer. The version of the manifest file is used to determine the
-	// schema to use for writing the entries.
-	WriteEntries(out io.Writer, entries []ManifestEntry) error
+	// // WriteEntries writes a list of manifest entries to a provided
+	// // io.Writer. The version of the manifest file is used to determine the
+	// // schema to use for writing the entries.
+	// WriteEntries(out io.Writer, entries []ManifestEntry) error
 }
 
 // ReadManifestList reads in an avro manifest list file and returns a slice
@@ -596,48 +634,566 @@ func ReadManifestList(in io.Reader) ([]ManifestFile, error) {
 	return out, dec.Error()
 }
 
-// WriteManifestList writes a list of v2 manifest files to an avro file.
-func WriteManifestList(out io.Writer, files []ManifestFile) error {
-	if len(files) == 0 {
-		return nil
+type writerImpl interface {
+	content() ManifestContent
+	prepareEntry(ManifestEntry, int64) (ManifestEntry, error)
+}
+
+type v1writerImpl struct{}
+
+func (v1writerImpl) content() ManifestContent { return ManifestContentData }
+func (v1writerImpl) prepareEntry(entry ManifestEntry, _ int64) (ManifestEntry, error) {
+	return entry, nil
+}
+
+type v2writerImpl struct{}
+
+func (v2writerImpl) content() ManifestContent { return ManifestContentData }
+func (v2writerImpl) prepareEntry(entry ManifestEntry, snapshotID int64) (ManifestEntry, error) {
+	if entry.SequenceNum() <= 0 {
+		if entry.SnapshotID() > 0 && entry.SnapshotID() != snapshotID {
+			return nil, fmt.Errorf("found unassigned sequence number for entry from snapshot: %d", entry.SnapshotID())
+		}
+
+		if entry.Status() != EntryStatusADDED {
+			return nil, errors.New("only entries with status ADDED can be missing a sequence number")
+		}
+	}
+
+	return entry, nil
+}
+
+type fieldStats interface {
+	toSummary() FieldSummary
+	update(value any)
+}
+
+type partitionFieldStats[T LiteralType] struct {
+	containsNull bool
+	containsNan  bool
+	min          *T
+	max          *T
+
+	cmp Comparator[T]
+}
+
+func newPartitionFieldStat(typ PrimitiveType) fieldStats {
+	switch typ.(type) {
+	case Int32Type:
+		return &partitionFieldStats[int32]{cmp: getComparator[int32]()}
+	case Int64Type:
+		return &partitionFieldStats[int64]{cmp: getComparator[int64]()}
+	case Float32Type:
+		return &partitionFieldStats[float32]{cmp: getComparator[float32]()}
+	case Float64Type:
+		return &partitionFieldStats[float64]{cmp: getComparator[float64]()}
+	case StringType:
+		return &partitionFieldStats[string]{cmp: getComparator[string]()}
+	case DateType:
+		return &partitionFieldStats[Date]{cmp: getComparator[Date]()}
+	case TimeType:
+		return &partitionFieldStats[Time]{cmp: getComparator[Time]()}
+	case TimestampType:
+		return &partitionFieldStats[Timestamp]{cmp: getComparator[Timestamp]()}
+	case UUIDType:
+		return &partitionFieldStats[uuid.UUID]{cmp: getComparator[uuid.UUID]()}
+	case BinaryType:
+		return &partitionFieldStats[[]byte]{cmp: getComparator[[]byte]()}
+	case FixedType:
+		return &partitionFieldStats[[]byte]{cmp: getComparator[[]byte]()}
+	case DecimalType:
+		return &partitionFieldStats[Decimal]{cmp: getComparator[Decimal]()}
+	default:
+		panic(fmt.Sprintf("expected primitive type for partition type: %s", typ))
+	}
+}
+
+func (p *partitionFieldStats[T]) toSummary() FieldSummary {
+	var (
+		lowerBound *[]byte
+		upperBound *[]byte
+		lit        Literal
+	)
+
+	if p.min != nil {
+		lit = NewLiteral(*p.min)
+		lb, _ := lit.MarshalBinary()
+		lowerBound = &lb
+	}
+
+	if p.max != nil {
+		lit = NewLiteral(*p.max)
+		ub, _ := lit.MarshalBinary()
+		upperBound = &ub
 	}
 
-	version := files[0].Version()
+	return FieldSummary{
+		ContainsNull: p.containsNull,
+		ContainsNaN:  &p.containsNan,
+		LowerBound:   lowerBound,
+		UpperBound:   upperBound,
+	}
+}
+
+func (p *partitionFieldStats[T]) update(value any) {
+	if value == nil {
+		p.containsNull = true
+		return
+	}
 
-	for _, file := range files[1:] {
-		if file.Version() != version {
-			return fmt.Errorf(
-				"%w: ManifestFile '%s' has non-matching version %d instead of %d",
-				ErrInvalidArgument, file.FilePath(), file.Version(), version,
-			)
+	var actualVal T
+	v := reflect.ValueOf(value)
+	if !v.CanConvert(reflect.TypeOf(actualVal)) {
+		panic(fmt.Sprintf("expected type %T, got %T", actualVal, value))
+	}
+
+	actualVal = v.Convert(reflect.TypeOf(actualVal)).Interface().(T)
+
+	switch f := any(actualVal).(type) {
+	case float32:
+		if math.IsNaN(float64(f)) {
+			p.containsNan = true
+			return
+		}
+	case float64:
+		if math.IsNaN(f) {
+			p.containsNan = true
+			return
 		}
 	}
 
-	var key string
+	if p.min == nil {
+		p.min = &actualVal
+		p.max = &actualVal
+	} else {
+		if p.cmp(actualVal, *p.min) < 0 {
+			p.min = &actualVal
+		}
+
+		if p.cmp(actualVal, *p.max) > 0 {
+			p.max = &actualVal
+		}
+	}
+}
+
+func constructPartitionSummaries(spec PartitionSpec, schema *Schema, partitions []map[string]any) ([]FieldSummary, error) {
+	partType := spec.PartitionType(schema)
+	fieldStats := make([]fieldStats, len(partType.FieldList))
+	for i, field := range partType.FieldList {
+		pt, ok := field.Type.(PrimitiveType)
+		if !ok {
+			return nil, fmt.Errorf("expected primitive type for partition field, got %s", field.Type)
+		}
+
+		fieldStats[i] = newPartitionFieldStat(pt)
+	}
+
+	for _, part := range partitions {
+		for i, field := range partType.FieldList {
+			fieldStats[i].update(part[field.Name])
+		}
+	}
+
+	summaries := make([]FieldSummary, len(fieldStats))
+	for i, stat := range fieldStats {
+		summaries[i] = stat.toSummary()
+	}
+	return summaries, nil
+}
+
+type ManifestWriter struct {
+	closed  bool
+	version int
+	impl    writerImpl
+
+	output io.Writer
+	writer *ocf.Encoder
+
+	spec   PartitionSpec
+	schema *Schema
+
+	snapshotID    int64
+	addedFiles    int32
+	addedRows     int64
+	existingFiles int32
+	existingRows  int64
+	deletedFiles  int32
+	deletedRows   int64
+
+	partitions  []map[string]any
+	minSeqNum   int64
+	reusedEntry ManifestEntry
+}
+
+func NewManifestWriter(version int, out io.Writer, spec PartitionSpec, schema *Schema, snapshotID int64) (*ManifestWriter, error) {
+	var (
+		impl        writerImpl
+		reusedEntry ManifestEntry
+	)
+
 	switch version {
 	case 1:
-		key = internal.ManifestListV1Key
+		impl = v1writerImpl{}
+		reusedEntry = &manifestEntryV1{}
 	case 2:
-		key = internal.ManifestListV2Key
+		impl = v2writerImpl{}
+		reusedEntry = &manifestEntryV2{}
 	default:
-		return fmt.Errorf("%w: non-recognized version %d", ErrInvalidArgument, version)
+		return nil, fmt.Errorf("unsupported manifest version: %d", version)
+	}
+
+	sc, err := partitionTypeToAvroSchema(spec.PartitionType(schema))
+	if err != nil {
+		return nil, err
+	}
+
+	fileSchema, err := internal.NewManifestEntrySchema(sc, version)
+	if err != nil {
+		return nil, err
+	}
+
+	w := &ManifestWriter{
+		impl:        impl,
+		version:     version,
+		output:      out,
+		spec:        spec,
+		schema:      schema,
+		snapshotID:  snapshotID,
+		minSeqNum:   -1,
+		partitions:  make([]map[string]any, 0),
+		reusedEntry: reusedEntry,
+	}
+
+	md, err := w.meta()
+	if err != nil {
+		return nil, err
+	}
+
+	enc, err := ocf.NewEncoderWithSchema(fileSchema, out,
+		ocf.WithMetadata(md),
+		ocf.WithCodec(ocf.Deflate))
+
+	w.writer = enc
+	return w, err
+}
+
+func (w *ManifestWriter) Close() error {
+	if w.closed {
+		return nil
+	}
+
+	if w.addedFiles+w.existingFiles+w.deletedFiles == 0 {
+		return errors.New("empty manifest file has been written")
 	}
 
-	return avroEncode(key, version, files, out)
+	w.closed = true
+	return w.writer.Close()
 }
 
-func writeManifestEntries(out io.Writer, entries []ManifestEntry, version int) error {
-	var key string
+func (w *ManifestWriter) ToManifestFile(location string, length int64) (ManifestFile, error) {
+	if err := w.Close(); err != nil {
+		return nil, err
+	}
+
+	if w.minSeqNum == 0 {
+		w.minSeqNum = -1
+	}
+
+	partitions, err := constructPartitionSummaries(w.spec, w.schema, w.partitions)
+	if err != nil {
+		return nil, err
+	}
+
+	switch w.version {
+	case 1:
+		return &manifestFileV1{
+			Path:               location,
+			Len:                length,
+			SpecID:             int32(w.spec.id),
+			AddedSnapshotID:    w.snapshotID,
+			AddedFilesCount:    &w.addedFiles,
+			ExistingFilesCount: &w.existingFiles,
+			DeletedFilesCount:  &w.deletedFiles,
+			AddedRowsCount:     &w.addedRows,
+			ExistingRowsCount:  &w.existingRows,
+			DeletedRowsCount:   &w.deletedRows,
+			PartitionList:      &partitions,
+			Key:                nil,
+		}, nil
+	case 2:
+		return &manifestFileV2{
+			Path:               location,
+			Len:                length,
+			SpecID:             int32(w.spec.id),
+			Content:            ManifestContentData,
+			SeqNumber:          -1,
+			MinSeqNumber:       w.minSeqNum,
+			AddedSnapshotID:    w.snapshotID,
+			AddedFilesCount:    w.addedFiles,
+			ExistingFilesCount: w.existingFiles,
+			DeletedFilesCount:  w.deletedFiles,
+			AddedRowsCount:     w.addedRows,
+			ExistingRowsCount:  w.existingRows,
+			DeletedRowsCount:   w.deletedRows,
+			PartitionList:      &partitions,
+			Key:                nil,
+		}, nil
+	default:
+		return nil, fmt.Errorf("unsupported manifest version: %d", w.version)
+	}
+}
+
+func (w *ManifestWriter) meta() (map[string][]byte, error) {
+	schemaJson, err := json.Marshal(w.schema)
+	if err != nil {
+		return nil, err
+	}
+
+	specFieldsJson, err := json.Marshal(w.spec.fields)
+	if err != nil {
+		return nil, err
+	}
+
+	return map[string][]byte{
+		"schema":            schemaJson,
+		"schema-id":         []byte(strconv.Itoa(w.schema.ID)),
+		"partition-spec":    specFieldsJson,
+		"partition-spec-id": []byte(strconv.Itoa(w.spec.ID())),
+		"format-version":    []byte(strconv.Itoa(w.version)),
+		"content":           []byte(w.impl.content().String()),
+	}, nil
+}
+
+func (w *ManifestWriter) addEntry(entry ManifestEntry) error {
+	if w.closed {
+		return errors.New("cannot add entry to closed manifest writer")
+	}
+
+	switch entry.Status() {
+	case EntryStatusADDED:
+		w.addedFiles++
+		w.addedRows += entry.DataFile().Count()
+	case EntryStatusEXISTING:
+		w.existingFiles++
+		w.existingRows += entry.DataFile().Count()
+	case EntryStatusDELETED:
+		w.deletedFiles++
+		w.deletedRows += entry.DataFile().Count()
+	default:
+		return fmt.Errorf("unknown entry status: %v", entry.Status())
+	}
+
+	w.partitions = append(w.partitions, entry.DataFile().Partition())
+	if (entry.Status() == EntryStatusADDED || entry.Status() == EntryStatusEXISTING) &&
+		entry.SequenceNum() > 0 && (w.minSeqNum < 0 || entry.SequenceNum() < w.minSeqNum) {
+		w.minSeqNum = entry.SequenceNum()
+	}
+
+	entry, err := w.impl.prepareEntry(entry, w.snapshotID)
+	if err != nil {
+		return err
+	}
+
+	return w.writer.Encode(entry)
+}
+
+func (w *ManifestWriter) Add(entry ManifestEntry) error {
+	w.reusedEntry.wrap(EntryStatusADDED, w.snapshotID, entry.SequenceNum(), nil, entry.DataFile())
+	return w.addEntry(w.reusedEntry)
+}
+
+func (w *ManifestWriter) Delete(entry ManifestEntry) error {
+	w.reusedEntry.wrap(EntryStatusDELETED, w.snapshotID, entry.SequenceNum(), entry.FileSequenceNum(), entry.DataFile())
+	return w.addEntry(w.reusedEntry)
+}
+
+func (w *ManifestWriter) Existing(entry ManifestEntry) error {
+	w.reusedEntry.wrap(EntryStatusEXISTING, w.snapshotID, entry.SequenceNum(), entry.FileSequenceNum(), entry.DataFile())
+	return w.addEntry(w.reusedEntry)
+}
+
+type ManifestListWriter struct {
+	version          int
+	out              io.Writer
+	commitSnapshotID int64
+	sequenceNumber   int64
+	writer           *ocf.Encoder
+}
+
+func NewManifestListWriterV1(out io.Writer, snapshotID int64, parentSnapshot *int64) (*ManifestListWriter, error) {
+	m := &ManifestListWriter{
+		version:          1,
+		out:              out,
+		commitSnapshotID: snapshotID,
+		sequenceNumber:   -1,
+	}
+
+	parentSnapshotStr := "null"
+	if parentSnapshot != nil {
+		parentSnapshotStr = strconv.Itoa(int(*parentSnapshot))
+	}
+
+	return m, m.init(map[string][]byte{
+		"format-version":     []byte(strconv.Itoa(m.version)),
+		"snapshot-id":        []byte(strconv.Itoa(int(snapshotID))),
+		"parent-snapshot-id": []byte(parentSnapshotStr),
+	})
+}
+
+func NewManifestListWriterV2(out io.Writer, snapshotID, sequenceNumber int64, parentSnapshot *int64) (*ManifestListWriter, error) {
+	m := &ManifestListWriter{
+		version:          2,
+		out:              out,
+		commitSnapshotID: snapshotID,
+		sequenceNumber:   sequenceNumber,
+	}
+
+	parentSnapshotStr := "null"
+	if parentSnapshot != nil {
+		parentSnapshotStr = strconv.Itoa(int(*parentSnapshot))
+	}
+
+	return m, m.init(map[string][]byte{
+		"format-version":     []byte(strconv.Itoa(m.version)),
+		"snapshot-id":        []byte(strconv.Itoa(int(snapshotID))),
+		"sequence-number":    []byte(strconv.Itoa(int(sequenceNumber))),
+		"parent-snapshot-id": []byte(parentSnapshotStr),
+	})
+}
+
+func (m *ManifestListWriter) init(meta map[string][]byte) error {
+	fileSchema, err := internal.NewManifestFileSchema(m.version)
+	if err != nil {
+		return err
+	}
+
+	enc, err := ocf.NewEncoderWithSchema(fileSchema, m.out,
+		ocf.WithMetadata(meta),
+		ocf.WithCodec(ocf.Deflate))
+	if err != nil {
+		return err
+	}
+
+	m.writer = enc
+	return nil
+}
+
+func (m *ManifestListWriter) Close() error {
+	if m.writer == nil {
+		return nil
+	}
+
+	return m.writer.Close()
+}
+
+func (m *ManifestListWriter) AddManifests(files []ManifestFile) error {
+	if len(files) == 0 {
+		return nil
+	}
+
+	switch m.version {
+	case 1:
+		if slices.ContainsFunc(files, func(f ManifestFile) bool {
+			return f.Version() != 1
+		}) {
+			return fmt.Errorf("%w: ManifestListWriter only supports version 1 manifest files", ErrInvalidArgument)
+		}
+
+		for _, file := range files {
+			if err := m.writer.Encode(file); err != nil {
+				return err
+			}
+		}
+
+	case 2:
+		for _, file := range files {
+			if file.Version() != 2 {
+				return fmt.Errorf("%w: ManifestListWriter only supports version 2 manifest files", ErrInvalidArgument)
+			}
+
+			wrapped := *(file.(*manifestFileV2))
+			if wrapped.SeqNumber == -1 {
+				// if the sequence number is being assigned here,
+				// then the manifest must be created by the current
+				// operation.
+				// to validate this, check the snapshot id matches the current commmit
+				if m.commitSnapshotID != wrapped.AddedSnapshotID {
+					return fmt.Errorf("found unassigned sequence number for a manifest from snapshot %d != %d",
+						m.commitSnapshotID, wrapped.AddedSnapshotID)
+				}
+				wrapped.SeqNumber = m.sequenceNumber
+			}
+
+			if wrapped.MinSeqNumber == -1 {
+				if m.commitSnapshotID != wrapped.AddedSnapshotID {
+					return fmt.Errorf("found unassigned sequence number for a manifest from snapshot: %d", wrapped.AddedSnapshotID)
+				}
+				// if the min sequence number is not determined, then there was no assigned sequence number
+				// for any file written to the wrapped manifest. replace the unassigned sequence number with
+				// the one for this commit
+				wrapped.MinSeqNumber = m.sequenceNumber
+			}
+			if err := m.writer.Encode(wrapped); err != nil {
+				return err
+			}
+		}
+	default:
+		return fmt.Errorf("unsupported manifest version: %d", m.version)
+	}
+
+	return nil
+}
+
+// WriteManifestList writes a list of manifest files to an avro file.
+func WriteManifestList(version int, out io.Writer, snapshotID int64, parentSnapshotID, sequenceNumber *int64, files []ManifestFile) error {
+	var (
+		writer *ManifestListWriter
+		err    error
+	)
+
 	switch version {
 	case 1:
-		key = internal.ManifestEntryV1Key
+		writer, err = NewManifestListWriterV1(out, snapshotID, parentSnapshotID)
 	case 2:
-		key = internal.ManifestEntryV2Key
+		if sequenceNumber == nil {
+			return errors.New("sequence number is required for V2 tables")
+		}
+		writer, err = NewManifestListWriterV2(out, snapshotID, *sequenceNumber, parentSnapshotID)
 	default:
-		return fmt.Errorf("%w: non-recognized version %d", ErrInvalidArgument, version)
+		return fmt.Errorf("unsupported manifest version: %d", version)
+	}
+
+	if err != nil {
+		return err
+	}
+
+	if err = writer.AddManifests(files); err != nil {
+		return err
 	}
 
-	return avroEncode(key, version, entries, out)
+	return writer.Close()
+}
+
+func WriteManifest(
+	out io.Writer,
+	version int,
+	spec PartitionSpec,
+	schema *Schema,
+	snapshotID int64,
+	entries []ManifestEntry,
+) (func(string, int64) (ManifestFile, error), error) {
+	w, err := NewManifestWriter(version, out, spec, schema, snapshotID)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, entry := range entries {
+		if err := w.addEntry(entry); err != nil {
+			return nil, err
+		}
+	}
+
+	return w.ToManifestFile, w.Close()
 }
 
 // ManifestEntryStatus defines constants for the entry status of
@@ -922,6 +1478,19 @@ func (m *manifestEntryV1) FileSequenceNum() *int64 {
 
 func (m *manifestEntryV1) DataFile() DataFile { return m.Data }
 
+func (m *manifestEntryV1) wrap(status ManifestEntryStatus, snapshotID int64, seqNum int64, fileSeqNum *int64, datafile DataFile) ManifestEntry {
+	m.EntryStatus = status
+	m.Snapshot = snapshotID
+	if seqNum > 0 {
+		m.SeqNum = &seqNum
+	}
+
+	m.FileSeqNum = fileSeqNum
+
+	m.Data = datafile
+	return m
+}
+
 // ManifestEntryV2Builder is a helper for building a V2 manifest entry
 // struct which will conform to the ManifestEntry interface.
 type ManifestEntryV2Builder struct {
@@ -1006,6 +1575,20 @@ func (m *manifestEntryV2) FileSequenceNum() *int64 {
 
 func (m *manifestEntryV2) DataFile() DataFile { return m.Data }
 
+func (m *manifestEntryV2) wrap(status ManifestEntryStatus, snapshotID int64, seqNum int64, fileSeqNum *int64, datafile DataFile) ManifestEntry {
+	m.EntryStatus = status
+	if snapshotID > 0 {
+		m.Snapshot = &snapshotID
+	}
+	if seqNum > 0 {
+		m.SeqNum = &seqNum
+	}
+	m.FileSeqNum = fileSeqNum
+
+	m.Data = datafile
+	return m
+}
+
 // DataFileBuilder is a helper for building a data file struct which will
 // conform to the DataFile interface.
 type DataFileBuilder struct {
@@ -1219,6 +1802,7 @@ type ManifestEntry interface {
 	DataFile() DataFile
 
 	inheritSeqNum(manifest ManifestFile)
+	wrap(status ManifestEntryStatus, snapshotID int64, seqNum int64, fileSeqNum *int64, datafile DataFile) ManifestEntry
 }
 
 var PositionalDeleteSchema = NewSchema(0,
diff --git a/schema_conversions.go b/schema_conversions.go
new file mode 100644
index 0000000..05b6879
--- /dev/null
+++ b/schema_conversions.go
@@ -0,0 +1,68 @@
+// 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 iceberg
+
+import (
+	"fmt"
+
+	"github.com/apache/iceberg-go/internal"
+	"github.com/hamba/avro/v2"
+)
+
+func partitionTypeToAvroSchema(t *StructType) (avro.Schema, error) {
+	fields := make([]*avro.Field, len(t.FieldList))
+	for i, f := range t.FieldList {
+		var sc avro.Schema
+		switch typ := f.Type.(type) {
+		case Int32Type:
+			sc = internal.IntSchema
+		case Int64Type:
+			sc = internal.LongSchema
+		case Float32Type:
+			sc = internal.FloatSchema
+		case Float64Type:
+			sc = internal.DoubleSchema
+		case StringType:
+			sc = internal.StringSchema
+		case DateType:
+			sc = internal.DateSchema
+		case TimeType:
+			sc = internal.TimeSchema
+		case TimestampType:
+			sc = internal.TimestampSchema
+		case TimestampTzType:
+			sc = internal.TimestampTzSchema
+		case UUIDType:
+			sc = internal.UUIDSchema
+		case BooleanType:
+			sc = internal.BoolSchema
+		case BinaryType:
+			sc = internal.BinarySchema
+		case FixedType:
+			sc = internal.Must(avro.NewFixedSchema("fixed", "", typ.len, nil))
+		case DecimalType:
+			sc = internal.DecimalSchema(typ.precision, typ.scale)
+		default:
+			return nil, fmt.Errorf("unsupported partition type: %s", f.Type.String())
+		}
+
+		fields[i], _ = avro.NewField(f.Name, sc, internal.WithFieldID(f.ID))
+	}
+
+	return avro.NewRecordSchema("r102", "", fields)
+}
diff --git a/table/metadata.go b/table/metadata.go
index bd4896d..43a348c 100644
--- a/table/metadata.go
+++ b/table/metadata.go
@@ -176,10 +176,13 @@ func MetadataBuilderFromBase(metadata Metadata) (*MetadataBuilder, error) {
 	b.lastPartitionID = metadata.LastPartitionSpecID()
 	b.props = metadata.Properties()
 	b.snapshotList = metadata.Snapshots()
-	b.currentSnapshotID = &metadata.CurrentSnapshot().SnapshotID
 	b.sortOrderList = metadata.SortOrders()
 	b.defaultSortOrderID = metadata.DefaultSortOrder()
 
+	if metadata.CurrentSnapshot() != nil {
+		b.currentSnapshotID = &metadata.CurrentSnapshot().SnapshotID
+	}
+
 	b.refs = make(map[string]SnapshotRef)
 	for name, ref := range metadata.Refs() {
 		b.refs[name] = ref
@@ -262,8 +265,8 @@ func (b *MetadataBuilder) AddSnapshot(snapshot *Snapshot) (*MetadataBuilder, err
 		return nil, fmt.Errorf("can't add snapshot with id %d, already exists", snapshot.SnapshotID)
 	} else if b.formatVersion == 2 &&
 		snapshot.SequenceNumber > 0 &&
-		snapshot.SequenceNumber <= *b.lastSequenceNumber &&
-		snapshot.ParentSnapshotID != nil {
+		snapshot.ParentSnapshotID != nil &&
+		snapshot.SequenceNumber <= *b.lastSequenceNumber {
 		return nil, fmt.Errorf("can't add snapshot with sequence number %d, must be > than last sequence number %d",
 			snapshot.SequenceNumber, b.lastSequenceNumber)
 	}
@@ -490,7 +493,7 @@ func (b *MetadataBuilder) SetSnapshotRef(
 		return nil, fmt.Errorf("can't set snapshot ref %s to unknown snapshot %d: %w", name, snapshotID, err)
 	}
 
-	if refType == MainBranch {
+	if name == MainBranch {
 		b.updates = append(b.updates, NewSetSnapshotRefUpdate(name, snapshotID, refType, maxRefAgeMs, maxSnapshotAgeMs, minSnapshotsToKeep))
 		b.currentSnapshotID = &snapshotID
 		b.snapshotLog = append(b.snapshotLog, SnapshotLogEntry{
@@ -521,6 +524,10 @@ func (b *MetadataBuilder) SetUUID(uuid uuid.UUID) (*MetadataBuilder, error) {
 }
 
 func (b *MetadataBuilder) buildCommonMetadata() *commonMetadata {
+	if b.lastUpdatedMS == 0 {
+		b.lastUpdatedMS = time.Now().UnixMilli()
+	}
+
 	return &commonMetadata{
 		FormatVersion:      b.formatVersion,
 		UUID:               b.uuid,
@@ -609,8 +616,14 @@ func (b *MetadataBuilder) Build() (Metadata, error) {
 		}, nil
 
 	case 2:
+		var lastSequenceNumber int64
+
+		if b.lastSequenceNumber != nil {
+			lastSequenceNumber = *b.lastSequenceNumber
+		}
+
 		return &metadataV2{
-			LastSequenceNumber: *b.lastSequenceNumber,
+			LastSequenceNumber: lastSequenceNumber,
 			commonMetadata:     *common,
 		}, nil
 
diff --git a/utils.go b/utils.go
index 087fb25..7807acb 100644
--- a/utils.go
+++ b/utils.go
@@ -21,14 +21,9 @@ import (
 	"cmp"
 	"fmt"
 	"hash/maphash"
-	"io"
 	"maps"
 	"runtime/debug"
-	"strconv"
 	"strings"
-
-	"github.com/apache/iceberg-go/internal"
-	"github.com/hamba/avro/v2/ocf"
 )
 
 var version string
@@ -217,24 +212,3 @@ func Difference(a, b []string) []string {
 	}
 	return diff
 }
-
-func avroEncode[T any](key string, version int, vals []T, out io.Writer) error {
-	enc, err := ocf.NewEncoderWithSchema(
-		internal.AvroSchemaCache.Get(key),
-		out, ocf.WithMetadata(map[string][]byte{
-			"format-version": []byte(strconv.Itoa(version)),
-		}),
-		ocf.WithCodec(ocf.Deflate),
-	)
-	if err != nil {
-		return err
-	}
-
-	for _, file := range vals {
-		if err := enc.Encode(file); err != nil {
-			return err
-		}
-	}
-
-	return enc.Close()
-}


Test patch:
diff --git a/catalog/glue/glue_test.go b/catalog/glue/glue_test.go
index 4c8aa48..2c6516f 100644
--- a/catalog/glue/glue_test.go
+++ b/catalog/glue/glue_test.go
@@ -20,10 +20,12 @@ package glue
 import (
 	"context"
 	"errors"
+	"fmt"
 	"os"
 	"testing"
 
 	"github.com/apache/iceberg-go/catalog"
+	"github.com/apache/iceberg-go/table"
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/config"
 	"github.com/aws/aws-sdk-go-v2/service/glue"
@@ -81,13 +83,42 @@ func (m *mockGlueClient) UpdateDatabase(ctx context.Context, params *glue.Update
 	return args.Get(0).(*glue.UpdateDatabaseOutput), args.Error(1)
 }
 
-var testIcebergGlueTable = types.Table{
+var testIcebergGlueTable1 = types.Table{
 	Name: aws.String("test_table"),
 	Parameters: map[string]string{
 		tableTypePropsKey:        "ICEBERG",
 		metadataLocationPropsKey: "s3://test-bucket/test_table/metadata/abc123-123.metadata.json",
 	},
 }
+var testIcebergGlueTable2 = types.Table{
+	Name: aws.String("test_table2"),
+	Parameters: map[string]string{
+		tableTypePropsKey:        "ICEBERG",
+		metadataLocationPropsKey: "s3://test-bucket/test_table/metadata/abc456-456.metadata.json",
+	},
+}
+
+var testIcebergGlueTable3 = types.Table{
+	Name: aws.String("test_table3"),
+	Parameters: map[string]string{
+		tableTypePropsKey:        "ICEBERG",
+		metadataLocationPropsKey: "s3://test-bucket/test_table/metadata/abc789-789.metadata.json",
+	},
+}
+var testIcebergGlueTable4 = types.Table{
+	Name: aws.String("test_table4"),
+	Parameters: map[string]string{
+		tableTypePropsKey:        "ICEBERG",
+		metadataLocationPropsKey: "s3://test-bucket/test_table/metadata/abc123-789.metadata.json",
+	},
+}
+var testIcebergGlueTable5 = types.Table{
+	Name: aws.String("test_table5"),
+	Parameters: map[string]string{
+		tableTypePropsKey:        "ICEBERG",
+		metadataLocationPropsKey: "s3://test-bucket/test_table/metadata/abc12345-789.metadata.json",
+	},
+}
 
 var testNonIcebergGlueTable = types.Table{
 	Name: aws.String("other_table"),
@@ -104,7 +135,7 @@ func TestGlueGetTable(t *testing.T) {
 	mockGlueSvc.On("GetTable", mock.Anything, &glue.GetTableInput{
 		DatabaseName: aws.String("test_database"),
 		Name:         aws.String("test_table"),
-	}, mock.Anything).Return(&glue.GetTableOutput{Table: &testIcebergGlueTable}, nil)
+	}, mock.Anything).Return(&glue.GetTableOutput{Table: &testIcebergGlueTable1}, nil)
 
 	glueCatalog := &Catalog{
 		glueSvc: mockGlueSvc,
@@ -123,19 +154,136 @@ func TestGlueListTables(t *testing.T) {
 	mockGlueSvc.On("GetTables", mock.Anything, &glue.GetTablesInput{
 		DatabaseName: aws.String("test_database"),
 	}, mock.Anything).Return(&glue.GetTablesOutput{
-		TableList: []types.Table{testIcebergGlueTable, testNonIcebergGlueTable},
+		TableList: []types.Table{testIcebergGlueTable1, testNonIcebergGlueTable},
 	}, nil).Once()
 
 	glueCatalog := &Catalog{
 		glueSvc: mockGlueSvc,
 	}
 
-	tables, err := glueCatalog.ListTables(context.TODO(), DatabaseIdentifier("test_database"))
-	assert.NoError(err)
-	assert.Len(tables, 1)
-	assert.Equal([]string{"test_database", "test_table"}, tables[0])
+	var lastErr error
+	tbls := make([]table.Identifier, 0)
+	iter := glueCatalog.ListTables(context.TODO(), DatabaseIdentifier("test_database"))
+
+	for tbl, err := range iter {
+		tbls = append(tbls, tbl)
+		if err != nil {
+			lastErr = err
+		}
+	}
+	assert.NoError(lastErr)
+	assert.Len(tbls, 1)
+	assert.Equal([]string{"test_database", "test_table"}, tbls[0])
+}
+
+func TestGlueListTablesPagination(t *testing.T) {
+	assert := require.New(t)
+
+	mockGlueSvc := &mockGlueClient{}
+
+	// First page
+	mockGlueSvc.On("GetTables", mock.Anything, &glue.GetTablesInput{
+		DatabaseName: aws.String("test_database"),
+	}, mock.Anything).Return(&glue.GetTablesOutput{
+		TableList: []types.Table{
+			testIcebergGlueTable1,
+			testIcebergGlueTable2,
+		},
+		NextToken: aws.String("token1"),
+	}, nil).Once()
+
+	// Second page
+	mockGlueSvc.On("GetTables", mock.Anything, &glue.GetTablesInput{
+		DatabaseName: aws.String("test_database"),
+		NextToken:    aws.String("token1"),
+	}, mock.Anything).Return(&glue.GetTablesOutput{
+		TableList: []types.Table{
+			testIcebergGlueTable3,
+			testIcebergGlueTable4,
+		},
+		NextToken: aws.String("token2"),
+	}, nil).Once()
+
+	// Third page
+	mockGlueSvc.On("GetTables", mock.Anything, &glue.GetTablesInput{
+		DatabaseName: aws.String("test_database"),
+		NextToken:    aws.String("token2"),
+	}, mock.Anything).Return(&glue.GetTablesOutput{
+		TableList: []types.Table{
+			testIcebergGlueTable5,
+			testNonIcebergGlueTable,
+		},
+	}, nil).Once()
+
+	glueCatalog := &Catalog{
+		glueSvc: mockGlueSvc,
+	}
+
+	var lastErr error
+	tbls := make([]table.Identifier, 0)
+	iter := glueCatalog.ListTables(context.TODO(), DatabaseIdentifier("test_database"))
+
+	for tbl, err := range iter {
+		tbls = append(tbls, tbl)
+		if err != nil {
+			lastErr = err
+		}
+	}
+
+	assert.NoError(lastErr)
+	assert.Len(tbls, 5) // Only Iceberg tables should be included
+	assert.Equal([]string{"test_database", "test_table"}, tbls[0])
+	assert.Equal([]string{"test_database", "test_table2"}, tbls[1])
+	assert.Equal([]string{"test_database", "test_table3"}, tbls[2])
+	assert.Equal([]string{"test_database", "test_table4"}, tbls[3])
+	assert.Equal([]string{"test_database", "test_table5"}, tbls[4])
+
+	mockGlueSvc.AssertExpectations(t)
 }
 
+func TestGlueListTablesError(t *testing.T) {
+	assert := require.New(t)
+
+	mockGlueSvc := &mockGlueClient{}
+
+	// First page succeeds
+	mockGlueSvc.On("GetTables", mock.Anything, &glue.GetTablesInput{
+		DatabaseName: aws.String("test_database"),
+	}, mock.Anything).Return(&glue.GetTablesOutput{
+		TableList: []types.Table{
+			testIcebergGlueTable1,
+		},
+		NextToken: aws.String("token1"),
+	}, nil).Once()
+
+	mockGlueSvc.On("GetTables", mock.Anything, &glue.GetTablesInput{
+		DatabaseName: aws.String("test_database"),
+		NextToken:    aws.String("token1"),
+	}, mock.Anything).Return(&glue.GetTablesOutput{}, fmt.Errorf("token expired")).Once()
+
+	glueCatalog := &Catalog{
+		glueSvc: mockGlueSvc,
+	}
+
+	var lastErr error
+	tbls := make([]table.Identifier, 0)
+	iter := glueCatalog.ListTables(context.TODO(), DatabaseIdentifier("test_database"))
+
+	for tbl, err := range iter {
+		if err != nil {
+			lastErr = err
+			break
+		}
+		tbls = append(tbls, tbl)
+	}
+
+	assert.Error(lastErr)
+	assert.Contains(lastErr.Error(), "token expired")
+	assert.Len(tbls, 1)
+	assert.Equal([]string{"test_database", "test_table"}, tbls[0])
+
+	mockGlueSvc.AssertExpectations(t)
+}
 func TestGlueListNamespaces(t *testing.T) {
 	assert := require.New(t)
 
@@ -175,7 +323,7 @@ func TestGlueDropTable(t *testing.T) {
 		DatabaseName: aws.String("test_database"),
 		Name:         aws.String("test_table"),
 	}, mock.Anything).Return(&glue.GetTableOutput{
-		Table: &testIcebergGlueTable,
+		Table: &testIcebergGlueTable1,
 	}, nil).Once()
 
 	mockGlueSvc.On("DeleteTable", mock.Anything, &glue.DeleteTableInput{
@@ -248,6 +396,47 @@ func TestGlueDropNamespace(t *testing.T) {
 	assert.NoError(err)
 }
 
+func TestGlueCheckNamespaceExists(t *testing.T) {
+	assert := require.New(t)
+	mockGlueSvc := &mockGlueClient{}
+	mockGlueSvc.On("GetDatabase", mock.Anything, &glue.GetDatabaseInput{
+		Name: aws.String("test_namespace"),
+	}, mock.Anything).Return(&glue.GetDatabaseOutput{
+		Database: &types.Database{
+			Name: aws.String("test_namespace"),
+			Parameters: map[string]string{
+				"database_type": "ICEBERG",
+			},
+		},
+	}, nil).Once()
+	glueCatalog := &Catalog{
+		glueSvc: mockGlueSvc,
+	}
+	exists, err := glueCatalog.CheckNamespaceExists(context.TODO(), DatabaseIdentifier("test_namespace"))
+	assert.NoError(err)
+	assert.True(exists)
+
+}
+
+func TestGlueCheckNamespaceNotExists(t *testing.T) {
+	assert := require.New(t)
+	mockGlueSvc := &mockGlueClient{}
+
+	mockGlueSvc.On("GetDatabase", mock.Anything, &glue.GetDatabaseInput{
+		Name: aws.String("nonexistent_namespace"),
+	}, mock.Anything).Return(&glue.GetDatabaseOutput{},
+		&types.EntityNotFoundException{Message: aws.String("Database not found")}).Once()
+
+	glueCatalog := &Catalog{
+		glueSvc: mockGlueSvc,
+	}
+
+	exists, err := glueCatalog.CheckNamespaceExists(context.TODO(), DatabaseIdentifier("nonexistent_namespace"))
+	assert.Error(err)
+	assert.False(exists)
+	assert.ErrorContains(err, "Database not found")
+}
+
 func TestGlueUpdateNamespaceProperties(t *testing.T) {
 	tests := []struct {
 		name        string
@@ -534,9 +723,18 @@ func TestGlueListTablesIntegration(t *testing.T) {
 
 	catalog := NewCatalog(WithAwsConfig(awscfg))
 
-	tables, err := catalog.ListTables(context.TODO(), DatabaseIdentifier(os.Getenv("TEST_DATABASE_NAME")))
-	assert.NoError(err)
-	assert.Equal([]string{os.Getenv("TEST_DATABASE_NAME"), os.Getenv("TEST_TABLE_NAME")}, tables[1])
+	iter := catalog.ListTables(context.TODO(), DatabaseIdentifier(os.Getenv("TEST_DATABASE_NAME")))
+	var lastErr error
+	tbls := make([]table.Identifier, 0)
+	for tbl, err := range iter {
+		tbls = append(tbls, tbl)
+		if err != nil {
+			lastErr = err
+		}
+	}
+
+	assert.NoError(lastErr)
+	assert.Equal([]string{os.Getenv("TEST_DATABASE_NAME"), os.Getenv("TEST_TABLE_NAME")}, tbls[1])
 }
 
 func TestGlueLoadTableIntegration(t *testing.T) {
diff --git a/catalog/rest/rest_integration_test.go b/catalog/rest/rest_integration_test.go
new file mode 100644
index 0000000..b17725a
--- /dev/null
+++ b/catalog/rest/rest_integration_test.go
@@ -0,0 +1,180 @@
+// 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.
+
+//go:build integration
+
+package rest_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/catalog"
+	"github.com/apache/iceberg-go/catalog/rest"
+	"github.com/apache/iceberg-go/io"
+	"github.com/stretchr/testify/suite"
+)
+
+type RestIntegrationSuite struct {
+	suite.Suite
+
+	ctx context.Context
+	cat *rest.Catalog
+}
+
+const TestNamespaceIdent = "TEST NS"
+
+func (s *RestIntegrationSuite) loadCatalog(ctx context.Context) *rest.Catalog {
+	cat, err := catalog.Load(ctx, "local", iceberg.Properties{
+		"type":               "rest",
+		"uri":                "http://localhost:8181",
+		io.S3AccessKeyID:     "admin",
+		io.S3SecretAccessKey: "password",
+	})
+	s.Require().NoError(err)
+	s.Require().IsType(&rest.Catalog{}, cat)
+	return cat.(*rest.Catalog)
+}
+
+func (s *RestIntegrationSuite) SetupTest() {
+	s.ctx = context.Background()
+	s.cat = s.loadCatalog(s.ctx)
+}
+
+var (
+	tableSchemaNested = iceberg.NewSchemaWithIdentifiers(1,
+		[]int{1},
+		iceberg.NestedField{
+			ID: 1, Name: "foo", Type: iceberg.PrimitiveTypes.String, Required: false},
+		iceberg.NestedField{
+			ID: 2, Name: "bar", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+		iceberg.NestedField{
+			ID: 3, Name: "baz", Type: iceberg.PrimitiveTypes.Bool, Required: false},
+		iceberg.NestedField{
+			ID: 4, Name: "qux", Required: true, Type: &iceberg.ListType{
+				ElementID: 5, Element: iceberg.PrimitiveTypes.String, ElementRequired: true}},
+		iceberg.NestedField{
+			ID: 6, Name: "quux",
+			Type: &iceberg.MapType{
+				KeyID:   7,
+				KeyType: iceberg.PrimitiveTypes.String,
+				ValueID: 8,
+				ValueType: &iceberg.MapType{
+					KeyID:         9,
+					KeyType:       iceberg.PrimitiveTypes.String,
+					ValueID:       10,
+					ValueType:     iceberg.PrimitiveTypes.Int32,
+					ValueRequired: true,
+				},
+				ValueRequired: true,
+			},
+			Required: true},
+		iceberg.NestedField{
+			ID: 11, Name: "location", Type: &iceberg.ListType{
+				ElementID: 12, Element: &iceberg.StructType{
+					FieldList: []iceberg.NestedField{
+						{ID: 13, Name: "latitude", Type: iceberg.PrimitiveTypes.Float32, Required: false},
+						{ID: 14, Name: "longitude", Type: iceberg.PrimitiveTypes.Float32, Required: false},
+					},
+				},
+				ElementRequired: true},
+			Required: true},
+		iceberg.NestedField{
+			ID:   15,
+			Name: "person",
+			Type: &iceberg.StructType{
+				FieldList: []iceberg.NestedField{
+					{ID: 16, Name: "name", Type: iceberg.PrimitiveTypes.String, Required: false},
+					{ID: 17, Name: "age", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+				},
+			},
+			Required: false,
+		},
+	)
+)
+
+func (s *RestIntegrationSuite) ensureNamespace() {
+	exists, err := s.cat.CheckNamespaceExists(s.ctx, catalog.ToIdentifier(TestNamespaceIdent))
+	s.Require().NoError(err)
+	if exists {
+		s.Require().NoError(s.cat.DropNamespace(s.ctx, catalog.ToIdentifier(TestNamespaceIdent)))
+	}
+
+	s.NoError(s.cat.CreateNamespace(s.ctx, catalog.ToIdentifier(TestNamespaceIdent),
+		iceberg.Properties{"foo": "bar", "prop": "yes"}))
+}
+
+func (s *RestIntegrationSuite) TestCreateNamespace() {
+	exists, err := s.cat.CheckNamespaceExists(s.ctx, catalog.ToIdentifier(TestNamespaceIdent))
+	s.Require().NoError(err)
+	if exists {
+		s.Require().NoError(s.cat.DropNamespace(s.ctx, catalog.ToIdentifier(TestNamespaceIdent)))
+	}
+
+	s.NoError(s.cat.CreateNamespace(s.ctx, catalog.ToIdentifier(TestNamespaceIdent), nil))
+	s.NoError(s.cat.DropNamespace(s.ctx, catalog.ToIdentifier(TestNamespaceIdent)))
+}
+
+func (s *RestIntegrationSuite) TestLoadNamespaceProps() {
+	s.ensureNamespace()
+
+	props, err := s.cat.LoadNamespaceProperties(s.ctx, catalog.ToIdentifier(TestNamespaceIdent))
+	s.Require().NoError(err)
+	s.Require().NotNil(props)
+	s.Equal("bar", props["foo"])
+	s.Equal("yes", props["prop"])
+}
+
+func (s *RestIntegrationSuite) TestUpdateNamespaceProps() {
+	s.ensureNamespace()
+
+	summary, err := s.cat.UpdateNamespaceProperties(s.ctx, catalog.ToIdentifier(TestNamespaceIdent),
+		[]string{"abc"}, iceberg.Properties{"prop": "no"})
+	s.Require().NoError(err)
+	s.Equal(catalog.PropertiesUpdateSummary{
+		Removed: []string{},
+		Updated: []string{"prop"},
+		Missing: []string{"abc"},
+	}, summary)
+}
+
+func (s *RestIntegrationSuite) TestCreateTable() {
+	s.ensureNamespace()
+
+	const location = "s3://warehouse/iceberg"
+
+	tbl, err := s.cat.CreateTable(s.ctx,
+		catalog.ToIdentifier(TestNamespaceIdent, "test-table"),
+		tableSchemaSimple, catalog.WithProperties(iceberg.Properties{"foobar": "baz"}),
+		catalog.WithLocation(location))
+	s.Require().NoError(err)
+	s.Require().NotNil(tbl)
+
+	s.Equal(location, tbl.Location())
+	s.Equal("baz", tbl.Properties()["foobar"])
+
+	exists, err := s.cat.CheckTableExists(s.ctx, catalog.ToIdentifier(TestNamespaceIdent, "test-table"))
+	s.Require().NoError(err)
+	s.True(exists)
+
+	s.Require().NoError(s.cat.DropTable(s.ctx, catalog.ToIdentifier(TestNamespaceIdent, "test-table")))
+}
+
+func TestRestIntegration(t *testing.T) {
+	suite.Run(t, new(RestIntegrationSuite))
+}
diff --git a/catalog/rest/rest_test.go b/catalog/rest/rest_test.go
index 32d96ad..6962f60 100644
--- a/catalog/rest/rest_test.go
+++ b/catalog/rest/rest_test.go
@@ -37,8 +37,9 @@ import (
 )
 
 const (
-	TestCreds = "client:secret"
-	TestToken = "some_jwt_token"
+	TestCreds       = "client:secret"
+	TestToken       = "some_jwt_token"
+	defaultPageSize = 20
 )
 
 var (
@@ -234,13 +235,17 @@ func (r *RestCatalogSuite) TestToken401() {
 
 func (r *RestCatalogSuite) TestListTables200() {
 	namespace := "examples"
+	customPageSize := 100
 	r.mux.HandleFunc("/v1/namespaces/"+namespace+"/tables", func(w http.ResponseWriter, req *http.Request) {
 		r.Require().Equal(http.MethodGet, req.Method)
 
 		for k, v := range TestHeaders {
 			r.Equal(v, req.Header.Values(k))
 		}
-
+		pageToken := req.URL.Query().Get("page-token")
+		pageSize := req.URL.Query().Get("page-size")
+		r.Equal("", pageToken)
+		r.Equal(strconv.Itoa(customPageSize), pageSize)
 		json.NewEncoder(w).Encode(map[string]any{
 			"identifiers": []any{
 				map[string]any{
@@ -254,9 +259,22 @@ func (r *RestCatalogSuite) TestListTables200() {
 	cat, err := rest.NewCatalog(context.Background(), "rest", r.srv.URL, rest.WithOAuthToken(TestToken))
 	r.Require().NoError(err)
 
-	tables, err := cat.ListTables(context.Background(), catalog.ToIdentifier(namespace))
-	r.Require().NoError(err)
-	r.Equal([]table.Identifier{{"examples", "fooshare"}}, tables)
+	ctx := cat.SetPageSize(context.Background(), customPageSize)
+
+	var lastErr error
+	tbls := make([]table.Identifier, 0)
+
+	iter := cat.ListTables(ctx, catalog.ToIdentifier(namespace))
+
+	for tbl, err := range iter {
+		tbls = append(tbls, tbl)
+		if err != nil {
+			lastErr = err
+			r.FailNow("unexpected error:", err)
+		}
+	}
+	r.Require().NoError(lastErr)
+	r.Equal([]table.Identifier{{"examples", "fooshare"}}, tbls)
 }
 
 func (r *RestCatalogSuite) TestListTablesPrefixed200() {
@@ -289,6 +307,10 @@ func (r *RestCatalogSuite) TestListTablesPrefixed200() {
 		for k, v := range TestHeaders {
 			r.Equal(v, req.Header.Values(k))
 		}
+		pageToken := req.URL.Query().Get("page-token")
+		pageSize := req.URL.Query().Get("page-size")
+		r.Equal("", pageToken)
+		r.Equal(strconv.Itoa(defaultPageSize), pageSize)
 
 		json.NewEncoder(w).Encode(map[string]any{
 			"identifiers": []any{
@@ -309,11 +331,172 @@ func (r *RestCatalogSuite) TestListTablesPrefixed200() {
 	r.NotNil(cat)
 	r.Equal(r.configVals.Get("warehouse"), "s3://some-bucket")
 
-	tables, err := cat.ListTables(context.Background(), catalog.ToIdentifier(namespace))
+	var lastErr error
+	tbls := make([]table.Identifier, 0)
+
+	iter := cat.ListTables(context.Background(), catalog.ToIdentifier(namespace))
+
+	for tbl, err := range iter {
+		tbls = append(tbls, tbl)
+		if err != nil {
+			lastErr = err
+			r.FailNow("unexpected error:", err)
+		}
+	}
+	r.Require().NoError(lastErr)
+	r.Equal([]table.Identifier{{"examples", "fooshare"}}, tbls)
+}
+func (r *RestCatalogSuite) TestListTablesPagination() {
+	defaultPageSize := 20
+	namespace := "accounting"
+	r.mux.HandleFunc("/v1/namespaces/"+namespace+"/tables", func(w http.ResponseWriter, req *http.Request) {
+		r.Require().Equal(http.MethodGet, req.Method)
+
+		for k, v := range TestHeaders {
+			r.Equal(v, req.Header.Values(k))
+		}
+
+		pageToken := req.URL.Query().Get("page-token")
+		pageSize := req.URL.Query().Get("page-size")
+		r.Equal(strconv.Itoa(defaultPageSize), pageSize)
+
+		var response map[string]any
+		if pageToken == "" {
+			response = map[string]any{
+				"identifiers": []any{
+					map[string]any{
+						"namespace": []string{namespace},
+						"name":      "paid1",
+					},
+					map[string]any{
+						"namespace": []string{namespace},
+						"name":      "paid2",
+					},
+				},
+				"next-page-token": "token1",
+			}
+		} else if pageToken == "token1" {
+			r.Equal("token1", pageToken)
+			response = map[string]any{
+				"identifiers": []any{
+					map[string]any{
+						"namespace": []string{namespace},
+						"name":      "pending1",
+					},
+					map[string]any{
+						"namespace": []string{namespace},
+						"name":      "pending2",
+					},
+				},
+				"next-page-token": "token2",
+			}
+		} else {
+			r.Equal("token2", pageToken)
+			response = map[string]any{
+				"identifiers": []any{
+					map[string]any{
+						"namespace": []string{namespace},
+						"name":      "owned",
+					},
+				},
+			}
+		}
+
+		json.NewEncoder(w).Encode(response)
+	})
+
+	cat, err := rest.NewCatalog(context.Background(), "rest", r.srv.URL, rest.WithOAuthToken(TestToken))
 	r.Require().NoError(err)
-	r.Equal([]table.Identifier{{"examples", "fooshare"}}, tables)
+
+	var lastErr error
+	tbls := make([]table.Identifier, 0)
+	iter := cat.ListTables(context.Background(), catalog.ToIdentifier(namespace))
+	for tbl, err := range iter {
+		tbls = append(tbls, tbl)
+		if err != nil {
+			lastErr = err
+			r.FailNow("unexpected error:", err)
+		}
+	}
+
+	r.Equal([]table.Identifier{
+		{"accounting", "paid1"},
+		{"accounting", "paid2"},
+		{"accounting", "pending1"},
+		{"accounting", "pending2"},
+		{"accounting", "owned"},
+	}, tbls)
+	r.Require().NoError(lastErr)
 }
+func (r *RestCatalogSuite) TestListTablesPaginationErrorOnSubsequentPage() {
+	namespace := "accounting"
+	r.mux.HandleFunc("/v1/namespaces/"+namespace+"/tables", func(w http.ResponseWriter, req *http.Request) {
+		r.Require().Equal(http.MethodGet, req.Method)
+
+		for k, v := range TestHeaders {
+			r.Equal(v, req.Header.Values(k))
+		}
+
+		pageToken := req.URL.Query().Get("page-token")
+
+		// First page succeeds
+		if pageToken == "" {
+			json.NewEncoder(w).Encode(map[string]any{
+				"identifiers": []any{
+					map[string]any{
+						"namespace": []string{namespace},
+						"name":      "paid1",
+					},
+					map[string]any{
+						"namespace": []string{namespace},
+						"name":      "paid2",
+					},
+				},
+				"next-page-token": "token1",
+			})
+			return
+		}
+
+		// Second page fails with an error
+		if pageToken == "token1" {
+			w.WriteHeader(http.StatusNotFound)
+			json.NewEncoder(w).Encode(map[string]any{
+				"error": map[string]any{
+					"message": "Token expired or invalid",
+					"type":    "NoSuchPageTokenException",
+					"code":    404,
+				},
+			})
+			return
+		}
+
+		r.FailNow("unexpected page token:", pageToken)
+	})
+
+	cat, err := rest.NewCatalog(context.Background(), "rest", r.srv.URL, rest.WithOAuthToken(TestToken))
+	r.Require().NoError(err)
+
+	tbls := make([]table.Identifier, 0)
+	var lastErr error
+	iter := cat.ListTables(context.Background(), catalog.ToIdentifier(namespace))
+	for tbl, err := range iter {
+		if err != nil {
+			lastErr = err
+			break
+		}
+		tbls = append(tbls, tbl)
+	}
 
+	// Check that we got the views from the first page
+	r.Equal([]table.Identifier{
+		{"accounting", "paid1"},
+		{"accounting", "paid2"},
+	}, tbls)
+
+	// Check that we got the error from the second page
+	r.Error(lastErr)
+	r.ErrorContains(lastErr, "Token expired or invalid")
+}
 func (r *RestCatalogSuite) TestListTables404() {
 	namespace := "examples"
 	r.mux.HandleFunc("/v1/namespaces/"+namespace+"/tables", func(w http.ResponseWriter, req *http.Request) {
@@ -323,6 +506,10 @@ func (r *RestCatalogSuite) TestListTables404() {
 			r.Equal(v, req.Header.Values(k))
 		}
 
+		pageToken := req.URL.Query().Get("page-token")
+		pageSize := req.URL.Query().Get("page-size")
+		r.Equal("", pageToken)
+		r.Equal(strconv.Itoa(defaultPageSize), pageSize)
 		w.WriteHeader(http.StatusNotFound)
 		json.NewEncoder(w).Encode(map[string]any{
 			"error": map[string]any{
@@ -336,9 +523,19 @@ func (r *RestCatalogSuite) TestListTables404() {
 	cat, err := rest.NewCatalog(context.Background(), "rest", r.srv.URL, rest.WithOAuthToken(TestToken))
 	r.Require().NoError(err)
 
-	_, err = cat.ListTables(context.Background(), catalog.ToIdentifier(namespace))
-	r.ErrorIs(err, catalog.ErrNoSuchNamespace)
-	r.ErrorContains(err, "Namespace does not exist: personal in warehouse 8bcb0838-50fc-472d-9ddb-8feb89ef5f1e")
+	var lastErr error
+	tbls := make([]table.Identifier, 0)
+
+	iter := cat.ListTables(context.Background(), catalog.ToIdentifier(namespace))
+
+	for tbl, err := range iter {
+		tbls = append(tbls, tbl)
+		if err != nil {
+			lastErr = err
+		}
+	}
+	r.ErrorIs(lastErr, catalog.ErrNoSuchNamespace)
+	r.ErrorContains(lastErr, "Namespace does not exist")
 }
 
 func (r *RestCatalogSuite) TestListNamespaces200() {
@@ -1018,20 +1215,20 @@ func (r *RestCatalogSuite) TestRenameTable200() {
 		}
 
 		var payload struct {
-			From struct {
+			Source struct {
 				Namespace []string `json:"namespace"`
 				Name      string   `json:"name"`
-			} `json:"from"`
-			To struct {
+			} `json:"source"`
+			Destination struct {
 				Namespace []string `json:"namespace"`
 				Name      string   `json:"name"`
-			} `json:"to"`
+			} `json:"destination"`
 		}
 		r.NoError(json.NewDecoder(req.Body).Decode(&payload))
-		r.Equal([]string{"fokko"}, payload.From.Namespace)
-		r.Equal("source", payload.From.Name)
-		r.Equal([]string{"fokko"}, payload.To.Namespace)
-		r.Equal("destination", payload.To.Name)
+		r.Equal([]string{"fokko"}, payload.Source.Namespace)
+		r.Equal("source", payload.Source.Name)
+		r.Equal([]string{"fokko"}, payload.Destination.Namespace)
+		r.Equal("destination", payload.Destination.Name)
 
 		w.WriteHeader(http.StatusOK)
 	})
diff --git a/catalog/sql/sql_test.go b/catalog/sql/sql_test.go
index 2cc4c61..c79a5e0 100644
--- a/catalog/sql/sql_test.go
+++ b/catalog/sql/sql_test.go
@@ -22,6 +22,7 @@ import (
 	"database/sql"
 	"fmt"
 	"maps"
+	"math/rand/v2"
 	"os"
 	"path/filepath"
 	"strings"
@@ -34,7 +35,6 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/suite"
 	"github.com/uptrace/bun/driver/sqliteshim"
-	"golang.org/x/exp/rand"
 )
 
 var (
@@ -110,7 +110,7 @@ func randomString(n int) string {
 	var b strings.Builder
 	b.Grow(n)
 	for range n {
-		b.WriteByte(letters[rand.Intn(len(letters))])
+		b.WriteByte(letters[rand.IntN(len(letters))])
 	}
 	return b.String()
 }
@@ -486,18 +486,41 @@ func (s *SqliteCatalogTestSuite) TestListTables() {
 		_, err = cat.CreateTable(ctx, tbl2, tableSchemaNested)
 		s.Require().NoError(err)
 
-		tables, err := cat.ListTables(ctx, ns1)
-		s.Require().NoError(err)
-		s.Len(tables, 1)
-		s.Equal(tbl1, tables[0])
+		var lastErr error
+		tables := make([]table.Identifier, 0)
+		iter := cat.ListTables(ctx, ns1)
+		for tbl, err := range iter {
+			tables = append(tables, tbl)
+			if err != nil {
+				lastErr = err
+			}
+		}
 
-		tables, err = cat.ListTables(ctx, ns2)
-		s.Require().NoError(err)
+		s.Require().NoError(lastErr)
 		s.Len(tables, 1)
-		s.Equal(tbl2, tables[0])
+		s.Equal(tbl1, tables[0])
 
-		_, err = cat.ListTables(ctx, table.Identifier{"does_not_exist"})
-		s.ErrorIs(err, catalog.ErrNoSuchNamespace)
+		tables2 := make([]table.Identifier, 0)
+		iter = cat.ListTables(ctx, ns2)
+		for tbl, err := range iter {
+			tables2 = append(tables2, tbl)
+			if err != nil {
+				lastErr = err
+			}
+		}
+		s.Require().NoError(lastErr)
+		s.Len(tables2, 1)
+		s.Equal(tbl2, tables2[0])
+
+		table3 := make([]table.Identifier, 0)
+		iter = cat.ListTables(ctx, table.Identifier{"does_not_exist"})
+		for tbl, err := range iter {
+			table3 = append(table3, tbl)
+			if err != nil {
+				lastErr = err
+			}
+		}
+		s.ErrorIs(lastErr, catalog.ErrNoSuchNamespace)
 	}
 }
 
diff --git a/manifest_test.go b/manifest_test.go
index 0e6e04c..bf3515d 100644
--- a/manifest_test.go
+++ b/manifest_test.go
@@ -76,7 +76,7 @@ var (
 				Content:          EntryContentEqDeletes,
 				Path:             "/home/iceberg/warehouse/nyc/taxis_partitioned/data/VendorID=null/00000-633-d8a4223e-dc97-45a1-86e1-adaba6e8abd7-00001.parquet",
 				Format:           ParquetFile,
-				PartitionData:    map[string]any{"VendorID": int(1), "tpep_pickup_datetime": time.Unix(1925, 0)},
+				PartitionData:    map[string]any{"VendorID": int(1), "tpep_pickup_datetime": time.Unix(1925, 0).UnixMicro()},
 				RecordCount:      19513,
 				FileSize:         388872,
 				BlockSizeInBytes: 67108864,
@@ -195,7 +195,7 @@ var (
 			Data: &dataFile{
 				Path:             "/home/iceberg/warehouse/nyc/taxis_partitioned/data/VendorID=1/00000-633-d8a4223e-dc97-45a1-86e1-adaba6e8abd7-00002.parquet",
 				Format:           ParquetFile,
-				PartitionData:    map[string]any{"VendorID": int(1), "tpep_pickup_datetime": time.Unix(1925, 0)},
+				PartitionData:    map[string]any{"VendorID": int(1), "tpep_pickup_datetime": time.Unix(1925, 0).UnixMicro()},
 				RecordCount:      95050,
 				FileSize:         1265950,
 				BlockSizeInBytes: 67108864,
@@ -365,6 +365,28 @@ var (
 			},
 		},
 	}
+
+	testSchema = NewSchema(0,
+		NestedField{ID: 1, Name: "VendorID", Type: PrimitiveTypes.Int32, Required: true},
+		NestedField{ID: 2, Name: "tpep_pickup_datetime", Type: PrimitiveTypes.Timestamp, Required: true},
+		NestedField{ID: 3, Name: "tpep_dropoff_datetime", Type: PrimitiveTypes.Timestamp, Required: true},
+		NestedField{ID: 4, Name: "passenger_count", Type: PrimitiveTypes.Int64, Required: false},
+		NestedField{ID: 5, Name: "trip_distance", Type: PrimitiveTypes.Float64, Required: true},
+		NestedField{ID: 6, Name: "RatecodeID", Type: PrimitiveTypes.Int64, Required: false},
+		NestedField{ID: 7, Name: "store_and_fwd_flag", Type: PrimitiveTypes.String, Required: false},
+		NestedField{ID: 8, Name: "PULocationID", Type: PrimitiveTypes.Int32, Required: false},
+		NestedField{ID: 9, Name: "DOLocationID", Type: PrimitiveTypes.Int32, Required: false},
+		NestedField{ID: 10, Name: "payment_type", Type: PrimitiveTypes.Int64, Required: true},
+		NestedField{ID: 11, Name: "fare_amount", Type: PrimitiveTypes.Float64, Required: true},
+		NestedField{ID: 12, Name: "extra", Type: PrimitiveTypes.Float64, Required: false},
+		NestedField{ID: 13, Name: "mta_tax", Type: PrimitiveTypes.Float64, Required: false},
+		NestedField{ID: 14, Name: "tip_amount", Type: PrimitiveTypes.Float64, Required: false},
+		NestedField{ID: 15, Name: "tolls_amount", Type: PrimitiveTypes.Float64, Required: false},
+		NestedField{ID: 16, Name: "improvement_surcharge", Type: PrimitiveTypes.Float64, Required: false},
+		NestedField{ID: 17, Name: "total_amount", Type: PrimitiveTypes.Float64, Required: true},
+		NestedField{ID: 18, Name: "congestion_surcharge", Type: PrimitiveTypes.Float64, Required: false},
+		NestedField{ID: 19, Name: "VendorID", Type: PrimitiveTypes.Int32, Required: false},
+	)
 )
 
 type ManifestTestSuite struct {
@@ -378,8 +400,9 @@ type ManifestTestSuite struct {
 }
 
 func (m *ManifestTestSuite) writeManifestList() {
-	m.Require().NoError(WriteManifestList(&m.v1ManifestList, manifestFileRecordsV1))
-	m.Require().NoError(WriteManifestList(&m.v2ManifestList, manifestFileRecordsV2))
+	m.Require().NoError(WriteManifestList(1, &m.v1ManifestList, snapshotID, nil, nil, manifestFileRecordsV1))
+	unassignedSequenceNum := int64(-1)
+	m.Require().NoError(WriteManifestList(2, &m.v2ManifestList, snapshotID, nil, &unassignedSequenceNum, manifestFileRecordsV2))
 }
 
 func (m *ManifestTestSuite) writeManifestEntries() {
@@ -393,8 +416,21 @@ func (m *ManifestTestSuite) writeManifestEntries() {
 		manifestEntryV2Recs[i] = rec
 	}
 
-	m.Require().NoError(writeManifestEntries(&m.v1ManifestEntries, manifestEntryV1Recs, 1))
-	m.Require().NoError(writeManifestEntries(&m.v2ManifestEntries, manifestEntryV2Recs, 2))
+	var partitionSpec = NewPartitionSpecID(1,
+		PartitionField{FieldID: 1000, SourceID: 1, Name: "VendorID", Transform: IdentityTransform{}},
+		PartitionField{FieldID: 1001, SourceID: 2, Name: "tpep_pickup_datetime", Transform: IdentityTransform{}})
+
+	fn, err := WriteManifest(&m.v1ManifestEntries, 1, partitionSpec, testSchema, entrySnapshotID, manifestEntryV1Recs)
+	m.Require().NoError(err)
+
+	_, err = fn("/home/iceberg/warehouse/nyc/taxis_partitioned/metadata/0125c686-8aa6-4502-bdcc-b6d17ca41a3b-m0.avro", int64(m.v1ManifestList.Len()))
+	m.Require().NoError(err)
+
+	fn, err = WriteManifest(&m.v2ManifestEntries, 2, partitionSpec, testSchema, entrySnapshotID, manifestEntryV2Recs)
+	m.Require().NoError(err)
+
+	_, err = fn("/home/iceberg/warehouse/nyc/taxis_partitioned/metadata/0125c686-8aa6-4502-bdcc-b6d17ca41a3b-m0.avro", int64(m.v1ManifestList.Len()))
+	m.Require().NoError(err)
 }
 
 func (m *ManifestTestSuite) SetupSuite() {
@@ -747,7 +783,7 @@ func (m *ManifestTestSuite) TestManifestEntryBuilder() {
 		EntryContentData,
 		"sample.parquet",
 		ParquetFile,
-		map[string]any{"int": int(1), "datetime": time.Unix(1925, 0)},
+		map[string]any{"int": int(1), "datetime": time.Unix(1925, 0).UnixMicro()},
 		1,
 		2,
 	)