Repository: apache/iceberg-go
Pull request #146 (2024-09-09 18:58:19)

Impl rest catalog + table updates & requirements
Hi @zeroshade 

I think it's really cool that you are working on a golang-iceberg implementation, and I would like to contribute if I can.
I have tried to finish the rest catalog implementation, and then I have added table updates and requirements, using a new MetadataBuilder struct that can simplify updating metadata. Like the existing implementation, I have tried to stay close to the pyIceberg implementation. 

I thought this could be a good starting point for also implementing transactions and table updates. I would love to get your input and change/adjust the implementation accordingly.

jwtryg: Hi @zeroshade 

I sincerely apologise for not getting back to you sooner, I suddenly found myself with very little time on my hands.
Once more, I thank you for your feedback - I will make sure to address all of your comments above this week!
zeroshade: @jwtryg any updates?
jwtryg: @zeroshade thank you for your patience :)

I have implemented your feedback and made some other changes - specifically, being more mindful of what should be exported, and then undoing the manic nil-pointer checking I somehow got myself into.
zeroshade: My apologies for the long delay here, could you resolve the conflicts? I should be able to give this a new pass of review in the next day or so.

Patch:
diff --git a/README.md b/README.md
index 58dc718..1c40aa3 100644
--- a/README.md
+++ b/README.md
@@ -25,7 +25,7 @@
 
 ### Prerequisites
 
-* Go 1.21 or later
+* Go 1.23 or later
 
 ### Build
 
@@ -62,18 +62,18 @@ $ cd iceberg-go/cmd/iceberg && go build .
 
 | Operation                | REST | Hive | DynamoDB | Glue |
 | :----------------------- | :--: | :--: | :------: | :--: |
-| Load Table               |      |      |          |  X   |
-| List Tables              |      |      |          |  X   |
-| Create Table             |      |      |          |      |
+| Load Table               |  X   |      |          |  X   |
+| List Tables              |  X   |      |          |  X   |
+| Create Table             |      |      |          |  X   |
 | Update Current Snapshot  |      |      |          |      |
 | Create New Snapshot      |      |      |          |      |
-| Rename Table             |      |      |          |      |
-| Drop Table               |      |      |          |      |
+| Rename Table             |      |      |          |  X   |
+| Drop Table               |      |      |          |  X   |
 | Alter Table              |      |      |          |      |
 | Set Table Properties     |      |      |          |      |
-| Create Namespace         |      |      |          |      |
-| Drop Namespace           |      |      |          |      |
-| Set Namespace Properties |      |      |          |      |
+| Create Namespace         |  X   |      |          |  X   |
+| Drop Namespace           |  X   |      |          |  X   |
+| Set Namespace Properties |  X   |      |          |  X   |
 
 ### Read/Write Data Support
 
@@ -83,4 +83,4 @@ $ cd iceberg-go/cmd/iceberg && go build .
 
 # Get in Touch
 
-- [Iceberg community](https://iceberg.apache.org/community/)
\ No newline at end of file
+- [Iceberg community](https://iceberg.apache.org/community/)
diff --git a/catalog/catalog.go b/catalog/catalog.go
index 65da7e5..9b5776d 100644
--- a/catalog/catalog.go
+++ b/catalog/catalog.go
@@ -45,6 +45,7 @@ var (
 	ErrNoSuchTable            = errors.New("table does not exist")
 	ErrNoSuchNamespace        = errors.New("namespace does not exist")
 	ErrNamespaceAlreadyExists = errors.New("namespace already exists")
+	ErrTableAlreadyExists     = errors.New("table already exists")
 )
 
 // WithAwsConfig sets the AWS configuration for the catalog.
@@ -156,7 +157,7 @@ type Catalog interface {
 	ListTables(ctx context.Context, namespace table.Identifier) ([]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
+	// DropTable tells the catalog to drop the table entirely.
 	DropTable(ctx context.Context, identifier table.Identifier) error
 	// RenameTable tells the catalog to rename a given table by the identifiers
 	// provided, and then loads and returns the destination table
diff --git a/catalog/rest.go b/catalog/rest.go
index ef9c332..47be1eb 100644
--- a/catalog/rest.go
+++ b/catalog/rest.go
@@ -84,6 +84,86 @@ func (e errorResponse) Error() string {
 	return e.Type + ": " + e.Message
 }
 
+type identifier struct {
+	Namespace []string `json:"namespace"`
+	Name      string   `json:"name"`
+}
+
+type commitTableResponse struct {
+	MetadataLoc string          `json:"metadata-location"`
+	RawMetadata json.RawMessage `json:"metadata"`
+	Metadata    table.Metadata  `json:"-"`
+}
+
+func (t *commitTableResponse) UnmarshalJSON(b []byte) (err error) {
+	type Alias commitTableResponse
+	if err = json.Unmarshal(b, (*Alias)(t)); err != nil {
+		return err
+	}
+
+	t.Metadata, err = table.ParseMetadataBytes(t.RawMetadata)
+	return
+}
+
+type loadTableResponse struct {
+	MetadataLoc string             `json:"metadata-location"`
+	RawMetadata json.RawMessage    `json:"metadata"`
+	Config      iceberg.Properties `json:"config"`
+	Metadata    table.Metadata     `json:"-"`
+}
+
+func (t *loadTableResponse) UnmarshalJSON(b []byte) (err error) {
+	type Alias loadTableResponse
+	if err = json.Unmarshal(b, (*Alias)(t)); err != nil {
+		return err
+	}
+
+	t.Metadata, err = table.ParseMetadataBytes(t.RawMetadata)
+	return
+}
+
+type createTableOption func(*createTableRequest)
+
+func WithLocation(loc string) createTableOption {
+	return func(req *createTableRequest) {
+		req.Location = strings.TrimRight(loc, "/")
+	}
+}
+
+func WithPartitionSpec(spec *iceberg.PartitionSpec) createTableOption {
+	return func(req *createTableRequest) {
+		req.PartitionSpec = spec
+	}
+}
+
+func WithWriteOrder(order *table.SortOrder) createTableOption {
+	return func(req *createTableRequest) {
+		req.WriteOrder = order
+	}
+}
+
+func WithStageCreate() createTableOption {
+	return func(req *createTableRequest) {
+		req.StageCreate = true
+	}
+}
+
+func WithProperties(props iceberg.Properties) createTableOption {
+	return func(req *createTableRequest) {
+		req.Props = props
+	}
+}
+
+type createTableRequest struct {
+	Name          string                 `json:"name"`
+	Schema        *iceberg.Schema        `json:"schema"`
+	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"`
+	Props         iceberg.Properties     `json:"properties,omitempty"`
+}
+
 type oauthTokenResponse struct {
 	AccessToken  string `json:"access_token"`
 	TokenType    string `json:"token_type"`
@@ -537,6 +617,20 @@ func checkValidNamespace(ident table.Identifier) error {
 	return nil
 }
 
+func (r *RestCatalog) tableFromResponse(identifier []string, metadata table.Metadata, loc string, config iceberg.Properties) (*table.Table, error) {
+	id := identifier
+	if r.name != "" {
+		id = append([]string{r.name}, identifier...)
+	}
+
+	iofs, err := iceio.LoadFS(config, loc)
+	if err != nil {
+		return nil, err
+	}
+
+	return table.New(id, metadata, loc, iofs), nil
+}
+
 func (r *RestCatalog) ListTables(ctx context.Context, namespace table.Identifier) ([]table.Identifier, error) {
 	if err := checkValidNamespace(namespace); err != nil {
 		return nil, err
@@ -546,12 +640,8 @@ func (r *RestCatalog) ListTables(ctx context.Context, namespace table.Identifier
 	path := []string{"namespaces", ns, "tables"}
 
 	type resp struct {
-		Identifiers []struct {
-			Namespace []string `json:"namespace"`
-			Name      string   `json:"name"`
-		} `json:"identifiers"`
+		Identifiers []identifier `json:"identifiers"`
 	}
-
 	rsp, err := doGet[resp](ctx, r.baseURI, path, r.cl, map[int]error{http.StatusNotFound: ErrNoSuchNamespace})
 	if err != nil {
 		return nil, err
@@ -573,64 +663,156 @@ func splitIdentForPath(ident table.Identifier) (string, string, error) {
 	return strings.Join(NamespaceFromIdent(ident), namespaceSeparator), TableNameFromIdent(ident), nil
 }
 
-type tblResponse struct {
-	MetadataLoc string             `json:"metadata-location"`
-	RawMetadata json.RawMessage    `json:"metadata"`
-	Config      iceberg.Properties `json:"config"`
-	Metadata    table.Metadata     `json:"-"`
-}
+func (r *RestCatalog) CreateTable(ctx context.Context, identifier table.Identifier, schema *iceberg.Schema, opts ...createTableOption) (*table.Table, error) {
+	ns, tbl, err := splitIdentForPath(identifier)
+	if err != nil {
+		return nil, err
+	}
 
-func (t *tblResponse) UnmarshalJSON(b []byte) (err error) {
-	type Alias tblResponse
-	if err = json.Unmarshal(b, (*Alias)(t)); err != nil {
-		return err
+	payload := createTableRequest{
+		Name:   tbl,
+		Schema: schema,
+	}
+	for _, o := range opts {
+		o(&payload)
 	}
 
-	t.Metadata, err = table.ParseMetadataBytes(t.RawMetadata)
-	return
+	ret, err := doPost[createTableRequest, loadTableResponse](ctx, r.baseURI, []string{"namespaces", ns, "tables"}, payload,
+		r.cl, map[int]error{http.StatusNotFound: ErrNoSuchNamespace, http.StatusConflict: ErrTableAlreadyExists})
+	if err != nil {
+		return nil, err
+	}
+
+	config := maps.Clone(r.props)
+	maps.Copy(config, ret.Metadata.Properties())
+	maps.Copy(config, ret.Config)
+
+	return r.tableFromResponse(identifier, ret.Metadata, ret.MetadataLoc, config)
 }
 
-func (r *RestCatalog) LoadTable(ctx context.Context, identifier table.Identifier, props iceberg.Properties) (*table.Table, error) {
+func (r *RestCatalog) RegisterTable(ctx context.Context, identifier table.Identifier, metadataLoc string) (*table.Table, error) {
 	ns, tbl, err := splitIdentForPath(identifier)
 	if err != nil {
 		return nil, err
 	}
 
-	if props == nil {
-		props = iceberg.Properties{}
+	type payload struct {
+		Name        string `json:"name"`
+		MetadataLoc string `json:"metadata-location"`
 	}
 
-	ret, err := doGet[tblResponse](ctx, r.baseURI, []string{"namespaces", ns, "tables", tbl},
-		r.cl, map[int]error{http.StatusNotFound: ErrNoSuchTable})
+	ret, err := doPost[payload, loadTableResponse](ctx, r.baseURI, []string{"namespaces", ns, "tables", tbl},
+		payload{Name: tbl, MetadataLoc: metadataLoc}, r.cl, map[int]error{http.StatusNotFound: ErrNoSuchNamespace, http.StatusConflict: ErrTableAlreadyExists})
 	if err != nil {
 		return nil, err
 	}
 
-	id := identifier
-	if r.name != "" {
-		id = append([]string{r.name}, identifier...)
+	config := maps.Clone(r.props)
+	maps.Copy(config, ret.Metadata.Properties())
+	maps.Copy(config, ret.Config)
+	return r.tableFromResponse(identifier, ret.Metadata, ret.MetadataLoc, config)
+}
+
+func (r *RestCatalog) LoadTable(ctx context.Context, identifier table.Identifier, props iceberg.Properties) (*table.Table, error) {
+	ns, tbl, err := splitIdentForPath(identifier)
+	if err != nil {
+		return nil, err
 	}
 
-	tblProps := maps.Clone(r.props)
-	maps.Copy(tblProps, props)
-	maps.Copy(tblProps, ret.Metadata.Properties())
+	ret, err := doGet[loadTableResponse](ctx, r.baseURI, []string{"namespaces", ns, "tables", tbl},
+		r.cl, map[int]error{http.StatusNotFound: ErrNoSuchTable})
+	if err != nil {
+		return nil, err
+	}
+
+	config := maps.Clone(r.props)
+	maps.Copy(config, props)
+	maps.Copy(config, ret.Metadata.Properties())
 	for k, v := range ret.Config {
-		tblProps[k] = v
+		config[k] = v
 	}
 
-	iofs, err := iceio.LoadFS(tblProps, ret.MetadataLoc)
+	return r.tableFromResponse(identifier, ret.Metadata, ret.MetadataLoc, config)
+}
+
+func (r *RestCatalog) UpdateTable(ctx context.Context, ident table.Identifier, requirements []table.Requirement, updates []table.Update) (*table.Table, error) {
+	ns, tbl, err := splitIdentForPath(ident)
 	if err != nil {
 		return nil, err
 	}
-	return table.New(id, ret.Metadata, ret.MetadataLoc, iofs), nil
+
+	restIdentifier := identifier{
+		Namespace: NamespaceFromIdent(ident),
+		Name:      tbl,
+	}
+	type payload struct {
+		Identifier   identifier          `json:"identifier"`
+		Requirements []table.Requirement `json:"requirements"`
+		Updates      []table.Update      `json:"updates"`
+	}
+	ret, err := doPost[payload, commitTableResponse](ctx, r.baseURI, []string{"namespaces", ns, "tables", tbl},
+		payload{Identifier: restIdentifier, Requirements: requirements, Updates: updates}, r.cl,
+		map[int]error{http.StatusNotFound: ErrNoSuchTable, http.StatusConflict: ErrCommitFailed})
+	if err != nil {
+		return nil, err
+	}
+
+	config := maps.Clone(r.props)
+	maps.Copy(config, ret.Metadata.Properties())
+
+	return r.tableFromResponse(ident, ret.Metadata, ret.MetadataLoc, config)
 }
 
 func (r *RestCatalog) DropTable(ctx context.Context, identifier table.Identifier) error {
-	return fmt.Errorf("%w: [Rest Catalog] drop table", iceberg.ErrNotImplemented)
+	ns, tbl, err := splitIdentForPath(identifier)
+	if err != nil {
+		return err
+	}
+
+	_, err = doDelete[struct{}](ctx, r.baseURI, []string{"namespaces", ns, "tables", tbl}, r.cl,
+		map[int]error{http.StatusNotFound: ErrNoSuchTable})
+
+	return err
+}
+
+func (r *RestCatalog) PurgeTable(ctx context.Context, identifier table.Identifier) error {
+	ns, tbl, err := splitIdentForPath(identifier)
+	if err != nil {
+		return err
+	}
+
+	uri := r.baseURI.JoinPath("namespaces", ns, "tables", tbl)
+	v := url.Values{}
+	v.Set("purgeRequested", "true")
+	uri.RawQuery = v.Encode()
+
+	_, err = doDelete[struct{}](ctx, uri, []string{}, r.cl,
+		map[int]error{http.StatusNotFound: ErrNoSuchTable})
+
+	return err
 }
 
 func (r *RestCatalog) RenameTable(ctx context.Context, from, to table.Identifier) (*table.Table, error) {
-	return nil, fmt.Errorf("%w: [Rest Catalog] rename table", iceberg.ErrNotImplemented)
+	type payload struct {
+		From identifier `json:"from"`
+		To   identifier `json:"to"`
+	}
+	f := identifier{
+		Namespace: NamespaceFromIdent(from),
+		Name:      TableNameFromIdent(from),
+	}
+	t := identifier{
+		Namespace: NamespaceFromIdent(to),
+		Name:      TableNameFromIdent(to),
+	}
+
+	_, err := doPost[payload, any](ctx, r.baseURI, []string{"tables", "rename"}, payload{From: f, To: t}, r.cl,
+		map[int]error{http.StatusNotFound: ErrNoSuchTable})
+	if err != nil {
+		return nil, err
+	}
+
+	return r.LoadTable(ctx, to, nil)
 }
 
 func (r *RestCatalog) CreateNamespace(ctx context.Context, namespace table.Identifier, props iceberg.Properties) error {
@@ -710,3 +892,20 @@ func (r *RestCatalog) UpdateNamespaceProperties(ctx context.Context, namespace t
 	return doPost[payload, PropertiesUpdateSummary](ctx, r.baseURI, []string{"namespaces", ns, "properties"},
 		payload{Remove: removals, Updates: updates}, r.cl, map[int]error{http.StatusNotFound: ErrNoSuchNamespace})
 }
+
+func (r *RestCatalog) CheckNamespaceExists(ctx context.Context, namespace table.Identifier) (bool, error) {
+	if err := checkValidNamespace(namespace); err != nil {
+		return false, err
+	}
+
+	_, err := doGet[struct{}](ctx, r.baseURI, []string{"namespaces", strings.Join(namespace, namespaceSeparator)},
+		r.cl, map[int]error{http.StatusNotFound: ErrNoSuchNamespace})
+	if err != nil {
+		if errors.Is(err, ErrNoSuchNamespace) {
+			return false, nil
+		}
+		return false, err
+	}
+
+	return true, nil
+}
diff --git a/dev/release/release_rc.sh b/dev/release/release_rc.sh
index ac6f0d1..c09b218 100755
--- a/dev/release/release_rc.sh
+++ b/dev/release/release_rc.sh
@@ -88,23 +88,28 @@ if [ "${RELEASE_SIGN}" -gt 0 ]; then
   echo "Found GitHub Actions workflow with ID: ${run_id}"
   gh run watch --repo "${repository}" --exit-status "${run_id}"
 
+  # Create release candidate directory structure
+  rc_dir="apache-iceberg-go-${version}-rc${rc}"
+  mkdir -p "${rc_dir}"
+
   echo "Downloading .tar.gz from GitHub Releases"
   gh release download "${rc_tag}" \
-    --dir . \
+    --dir "${rc_dir}" \
     --pattern "${tar_gz}" \
     --repo "${repository}" \
     --skip-existing
 
   echo "Signing tar.gz and creating checksums"
+  cd "${rc_dir}"
   gpg --armor --output "${tar_gz}.asc" --detach-sig "${tar_gz}"
+  sha512sum "${tar_gz}" > "${tar_gz}.sha512"
+  cd ..
 fi
 
 if [ "${RELEASE_UPLOAD}" -gt 0 ]; then
-  echo "Uploading signature"
-  gh release upload "${rc_tag}" \
-    --clobber \
-    --repo "${repository}" \
-    "${tar_gz}.asc"
+  echo "Uploading to ASF dist/dev..."
+  svn mkdir -p "https://dist.apache.org/repos/dist/dev/iceberg/${rc_dir}" --parents
+  svn import "${rc_dir}" "https://dist.apache.org/repos/dist/dev/iceberg/${rc_dir}" -m "Apache Iceberg Go ${version} RC${rc}"
 fi
 
 echo "Draft email for dev@iceberg.apache.org mailing list"
@@ -134,7 +139,7 @@ The vote will be open for at least 72 hours.
 [ ] -1 Do not release this as Apache Iceberg Go ${version} because...
 
 [1]: https://github.com/apache/iceberg-go/tree/${rc_hash}
-[2]: https://github.com/apache/iceberg-go/releases/${rc_tag}
+[2]: https://dist.apache.org/repos/dist/dev/iceberg/apache-iceberg-go-${version}-rc${rc}
 [3]: https://github.com/apache/iceberg-go/blob/main/dev/release/README.md#verify
 MAIL
 echo "---------------------------------------------------------"
diff --git a/go.mod b/go.mod
index a097645..0c3f744 100644
--- a/go.mod
+++ b/go.mod
@@ -23,11 +23,11 @@ toolchain go1.23.2
 
 require (
 	github.com/apache/arrow-go/v18 v18.0.1-0.20241029153821-f0c5d9939d3f
-	github.com/aws/aws-sdk-go-v2 v1.32.6
-	github.com/aws/aws-sdk-go-v2/config v1.28.6
-	github.com/aws/aws-sdk-go-v2/credentials v1.17.47
-	github.com/aws/aws-sdk-go-v2/service/glue v1.104.0
-	github.com/aws/aws-sdk-go-v2/service/s3 v1.71.0
+	github.com/aws/aws-sdk-go-v2 v1.32.7
+	github.com/aws/aws-sdk-go-v2/config v1.28.7
+	github.com/aws/aws-sdk-go-v2/credentials v1.17.48
+	github.com/aws/aws-sdk-go-v2/service/glue v1.105.0
+	github.com/aws/aws-sdk-go-v2/service/s3 v1.71.1
 	github.com/aws/smithy-go v1.22.1
 	github.com/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815
 	github.com/google/uuid v1.6.0
@@ -38,7 +38,7 @@ require (
 	github.com/twmb/murmur3 v1.1.8
 	gocloud.dev v0.40.0
 	golang.org/x/sync v0.10.0
-	google.golang.org/api v0.211.0
+	google.golang.org/api v0.214.0
 	gopkg.in/yaml.v3 v3.0.1
 )
 
@@ -47,9 +47,9 @@ require (
 	atomicgo.dev/keyboard v0.2.9 // indirect
 	atomicgo.dev/schedule v0.1.0 // indirect
 	cloud.google.com/go v0.116.0 // indirect
-	cloud.google.com/go/auth v0.12.1 // indirect
+	cloud.google.com/go/auth v0.13.0 // indirect
 	cloud.google.com/go/auth/oauth2adapt v0.2.6 // indirect
-	cloud.google.com/go/compute/metadata v0.5.2 // indirect
+	cloud.google.com/go/compute/metadata v0.6.0 // indirect
 	cloud.google.com/go/iam v1.2.2 // indirect
 	cloud.google.com/go/storage v1.43.0 // indirect
 	github.com/JohnCGriffin/overflow v0.0.0-20211019200055-46fa312c352c // indirect
@@ -58,19 +58,19 @@ require (
 	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.7 // indirect
-	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.21 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.22 // 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.25 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.25 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.26 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.26 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.1 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.25 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.26 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.1 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.4.6 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.6 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.18.6 // indirect
-	github.com/aws/aws-sdk-go-v2/service/sso v1.24.7 // indirect
-	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.28.6 // indirect
-	github.com/aws/aws-sdk-go-v2/service/sts v1.33.2 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.4.7 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.7 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.18.7 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.24.8 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.28.7 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sts v1.33.3 // indirect
 	github.com/containerd/console v1.0.3 // indirect
 	github.com/creasty/defaults v1.8.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
@@ -104,7 +104,8 @@ require (
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/pierrec/lz4/v4 v4.1.21 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	github.com/rivo/uniseg v0.4.4 // indirect
+	github.com/rivo/uniseg v0.4.7 // indirect
+	github.com/rogpeppe/go-internal v1.12.0 // indirect
 	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/substrait-io/substrait v0.57.1 // indirect
 	github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e // indirect
@@ -118,7 +119,7 @@ require (
 	golang.org/x/crypto v0.31.0 // indirect
 	golang.org/x/exp v0.0.0-20240909161429-701f63a606c0 // indirect
 	golang.org/x/mod v0.22.0 // indirect
-	golang.org/x/net v0.32.0 // indirect
+	golang.org/x/net v0.33.0 // indirect
 	golang.org/x/oauth2 v0.24.0 // indirect
 	golang.org/x/sys v0.28.0 // indirect
 	golang.org/x/term v0.27.0 // indirect
@@ -128,7 +129,7 @@ require (
 	golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da // indirect
 	google.golang.org/genproto v0.0.0-20241113202542-65e8d215514f // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20241118233622-e639e219e697 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20241206012308-a4fef0638583 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20241209162323-e6fa225c2576 // indirect
 	google.golang.org/grpc v1.67.1 // indirect
 	google.golang.org/protobuf v1.35.2 // indirect
 )
diff --git a/go.sum b/go.sum
index 074d534..2ecff2e 100644
--- a/go.sum
+++ b/go.sum
@@ -9,12 +9,12 @@ atomicgo.dev/schedule v0.1.0/go.mod h1:xeUa3oAkiuHYh8bKiQBRojqAMq3PXXbJujjb0hw8p
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 cloud.google.com/go v0.116.0 h1:B3fRrSDkLRt5qSHWe40ERJvhvnQwdZiHu0bJOpldweE=
 cloud.google.com/go v0.116.0/go.mod h1:cEPSRWPzZEswwdr9BxE6ChEn01dWlTaF05LiC2Xs70U=
-cloud.google.com/go/auth v0.12.1 h1:n2Bj25BUMM0nvE9D2XLTiImanwZhO3DkfWSYS/SAJP4=
-cloud.google.com/go/auth v0.12.1/go.mod h1:BFMu+TNpF3DmvfBO9ClqTR/SiqVIm7LukKF9mbendF4=
+cloud.google.com/go/auth v0.13.0 h1:8Fu8TZy167JkW8Tj3q7dIkr2v4cndv41ouecJx0PAHs=
+cloud.google.com/go/auth v0.13.0/go.mod h1:COOjD9gwfKNKz+IIduatIhYJQIc0mG3H102r/EMxX6Q=
 cloud.google.com/go/auth/oauth2adapt v0.2.6 h1:V6a6XDu2lTwPZWOawrAa9HUK+DB2zfJyTuciBG5hFkU=
 cloud.google.com/go/auth/oauth2adapt v0.2.6/go.mod h1:AlmsELtlEBnaNTL7jCj8VQFLy6mbZv0s4Q7NGBeQ5E8=
-cloud.google.com/go/compute/metadata v0.5.2 h1:UxK4uu/Tn+I3p2dYWTfiX4wva7aYlKixAHn3fyqngqo=
-cloud.google.com/go/compute/metadata v0.5.2/go.mod h1:C66sj2AluDcIqakBq/M8lw8/ybHgOZqin2obFxa/E5k=
+cloud.google.com/go/compute/metadata v0.6.0 h1:A6hENjEsCDtC1k8byVsgwvVcioamEHvZ4j01OwKxG9I=
+cloud.google.com/go/compute/metadata v0.6.0/go.mod h1:FjyFAW1MW0C203CEOMDTu3Dk1FlqW3Rga40jzHL4hfg=
 cloud.google.com/go/iam v1.2.2 h1:ozUSofHUGf/F4tCNy/mu9tHLTaxZFLOUiKzjcgWHGIA=
 cloud.google.com/go/iam v1.2.2/go.mod h1:0Ys8ccaZHdI1dEUilwzqng/6ps2YB6vRsjIe00/+6JY=
 cloud.google.com/go/longrunning v0.6.2 h1:xjDfh1pQcWPEvnfjZmwjKQEcHnpz6lHjfy7Fo0MK+hc=
@@ -48,44 +48,44 @@ 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.32.6 h1:7BokKRgRPuGmKkFMhEg/jSul+tB9VvXhcViILtfG8b4=
-github.com/aws/aws-sdk-go-v2 v1.32.6/go.mod h1:P5WJBrYqqbWVaOxgH0X/FYYD47/nooaPOZPlQdmiN2U=
+github.com/aws/aws-sdk-go-v2 v1.32.7 h1:ky5o35oENWi0JYWUZkB7WYvVPP+bcRF5/Iq7JWSb5Rw=
+github.com/aws/aws-sdk-go-v2 v1.32.7/go.mod h1:P5WJBrYqqbWVaOxgH0X/FYYD47/nooaPOZPlQdmiN2U=
 github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.7 h1:lL7IfaFzngfx0ZwUGOZdsFFnQ5uLvR0hWqqhyE7Q9M8=
 github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.7/go.mod h1:QraP0UcVlQJsmHfioCrveWOC1nbiWUl3ej08h4mXWoc=
-github.com/aws/aws-sdk-go-v2/config v1.28.6 h1:D89IKtGrs/I3QXOLNTH93NJYtDhm8SYa9Q5CsPShmyo=
-github.com/aws/aws-sdk-go-v2/config v1.28.6/go.mod h1:GDzxJ5wyyFSCoLkS+UhGB0dArhb9mI+Co4dHtoTxbko=
-github.com/aws/aws-sdk-go-v2/credentials v1.17.47 h1:48bA+3/fCdi2yAwVt+3COvmatZ6jUDNkDTIsqDiMUdw=
-github.com/aws/aws-sdk-go-v2/credentials v1.17.47/go.mod h1:+KdckOejLW3Ks3b0E3b5rHsr2f9yuORBum0WPnE5o5w=
-github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.21 h1:AmoU1pziydclFT/xRV+xXE/Vb8fttJCLRPv8oAkprc0=
-github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.21/go.mod h1:AjUdLYe4Tgs6kpH4Bv7uMZo7pottoyHMn4eTcIcneaY=
+github.com/aws/aws-sdk-go-v2/config v1.28.7 h1:GduUnoTXlhkgnxTD93g1nv4tVPILbdNQOzav+Wpg7AE=
+github.com/aws/aws-sdk-go-v2/config v1.28.7/go.mod h1:vZGX6GVkIE8uECSUHB6MWAUsd4ZcG2Yq/dMa4refR3M=
+github.com/aws/aws-sdk-go-v2/credentials v1.17.48 h1:IYdLD1qTJ0zanRavulofmqut4afs45mOWEI+MzZtTfQ=
+github.com/aws/aws-sdk-go-v2/credentials v1.17.48/go.mod h1:tOscxHN3CGmuX9idQ3+qbkzrjVIx32lqDSU1/0d/qXs=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.22 h1:kqOrpojG71DxJm/KDPO+Z/y1phm1JlC8/iT+5XRmAn8=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.22/go.mod h1:NtSFajXVVL8TA2QNngagVZmUtXciyrHOt7xgz4faS/M=
 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.25 h1:s/fF4+yDQDoElYhfIVvSNyeCydfbuTKzhxSXDXCPasU=
-github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.25/go.mod h1:IgPfDv5jqFIzQSNbUEMoitNooSMXjRSDkhXv8jiROvU=
-github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.25 h1:ZntTCl5EsYnhN/IygQEUugpdwbhdkom9uHcbCftiGgA=
-github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.25/go.mod h1:DBdPrgeocww+CSl1C8cEV8PN1mHMBhuCDLpXezyvWkE=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.26 h1:I/5wmGMffY4happ8NOCuIUEWGUvvFp5NSeQcXl9RHcI=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.26/go.mod h1:FR8f4turZtNy6baO0KJ5FJUmXH/cSkI9fOngs0yl6mA=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.26 h1:zXFLuEuMMUOvEARXFUVJdfqZ4bvvSgdGRq/ATcrQxzM=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.26/go.mod h1:3o2Wpy0bogG1kyOPrgkXA8pgIfEEv0+m19O9D5+W8y8=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.8.1 h1:VaRN3TlFdd6KxX1x3ILT5ynH6HvKgqdiXoTxAF4HQcQ=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.8.1/go.mod h1:FbtygfRFze9usAadmnGJNc8KsP346kEe+y2/oyhGAGc=
-github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.25 h1:r67ps7oHCYnflpgDy2LZU0MAQtQbYIOqNNnqGO6xQkE=
-github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.25/go.mod h1:GrGY+Q4fIokYLtjCVB/aFfCVL6hhGUFl8inD18fDalE=
-github.com/aws/aws-sdk-go-v2/service/glue v1.104.0 h1:H6ZuOpWZpXqLieLZ2shNSJI+deobB1+FKL6plTOaqUo=
-github.com/aws/aws-sdk-go-v2/service/glue v1.104.0/go.mod h1:ajiRue7mZ0vQjVHQkQG2KBaPHW8lL5GtvmjRTHWDaqk=
+github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.26 h1:GeNJsIFHB+WW5ap2Tec4K6dzcVTsRbsT1Lra46Hv9ME=
+github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.26/go.mod h1:zfgMpwHDXX2WGoG84xG2H+ZlPTkJUU4YUvx2svLQYWo=
+github.com/aws/aws-sdk-go-v2/service/glue v1.105.0 h1:raq38Qb6iJJtzADr7Z4IYHOFp5E1NVpHDGoTOsGLHNM=
+github.com/aws/aws-sdk-go-v2/service/glue v1.105.0/go.mod h1:FyYpmVnMux6fzG2kcLnVwT/swhs8DNtleGIkc8gh63c=
 github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.1 h1:iXtILhvDxB6kPvEXgsDhGaZCSC6LQET5ZHSdJozeI0Y=
 github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.1/go.mod h1:9nu0fVANtYiAePIBh2/pFUSwtJ402hLnp854CNoDOeE=
-github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.4.6 h1:HCpPsWqmYQieU7SS6E9HXfdAMSud0pteVXieJmcpIRI=
-github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.4.6/go.mod h1:ngUiVRCco++u+soRRVBIvBZxSMMvOVMXA4PJ36JLfSw=
-github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.6 h1:50+XsN70RS7dwJ2CkVNXzj7U2L1HKP8nqTd3XWEXBN4=
-github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.6/go.mod h1:WqgLmwY7so32kG01zD8CPTJWVWM+TzJoOVHwTg4aPug=
-github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.18.6 h1:BbGDtTi0T1DYlmjBiCr/le3wzhA37O8QTC5/Ab8+EXk=
-github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.18.6/go.mod h1:hLMJt7Q8ePgViKupeymbqI0la+t9/iYFBjxQCFwuAwI=
-github.com/aws/aws-sdk-go-v2/service/s3 v1.71.0 h1:nyuzXooUNJexRT0Oy0UQY6AhOzxPxhtt4DcBIHyCnmw=
-github.com/aws/aws-sdk-go-v2/service/s3 v1.71.0/go.mod h1:sT/iQz8JK3u/5gZkT+Hmr7GzVZehUMkRZpOaAwYXeGY=
-github.com/aws/aws-sdk-go-v2/service/sso v1.24.7 h1:rLnYAfXQ3YAccocshIH5mzNNwZBkBo+bP6EhIxak6Hw=
-github.com/aws/aws-sdk-go-v2/service/sso v1.24.7/go.mod h1:ZHtuQJ6t9A/+YDuxOLnbryAmITtr8UysSny3qcyvJTc=
-github.com/aws/aws-sdk-go-v2/service/ssooidc v1.28.6 h1:JnhTZR3PiYDNKlXy50/pNeix9aGMo6lLpXwJ1mw8MD4=
-github.com/aws/aws-sdk-go-v2/service/ssooidc v1.28.6/go.mod h1:URronUEGfXZN1VpdktPSD1EkAL9mfrV+2F4sjH38qOY=
-github.com/aws/aws-sdk-go-v2/service/sts v1.33.2 h1:s4074ZO1Hk8qv65GqNXqDjmkf4HSQqJukaLuuW0TpDA=
-github.com/aws/aws-sdk-go-v2/service/sts v1.33.2/go.mod h1:mVggCnIWoM09jP71Wh+ea7+5gAp53q+49wDFs1SW5z8=
+github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.4.7 h1:tB4tNw83KcajNAzaIMhkhVI2Nt8fAZd5A5ro113FEMY=
+github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.4.7/go.mod h1:lvpyBGkZ3tZ9iSsUIcC2EWp+0ywa7aK3BLT+FwZi+mQ=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.7 h1:8eUsivBQzZHqe/3FE+cqwfH+0p5Jo8PFM/QYQSmeZ+M=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.7/go.mod h1:kLPQvGUmxn/fqiCrDeohwG33bq2pQpGeY62yRO6Nrh0=
+github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.18.7 h1:Hi0KGbrnr57bEHWM0bJ1QcBzxLrL/k2DHvGYhb8+W1w=
+github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.18.7/go.mod h1:wKNgWgExdjjrm4qvfbTorkvocEstaoDl4WCvGfeCy9c=
+github.com/aws/aws-sdk-go-v2/service/s3 v1.71.1 h1:aOVVZJgWbaH+EJYPvEgkNhCEbXXvH7+oML36oaPK3zE=
+github.com/aws/aws-sdk-go-v2/service/s3 v1.71.1/go.mod h1:r+xl5yzMk9083rMR+sJ5TYj9Tihvf/l1oxzZXDgGj2Q=
+github.com/aws/aws-sdk-go-v2/service/sso v1.24.8 h1:CvuUmnXI7ebaUAhbJcDy9YQx8wHR69eZ9I7q5hszt/g=
+github.com/aws/aws-sdk-go-v2/service/sso v1.24.8/go.mod h1:XDeGv1opzwm8ubxddF0cgqkZWsyOtw4lr6dxwmb6YQg=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.28.7 h1:F2rBfNAL5UyswqoeWv9zs74N/NanhK16ydHW1pahX6E=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.28.7/go.mod h1:JfyQ0g2JG8+Krq0EuZNnRwX0mU0HrwY/tG6JNfcqh4k=
+github.com/aws/aws-sdk-go-v2/service/sts v1.33.3 h1:Xgv/hyNgvLda/M9l9qxXc4UFSgppnRczLxlMs5Ae/QY=
+github.com/aws/aws-sdk-go-v2/service/sts v1.33.3/go.mod h1:5Gn+d+VaaRgsjewpMvGazt0WfcFO+Md4wLOuBfGR9Bc=
 github.com/aws/smithy-go v1.22.1 h1:/HPHZQ0g7f4eUeK6HKglFz8uwVfZKgoI25rb/J+dnro=
 github.com/aws/smithy-go v1.22.1/go.mod h1:irrKGvNn1InZwb2d7fkIRNucdfwR8R+Ts3wxYa/cJHg=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
@@ -238,10 +238,10 @@ github.com/pterm/pterm v0.12.40/go.mod h1:ffwPLwlbXxP+rxT0GsgDTzS3y3rmpAO1NMjUkG
 github.com/pterm/pterm v0.12.80 h1:mM55B+GnKUnLMUSqhdINe4s6tOuVQIetQ3my8JGyAIg=
 github.com/pterm/pterm v0.12.80/go.mod h1:c6DeF9bSnOSeFPZlfs4ZRAFcf5SCoTwvwQ5xaKGQlHo=
 github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
-github.com/rivo/uniseg v0.4.4 h1:8TfxU8dW6PdqD27gjM8MVNuicgxIjxpm4K7x4jp8sis=
-github.com/rivo/uniseg v0.4.4/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88=
-github.com/rogpeppe/go-internal v1.9.0 h1:73kH8U+JUqXU8lRuOHeVHaa/SZPifC7BkcraZVejAe8=
-github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
+github.com/rivo/uniseg v0.4.7 h1:WUdvkW8uEhrYfLC4ZzdpI2ztxP1I582+49Oc5Mq64VQ=
+github.com/rivo/uniseg v0.4.7/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88=
+github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8=
+github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4=
 github.com/sergi/go-diff v1.2.0 h1:XU+rvMAioB0UC3q1MFrIQy4Vo5/4VsRDQQXHsEya6xQ=
 github.com/sergi/go-diff v1.2.0/go.mod h1:STckp+ISIX8hZLjrqAeVduY0gWCT9IjLuqbuNXdaHfM=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
@@ -322,8 +322,8 @@ golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
 golang.org/x/net v0.15.0/go.mod h1:idbUs1IY1+zTqbi8yxTbhexhEEk5ur9LInksu6HrEpk=
 golang.org/x/net v0.20.0/go.mod h1:z8BVo6PvndSri0LbOE3hAn0apkU+1YvI6E70E9jsnvY=
-golang.org/x/net v0.32.0 h1:ZqPmj8Kzc+Y6e0+skZsuACbx+wzMgo5MQsJh9Qd6aYI=
-golang.org/x/net v0.32.0/go.mod h1:CwU0IoeOlnQQWJ6ioyFrfRuomB8GKF6KbYXZVyeXNfs=
+golang.org/x/net v0.33.0 h1:74SYHlV8BIgHIFC/LrYkOGIwL19eTYXQ5wc6TBuO36I=
+golang.org/x/net v0.33.0/go.mod h1:HXLR5J+9DxmrqMwG9qjGCxZ+zKXxBru04zlTvWlWuN4=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.24.0 h1:KTBBxWqUa0ykRPLtV69rRto9TLXcqYkeswu48x/gvNE=
 golang.org/x/oauth2 v0.24.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI=
@@ -395,8 +395,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.211.0 h1:IUpLjq09jxBSV1lACO33CGY3jsRcbctfGzhj+ZSE/Bg=
-google.golang.org/api v0.211.0/go.mod h1:XOloB4MXFH4UTlQSGuNUxw0UT74qdENK8d6JNsXKLi0=
+google.golang.org/api v0.214.0 h1:h2Gkq07OYi6kusGOaT/9rnNljuXmqPnaig7WGPmKbwA=
+google.golang.org/api v0.214.0/go.mod h1:bYPpLG8AyeMWwDU6NXoB00xC0DFkikVvd5MfwoxjLqE=
 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=
@@ -406,8 +406,8 @@ google.golang.org/genproto v0.0.0-20241113202542-65e8d215514f h1:zDoHYmMzMacIdjN
 google.golang.org/genproto v0.0.0-20241113202542-65e8d215514f/go.mod h1:Q5m6g8b5KaFFzsQFIGdJkSJDGeJiybVenoYFMMa3ohI=
 google.golang.org/genproto/googleapis/api v0.0.0-20241118233622-e639e219e697 h1:pgr/4QbFyktUv9CtQ/Fq4gzEE6/Xs7iCXbktaGzLHbQ=
 google.golang.org/genproto/googleapis/api v0.0.0-20241118233622-e639e219e697/go.mod h1:+D9ySVjN8nY8YCVjc5O7PZDIdZporIDY3KaGfJunh88=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20241206012308-a4fef0638583 h1:IfdSdTcLFy4lqUQrQJLkLt1PB+AsqVz6lwkWPzWEz10=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20241206012308-a4fef0638583/go.mod h1:5uTbfoYQed2U9p3KIj2/Zzm02PYhndfdmML0qC3q3FU=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20241209162323-e6fa225c2576 h1:8ZmaLZE4XWrtU3MyClkYqqtl6Oegr3235h7jxsDyqCY=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20241209162323-e6fa225c2576/go.mod h1:5uTbfoYQed2U9p3KIj2/Zzm02PYhndfdmML0qC3q3FU=
 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=
diff --git a/partitions.go b/partitions.go
index 48b730c..9416d70 100644
--- a/partitions.go
+++ b/partitions.go
@@ -20,6 +20,7 @@ package iceberg
 import (
 	"encoding/json"
 	"fmt"
+	"iter"
 	"slices"
 	"strings"
 )
@@ -116,6 +117,11 @@ func (ps PartitionSpec) Equals(other PartitionSpec) bool {
 	return ps.id == other.id && slices.Equal(ps.fields, other.fields)
 }
 
+// Fields returns a clone of the partition fields in this spec.
+func (ps *PartitionSpec) Fields() iter.Seq[PartitionField] {
+	return slices.Values(ps.fields)
+}
+
 func (ps PartitionSpec) MarshalJSON() ([]byte, error) {
 	if ps.fields == nil {
 		ps.fields = []PartitionField{}
diff --git a/table/metadata.go b/table/metadata.go
index 47b3ffe..73021c5 100644
--- a/table/metadata.go
+++ b/table/metadata.go
@@ -22,14 +22,26 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"iter"
 	"maps"
 	"slices"
+	"time"
 
 	"github.com/apache/iceberg-go"
 
 	"github.com/google/uuid"
 )
 
+const (
+	partitionFieldStartID       = 1000
+	supportedTableFormatVersion = 2
+
+	addPartionSpecAction = "add-partition-spec"
+	addSchemaAction      = "add-schema"
+	addSnapshotAction    = "add-snapshot"
+	addSortOrderAction   = "add-sort-order"
+)
+
 // Metadata for an iceberg table as specified in the Iceberg spec
 //
 // https://iceberg.apache.org/spec/#iceberg-table-spec
@@ -80,20 +92,542 @@ type Metadata interface {
 	SnapshotByName(name string) *Snapshot
 	// CurrentSnapshot returns the table's current snapshot.
 	CurrentSnapshot() *Snapshot
+	// Ref returns the snapshot ref for the main branch.
+	Ref() SnapshotRef
+	// Refs returns a list of snapshot name/reference pairs.
+	Refs() iter.Seq2[string, SnapshotRef]
+	// SnapshotLogs returns the list of snapshot logs for the table.
+	SnapshotLogs() iter.Seq[SnapshotLogEntry]
 	// SortOrder returns the table's current sort order, ie: the one with the
 	// ID that matches the default-sort-order-id.
 	SortOrder() SortOrder
 	// SortOrders returns the list of sort orders in the table.
 	SortOrders() []SortOrder
+	// DefaultSortOrder returns the ID of the current sort order that writers
+	// should use by default.
+	DefaultSortOrder() int
 	// Properties is a string to string map of table properties. This is used
 	// to control settings that affect reading and writing and is not intended
 	// to be used for arbitrary metadata. For example, commit.retry.num-retries
 	// is used to control the number of commit retries.
 	Properties() iceberg.Properties
+	// PreviousFiles returns the list of metadata log entries for the table.
+	PreviousFiles() iter.Seq[MetadataLogEntry]
 
 	Equals(Metadata) bool
 }
 
+type MetadataBuilder struct {
+	base    Metadata
+	updates []Update
+
+	// common fields
+	formatVersion      int
+	uuid               uuid.UUID
+	loc                string
+	lastUpdatedMS      int64
+	lastColumnId       int
+	schemaList         []*iceberg.Schema
+	currentSchemaID    int
+	specs              []iceberg.PartitionSpec
+	defaultSpecID      int
+	lastPartitionID    *int
+	props              iceberg.Properties
+	snapshotList       []Snapshot
+	currentSnapshotID  *int64
+	snapshotLog        []SnapshotLogEntry
+	metadataLog        []MetadataLogEntry
+	sortOrderList      []SortOrder
+	defaultSortOrderID int
+	refs               map[string]SnapshotRef
+
+	// V2 specific
+	lastSequenceNumber *int64
+}
+
+func NewMetadataBuilder() (*MetadataBuilder, error) {
+	return &MetadataBuilder{
+		updates:       make([]Update, 0),
+		schemaList:    make([]*iceberg.Schema, 0),
+		specs:         make([]iceberg.PartitionSpec, 0),
+		props:         make(iceberg.Properties),
+		snapshotList:  make([]Snapshot, 0),
+		snapshotLog:   make([]SnapshotLogEntry, 0),
+		metadataLog:   make([]MetadataLogEntry, 0),
+		sortOrderList: make([]SortOrder, 0),
+		refs:          make(map[string]SnapshotRef),
+	}, nil
+}
+
+func MetadataBuilderFromBase(metadata Metadata) (*MetadataBuilder, error) {
+	b := &MetadataBuilder{}
+	b.base = metadata
+
+	b.formatVersion = metadata.Version()
+	b.uuid = metadata.TableUUID()
+	b.loc = metadata.Location()
+	b.lastUpdatedMS = metadata.LastUpdatedMillis()
+	b.lastColumnId = metadata.LastColumnID()
+	b.schemaList = metadata.Schemas()
+	b.currentSchemaID = metadata.CurrentSchema().ID
+	b.specs = metadata.PartitionSpecs()
+	b.defaultSpecID = metadata.DefaultPartitionSpec()
+	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()
+
+	b.refs = make(map[string]SnapshotRef)
+	for name, ref := range metadata.Refs() {
+		b.refs[name] = ref
+	}
+
+	b.snapshotLog = make([]SnapshotLogEntry, 0)
+	for log := range metadata.SnapshotLogs() {
+		b.snapshotLog = append(b.snapshotLog, log)
+	}
+
+	b.metadataLog = make([]MetadataLogEntry, 0)
+	for entry := range metadata.PreviousFiles() {
+		b.metadataLog = append(b.metadataLog, entry)
+	}
+
+	return b, nil
+}
+
+func (b *MetadataBuilder) AddSchema(schema *iceberg.Schema, newLastColumnID int, initial bool) (*MetadataBuilder, error) {
+	if newLastColumnID < b.lastColumnId {
+		return nil, fmt.Errorf("%w: newLastColumnID %d, must be >= %d", iceberg.ErrInvalidArgument, newLastColumnID, b.lastColumnId)
+	}
+
+	var schemas []*iceberg.Schema
+	if initial {
+		schemas = []*iceberg.Schema{schema}
+	} else {
+		schemas = append(b.schemaList, schema)
+	}
+
+	b.lastColumnId = newLastColumnID
+	b.schemaList = schemas
+	b.updates = append(b.updates, NewAddSchemaUpdate(schema, newLastColumnID, initial))
+
+	return b, nil
+}
+
+func (b *MetadataBuilder) AddPartitionSpec(spec *iceberg.PartitionSpec, initial bool) (*MetadataBuilder, error) {
+	for _, s := range b.specs {
+		if s.ID() == spec.ID() && !initial {
+			return nil, fmt.Errorf("partition spec with id %d already exists", spec.ID())
+		}
+	}
+
+	maxFieldID := 0
+	for f := range spec.Fields() {
+		maxFieldID = max(maxFieldID, f.FieldID)
+	}
+
+	prev := partitionFieldStartID - 1
+	if b.lastPartitionID != nil {
+		prev = *b.lastPartitionID
+	}
+	lastPartitionID := max(maxFieldID, prev)
+
+	var specs []iceberg.PartitionSpec
+	if initial {
+		specs = []iceberg.PartitionSpec{*spec}
+	} else {
+		specs = append(b.specs, *spec)
+	}
+
+	b.specs = specs
+	b.lastPartitionID = &lastPartitionID
+	b.updates = append(b.updates, NewAddPartitionSpecUpdate(spec, initial))
+
+	return b, nil
+}
+
+func (b *MetadataBuilder) AddSnapshot(snapshot *Snapshot) (*MetadataBuilder, error) {
+	if snapshot == nil {
+		return nil, nil
+	}
+
+	if len(b.schemaList) == 0 {
+		return nil, errors.New("can't add snapshot with no added schemas")
+	} else if len(b.specs) == 0 {
+		return nil, errors.New("can't add snapshot with no added partition specs")
+	} else if s, _ := b.SnapshotByID(snapshot.SnapshotID); s != nil {
+		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 {
+		return nil, fmt.Errorf("can't add snapshot with sequence number %d, must be > than last sequence number %d",
+			snapshot.SequenceNumber, b.lastSequenceNumber)
+	}
+
+	b.updates = append(b.updates, NewAddSnapshotUpdate(snapshot))
+	b.lastUpdatedMS = snapshot.TimestampMs
+	b.lastSequenceNumber = &snapshot.SequenceNumber
+	b.snapshotList = append(b.snapshotList, *snapshot)
+	return b, nil
+}
+
+func (b *MetadataBuilder) AddSortOrder(sortOrder *SortOrder, initial bool) (*MetadataBuilder, error) {
+	var sortOrders []SortOrder
+	if !initial {
+		sortOrders = append(sortOrders, b.sortOrderList...)
+	}
+
+	for _, s := range sortOrders {
+		if s.OrderID == sortOrder.OrderID {
+			return nil, fmt.Errorf("sort order with id %d already exists", sortOrder.OrderID)
+		}
+	}
+
+	b.sortOrderList = append(sortOrders, *sortOrder)
+	b.updates = append(b.updates, NewAddSortOrderUpdate(sortOrder, initial))
+
+	return b, nil
+}
+
+func (b *MetadataBuilder) RemoveProperties(keys []string) (*MetadataBuilder, error) {
+	if len(keys) == 0 {
+		return b, nil
+	}
+
+	b.updates = append(b.updates, NewRemovePropertiesUpdate(keys))
+	for _, key := range keys {
+		delete(b.props, key)
+	}
+
+	return b, nil
+}
+
+func (b *MetadataBuilder) SetCurrentSchemaID(currentSchemaID int) (*MetadataBuilder, error) {
+	if currentSchemaID == -1 {
+		currentSchemaID = maxBy(b.schemaList, func(s *iceberg.Schema) int {
+			return s.ID
+		})
+		if !slices.ContainsFunc(b.updates, func(u Update) bool {
+			return u.Action() == addSchemaAction && u.(*addSchemaUpdate).Schema.ID == currentSchemaID
+		}) {
+			return nil, errors.New("can't set current schema to last added schema, no schema has been added")
+		}
+	}
+
+	if currentSchemaID == b.currentSchemaID {
+		return b, nil
+	}
+
+	_, err := b.GetSchemaByID(currentSchemaID)
+	if err != nil {
+		return nil, fmt.Errorf("can't set current schema to schema with id %d: %w", currentSchemaID, err)
+	}
+
+	b.updates = append(b.updates, NewSetCurrentSchemaUpdate(currentSchemaID))
+	b.currentSchemaID = currentSchemaID
+	return b, nil
+}
+
+func (b *MetadataBuilder) SetDefaultSortOrderID(defaultSortOrderID int) (*MetadataBuilder, error) {
+	if defaultSortOrderID == -1 {
+		defaultSortOrderID = maxBy(b.sortOrderList, func(s SortOrder) int {
+			return s.OrderID
+		})
+		if !slices.ContainsFunc(b.updates, func(u Update) bool {
+			return u.Action() == addSortOrderAction && u.(*addSortOrderUpdate).SortOrder.OrderID == defaultSortOrderID
+		}) {
+			return nil, fmt.Errorf("can't set default sort order to last added with no added sort orders")
+		}
+	}
+
+	if defaultSortOrderID == b.defaultSortOrderID {
+		return b, nil
+	}
+
+	if _, err := b.GetSortOrderByID(defaultSortOrderID); err != nil {
+		return nil, fmt.Errorf("can't set default sort order to sort order with id %d: %w", defaultSortOrderID, err)
+	}
+
+	b.updates = append(b.updates, NewSetDefaultSortOrderUpdate(defaultSortOrderID))
+	b.defaultSortOrderID = defaultSortOrderID
+	return b, nil
+}
+
+func (b *MetadataBuilder) SetDefaultSpecID(defaultSpecID int) (*MetadataBuilder, error) {
+	if defaultSpecID == -1 {
+		defaultSpecID = maxBy(b.specs, func(s iceberg.PartitionSpec) int {
+			return s.ID()
+		})
+		if !slices.ContainsFunc(b.updates, func(u Update) bool {
+			return u.Action() == addPartionSpecAction && u.(*addPartitionSpecUpdate).Spec.ID() == defaultSpecID
+		}) {
+			return nil, fmt.Errorf("can't set default spec to last added with no added partition specs")
+		}
+	}
+
+	if defaultSpecID == b.defaultSpecID {
+		return b, nil
+	}
+
+	if _, err := b.GetSpecByID(defaultSpecID); err != nil {
+		return nil, fmt.Errorf("can't set default spec to spec with id %d: %w", defaultSpecID, err)
+	}
+
+	b.updates = append(b.updates, NewSetDefaultSpecUpdate(defaultSpecID))
+	b.defaultSpecID = defaultSpecID
+	return b, nil
+}
+
+func (b *MetadataBuilder) SetFormatVersion(formatVersion int) (*MetadataBuilder, error) {
+	if formatVersion < b.formatVersion {
+		return nil, fmt.Errorf("downgrading format version from %d to %d is not allowed",
+			b.formatVersion, formatVersion)
+	}
+
+	if formatVersion > supportedTableFormatVersion {
+		return nil, fmt.Errorf("unsupported format version %d", formatVersion)
+	}
+
+	if formatVersion == b.formatVersion {
+		return b, nil
+	}
+
+	b.updates = append(b.updates, NewUpgradeFormatVersionUpdate(formatVersion))
+	b.formatVersion = formatVersion
+	return b, nil
+}
+
+func (b *MetadataBuilder) SetLoc(loc string) (*MetadataBuilder, error) {
+	if b.loc == loc {
+		return b, nil
+	}
+
+	b.updates = append(b.updates, NewSetLocationUpdate(loc))
+	b.loc = loc
+	return b, nil
+}
+
+func (b *MetadataBuilder) SetProperties(props iceberg.Properties) (*MetadataBuilder, error) {
+	if len(props) == 0 {
+		return b, nil
+	}
+
+	b.updates = append(b.updates, NewSetPropertiesUpdate(props))
+	maps.Copy(b.props, props)
+	return b, nil
+}
+
+type setSnapshotRefOption func(*SnapshotRef) error
+
+func WithMaxRefAgeMs(maxRefAgeMs int64) setSnapshotRefOption {
+	return func(ref *SnapshotRef) error {
+		if maxRefAgeMs <= 0 {
+			return fmt.Errorf("%w: maxRefAgeMs %d, must be > 0", iceberg.ErrInvalidArgument, maxRefAgeMs)
+		}
+		ref.MaxRefAgeMs = &maxRefAgeMs
+		return nil
+	}
+}
+
+func WithMaxSnapshotAgeMs(maxSnapshotAgeMs int64) setSnapshotRefOption {
+	return func(ref *SnapshotRef) error {
+		if maxSnapshotAgeMs <= 0 {
+			return fmt.Errorf("%w: maxSnapshotAgeMs %d, must be > 0", iceberg.ErrInvalidArgument, maxSnapshotAgeMs)
+		}
+		ref.MaxSnapshotAgeMs = &maxSnapshotAgeMs
+		return nil
+	}
+}
+
+func WithMinSnapshotsToKeep(minSnapshotsToKeep int) setSnapshotRefOption {
+	return func(ref *SnapshotRef) error {
+		if minSnapshotsToKeep <= 0 {
+			return fmt.Errorf("%w: minSnapshotsToKeep %d, must be > 0", iceberg.ErrInvalidArgument, minSnapshotsToKeep)
+		}
+		ref.MinSnapshotsToKeep = &minSnapshotsToKeep
+		return nil
+	}
+}
+
+func (b *MetadataBuilder) SetSnapshotRef(
+	name string,
+	snapshotID int64,
+	refType RefType,
+	options ...setSnapshotRefOption,
+) (*MetadataBuilder, error) {
+	ref := SnapshotRef{
+		SnapshotID:      snapshotID,
+		SnapshotRefType: refType,
+	}
+	for _, opt := range options {
+		if err := opt(&ref); err != nil {
+			return nil, fmt.Errorf("invalid snapshot ref option: %w", err)
+		}
+	}
+
+	var maxRefAgeMs, maxSnapshotAgeMs int64
+	var minSnapshotsToKeep int
+	if ref.MaxRefAgeMs != nil {
+		maxRefAgeMs = *ref.MaxRefAgeMs
+	}
+	if ref.MaxSnapshotAgeMs != nil {
+		maxSnapshotAgeMs = *ref.MaxSnapshotAgeMs
+	}
+	if ref.MinSnapshotsToKeep != nil {
+		minSnapshotsToKeep = *ref.MinSnapshotsToKeep
+	}
+
+	if existingRef, ok := b.refs[name]; ok && existingRef.Equals(ref) {
+		return b, nil
+	}
+
+	snapshot, err := b.SnapshotByID(snapshotID)
+	if err != nil {
+		return nil, fmt.Errorf("can't set snapshot ref %s to unknown snapshot %d: %w", name, snapshotID, err)
+	}
+
+	if refType == MainBranch {
+		b.updates = append(b.updates, NewSetSnapshotRefUpdate(name, snapshotID, refType, maxRefAgeMs, maxSnapshotAgeMs, minSnapshotsToKeep))
+		b.currentSnapshotID = &snapshotID
+		b.snapshotLog = append(b.snapshotLog, SnapshotLogEntry{
+			SnapshotID:  snapshotID,
+			TimestampMs: snapshot.TimestampMs,
+		})
+		b.lastUpdatedMS = time.Now().Local().UnixMilli()
+	}
+
+	if slices.ContainsFunc(b.updates, func(u Update) bool {
+		return u.Action() == addSnapshotAction && u.(*addSnapshotUpdate).Snapshot.SnapshotID == snapshotID
+	}) {
+		b.lastUpdatedMS = snapshot.TimestampMs
+	}
+
+	b.refs[name] = ref
+	return b, nil
+}
+
+func (b *MetadataBuilder) SetUUID(uuid uuid.UUID) (*MetadataBuilder, error) {
+	if b.uuid == uuid {
+		return b, nil
+	}
+
+	b.updates = append(b.updates, NewAssignUUIDUpdate(uuid))
+	b.uuid = uuid
+	return b, nil
+}
+
+func (b *MetadataBuilder) buildCommonMetadata() *commonMetadata {
+	return &commonMetadata{
+		FormatVersion:      b.formatVersion,
+		UUID:               b.uuid,
+		Loc:                b.loc,
+		LastUpdatedMS:      b.lastUpdatedMS,
+		LastColumnId:       b.lastColumnId,
+		SchemaList:         b.schemaList,
+		CurrentSchemaID:    b.currentSchemaID,
+		Specs:              b.specs,
+		DefaultSpecID:      b.defaultSpecID,
+		LastPartitionID:    b.lastPartitionID,
+		Props:              b.props,
+		SnapshotList:       b.snapshotList,
+		CurrentSnapshotID:  b.currentSnapshotID,
+		SnapshotLog:        b.snapshotLog,
+		MetadataLog:        b.metadataLog,
+		SortOrderList:      b.sortOrderList,
+		DefaultSortOrderID: b.defaultSortOrderID,
+		SnapshotRefs:       b.refs,
+	}
+}
+
+func (b *MetadataBuilder) GetSchemaByID(id int) (*iceberg.Schema, error) {
+	for _, s := range b.schemaList {
+		if s.ID == id {
+			return s, nil
+		}
+	}
+
+	return nil, fmt.Errorf("%w: schema with id %d not found", iceberg.ErrInvalidArgument, id)
+}
+
+func (b *MetadataBuilder) GetSpecByID(id int) (*iceberg.PartitionSpec, error) {
+	for _, s := range b.specs {
+		if s.ID() == id {
+			return &s, nil
+		}
+	}
+
+	return nil, fmt.Errorf("partition spec with id %d not found", id)
+}
+
+func (b *MetadataBuilder) GetSortOrderByID(id int) (*SortOrder, error) {
+	for _, s := range b.sortOrderList {
+		if s.OrderID == id {
+			return &s, nil
+		}
+	}
+
+	return nil, fmt.Errorf("sort order with id %d not found", id)
+}
+
+func (b *MetadataBuilder) SnapshotByID(id int64) (*Snapshot, error) {
+	for _, s := range b.snapshotList {
+		if s.SnapshotID == id {
+			return &s, nil
+		}
+	}
+
+	return nil, fmt.Errorf("snapshot with id %d not found", id)
+}
+
+func (b *MetadataBuilder) Build() (Metadata, error) {
+	common := b.buildCommonMetadata()
+	switch b.formatVersion {
+	case 1:
+		schema, err := b.GetSchemaByID(b.currentSchemaID)
+		if err != nil {
+			return nil, fmt.Errorf("can't build metadata, missing schema for schema ID %d: %w", b.currentSchemaID, err)
+		}
+
+		partition, err := b.GetSpecByID(b.defaultSpecID)
+		if err != nil {
+			return nil, fmt.Errorf("can't build metadata, missing partition spec for spec ID %d: %w", b.defaultSpecID, err)
+		}
+
+		partitionFields := make([]iceberg.PartitionField, 0)
+		for f := range partition.Fields() {
+			partitionFields = append(partitionFields, f)
+		}
+
+		return &metadataV1{
+			Schema:         schema,
+			Partition:      partitionFields,
+			commonMetadata: *common,
+		}, nil
+
+	case 2:
+		return &metadataV2{
+			LastSequenceNumber: *b.lastSequenceNumber,
+			commonMetadata:     *common,
+		}, nil
+
+	default:
+		panic("unreachable: invalid format version")
+	}
+}
+
+// maxBy returns the maximum value of extract(e) for all e in elems.
+// If elems is empty, returns 0.
+func maxBy[S ~[]E, E any](elems S, extract func(e E) int) int {
+	m := 0
+	for _, e := range elems {
+		m = max(m, extract(e))
+	}
+	return m
+}
+
 var (
 	ErrInvalidMetadataFormatVersion = errors.New("invalid or missing format-version in table metadata")
 	ErrInvalidMetadata              = errors.New("invalid metadata")
@@ -128,9 +662,9 @@ func ParseMetadataBytes(b []byte) (Metadata, error) {
 	var ret Metadata
 	switch ver.FormatVersion {
 	case 1:
-		ret = &MetadataV1{}
+		ret = &metadataV1{}
 	case 2:
-		ret = &MetadataV2{}
+		ret = &metadataV2{}
 	default:
 		return nil, ErrInvalidMetadataFormatVersion
 	}
@@ -163,10 +697,28 @@ type commonMetadata struct {
 	MetadataLog        []MetadataLogEntry      `json:"metadata-log"`
 	SortOrderList      []SortOrder             `json:"sort-orders"`
 	DefaultSortOrderID int                     `json:"default-sort-order-id"`
-	Refs               map[string]SnapshotRef  `json:"refs"`
+	SnapshotRefs       map[string]SnapshotRef  `json:"refs"`
+}
+
+func (c *commonMetadata) Ref() SnapshotRef                     { return c.SnapshotRefs[MainBranch] }
+func (c *commonMetadata) Refs() iter.Seq2[string, SnapshotRef] { return maps.All(c.SnapshotRefs) }
+func (c *commonMetadata) SnapshotLogs() iter.Seq[SnapshotLogEntry] {
+	return slices.Values(c.SnapshotLog)
+}
+
+func (c *commonMetadata) PreviousFiles() iter.Seq[MetadataLogEntry] {
+	return slices.Values(c.MetadataLog)
 }
 
 func (c *commonMetadata) Equals(other *commonMetadata) bool {
+	if other == nil {
+		return false
+	}
+
+	if c == other {
+		return true
+	}
+
 	switch {
 	case c.LastPartitionID == nil && other.LastPartitionID != nil:
 		fallthrough
@@ -187,7 +739,7 @@ func (c *commonMetadata) Equals(other *commonMetadata) bool {
 		fallthrough
 	case !maps.Equal(c.Props, other.Props):
 		fallthrough
-	case !maps.EqualFunc(c.Refs, other.Refs, func(sr1, sr2 SnapshotRef) bool { return sr1.Equals(sr2) }):
+	case !maps.EqualFunc(c.SnapshotRefs, other.SnapshotRefs, func(sr1, sr2 SnapshotRef) bool { return sr1.Equals(sr2) }):
 		return false
 	}
 
@@ -245,7 +797,7 @@ func (c *commonMetadata) SnapshotByID(id int64) *Snapshot {
 }
 
 func (c *commonMetadata) SnapshotByName(name string) *Snapshot {
-	if ref, ok := c.Refs[name]; ok {
+	if ref, ok := c.SnapshotRefs[name]; ok {
 		return c.SnapshotByID(ref.SnapshotID)
 	}
 	return nil
@@ -268,6 +820,10 @@ func (c *commonMetadata) SortOrder() SortOrder {
 	return UnsortedSortOrder
 }
 
+func (c *commonMetadata) DefaultSortOrder() int {
+	return c.DefaultSortOrderID
+}
+
 func (c *commonMetadata) Properties() iceberg.Properties {
 	return c.Props
 }
@@ -284,8 +840,8 @@ func (c *commonMetadata) preValidate() {
 	}
 
 	if c.CurrentSnapshotID != nil {
-		if _, ok := c.Refs[MainBranch]; !ok {
-			c.Refs[MainBranch] = SnapshotRef{
+		if _, ok := c.SnapshotRefs[MainBranch]; !ok {
+			c.SnapshotRefs[MainBranch] = SnapshotRef{
 				SnapshotID:      *c.CurrentSnapshotID,
 				SnapshotRefType: BranchRef,
 			}
@@ -296,8 +852,8 @@ func (c *commonMetadata) preValidate() {
 		c.MetadataLog = []MetadataLogEntry{}
 	}
 
-	if c.Refs == nil {
-		c.Refs = make(map[string]SnapshotRef)
+	if c.SnapshotRefs == nil {
+		c.SnapshotRefs = make(map[string]SnapshotRef)
 	}
 
 	if c.SnapshotLog == nil {
@@ -370,26 +926,30 @@ func (c *commonMetadata) validate() error {
 
 func (c *commonMetadata) Version() int { return c.FormatVersion }
 
-type MetadataV1 struct {
-	Schema    iceberg.Schema           `json:"schema"`
+type metadataV1 struct {
+	Schema    *iceberg.Schema          `json:"schema"`
 	Partition []iceberg.PartitionField `json:"partition-spec"`
 
 	commonMetadata
 }
 
-func (m *MetadataV1) Equals(other Metadata) bool {
-	rhs, ok := other.(*MetadataV1)
+func (m *metadataV1) Equals(other Metadata) bool {
+	rhs, ok := other.(*metadataV1)
 	if !ok {
 		return false
 	}
 
-	return m.Schema.Equals(&rhs.Schema) && slices.Equal(m.Partition, rhs.Partition) &&
+	if m == rhs {
+		return true
+	}
+
+	return m.Schema.Equals(rhs.Schema) && slices.Equal(m.Partition, rhs.Partition) &&
 		m.commonMetadata.Equals(&rhs.commonMetadata)
 }
 
-func (m *MetadataV1) preValidate() {
-	if len(m.SchemaList) == 0 {
-		m.SchemaList = []*iceberg.Schema{&m.Schema}
+func (m *metadataV1) preValidate() {
+	if len(m.SchemaList) == 0 && m.Schema != nil {
+		m.SchemaList = []*iceberg.Schema{m.Schema}
 	}
 
 	if len(m.Specs) == 0 {
@@ -416,8 +976,8 @@ func (m *MetadataV1) preValidate() {
 	m.commonMetadata.preValidate()
 }
 
-func (m *MetadataV1) UnmarshalJSON(b []byte) error {
-	type Alias MetadataV1
+func (m *metadataV1) UnmarshalJSON(b []byte) error {
+	type Alias metadataV1
 	aux := (*Alias)(m)
 
 	if err := json.Unmarshal(b, aux); err != nil {
@@ -428,34 +988,38 @@ func (m *MetadataV1) UnmarshalJSON(b []byte) error {
 	return m.validate()
 }
 
-func (m *MetadataV1) ToV2() MetadataV2 {
+func (m *metadataV1) ToV2() metadataV2 {
 	commonOut := m.commonMetadata
 	commonOut.FormatVersion = 2
 	if commonOut.UUID.String() == "" {
 		commonOut.UUID = uuid.New()
 	}
 
-	return MetadataV2{commonMetadata: commonOut}
+	return metadataV2{commonMetadata: commonOut}
 }
 
-type MetadataV2 struct {
-	LastSequenceNumber int `json:"last-sequence-number"`
+type metadataV2 struct {
+	LastSequenceNumber int64 `json:"last-sequence-number"`
 
 	commonMetadata
 }
 
-func (m *MetadataV2) Equals(other Metadata) bool {
-	rhs, ok := other.(*MetadataV2)
+func (m *metadataV2) Equals(other Metadata) bool {
+	rhs, ok := other.(*metadataV2)
 	if !ok {
 		return false
 	}
 
+	if m == rhs {
+		return true
+	}
+
 	return m.LastSequenceNumber == rhs.LastSequenceNumber &&
 		m.commonMetadata.Equals(&rhs.commonMetadata)
 }
 
-func (m *MetadataV2) UnmarshalJSON(b []byte) error {
-	type Alias MetadataV2
+func (m *metadataV2) UnmarshalJSON(b []byte) error {
+	type Alias metadataV2
 	aux := (*Alias)(m)
 
 	if err := json.Unmarshal(b, aux); err != nil {
diff --git a/table/requirements.go b/table/requirements.go
new file mode 100644
index 0000000..5a4b883
--- /dev/null
+++ b/table/requirements.go
@@ -0,0 +1,267 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table
+
+import (
+	"fmt"
+
+	"github.com/google/uuid"
+)
+
+const (
+	reqAssertCreate                  = "assert-create"
+	reqAssertTableUUID               = "assert-table-uuid"
+	reqAssertRefSnapshotID           = "assert-ref-snapshot-id"
+	reqAssertDefaultSpecID           = "assert-default-spec-id"
+	reqAssertCurrentSchemaID         = "assert-current-schema-id"
+	reqAssertDefaultSortOrderID      = "assert-default-sort-order-id"
+	reqAssertLastAssignedFieldID     = "assert-last-assigned-field-id"
+	reqAssertLastAssignedPartitionID = "assert-last-assigned-partition-id"
+)
+
+// A Requirement is a validation rule that must be satisfied before attempting to
+// make and commit changes to a table. Requirements are used to ensure that the
+// table is in a valid state before making changes.
+type Requirement interface {
+	// Validate checks that the current table metadata satisfies the requirement.
+	Validate(Metadata) error
+}
+
+// baseRequirement is a common struct that all requirements embed. It is used to
+// identify the type of the requirement.
+type baseRequirement struct {
+	Type string `json:"type"`
+}
+
+type assertCreate struct {
+	baseRequirement
+}
+
+// AssertCreate creates a requirement that the table does not already exist.
+func AssertCreate() Requirement {
+	return &assertCreate{
+		baseRequirement: baseRequirement{Type: reqAssertCreate},
+	}
+}
+
+func (a *assertCreate) Validate(meta Metadata) error {
+	if meta != nil {
+		return fmt.Errorf("Table already exists")
+	}
+
+	return nil
+}
+
+type assertTableUuid struct {
+	baseRequirement
+	UUID uuid.UUID `json:"uuid"`
+}
+
+// AssertTableUUID creates a requirement that the table UUID matches the given UUID.
+func AssertTableUUID(uuid uuid.UUID) Requirement {
+	return &assertTableUuid{
+		baseRequirement: baseRequirement{Type: reqAssertTableUUID},
+		UUID:            uuid,
+	}
+}
+
+func (a *assertTableUuid) Validate(meta Metadata) error {
+	if meta == nil {
+		return fmt.Errorf("requirement failed: current table metadata does not exist")
+	}
+
+	if meta.TableUUID() != a.UUID {
+		return fmt.Errorf("UUID mismatch: %s != %s", meta.TableUUID(), a.UUID)
+	}
+
+	return nil
+}
+
+type assertRefSnapshotID struct {
+	baseRequirement
+	Ref        string `json:"ref"`
+	SnapshotID *int64 `json:"snapshot-id"`
+}
+
+// AssertRefSnapshotID creates a requirement which ensures that the table branch
+// or tag identified by the given ref must reference the given snapshot id.
+// If the id is nil, the ref must not already exist.
+func AssertRefSnapshotID(ref string, id *int64) Requirement {
+	return &assertRefSnapshotID{
+		baseRequirement: baseRequirement{Type: reqAssertRefSnapshotID},
+		Ref:             ref,
+		SnapshotID:      id,
+	}
+}
+
+func (a *assertRefSnapshotID) Validate(meta Metadata) error {
+	if meta == nil {
+		return fmt.Errorf("requirement failed: current table metadata does not exist")
+	}
+
+	var r *SnapshotRef
+	for name, ref := range meta.Refs() {
+		if name == a.Ref {
+			r = &ref
+			break
+		}
+	}
+	if r == nil {
+		return fmt.Errorf("requirement failed: branch or tag %s is missing, expected %d", a.Ref, a.SnapshotID)
+	}
+
+	if a.SnapshotID == nil {
+		return fmt.Errorf("requirement failed: %s %s was created concurrently", r.SnapshotRefType, a.Ref)
+	}
+
+	if r.SnapshotID != *a.SnapshotID {
+		return fmt.Errorf("requirement failed: %s %s has changed: expected id %d, found %d", r.SnapshotRefType, a.Ref, a.SnapshotID, r.SnapshotID)
+	}
+
+	return nil
+}
+
+type assertLastAssignedFieldId struct {
+	baseRequirement
+	LastAssignedFieldID int `json:"last-assigned-field-id"`
+}
+
+// AssertLastAssignedFieldID validates that the table's last assigned column ID
+// matches the given id.
+func AssertLastAssignedFieldID(id int) Requirement {
+	return &assertLastAssignedFieldId{
+		baseRequirement:     baseRequirement{Type: reqAssertLastAssignedFieldID},
+		LastAssignedFieldID: id,
+	}
+}
+
+func (a *assertLastAssignedFieldId) Validate(meta Metadata) error {
+	if meta == nil {
+		return fmt.Errorf("requirement failed: current table metadata does not exist")
+	}
+
+	if meta.LastColumnID() != a.LastAssignedFieldID {
+		return fmt.Errorf("requirement failed: last assigned field id has changed: expected %d, found %d", a.LastAssignedFieldID, meta.LastColumnID())
+	}
+
+	return nil
+}
+
+type assertCurrentSchemaId struct {
+	baseRequirement
+	CurrentSchemaID int `json:"current-schema-id"`
+}
+
+// AssertCurrentSchemaId creates a requirement that the table's current schema ID
+// matches the given id.
+func AssertCurrentSchemaID(id int) Requirement {
+	return &assertCurrentSchemaId{
+		baseRequirement: baseRequirement{Type: reqAssertCurrentSchemaID},
+		CurrentSchemaID: id,
+	}
+}
+
+func (a *assertCurrentSchemaId) Validate(meta Metadata) error {
+	if meta == nil {
+		return fmt.Errorf("requirement failed: current table metadata does not exist")
+	}
+
+	if meta.CurrentSchema().ID != a.CurrentSchemaID {
+		return fmt.Errorf("requirement failed: current schema id has changed: expected %d, found %d", a.CurrentSchemaID, meta.CurrentSchema().ID)
+	}
+
+	return nil
+}
+
+type assertLastAssignedPartitionId struct {
+	baseRequirement
+	LastAssignedPartitionID int `json:"last-assigned-partition-id"`
+}
+
+// AssertLastAssignedPartitionID creates a requriement that the table's last assigned partition ID
+// matches the given id.
+func AssertLastAssignedPartitionID(id int) Requirement {
+	return &assertLastAssignedPartitionId{
+		baseRequirement:         baseRequirement{Type: reqAssertLastAssignedPartitionID},
+		LastAssignedPartitionID: id,
+	}
+}
+
+func (a *assertLastAssignedPartitionId) Validate(meta Metadata) error {
+	if meta == nil {
+		return fmt.Errorf("requirement failed: current table metadata does not exist")
+	}
+
+	if *meta.LastPartitionSpecID() != a.LastAssignedPartitionID {
+		return fmt.Errorf("requirement failed: last assigned partition id has changed: expected %d, found %d", a.LastAssignedPartitionID, *meta.LastPartitionSpecID())
+	}
+
+	return nil
+}
+
+type assertDefaultSpecId struct {
+	baseRequirement
+	DefaultSpecID int `json:"default-spec-id"`
+}
+
+// AssertDefaultSpecID creates a requirement that the table's default partition spec ID
+// matches the given id.
+func AssertDefaultSpecID(id int) Requirement {
+	return &assertDefaultSpecId{
+		baseRequirement: baseRequirement{Type: reqAssertDefaultSpecID},
+		DefaultSpecID:   id,
+	}
+}
+
+func (a *assertDefaultSpecId) Validate(meta Metadata) error {
+	if meta == nil {
+		return fmt.Errorf("requirement failed: current table metadata does not exist")
+	}
+
+	if meta.DefaultPartitionSpec() != a.DefaultSpecID {
+		return fmt.Errorf("requirement failed: default spec id has changed: expected %d, found %d", a.DefaultSpecID, meta.DefaultPartitionSpec())
+	}
+
+	return nil
+}
+
+type assertDefaultSortOrderId struct {
+	baseRequirement
+	DefaultSortOrderID int `json:"default-sort-order-id"`
+}
+
+// AssertDefaultSortOrderID creates a requirement that the table's default sort order ID
+// matches the given id.
+func AssertDefaultSortOrderID(id int) Requirement {
+	return &assertDefaultSortOrderId{
+		baseRequirement:    baseRequirement{Type: reqAssertDefaultSortOrderID},
+		DefaultSortOrderID: id,
+	}
+}
+
+func (a *assertDefaultSortOrderId) Validate(meta Metadata) error {
+	if meta == nil {
+		return fmt.Errorf("requirement failed: current table metadata does not exist")
+	}
+
+	if meta.DefaultSortOrder() != a.DefaultSortOrderID {
+		return fmt.Errorf("requirement failed: default sort order id has changed: expected %d, found %d", a.DefaultSortOrderID, meta.DefaultSortOrder())
+	}
+
+	return nil
+}
diff --git a/table/updates.go b/table/updates.go
new file mode 100644
index 0000000..7b09a0f
--- /dev/null
+++ b/table/updates.go
@@ -0,0 +1,379 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table
+
+import (
+	"fmt"
+
+	"github.com/apache/iceberg-go"
+	"github.com/google/uuid"
+)
+
+const (
+	updateSpec             = "add-spec"
+	updateAddSchema        = "add-schema"
+	updateSortOrder        = "add-sort-order"
+	updateAssignUUID       = "assign-uuid"
+	updateDefaultSpec      = "set-default-spec"
+	updateCurrentSchema    = "set-current-schema"
+	updateDefaultSortOrder = "set-default-sort-order"
+	updateUpgradeFormat    = "upgrade-format-version"
+)
+
+// Update represents a change to a table's metadata.
+type Update interface {
+	// Action returns the name of the action that the update represents.
+	Action() string
+	// Apply applies the update to the given metadata builder.
+	Apply(*MetadataBuilder) error
+}
+
+// baseUpdate contains the common fields for all updates. It is used to identify the type
+// of the update.
+type baseUpdate struct {
+	ActionName string `json:"action"`
+}
+
+func (u *baseUpdate) Action() string {
+	return u.ActionName
+}
+
+type assignUUIDUpdate struct {
+	baseUpdate
+	UUID uuid.UUID `json:"uuid"`
+}
+
+// NewAssignUUIDUpdate creates a new update to assign a UUID to the table metadata.
+func NewAssignUUIDUpdate(uuid uuid.UUID) Update {
+	return &assignUUIDUpdate{
+		baseUpdate: baseUpdate{ActionName: updateAssignUUID},
+		UUID:       uuid,
+	}
+}
+
+func (u *assignUUIDUpdate) Apply(builder *MetadataBuilder) error {
+	_, err := builder.SetUUID(u.UUID)
+	return err
+}
+
+type upgradeFormatVersionUpdate struct {
+	baseUpdate
+	FormatVersion int `json:"format-version"`
+}
+
+// NewUpgradeFormatVersionUpdate creates a new update that upgrades the format version
+// of the table metadata to the given formatVersion.
+func NewUpgradeFormatVersionUpdate(formatVersion int) Update {
+	return &upgradeFormatVersionUpdate{
+		baseUpdate:    baseUpdate{ActionName: updateUpgradeFormat},
+		FormatVersion: formatVersion,
+	}
+}
+
+func (u *upgradeFormatVersionUpdate) Apply(builder *MetadataBuilder) error {
+	_, err := builder.SetFormatVersion(u.FormatVersion)
+	return err
+}
+
+type addSchemaUpdate struct {
+	baseUpdate
+	Schema       *iceberg.Schema `json:"schema"`
+	LastColumnID int             `json:"last-column-id"`
+	initial      bool
+}
+
+// NewAddSchemaUpdate creates a new update that adds the given schema and last column ID to
+// the table metadata. If the initial flag is set to true, the schema is considered the initial
+// schema of the table, and all previously added schemas in the metadata builder are removed.
+func NewAddSchemaUpdate(schema *iceberg.Schema, lastColumnID int, initial bool) Update {
+	return &addSchemaUpdate{
+		baseUpdate:   baseUpdate{ActionName: updateAddSchema},
+		Schema:       schema,
+		LastColumnID: lastColumnID,
+		initial:      initial,
+	}
+}
+
+func (u *addSchemaUpdate) Apply(builder *MetadataBuilder) error {
+	_, err := builder.AddSchema(u.Schema, u.LastColumnID, u.initial)
+	return err
+}
+
+type setCurrentSchemaUpdate struct {
+	baseUpdate
+	SchemaID int `json:"schema-id"`
+}
+
+// NewSetCurrentSchemaUpdate creates a new update that sets the current schema of the table
+// metadata to the given schema ID.
+func NewSetCurrentSchemaUpdate(id int) Update {
+	return &setCurrentSchemaUpdate{
+		baseUpdate: baseUpdate{ActionName: updateCurrentSchema},
+		SchemaID:   id,
+	}
+}
+
+func (u *setCurrentSchemaUpdate) Apply(builder *MetadataBuilder) error {
+	_, err := builder.SetCurrentSchemaID(u.SchemaID)
+	return err
+}
+
+type addPartitionSpecUpdate struct {
+	baseUpdate
+	Spec    *iceberg.PartitionSpec `json:"spec"`
+	initial bool
+}
+
+// NewAddPartitionSpecUpdate creates a new update that adds the given partition spec to the table
+// metadata. If the initial flag is set to true, the spec is considered the initial spec of the table,
+// and all other previously added specs in the metadata builder are removed.
+func NewAddPartitionSpecUpdate(spec *iceberg.PartitionSpec, initial bool) Update {
+	return &addPartitionSpecUpdate{
+		baseUpdate: baseUpdate{ActionName: updateSpec},
+		Spec:       spec,
+		initial:    initial,
+	}
+}
+
+func (u *addPartitionSpecUpdate) Apply(builder *MetadataBuilder) error {
+	_, err := builder.AddPartitionSpec(u.Spec, u.initial)
+	return err
+}
+
+type setDefaultSpecUpdate struct {
+	baseUpdate
+	SpecID int `json:"spec-id"`
+}
+
+// NewSetDefaultSpecUpdate creates a new update that sets the default partition spec of the
+// table metadata to the given spec ID.
+func NewSetDefaultSpecUpdate(id int) Update {
+	return &setDefaultSpecUpdate{
+		baseUpdate: baseUpdate{ActionName: updateDefaultSpec},
+		SpecID:     id,
+	}
+}
+
+func (u *setDefaultSpecUpdate) Apply(builder *MetadataBuilder) error {
+	_, err := builder.SetDefaultSpecID(u.SpecID)
+	return err
+}
+
+type addSortOrderUpdate struct {
+	baseUpdate
+	SortOrder *SortOrder `json:"sort-order"`
+	initial   bool
+}
+
+// NewAddSortOrderUpdate creates a new update that adds the given sort order to the table metadata.
+// If the initial flag is set to true, the sort order is considered the initial sort order of the table,
+// and all previously added sort orders in the metadata builder are removed.
+func NewAddSortOrderUpdate(sortOrder *SortOrder, initial bool) Update {
+	return &addSortOrderUpdate{
+		baseUpdate: baseUpdate{ActionName: updateSortOrder},
+		SortOrder:  sortOrder,
+		initial:    initial,
+	}
+}
+
+func (u *addSortOrderUpdate) Apply(builder *MetadataBuilder) error {
+	_, err := builder.AddSortOrder(u.SortOrder, u.initial)
+	return err
+}
+
+type setDefaultSortOrderUpdate struct {
+	baseUpdate
+	SortOrderID int `json:"sort-order-id"`
+}
+
+// NewSetDefaultSortOrderUpdate creates a new update that sets the default sort order of the table metadata
+// to the given sort order ID.
+func NewSetDefaultSortOrderUpdate(id int) Update {
+	return &setDefaultSortOrderUpdate{
+		baseUpdate:  baseUpdate{ActionName: updateDefaultSortOrder},
+		SortOrderID: id,
+	}
+}
+
+func (u *setDefaultSortOrderUpdate) Apply(builder *MetadataBuilder) error {
+	_, err := builder.SetDefaultSortOrderID(u.SortOrderID)
+	return err
+}
+
+type addSnapshotUpdate struct {
+	baseUpdate
+	Snapshot *Snapshot `json:"snapshot"`
+}
+
+// NewAddSnapshotUpdate creates a new update that adds the given snapshot to the table metadata.
+func NewAddSnapshotUpdate(snapshot *Snapshot) Update {
+	return &addSnapshotUpdate{
+		baseUpdate: baseUpdate{ActionName: "add-snapshot"},
+		Snapshot:   snapshot,
+	}
+}
+
+func (u *addSnapshotUpdate) Apply(builder *MetadataBuilder) error {
+	_, err := builder.AddSnapshot(u.Snapshot)
+	return err
+}
+
+type setSnapshotRefUpdate struct {
+	baseUpdate
+	RefName            string  `json:"ref-name"`
+	RefType            RefType `json:"type"`
+	SnapshotID         int64   `json:"snapshot-id"`
+	MaxRefAgeMs        int64   `json:"max-ref-age-ms,omitempty"`
+	MaxSnapshotAgeMs   int64   `json:"max-snapshot-age-ms,omitempty"`
+	MinSnapshotsToKeep int     `json:"min-snapshots-to-keep,omitempty"`
+}
+
+// NewSetSnapshotRefUpdate creates a new update that sets the given snapshot reference
+// as the current snapshot of the table metadata. MaxRefAgeMs, MaxSnapshotAgeMs,
+// and MinSnapshotsToKeep are optional, and any non-positive values are ignored.
+func NewSetSnapshotRefUpdate(
+	name string,
+	snapshotID int64,
+	refType RefType,
+	maxRefAgeMs, maxSnapshotAgeMs int64,
+	minSnapshotsToKeep int,
+) Update {
+	return &setSnapshotRefUpdate{
+		baseUpdate:         baseUpdate{ActionName: "set-snapshot-ref"},
+		RefName:            name,
+		RefType:            refType,
+		SnapshotID:         snapshotID,
+		MaxRefAgeMs:        maxRefAgeMs,
+		MaxSnapshotAgeMs:   maxSnapshotAgeMs,
+		MinSnapshotsToKeep: minSnapshotsToKeep,
+	}
+}
+
+func (u *setSnapshotRefUpdate) Apply(builder *MetadataBuilder) error {
+	opts := []setSnapshotRefOption{}
+	if u.MaxRefAgeMs >= 0 {
+		opts = append(opts, WithMaxRefAgeMs(u.MaxRefAgeMs))
+	}
+	if u.MaxSnapshotAgeMs >= 0 {
+		opts = append(opts, WithMaxSnapshotAgeMs(u.MaxSnapshotAgeMs))
+	}
+	if u.MinSnapshotsToKeep >= 0 {
+		opts = append(opts, WithMinSnapshotsToKeep(u.MinSnapshotsToKeep))
+	}
+
+	_, err := builder.SetSnapshotRef(
+		u.RefName,
+		u.SnapshotID,
+		u.RefType,
+		opts...,
+	)
+	return err
+}
+
+type setLocationUpdate struct {
+	baseUpdate
+	Location string `json:"location"`
+}
+
+// NewSetLocationUpdate creates a new update that sets the location of the table metadata.
+func NewSetLocationUpdate(loc string) Update {
+	return &setLocationUpdate{
+		baseUpdate: baseUpdate{ActionName: "set-location"},
+		Location:   loc,
+	}
+}
+
+func (u *setLocationUpdate) Apply(builder *MetadataBuilder) error {
+	_, err := builder.SetLoc(u.Location)
+	return err
+}
+
+type setPropertiesUpdate struct {
+	baseUpdate
+	Updates iceberg.Properties `json:"updates"`
+}
+
+// NewSetPropertiesUpdate creates a new update that sets the given properties in the
+// table metadata.
+func NewSetPropertiesUpdate(updates iceberg.Properties) *setPropertiesUpdate {
+	return &setPropertiesUpdate{
+		baseUpdate: baseUpdate{ActionName: "set-properties"},
+		Updates:    updates,
+	}
+}
+
+func (u *setPropertiesUpdate) Apply(builder *MetadataBuilder) error {
+	_, err := builder.SetProperties(u.Updates)
+	return err
+}
+
+type removePropertiesUpdate struct {
+	baseUpdate
+	Removals []string `json:"removals"`
+}
+
+// NewRemovePropertiesUpdate creates a new update that removes properties from the table metadata.
+// The properties are identified by their names, and if a property with the given name does not exist,
+// it is ignored.
+func NewRemovePropertiesUpdate(removals []string) Update {
+	return &removePropertiesUpdate{
+		baseUpdate: baseUpdate{ActionName: "remove-properties"},
+		Removals:   removals,
+	}
+}
+
+func (u *removePropertiesUpdate) Apply(builder *MetadataBuilder) error {
+	_, err := builder.RemoveProperties(u.Removals)
+	return err
+}
+
+type removeSnapshotsUpdate struct {
+	baseUpdate
+	SnapshotIDs []int64 `json:"snapshot-ids"`
+}
+
+// NewRemoveSnapshotsUpdate creates a new update that removes all snapshots from
+// the table metadata with the given snapshot IDs.
+func NewRemoveSnapshotsUpdate(ids []int64) Update {
+	return &removeSnapshotsUpdate{
+		baseUpdate:  baseUpdate{ActionName: "remove-snapshots"},
+		SnapshotIDs: ids,
+	}
+}
+
+func (u *removeSnapshotsUpdate) Apply(builder *MetadataBuilder) error {
+	return fmt.Errorf("%w: remove-snapshots", iceberg.ErrNotImplemented)
+}
+
+type removeSnapshotRefUpdate struct {
+	baseUpdate
+	RefName string `json:"ref-name"`
+}
+
+// NewRemoveSnapshotRefUpdate creates a new update that removes a snapshot reference
+// from the table metadata.
+func NewRemoveSnapshotRefUpdate(ref string) *removeSnapshotRefUpdate {
+	return &removeSnapshotRefUpdate{
+		baseUpdate: baseUpdate{ActionName: "remove-snapshot-ref"},
+		RefName:    ref,
+	}
+}
+
+func (u *removeSnapshotRefUpdate) Apply(builder *MetadataBuilder) error {
+	return fmt.Errorf("%w: remove-snapshot-ref", iceberg.ErrNotImplemented)
+}


Test patch:
diff --git a/catalog/rest_test.go b/catalog/rest_test.go
index 618c5e0..212d5a8 100644
--- a/catalog/rest_test.go
+++ b/catalog/rest_test.go
@@ -22,6 +22,7 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	"encoding/json"
+	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
@@ -624,6 +625,105 @@ func (r *RestCatalogSuite) TestUpdateNamespaceProps404() {
 	r.ErrorContains(err, "Namespace does not exist: does_not_exist in warehouse")
 }
 
+var (
+	exampleTableMetadataNoSnapshotV1 = `{
+	"format-version": 1,
+	"table-uuid": "bf289591-dcc0-4234-ad4f-5c3eed811a29",
+	"location": "s3://warehouse/database/table",
+	"last-updated-ms": 1657810967051,
+	"last-column-id": 3,
+	"schema": {
+		"type": "struct",
+		"schema-id": 0,
+		"identifier-field-ids": [2],
+		"fields": [
+			{"id": 1, "name": "foo", "required": false, "type": "string"},
+			{"id": 2, "name": "bar", "required": true, "type": "int"},
+			{"id": 3, "name": "baz", "required": false, "type": "boolean"}
+		]
+	},
+	"current-schema-id": 0,
+	"schemas": [
+		{
+			"type": "struct",
+			"schema-id": 0,
+			"identifier-field-ids": [2],
+			"fields": [
+				{"id": 1, "name": "foo", "required": false, "type": "string"},
+				{"id": 2, "name": "bar", "required": true, "type": "int"},
+				{"id": 3, "name": "baz", "required": false, "type": "boolean"}
+			]
+		}
+	],
+	"partition-spec": [],
+	"default-spec-id": 0,
+	"last-partition-id": 999,
+	"default-sort-order-id": 0,
+	"sort-orders": [{"order-id": 0, "fields": []}],
+	"properties": {
+		"write.delete.parquet.compression-codec": "zstd",
+		"write.metadata.compression-codec": "gzip",
+		"write.summary.partition-limit": "100",
+		"write.parquet.compression-codec": "zstd"
+	},
+	"current-snapshot-id": -1,
+	"refs": {},
+	"snapshots": [],
+	"snapshot-log": [],
+	"metadata-log": []
+}`
+
+	createTableRestExample = fmt.Sprintf(`{
+	"metadata-location": "s3://warehouse/database/table/metadata.json",
+	"metadata": %s,
+	"config": {
+		"client.factory": "io.tabular.iceberg.catalog.TabularAwsClientFactory",
+		"region": "us-west-2"
+	}
+}`, exampleTableMetadataNoSnapshotV1)
+
+	tableSchemaSimple = iceberg.NewSchemaWithIdentifiers(1, []int{2},
+		iceberg.NestedField{ID: 1, Name: "foo", Type: iceberg.StringType{}, 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},
+	)
+)
+
+func (r *RestCatalogSuite) TestCreateTable200() {
+	r.mux.HandleFunc("/v1/namespaces/fokko/tables", func(w http.ResponseWriter, req *http.Request) {
+		r.Require().Equal(http.MethodPost, req.Method)
+
+		for k, v := range TestHeaders {
+			r.Equal(v, req.Header.Values(k))
+		}
+
+		w.Write([]byte(createTableRestExample))
+	})
+
+	t := createTableRestExample
+	_ = t
+	cat, err := catalog.NewRestCatalog("rest", r.srv.URL, catalog.WithOAuthToken(TestToken))
+	r.Require().NoError(err)
+
+	tbl, err := cat.CreateTable(
+		context.Background(),
+		catalog.ToRestIdentifier("fokko", "fokko2"),
+		tableSchemaSimple,
+	)
+	r.Require().NoError(err)
+
+	r.Equal(catalog.ToRestIdentifier("rest", "fokko", "fokko2"), tbl.Identifier())
+	r.Equal("s3://warehouse/database/table/metadata.json", tbl.MetadataLocation())
+	r.EqualValues(1, tbl.Metadata().Version())
+	r.Equal("bf289591-dcc0-4234-ad4f-5c3eed811a29", tbl.Metadata().TableUUID().String())
+	r.EqualValues(1657810967051, tbl.Metadata().LastUpdatedMillis())
+	r.Equal(3, tbl.Metadata().LastColumnID())
+	r.Zero(tbl.Schema().ID)
+	r.Zero(tbl.Metadata().DefaultPartitionSpec())
+	r.Equal(999, *tbl.Metadata().LastPartitionSpecID())
+	r.Equal(table.UnsortedSortOrder, tbl.SortOrder())
+}
+
 func (r *RestCatalogSuite) TestLoadTable200() {
 	r.mux.HandleFunc("/v1/namespaces/fokko/tables/table", func(w http.ResponseWriter, req *http.Request) {
 		r.Require().Equal(http.MethodGet, req.Method)
diff --git a/table/metadata_test.go b/table/metadata_internal_test.go
similarity index 93%
rename from table/metadata_test.go
rename to table/metadata_internal_test.go
index e268d88..a02ac7f 100644
--- a/table/metadata_test.go
+++ b/table/metadata_internal_test.go
@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-package table_test
+package table
 
 import (
 	"encoding/json"
@@ -23,7 +23,6 @@ import (
 	"testing"
 
 	"github.com/apache/iceberg-go"
-	"github.com/apache/iceberg-go/table"
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -112,14 +111,14 @@ const ExampleTableMetadataV1 = `{
 }`
 
 func TestMetadataV1Parsing(t *testing.T) {
-	meta, err := table.ParseMetadataBytes([]byte(ExampleTableMetadataV1))
+	meta, err := ParseMetadataBytes([]byte(ExampleTableMetadataV1))
 	require.NoError(t, err)
 	require.NotNil(t, meta)
 
-	assert.IsType(t, (*table.MetadataV1)(nil), meta)
+	assert.IsType(t, (*metadataV1)(nil), meta)
 	assert.Equal(t, 1, meta.Version())
 
-	data := meta.(*table.MetadataV1)
+	data := meta.(*metadataV1)
 	assert.Equal(t, uuid.MustParse("d20125c8-7284-442c-9aea-15fee620737c"), meta.TableUUID())
 	assert.Equal(t, "s3://bucket/test/location", meta.Location())
 	assert.Equal(t, int64(1602638573874), meta.LastUpdatedMillis())
@@ -156,21 +155,21 @@ func TestMetadataV1Parsing(t *testing.T) {
 	assert.Nil(t, meta.SnapshotByID(0))
 	assert.Nil(t, meta.SnapshotByName("foo"))
 	assert.Zero(t, data.DefaultSortOrderID)
-	assert.Equal(t, table.UnsortedSortOrder, meta.SortOrder())
+	assert.Equal(t, UnsortedSortOrder, meta.SortOrder())
 }
 
 func TestMetadataV2Parsing(t *testing.T) {
-	meta, err := table.ParseMetadataBytes([]byte(ExampleTableMetadataV2))
+	meta, err := ParseMetadataBytes([]byte(ExampleTableMetadataV2))
 	require.NoError(t, err)
 	require.NotNil(t, meta)
 
-	assert.IsType(t, (*table.MetadataV2)(nil), meta)
+	assert.IsType(t, (*metadataV2)(nil), meta)
 	assert.Equal(t, 2, meta.Version())
 
-	data := meta.(*table.MetadataV2)
+	data := meta.(*metadataV2)
 	assert.Equal(t, uuid.MustParse("9c12d441-03fe-4693-9a96-a0705ddf69c1"), data.UUID)
 	assert.Equal(t, "s3://bucket/test/location", data.Location())
-	assert.Equal(t, 34, data.LastSequenceNumber)
+	assert.Equal(t, int64(34), data.LastSequenceNumber)
 	assert.Equal(t, int64(1602638573590), data.LastUpdatedMS)
 	assert.Equal(t, 3, data.LastColumnId)
 	assert.Equal(t, 0, data.SchemaList[0].ID)
@@ -192,7 +191,7 @@ func TestMetadataV2Parsing(t *testing.T) {
 }
 
 func TestParsingCorrectTypes(t *testing.T) {
-	var meta table.MetadataV2
+	var meta metadataV2
 	require.NoError(t, json.Unmarshal([]byte(ExampleTableMetadataV2), &meta))
 
 	assert.IsType(t, &iceberg.Schema{}, meta.SchemaList[0])
@@ -201,7 +200,7 @@ func TestParsingCorrectTypes(t *testing.T) {
 }
 
 func TestSerializeMetadataV1(t *testing.T) {
-	var meta table.MetadataV1
+	var meta metadataV1
 	require.NoError(t, json.Unmarshal([]byte(ExampleTableMetadataV1), &meta))
 
 	data, err := json.Marshal(&meta)
@@ -212,7 +211,7 @@ func TestSerializeMetadataV1(t *testing.T) {
 }
 
 func TestSerializeMetadataV2(t *testing.T) {
-	var meta table.MetadataV2
+	var meta metadataV2
 	require.NoError(t, json.Unmarshal([]byte(ExampleTableMetadataV2), &meta))
 
 	data, err := json.Marshal(&meta)
@@ -243,9 +242,9 @@ func TestInvalidFormatVersion(t *testing.T) {
         "snapshots": []
     }`
 
-	_, err := table.ParseMetadataBytes([]byte(metadataInvalidFormat))
+	_, err := ParseMetadataBytes([]byte(metadataInvalidFormat))
 	assert.Error(t, err)
-	assert.ErrorIs(t, err, table.ErrInvalidMetadataFormatVersion)
+	assert.ErrorIs(t, err, ErrInvalidMetadataFormatVersion)
 }
 
 func TestCurrentSchemaNotFound(t *testing.T) {
@@ -278,9 +277,9 @@ func TestCurrentSchemaNotFound(t *testing.T) {
         "snapshots": []
     }`
 
-	_, err := table.ParseMetadataBytes([]byte(schemaNotFound))
+	_, err := ParseMetadataBytes([]byte(schemaNotFound))
 	assert.Error(t, err)
-	assert.ErrorIs(t, err, table.ErrInvalidMetadata)
+	assert.ErrorIs(t, err, ErrInvalidMetadata)
 	assert.ErrorContains(t, err, "current-schema-id 2 can't be found in any schema")
 }
 
@@ -322,9 +321,9 @@ func TestSortOrderNotFound(t *testing.T) {
         "snapshots": []
     }`
 
-	_, err := table.ParseMetadataBytes([]byte(metadataSortOrderNotFound))
+	_, err := ParseMetadataBytes([]byte(metadataSortOrderNotFound))
 	assert.Error(t, err)
-	assert.ErrorIs(t, err, table.ErrInvalidMetadata)
+	assert.ErrorIs(t, err, ErrInvalidMetadata)
 	assert.ErrorContains(t, err, "default-sort-order-id 4 can't be found in [3: [\n2 asc nulls-first\nbucket[4](3) desc nulls-last\n]]")
 }
 
@@ -358,10 +357,10 @@ func TestSortOrderUnsorted(t *testing.T) {
         "snapshots": []
     }`
 
-	var meta table.MetadataV2
+	var meta metadataV2
 	require.NoError(t, json.Unmarshal([]byte(sortOrderUnsorted), &meta))
 
-	assert.Equal(t, table.UnsortedSortOrderID, meta.DefaultSortOrderID)
+	assert.Equal(t, UnsortedSortOrderID, meta.DefaultSortOrderID)
 	assert.Len(t, meta.SortOrderList, 0)
 }
 
@@ -394,28 +393,28 @@ func TestInvalidPartitionSpecID(t *testing.T) {
         "last-partition-id": 1000
     }`
 
-	var meta table.MetadataV2
+	var meta metadataV2
 	err := json.Unmarshal([]byte(invalidSpecID), &meta)
-	assert.ErrorIs(t, err, table.ErrInvalidMetadata)
+	assert.ErrorIs(t, err, ErrInvalidMetadata)
 	assert.ErrorContains(t, err, "default-spec-id 1 can't be found")
 }
 
 func TestV2RefCreation(t *testing.T) {
-	var meta table.MetadataV2
+	var meta metadataV2
 	require.NoError(t, json.Unmarshal([]byte(ExampleTableMetadataV2), &meta))
 
 	maxRefAge := int64(10000000)
-	assert.Equal(t, map[string]table.SnapshotRef{
+	assert.Equal(t, map[string]SnapshotRef{
 		"main": {
 			SnapshotID:      3055729675574597004,
-			SnapshotRefType: table.BranchRef,
+			SnapshotRefType: BranchRef,
 		},
 		"test": {
 			SnapshotID:      3051729675574597004,
-			SnapshotRefType: table.TagRef,
+			SnapshotRefType: TagRef,
 			MaxRefAgeMs:     &maxRefAge,
 		},
-	}, meta.Refs)
+	}, meta.SnapshotRefs)
 }
 
 func TestV1WriteMetadataToV2(t *testing.T) {
@@ -453,11 +452,11 @@ func TestV1WriteMetadataToV2(t *testing.T) {
 		"snapshots": [{"snapshot-id": 1925, "timestamp-ms": 1602638573822}]
 	}`
 
-	meta, err := table.ParseMetadataString(minimalV1Example)
+	meta, err := ParseMetadataString(minimalV1Example)
 	require.NoError(t, err)
-	assert.IsType(t, (*table.MetadataV1)(nil), meta)
+	assert.IsType(t, (*metadataV1)(nil), meta)
 
-	metaV2 := meta.(*table.MetadataV1).ToV2()
+	metaV2 := meta.(*metadataV1).ToV2()
 	metaV2Json, err := json.Marshal(metaV2)
 	require.NoError(t, err)
 
diff --git a/table/table_test.go b/table/table_test.go
index cde94ab..f09054c 100644
--- a/table/table_test.go
+++ b/table/table_test.go
@@ -41,7 +41,7 @@ func (t *TableTestSuite) SetupSuite() {
 	var mockfs internal.MockFS
 	mockfs.Test(t.T())
 	mockfs.On("Open", "s3://bucket/test/location/uuid.metadata.json").
-		Return(&internal.MockFile{Contents: bytes.NewReader([]byte(ExampleTableMetadataV2))}, nil)
+		Return(&internal.MockFile{Contents: bytes.NewReader([]byte(table.ExampleTableMetadataV2))}, nil)
 	defer mockfs.AssertExpectations(t.T())
 
 	tbl, err := table.NewFromLocation([]string{"foo"}, "s3://bucket/test/location/uuid.metadata.json", &mockfs)
@@ -59,7 +59,7 @@ func (t *TableTestSuite) TestNewTableFromReadFile() {
 	var mockfsReadFile internal.MockFSReadFile
 	mockfsReadFile.Test(t.T())
 	mockfsReadFile.On("ReadFile", "s3://bucket/test/location/uuid.metadata.json").
-		Return([]byte(ExampleTableMetadataV2), nil)
+		Return([]byte(table.ExampleTableMetadataV2), nil)
 	defer mockfsReadFile.AssertExpectations(t.T())
 
 	tbl2, err := table.NewFromLocation([]string{"foo"}, "s3://bucket/test/location/uuid.metadata.json", &mockfsReadFile)