Repository: apache/iceberg-go
Pull request #58 (2024-01-31 19:14:15)

feat(catalog): add initial rest catalog impl
Adding an initial implementation and unit tests for the Rest catalog.

zeroshade: CC @Fokko @wolfeidau @nastra @HonahX @jackye1995 
nastra: I think what's currently missing is having a way to configure the `warehouse` (which I hardcoded for testing) but also handling the signing part of requests against S3, similar to https://github.com/apache/iceberg-python/blob/f66e3652fdf9720d6c63a6fcec7bcd08d5bb186c/pyiceberg/io/fsspec.py#L70-L95

Listing files via `go run ./cmd/iceberg files iceberg124.foobar --catalog rest --uri https://api.dev.tabular.io/ws/ --credential <creds>` will fail with
```
2024/02/13 10:24:07 could not open manifest file: operation error S3: GetObject, https response error StatusCode: 403, RequestID: 066G7WZD23KHZCBJ, HostID: d4V0iCd2uzvp9gZJWDOWmljaREgSaL9Iro0XxOFsv38ECJpdCd/JHWG8Y6/i7oSal8cONZ87Tis=, api error AccessDenied: Access Denied
exit status 1
```
I believe this is because `FileIO` isn't configured with the `TOKEN` in the authorization header that's coming back from the `config` inside `tblResponse` [here](https://github.com/apache/iceberg-go/blob/e530020e2cc06637bda293fc1ad49e99bd376b94/catalog/rest.go#L616). Reading all other metadata of tables work via CLI, but this is because those never use `FileIO` and only `files` does that atm.
zeroshade: @nastra 

Hmm.  So, setting the env vars `AWS_REGION`, `AWS_ACCESS_KEY_ID` and `AWS_SECRET_ACCESS_KEY` should all work and get picked up by the FileIO. But I haven't tried testing with `https://api.dev.tabular.io/ws/` before. 

There is the ability to set a session token via the `s3.session-token` property but you're right that I don't think it gets propagated. Is there any special configuration I need to set up in order to try testing out the api.dev.tabular.io/ws/ uri myself?
zeroshade: @nastra So I've figured out the issue:

The properties are correctly being propagated to the FileIO object, however it looks like the tabular api doesn't like the Go Iceberg user-agent.

I loaded up `pyiceberg` to see what it does differently and how it works, and saw that the request for the table included in its response a series of s3 properties including an access-key-id, session-token, and secret-access-key in the config. When I looked at the same request from the Go cli those properties weren't there. If I hardcode and change the User-Agent that the Go CLI passes to be `PyIceberg/0.5.1` suddenly those properties are returned and loading the manifests works just fine. So the problem is definitely the fact that the User-Agent isn't recognized by the tabular rest catalog enough for it to send the s3 key properties.

Anything we can do on the tabular side?
 During `RestCatalog.LoadTable` 

Patch:
diff --git a/.github/workflows/go-ci.yml b/.github/workflows/go-ci.yml
index 5f4c42a..a6014fd 100644
--- a/.github/workflows/go-ci.yml
+++ b/.github/workflows/go-ci.yml
@@ -39,7 +39,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        go: [ '1.20', '1.21' ]
+        go: [ '1.21' ]
         os: [ 'ubuntu-latest', 'windows-latest', 'macos-latest' ]
     steps:
     - uses: actions/checkout@v3
diff --git a/.gitignore b/.gitignore
index 6c1bcf4..8b398c6 100644
--- a/.gitignore
+++ b/.gitignore
@@ -46,3 +46,7 @@ lib/
 *.iml
 
 .envrc*
+
+# local catalog environment via docker
+dev/notebooks
+dev/warehouse
\ No newline at end of file
diff --git a/catalog/README.md b/catalog/README.md
new file mode 100644
index 0000000..561cccb
--- /dev/null
+++ b/catalog/README.md
@@ -0,0 +1,123 @@
+<!--
+ - 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.
+ -->
+
+# Catalog Implementations
+
+## Integration Testing
+
+The Catalog implementations can be manually tested using the CLI implemented
+in the `cmd/iceberg` folder.
+
+### REST Catalog
+
+To test the REST catalog implementation, we have a docker configuration
+for a Minio container and tabluario/iceberg-rest container.
+
+You can spin up the local catalog by going to the `dev/` folder and running
+`docker-compose up`. You can then follow the steps of the Iceberg [Quickstart](https://iceberg.apache.org/spark-quickstart/#creating-a-table) 
+tutorial, which we've summarized below.
+
+#### Setup your Iceberg catalog
+
+First launch a pyspark console by running:
+
+```bash
+docker exec -it spark-iceberg pyspark
+```
+
+Once in the pyspark shell, we create a simple table with a namespace of 
+"demo.nyc" called "taxis":
+
+```python
+from pyspark.sql.types import DoubleType, FloatType, LongType, StructType,StructField, StringType
+schema = StructType([
+  StructField("vendor_id", LongType(), True),
+  StructField("trip_id", LongType(), True),
+  StructField("trip_distance", FloatType(), True),
+  StructField("fare_amount", DoubleType(), True),
+  StructField("store_and_fwd_flag", StringType(), True)
+])
+
+df = spark.createDataFrame([], schema)
+df.writeTo("demo.nyc.taxis").create()
+```
+
+Finally, we write another data-frame to the table to add new files:
+
+```python
+schema = spark.table("demo.nyc.taxis").schema
+data = [
+    (1, 1000371, 1.8, 15.32, "N"),
+    (2, 1000372, 2.5, 22.15, "N"),
+    (2, 1000373, 0.9, 9.01, "N"),
+    (1, 1000374, 8.4, 42.13, "Y")
+  ]
+df = spark.createDataFrame(data, schema)
+df.writeTo("demo.nyc.taxis").append()
+```
+
+#### Testing with the CLI
+
+Now that we have a table in the catalog which is running. You can use the 
+CLI which is implemented in the `cmd/iceberg` folder. You will need to set
+the following environment variables (which can also be found in the 
+docker-compose.yml):
+
+```
+AWS_S3_ENDPOINT=http://localhost:9000
+AWS_REGION=us-east-1
+AWS_ACCESS_KEY_ID=admin
+AWS_SECRET_ACCESS_KEY=password
+```
+
+With those environment variables set you can now run the CLI:
+
+```bash
+$ go run ./cmd/iceberg list --catalog rest --uri http://localhost:8181
+┌──────┐
+| IDs  |
+| ---- |
+| demo |
+└──────┘
+```
+
+You can retrieve the schema of the table:
+
+```bash
+$ go run ./cmd/iceberg schema --catalog rest --uri http://localhost:8181 demo.nyc.taxis
+Current Schema, id=0
+├──1: vendor_id: optional long
+├──2: trip_id: optional long
+├──3: trip_distance: optional float
+├──4: fare_amount: optional double
+└──5: store_and_fwd_flag: optional string
+```
+
+You can get the file list:
+
+```bash
+$ go run ./cmd/iceberg files --catalog rest --uri http://localhost:8181 demo.nyc.taxis
+Snapshots: rest.demo.nyc.taxis
+└─┬Snapshot 7004656639550124164, schema 0: s3://warehouse/demo/nyc/taxis/metadata/snap-7004656639550124164-1-0d533cd4-f0c1-45a6-a691-f2be3abe5491.avro
+  └─┬Manifest: s3://warehouse/demo/nyc/taxis/metadata/0d533cd4-f0c1-45a6-a691-f2be3abe5491-m0.avro
+    ├──Datafile: s3://warehouse/demo/nyc/taxis/data/00004-24-244255d4-8bf6-41bd-8885-bf7d2136fddf-00001.parquet
+    ├──Datafile: s3://warehouse/demo/nyc/taxis/data/00009-29-244255d4-8bf6-41bd-8885-bf7d2136fddf-00001.parquet
+    ├──Datafile: s3://warehouse/demo/nyc/taxis/data/00014-34-244255d4-8bf6-41bd-8885-bf7d2136fddf-00001.parquet
+    └──Datafile: s3://warehouse/demo/nyc/taxis/data/00019-39-244255d4-8bf6-41bd-8885-bf7d2136fddf-00001.parquet
+```
+
+and so on, for the various options available in the CLI.
\ No newline at end of file
diff --git a/catalog/catalog.go b/catalog/catalog.go
index 7b04468..d6d7f1e 100644
--- a/catalog/catalog.go
+++ b/catalog/catalog.go
@@ -19,8 +19,11 @@ package catalog
 
 import (
 	"context"
+	"crypto/tls"
 	"errors"
+	"net/url"
 
+	"github.com/apache/iceberg-go"
 	"github.com/apache/iceberg-go/table"
 	"github.com/aws/aws-sdk-go-v2/aws"
 )
@@ -37,29 +40,148 @@ const (
 
 var (
 	// ErrNoSuchTable is returned when a table does not exist in the catalog.
-	ErrNoSuchTable = errors.New("table does not exist")
+	ErrNoSuchTable            = errors.New("table does not exist")
+	ErrNoSuchNamespace        = errors.New("namespace does not exist")
+	ErrNamespaceAlreadyExists = errors.New("namespace already exists")
 )
 
 // WithAwsConfig sets the AWS configuration for the catalog.
-func WithAwsConfig(cfg aws.Config) Option {
+func WithAwsConfig(cfg aws.Config) Option[GlueCatalog] {
 	return func(o *options) {
 		o.awsConfig = cfg
 	}
 }
 
-type Option func(*options)
+func WithCredential(cred string) Option[RestCatalog] {
+	return func(o *options) {
+		o.credential = cred
+	}
+}
+
+func WithOAuthToken(token string) Option[RestCatalog] {
+	return func(o *options) {
+		o.oauthToken = token
+	}
+}
+
+func WithTLSConfig(config *tls.Config) Option[RestCatalog] {
+	return func(o *options) {
+		o.tlsConfig = config
+	}
+}
+
+func WithWarehouseLocation(loc string) Option[RestCatalog] {
+	return func(o *options) {
+		o.warehouseLocation = loc
+	}
+}
+
+func WithMetadataLocation(loc string) Option[RestCatalog] {
+	return func(o *options) {
+		o.metadataLocation = loc
+	}
+}
+
+func WithSigV4() Option[RestCatalog] {
+	return func(o *options) {
+		o.enableSigv4 = true
+		o.sigv4Service = "execute-api"
+	}
+}
+
+func WithSigV4RegionSvc(region, service string) Option[RestCatalog] {
+	return func(o *options) {
+		o.enableSigv4 = true
+		o.sigv4Region = region
+
+		if service == "" {
+			o.sigv4Service = "execute-api"
+		} else {
+			o.sigv4Service = service
+		}
+	}
+}
+
+func WithAuthURI(uri *url.URL) Option[RestCatalog] {
+	return func(o *options) {
+		o.authUri = uri
+	}
+}
+
+func WithPrefix(prefix string) Option[RestCatalog] {
+	return func(o *options) {
+		o.prefix = prefix
+	}
+}
+
+type Option[T GlueCatalog | RestCatalog] func(*options)
 
 type options struct {
 	awsConfig aws.Config
+
+	tlsConfig         *tls.Config
+	credential        string
+	oauthToken        string
+	warehouseLocation string
+	metadataLocation  string
+	enableSigv4       bool
+	sigv4Region       string
+	sigv4Service      string
+	prefix            string
+	authUri           *url.URL
+}
+
+type PropertiesUpdateSummary struct {
+	Removed []string `json:"removed"`
+	Updated []string `json:"updated"`
+	Missing []string `json:"missing"`
 }
 
 // Catalog for iceberg table operations like create, drop, load, list and others.
 type Catalog interface {
+	// CatalogType returns the type of the catalog.
+	CatalogType() CatalogType
+
 	// ListTables returns a list of table identifiers in the catalog, with the returned
 	// identifiers containing the information required to load the table via that catalog.
 	ListTables(ctx context.Context, namespace table.Identifier) ([]table.Identifier, error)
 	// LoadTable loads a table from the catalog and returns a Table with the metadata.
-	LoadTable(ctx context.Context, identifier table.Identifier, props map[string]string) (*table.Table, error)
-	// CatalogType returns the type of the catalog.
-	CatalogType() CatalogType
+	LoadTable(ctx context.Context, identifier table.Identifier, props iceberg.Properties) (*table.Table, error)
+	// 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
+	RenameTable(ctx context.Context, from, to table.Identifier) (*table.Table, error)
+	// ListNamespaces returns the list of available namespaces, optionally filtering by a
+	// parent namespace
+	ListNamespaces(ctx context.Context, parent table.Identifier) ([]table.Identifier, error)
+	// CreateNamespace tells the catalog to create a new namespace with the given properties
+	CreateNamespace(ctx context.Context, namespace table.Identifier, props iceberg.Properties) error
+	// DropNamespace tells the catalog to drop the namespace and all tables in that namespace
+	DropNamespace(ctx context.Context, namespace table.Identifier) error
+	// LoadNamespaceProperties returns the current properties in the catalog for
+	// a given namespace
+	LoadNamespaceProperties(ctx context.Context, namespace table.Identifier) (iceberg.Properties, error)
+	// UpdateNamespaceProperties allows removing, adding, and/or updating properties of a namespace
+	UpdateNamespaceProperties(ctx context.Context, namespace table.Identifier,
+		removals []string, updates iceberg.Properties) (PropertiesUpdateSummary, error)
+}
+
+const (
+	keyOauthToken        = "token"
+	keyWarehouseLocation = "warehouse"
+	keyMetadataLocation  = "metadata_location"
+	keyOauthCredential   = "credential"
+)
+
+func TableNameFromIdent(ident table.Identifier) string {
+	if len(ident) == 0 {
+		return ""
+	}
+
+	return ident[len(ident)-1]
+}
+
+func NamespaceFromIdent(ident table.Identifier) table.Identifier {
+	return ident[:len(ident)-1]
 }
diff --git a/catalog/glue.go b/catalog/glue.go
index 3caca0f..b398185 100644
--- a/catalog/glue.go
+++ b/catalog/glue.go
@@ -22,6 +22,7 @@ import (
 	"errors"
 	"fmt"
 
+	"github.com/apache/iceberg-go"
 	"github.com/apache/iceberg-go/io"
 	"github.com/apache/iceberg-go/table"
 	"github.com/aws/aws-sdk-go-v2/aws"
@@ -44,7 +45,7 @@ type GlueCatalog struct {
 	glueSvc glueAPI
 }
 
-func NewGlueCatalog(opts ...Option) *GlueCatalog {
+func NewGlueCatalog(opts ...Option[GlueCatalog]) *GlueCatalog {
 	glueOps := &options{}
 
 	for _, o := range opts {
@@ -91,7 +92,7 @@ func (c *GlueCatalog) ListTables(ctx context.Context, namespace table.Identifier
 // LoadTable loads a table from the catalog table details.
 //
 // The identifier should contain the Glue database name, then glue table name.
-func (c *GlueCatalog) LoadTable(ctx context.Context, identifier table.Identifier, props map[string]string) (*table.Table, error) {
+func (c *GlueCatalog) LoadTable(ctx context.Context, identifier table.Identifier, props iceberg.Properties) (*table.Table, error) {
 	database, tableName, err := identifierToGlueTable(identifier)
 	if err != nil {
 		return nil, err
@@ -124,6 +125,35 @@ func (c *GlueCatalog) CatalogType() CatalogType {
 	return Glue
 }
 
+func (c *GlueCatalog) DropTable(ctx context.Context, identifier table.Identifier) error {
+	return fmt.Errorf("%w: [Glue Catalog] drop table", iceberg.ErrNotImplemented)
+}
+
+func (c *GlueCatalog) RenameTable(ctx context.Context, from, to table.Identifier) (*table.Table, error) {
+	return nil, fmt.Errorf("%w: [Glue Catalog] rename table", iceberg.ErrNotImplemented)
+}
+
+func (c *GlueCatalog) CreateNamespace(ctx context.Context, namespace table.Identifier, props iceberg.Properties) error {
+	return fmt.Errorf("%w: [Glue Catalog] create namespace", iceberg.ErrNotImplemented)
+}
+
+func (c *GlueCatalog) DropNamespace(ctx context.Context, namespace table.Identifier) error {
+	return fmt.Errorf("%w: [Glue Catalog] drop namespace", iceberg.ErrNotImplemented)
+}
+
+func (c *GlueCatalog) LoadNamespaceProperties(ctx context.Context, namespace table.Identifier) (iceberg.Properties, error) {
+	return nil, fmt.Errorf("%w: [Glue Catalog] load namespace properties", iceberg.ErrNotImplemented)
+}
+
+func (c *GlueCatalog) UpdateNamespaceProperties(ctx context.Context, namespace table.Identifier,
+	removals []string, updates iceberg.Properties) (PropertiesUpdateSummary, error) {
+	return PropertiesUpdateSummary{}, fmt.Errorf("%w: [Glue Catalog] update namespace properties", iceberg.ErrNotImplemented)
+}
+
+func (c *GlueCatalog) ListNamespaces(ctx context.Context, parent table.Identifier) ([]table.Identifier, error) {
+	return nil, fmt.Errorf("%w: [Glue Catalog] list namespaces", iceberg.ErrNotImplemented)
+}
+
 // GetTable loads a table from the Glue Catalog using the given database and table name.
 func (c *GlueCatalog) getTable(ctx context.Context, database, tableName string) (string, error) {
 	tblRes, err := c.glueSvc.GetTable(ctx,
diff --git a/catalog/rest.go b/catalog/rest.go
new file mode 100644
index 0000000..93030bd
--- /dev/null
+++ b/catalog/rest.go
@@ -0,0 +1,711 @@
+// 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 catalog
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"hash"
+	"io"
+	"maps"
+	"net/http"
+	"net/url"
+	"strings"
+	"time"
+
+	"github.com/apache/iceberg-go"
+	iceio "github.com/apache/iceberg-go/io"
+	"github.com/apache/iceberg-go/table"
+	"github.com/aws/aws-sdk-go-v2/aws"
+	v4 "github.com/aws/aws-sdk-go-v2/aws/signer/v4"
+	"github.com/aws/aws-sdk-go-v2/config"
+)
+
+var (
+	_ Catalog = (*RestCatalog)(nil)
+)
+
+const (
+	authorizationHeader = "Authorization"
+	bearerPrefix        = "Bearer"
+	namespaceSeparator  = "\x1F"
+	keyPrefix           = "prefix"
+
+	icebergRestSpecVersion = "0.14.1"
+
+	keyRestSigV4        = "rest.sigv4-enabled"
+	keyRestSigV4Region  = "rest.signing-region"
+	keyRestSigV4Service = "rest.signing-name"
+	keyAuthUrl          = "rest.authorization-url"
+)
+
+var (
+	ErrRESTError            = errors.New("REST error")
+	ErrBadRequest           = fmt.Errorf("%w: bad request", ErrRESTError)
+	ErrForbidden            = fmt.Errorf("%w: forbidden", ErrRESTError)
+	ErrUnauthorized         = fmt.Errorf("%w: unauthorized", ErrRESTError)
+	ErrAuthorizationExpired = fmt.Errorf("%w: authorization expired", ErrRESTError)
+	ErrServiceUnavailable   = fmt.Errorf("%w: service unavailable", ErrRESTError)
+	ErrServerError          = fmt.Errorf("%w: server error", ErrRESTError)
+	ErrCommitFailed         = fmt.Errorf("%w: commit failed, refresh and try again", ErrRESTError)
+	ErrCommitStateUnknown   = fmt.Errorf("%w: commit failed due to unknown reason", ErrRESTError)
+	ErrOAuthError           = fmt.Errorf("%w: oauth error", ErrRESTError)
+)
+
+type errorResponse struct {
+	Message string `json:"message"`
+	Type    string `json:"type"`
+	Code    int    `json:"code"`
+
+	wrapping error
+}
+
+func (e errorResponse) Unwrap() error { return e.wrapping }
+func (e errorResponse) Error() string {
+	return e.Type + ": " + e.Message
+}
+
+type oauthTokenResponse struct {
+	AccessToken  string `json:"access_token"`
+	TokenType    string `json:"token_type"`
+	ExpiresIn    int    `json:"expires_in"`
+	Scope        string `json:"scope"`
+	RefreshToken string `json:"refresh_token"`
+}
+
+type oauthErrorResponse struct {
+	Err     string `json:"error"`
+	ErrDesc string `json:"error_description"`
+	ErrURI  string `json:"error_uri"`
+}
+
+func (o oauthErrorResponse) Unwrap() error { return ErrOAuthError }
+func (o oauthErrorResponse) Error() string {
+	msg := o.Err
+	if o.ErrDesc != "" {
+		msg += ": " + o.ErrDesc
+	}
+
+	if o.ErrURI != "" {
+		msg += " (" + o.ErrURI + ")"
+	}
+	return msg
+}
+
+type configResponse struct {
+	Defaults  iceberg.Properties `json:"defaults"`
+	Overrides iceberg.Properties `json:"overrides"`
+}
+
+type sessionTransport struct {
+	http.Transport
+
+	defaultHeaders http.Header
+	signer         v4.HTTPSigner
+	cfg            aws.Config
+	service        string
+	h              hash.Hash
+}
+
+// from https://pkg.go.dev/github.com/aws/aws-sdk-go-v2/aws/signer/v4#Signer.SignHTTP
+const emptyStringHash = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
+
+func (s *sessionTransport) RoundTrip(r *http.Request) (*http.Response, error) {
+	for k, v := range s.defaultHeaders {
+		for _, hdr := range v {
+			r.Header.Add(k, hdr)
+		}
+	}
+
+	if s.signer != nil {
+		var payloadHash string
+		if r.Body == nil {
+			payloadHash = emptyStringHash
+		} else {
+			rdr, err := r.GetBody()
+			if err != nil {
+				return nil, err
+			}
+
+			if _, err = io.Copy(s.h, rdr); err != nil {
+				return nil, err
+			}
+
+			payloadHash = string(s.h.Sum(nil))
+			s.h.Reset()
+		}
+
+		creds, err := s.cfg.Credentials.Retrieve(r.Context())
+		if err != nil {
+			return nil, err
+		}
+
+		// modifies the request in place
+		err = s.signer.SignHTTP(r.Context(), creds, r, payloadHash,
+			s.service, s.cfg.Region, time.Now())
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return s.Transport.RoundTrip(r)
+}
+
+func do[T any](ctx context.Context, method string, baseURI *url.URL, path []string, cl *http.Client, override map[int]error, allowNoContent bool) (ret T, err error) {
+	var (
+		req *http.Request
+		rsp *http.Response
+	)
+
+	uri := baseURI.JoinPath(path...).String()
+	if req, err = http.NewRequestWithContext(ctx, method, uri, nil); err != nil {
+		return
+	}
+
+	if rsp, err = cl.Do(req); err != nil {
+		return
+	}
+
+	if allowNoContent && rsp.StatusCode == http.StatusNoContent {
+		return
+	}
+
+	if rsp.StatusCode != http.StatusOK {
+		return ret, handleNon200(rsp, override)
+	}
+
+	defer rsp.Body.Close()
+	if err = json.NewDecoder(rsp.Body).Decode(&ret); err != nil {
+		return ret, fmt.Errorf("%w: error decoding json payload: `%s`", ErrRESTError, err.Error())
+	}
+
+	return
+}
+
+func doGet[T any](ctx context.Context, baseURI *url.URL, path []string, cl *http.Client, override map[int]error) (ret T, err error) {
+	return do[T](ctx, http.MethodGet, baseURI, path, cl, override, false)
+}
+
+func doDelete[T any](ctx context.Context, baseURI *url.URL, path []string, cl *http.Client, override map[int]error) (ret T, err error) {
+	return do[T](ctx, http.MethodDelete, baseURI, path, cl, override, true)
+}
+
+func doPost[Payload, Result any](ctx context.Context, baseURI *url.URL, path []string, payload Payload, cl *http.Client, override map[int]error) (ret Result, err error) {
+	var (
+		req  *http.Request
+		rsp  *http.Response
+		data []byte
+	)
+
+	uri := baseURI.JoinPath(path...).String()
+	data, err = json.Marshal(payload)
+	if err != nil {
+		return
+	}
+
+	req, err = http.NewRequestWithContext(ctx, http.MethodPost, uri, bytes.NewReader(data))
+	if err != nil {
+		return
+	}
+
+	req.Header.Add("Content-Type", "application/json")
+
+	rsp, err = cl.Do(req)
+	if err != nil {
+		return
+	}
+
+	if rsp.StatusCode != http.StatusOK {
+		return ret, handleNon200(rsp, override)
+	}
+
+	if rsp.ContentLength == 0 {
+		return
+	}
+
+	defer rsp.Body.Close()
+	if err = json.NewDecoder(rsp.Body).Decode(&ret); err != nil {
+		return ret, fmt.Errorf("%w: error decoding json payload: `%s`", ErrRESTError, err.Error())
+	}
+
+	return
+}
+
+func handleNon200(rsp *http.Response, override map[int]error) error {
+	var e errorResponse
+
+	dec := json.NewDecoder(rsp.Body)
+	dec.Decode(&struct {
+		Error *errorResponse `json:"error"`
+	}{Error: &e})
+
+	if override != nil {
+		if err, ok := override[rsp.StatusCode]; ok {
+			e.wrapping = err
+			return e
+		}
+	}
+
+	switch rsp.StatusCode {
+	case http.StatusBadRequest:
+		e.wrapping = ErrBadRequest
+	case http.StatusUnauthorized:
+		e.wrapping = ErrUnauthorized
+	case http.StatusForbidden:
+		e.wrapping = ErrForbidden
+	case http.StatusUnprocessableEntity:
+		e.wrapping = ErrRESTError
+	case 419:
+		e.wrapping = ErrAuthorizationExpired
+	case http.StatusNotImplemented:
+		e.wrapping = iceberg.ErrNotImplemented
+	case http.StatusServiceUnavailable:
+		e.wrapping = ErrServiceUnavailable
+	default:
+		if 500 <= rsp.StatusCode && rsp.StatusCode < 600 {
+			e.wrapping = ErrServerError
+		} else {
+			e.wrapping = ErrRESTError
+		}
+	}
+
+	return e
+}
+
+func ToRestIdentifier(ident ...string) table.Identifier {
+	if len(ident) == 1 {
+		if ident[0] == "" {
+			return nil
+		}
+		return table.Identifier(strings.Split(ident[0], "."))
+	}
+
+	return table.Identifier(ident)
+}
+
+func fromProps(props iceberg.Properties) *options {
+	o := &options{}
+	for k, v := range props {
+		switch k {
+		case keyOauthToken:
+			o.oauthToken = v
+		case keyWarehouseLocation:
+			o.warehouseLocation = v
+		case keyMetadataLocation:
+			o.metadataLocation = v
+		case keyRestSigV4:
+			o.enableSigv4 = strings.ToLower(v) == "true"
+		case keyRestSigV4Region:
+			o.sigv4Region = v
+		case keyRestSigV4Service:
+			o.sigv4Service = v
+		case keyAuthUrl:
+			u, err := url.Parse(v)
+			if err != nil {
+				continue
+			}
+			o.authUri = u
+		case keyOauthCredential:
+			o.credential = v
+		case keyPrefix:
+			o.prefix = v
+		}
+	}
+	return o
+}
+
+func toProps(o *options) iceberg.Properties {
+	props := iceberg.Properties{}
+
+	setIf := func(key, v string) {
+		if v != "" {
+			props[key] = v
+		}
+	}
+
+	setIf(keyOauthCredential, o.credential)
+	setIf(keyOauthToken, o.oauthToken)
+	setIf(keyWarehouseLocation, o.warehouseLocation)
+	setIf(keyMetadataLocation, o.metadataLocation)
+	if o.enableSigv4 {
+		props[keyRestSigV4] = "true"
+		setIf(keyRestSigV4Region, o.sigv4Region)
+		setIf(keyRestSigV4Service, o.sigv4Service)
+	}
+
+	setIf(keyPrefix, o.prefix)
+	if o.authUri != nil {
+		setIf(keyAuthUrl, o.authUri.String())
+	}
+	return props
+}
+
+type RestCatalog struct {
+	baseURI *url.URL
+	cl      *http.Client
+
+	name  string
+	props iceberg.Properties
+}
+
+func NewRestCatalog(name, uri string, opts ...Option[RestCatalog]) (*RestCatalog, error) {
+	ops := &options{}
+	for _, o := range opts {
+		o(ops)
+	}
+
+	baseuri, err := url.Parse(uri)
+	if err != nil {
+		return nil, err
+	}
+
+	r := &RestCatalog{
+		name:    name,
+		baseURI: baseuri.JoinPath("v1"),
+	}
+
+	if ops, err = r.fetchConfig(ops); err != nil {
+		return nil, err
+	}
+
+	cl, err := r.createSession(ops)
+	if err != nil {
+		return nil, err
+	}
+
+	r.cl = cl
+	if ops.prefix != "" {
+		r.baseURI = r.baseURI.JoinPath(ops.prefix)
+	}
+	r.props = toProps(ops)
+	return r, nil
+}
+
+func (r *RestCatalog) fetchAccessToken(cl *http.Client, creds string, opts *options) (string, error) {
+	clientID, clientSecret, hasID := strings.Cut(creds, ":")
+	if !hasID {
+		clientID, clientSecret = "", clientID
+	}
+
+	data := url.Values{
+		"grant_type":    {"client_credentials"},
+		"client_id":     {clientID},
+		"client_secret": {clientSecret},
+		"scope":         {"catalog"},
+	}
+
+	uri := opts.authUri
+	if uri == nil {
+		uri = r.baseURI.JoinPath("oauth/tokens")
+	}
+
+	rsp, err := cl.PostForm(uri.String(), data)
+	if err != nil {
+		return "", err
+	}
+
+	if rsp.StatusCode == http.StatusOK {
+		defer rsp.Body.Close()
+		dec := json.NewDecoder(rsp.Body)
+		var tok oauthTokenResponse
+		if err := dec.Decode(&tok); err != nil {
+			return "", fmt.Errorf("failed to decode oauth token response: %w", err)
+		}
+
+		return tok.AccessToken, nil
+	}
+
+	switch rsp.StatusCode {
+	case http.StatusUnauthorized, http.StatusBadRequest:
+		defer rsp.Request.GetBody()
+		dec := json.NewDecoder(rsp.Body)
+		var oauthErr oauthErrorResponse
+		if err := dec.Decode(&oauthErr); err != nil {
+			return "", fmt.Errorf("failed to decode oauth error: %w", err)
+		}
+
+		return "", oauthErr
+	default:
+		return "", handleNon200(rsp, nil)
+	}
+}
+
+func (r *RestCatalog) createSession(opts *options) (*http.Client, error) {
+	session := &sessionTransport{
+		Transport:      http.Transport{TLSClientConfig: opts.tlsConfig},
+		defaultHeaders: http.Header{},
+	}
+	cl := &http.Client{Transport: session}
+
+	token := opts.oauthToken
+	if token == "" && opts.credential != "" {
+		var err error
+		if token, err = r.fetchAccessToken(cl, opts.credential, opts); err != nil {
+			return nil, fmt.Errorf("auth error: %w", err)
+		}
+	}
+
+	if token != "" {
+		session.defaultHeaders.Set(authorizationHeader, bearerPrefix+" "+token)
+	}
+
+	session.defaultHeaders.Set("X-Client-Version", icebergRestSpecVersion)
+	session.defaultHeaders.Set("Content-Type", "application/json")
+	session.defaultHeaders.Set("User-Agent", "GoIceberg/"+iceberg.Version())
+
+	if opts.enableSigv4 {
+		cfg, err := config.LoadDefaultConfig(context.Background())
+		if err != nil {
+			return nil, err
+		}
+
+		if opts.sigv4Region != "" {
+			cfg.Region = opts.sigv4Region
+		}
+
+		session.cfg, session.service = cfg, opts.sigv4Service
+		session.signer, session.h = v4.NewSigner(), sha256.New()
+	}
+
+	return cl, nil
+}
+
+func (r *RestCatalog) fetchConfig(opts *options) (*options, error) {
+	params := url.Values{}
+	if opts.warehouseLocation != "" {
+		params.Set(keyWarehouseLocation, opts.warehouseLocation)
+	}
+
+	route := r.baseURI.JoinPath("config")
+	route.RawQuery = params.Encode()
+
+	sess, err := r.createSession(opts)
+	if err != nil {
+		return nil, err
+	}
+
+	rsp, err := doGet[configResponse](context.Background(), route, []string{}, sess, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	cfg := rsp.Defaults
+	maps.Copy(cfg, toProps(opts))
+	maps.Copy(cfg, rsp.Overrides)
+
+	o := fromProps(cfg)
+	o.awsConfig = opts.awsConfig
+	o.tlsConfig = opts.tlsConfig
+
+	if uri, ok := cfg["uri"]; ok {
+		r.baseURI, err = url.Parse(uri)
+		if err != nil {
+			return nil, err
+		}
+		r.baseURI = r.baseURI.JoinPath("v1")
+	}
+
+	return o, nil
+}
+
+func (r *RestCatalog) CatalogType() CatalogType { return REST }
+
+func checkValidNamespace(ident table.Identifier) error {
+	if len(ident) < 1 {
+		return fmt.Errorf("%w: empty namespace identifier", ErrNoSuchNamespace)
+	}
+	return nil
+}
+
+func (r *RestCatalog) ListTables(ctx context.Context, namespace table.Identifier) ([]table.Identifier, error) {
+	if err := checkValidNamespace(namespace); err != nil {
+		return nil, err
+	}
+
+	ns := strings.Join(namespace, namespaceSeparator)
+	path := []string{"namespaces", ns, "tables"}
+
+	type resp struct {
+		Identifiers []struct {
+			Namespace []string `json:"namespace"`
+			Name      string   `json:"name"`
+		} `json:"identifiers"`
+	}
+
+	rsp, err := doGet[resp](ctx, r.baseURI, path, r.cl, map[int]error{http.StatusNotFound: ErrNoSuchNamespace})
+	if err != nil {
+		return nil, err
+	}
+
+	out := make([]table.Identifier, len(rsp.Identifiers))
+	for i, id := range rsp.Identifiers {
+		out[i] = append(id.Namespace, id.Name)
+	}
+	return out, nil
+}
+
+func splitIdentForPath(ident table.Identifier) (string, string, error) {
+	if len(ident) < 1 {
+		return "", "", fmt.Errorf("%w: missing namespace or invalid identifier %v",
+			ErrNoSuchTable, strings.Join(ident, "."))
+	}
+
+	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 (t *tblResponse) UnmarshalJSON(b []byte) (err error) {
+	type Alias tblResponse
+	if err = json.Unmarshal(b, (*Alias)(t)); err != nil {
+		return err
+	}
+
+	t.Metadata, err = table.ParseMetadataBytes(t.RawMetadata)
+	return
+}
+
+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
+	}
+
+	if props == nil {
+		props = iceberg.Properties{}
+	}
+
+	ret, err := doGet[tblResponse](ctx, r.baseURI, []string{"namespaces", ns, "tables", tbl},
+		r.cl, map[int]error{http.StatusNotFound: ErrNoSuchTable})
+	if err != nil {
+		return nil, err
+	}
+
+	id := identifier
+	if r.name != "" {
+		id = append([]string{r.name}, identifier...)
+	}
+
+	tblProps := maps.Clone(r.props)
+	maps.Copy(tblProps, props)
+	maps.Copy(tblProps, ret.Metadata.Properties())
+	for k, v := range ret.Config {
+		tblProps[k] = v
+	}
+
+	iofs, err := iceio.LoadFS(tblProps, ret.MetadataLoc)
+	if err != nil {
+		return nil, err
+	}
+	return table.New(id, ret.Metadata, ret.MetadataLoc, iofs), nil
+}
+
+func (r *RestCatalog) DropTable(ctx context.Context, identifier table.Identifier) error {
+	return fmt.Errorf("%w: [Rest Catalog] drop table", iceberg.ErrNotImplemented)
+}
+
+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)
+}
+
+func (r *RestCatalog) CreateNamespace(ctx context.Context, namespace table.Identifier, props iceberg.Properties) error {
+	if err := checkValidNamespace(namespace); err != nil {
+		return err
+	}
+
+	_, err := doPost[map[string]any, struct{}](ctx, r.baseURI, []string{"namespaces"},
+		map[string]any{"namespace": namespace, "properties": props}, r.cl, map[int]error{
+			http.StatusNotFound: ErrNoSuchNamespace, http.StatusConflict: ErrNamespaceAlreadyExists})
+	return err
+}
+
+func (r *RestCatalog) DropNamespace(ctx context.Context, namespace table.Identifier) error {
+	if err := checkValidNamespace(namespace); err != nil {
+		return err
+	}
+
+	_, err := doDelete[struct{}](ctx, r.baseURI, []string{"namespaces", strings.Join(namespace, namespaceSeparator)},
+		r.cl, map[int]error{http.StatusNotFound: ErrNoSuchNamespace})
+
+	return err
+}
+
+func (r *RestCatalog) ListNamespaces(ctx context.Context, parent table.Identifier) ([]table.Identifier, error) {
+	uri := r.baseURI.JoinPath("namespaces")
+	if len(parent) != 0 {
+		v := url.Values{}
+		v.Set("parent", strings.Join(parent, namespaceSeparator))
+		uri.RawQuery = v.Encode()
+	}
+
+	type rsptype struct {
+		Namespaces []table.Identifier `json:"namespaces"`
+	}
+
+	rsp, err := doGet[rsptype](ctx, uri, []string{}, r.cl, map[int]error{http.StatusNotFound: ErrNoSuchNamespace})
+	if err != nil {
+		return nil, err
+	}
+
+	return rsp.Namespaces, nil
+}
+
+func (r *RestCatalog) LoadNamespaceProperties(ctx context.Context, namespace table.Identifier) (iceberg.Properties, error) {
+	if err := checkValidNamespace(namespace); err != nil {
+		return nil, err
+	}
+
+	type nsresponse struct {
+		Namespace table.Identifier   `json:"namespace"`
+		Props     iceberg.Properties `json:"properties"`
+	}
+
+	rsp, err := doGet[nsresponse](ctx, r.baseURI, []string{"namespaces", strings.Join(namespace, namespaceSeparator)},
+		r.cl, map[int]error{http.StatusNotFound: ErrNoSuchNamespace})
+	if err != nil {
+		return nil, err
+	}
+
+	return rsp.Props, nil
+}
+
+func (r *RestCatalog) UpdateNamespaceProperties(ctx context.Context, namespace table.Identifier,
+	removals []string, updates iceberg.Properties) (PropertiesUpdateSummary, error) {
+
+	if err := checkValidNamespace(namespace); err != nil {
+		return PropertiesUpdateSummary{}, err
+	}
+
+	type payload struct {
+		Remove  []string           `json:"removals"`
+		Updates iceberg.Properties `json:"updates"`
+	}
+
+	ns := strings.Join(namespace, namespaceSeparator)
+	return doPost[payload, PropertiesUpdateSummary](ctx, r.baseURI, []string{"namespaces", ns, "properties"},
+		payload{Remove: removals, Updates: updates}, r.cl, map[int]error{http.StatusNotFound: ErrNoSuchNamespace})
+}
diff --git a/cmd/iceberg/main.go b/cmd/iceberg/main.go
new file mode 100644
index 0000000..fb25618
--- /dev/null
+++ b/cmd/iceberg/main.go
@@ -0,0 +1,343 @@
+// 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 main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"os"
+	"strings"
+
+	"github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/catalog"
+	"github.com/apache/iceberg-go/table"
+	"github.com/docopt/docopt-go"
+)
+
+const usage = `iceberg.
+
+Usage:
+  iceberg list [options] [PARENT]
+  iceberg describe [options] [namespace | table] IDENTIFIER
+  iceberg (schema | spec | uuid | location) [options] TABLE_ID
+  iceberg drop [options] (namespace | table) IDENTIFIER
+  iceberg files [options] TABLE_ID [--history]
+  iceberg rename [options] <from> <to>
+  iceberg properties [options] get (namespace | table) IDENTIFIER [PROPNAME]
+  iceberg properties [options] set (namespace | table) IDENTIFIER PROPNAME VALUE
+  iceberg properties [options] remove (namespace | table) IDENTIFIER PROPNAME
+  iceberg -h | --help | --version
+
+Arguments:
+  PARENT         Catalog parent namespace
+  IDENTIFIER     fully qualified namespace or table
+  TABLE_ID       full path to a table
+  PROPNAME       name of a property
+  VALUE          value to set
+
+Options:
+  -h --help          show this helpe messages and exit
+  --catalog TEXT     specify the catalog type [default: rest]
+  --uri TEXT         specify the catalog URI
+  --output TYPE      output type (json/text) [default: text]
+  --credential TEXT  specify credentials for the catalog
+  --warehouse TEXT   specify the warehouse to use`
+
+func main() {
+	args, err := docopt.ParseArgs(usage, os.Args[1:], iceberg.Version())
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	cfg := struct {
+		List     bool `docopt:"list"`
+		Describe bool `docopt:"describe"`
+		Schema   bool `docopt:"schema"`
+		Spec     bool `docopt:"spec"`
+		Uuid     bool `docopt:"uuid"`
+		Location bool `docopt:"location"`
+		Props    bool `docopt:"properties"`
+		Drop     bool `docopt:"drop"`
+		Files    bool `docopt:"files"`
+		Rename   bool `docopt:"rename"`
+
+		Get    bool `docopt:"get"`
+		Set    bool `docopt:"set"`
+		Remove bool `docopt:"remove"`
+
+		Namespace bool `docopt:"namespace"`
+		Table     bool `docopt:"table"`
+
+		RenameFrom string `docopt:"<from>"`
+		RenameTo   string `docopt:"<to>"`
+
+		Parent   string `docopt:"PARENT"`
+		Ident    string `docopt:"IDENTIFIER"`
+		TableID  string `docopt:"TABLE_ID"`
+		PropName string `docopt:"PROPNAME"`
+		Value    string `docopt:"VALUE"`
+
+		Catalog   string `docopt:"--catalog"`
+		URI       string `docopt:"--uri"`
+		Output    string `docopt:"--output"`
+		History   bool   `docopt:"--history"`
+		Cred      string `docopt:"--credential"`
+		Warehouse string `docopt:"--warehouse"`
+	}{}
+
+	if err := args.Bind(&cfg); err != nil {
+		log.Fatal(err)
+	}
+
+	var output Output
+	switch strings.ToLower(cfg.Output) {
+	case "text":
+		output = text{}
+	case "json":
+		fallthrough
+	default:
+		log.Fatal("unimplemented output type")
+	}
+
+	var cat catalog.Catalog
+	switch catalog.CatalogType(cfg.Catalog) {
+	case catalog.REST:
+		opts := []catalog.Option[catalog.RestCatalog]{}
+		if len(cfg.Cred) > 0 {
+			opts = append(opts, catalog.WithCredential(cfg.Cred))
+		}
+
+		if len(cfg.Warehouse) > 0 {
+			opts = append(opts, catalog.WithWarehouseLocation(cfg.Warehouse))
+		}
+
+		if cat, err = catalog.NewRestCatalog("rest", cfg.URI, opts...); err != nil {
+			log.Fatal(err)
+		}
+	default:
+		log.Fatal("unrecognized catalog type")
+	}
+
+	switch {
+	case cfg.List:
+		list(output, cat, cfg.Parent)
+	case cfg.Describe:
+		entityType := "any"
+		if cfg.Namespace {
+			entityType = "ns"
+		} else if cfg.Table {
+			entityType = "tbl"
+		}
+
+		describe(output, cat, cfg.Ident, entityType)
+	case cfg.Schema:
+		tbl := loadTable(output, cat, cfg.TableID)
+		output.Schema(tbl.Schema())
+	case cfg.Spec:
+		tbl := loadTable(output, cat, cfg.TableID)
+		output.Spec(tbl.Spec())
+	case cfg.Location:
+		tbl := loadTable(output, cat, cfg.TableID)
+		output.Text(tbl.Location())
+	case cfg.Uuid:
+		tbl := loadTable(output, cat, cfg.TableID)
+		output.Uuid(tbl.Metadata().TableUUID())
+	case cfg.Props:
+		properties(output, cat, propCmd{
+			get: cfg.Get, set: cfg.Set, remove: cfg.Remove,
+			namespace: cfg.Namespace, table: cfg.Table,
+			identifier: cfg.Ident,
+			propname:   cfg.PropName,
+			value:      cfg.Value,
+		})
+	case cfg.Rename:
+		_, err := cat.RenameTable(context.Background(),
+			catalog.ToRestIdentifier(cfg.RenameFrom), catalog.ToRestIdentifier(cfg.RenameTo))
+		if err != nil {
+			output.Error(err)
+			os.Exit(1)
+		}
+
+		output.Text("Renamed table from " + cfg.RenameFrom + " to " + cfg.RenameTo)
+	case cfg.Drop:
+		switch {
+		case cfg.Namespace:
+			err := cat.DropNamespace(context.Background(), catalog.ToRestIdentifier(cfg.Ident))
+			if err != nil {
+				output.Error(err)
+				os.Exit(1)
+			}
+		case cfg.Table:
+			err := cat.DropTable(context.Background(), catalog.ToRestIdentifier(cfg.Ident))
+			if err != nil {
+				output.Error(err)
+				os.Exit(1)
+			}
+		}
+	case cfg.Files:
+		tbl := loadTable(output, cat, cfg.TableID)
+		output.Files(tbl, cfg.History)
+	}
+}
+
+func list(output Output, cat catalog.Catalog, parent string) {
+	prnt := catalog.ToRestIdentifier(parent)
+
+	ids, err := cat.ListNamespaces(context.Background(), prnt)
+	if err != nil {
+		output.Error(err)
+		os.Exit(1)
+	}
+
+	if len(ids) == 0 && parent != "" {
+		ids, err = cat.ListTables(context.Background(), prnt)
+		if err != nil {
+			output.Error(err)
+			os.Exit(1)
+		}
+	}
+	output.Identifiers(ids)
+}
+
+func describe(output Output, cat catalog.Catalog, id string, entityType string) {
+	ctx := context.Background()
+
+	ident := catalog.ToRestIdentifier(id)
+
+	isNS, isTbl := false, false
+	if (entityType == "any" || entityType == "ns") && len(ident) > 0 {
+		nsprops, err := cat.LoadNamespaceProperties(ctx, ident)
+		if err != nil {
+			if errors.Is(err, catalog.ErrNoSuchNamespace) {
+				if entityType != "any" || len(ident) == 1 {
+					output.Error(err)
+					os.Exit(1)
+				}
+			} else {
+				output.Error(err)
+				os.Exit(1)
+			}
+		} else {
+			isNS = true
+			output.DescribeProperties(nsprops)
+		}
+	}
+
+	if (entityType == "any" || entityType == "tbl") && len(ident) > 1 {
+		tbl, err := cat.LoadTable(ctx, ident, nil)
+		if err != nil {
+			if !errors.Is(err, catalog.ErrNoSuchTable) || entityType != "any" {
+				output.Error(err)
+				os.Exit(1)
+			}
+		} else {
+			isTbl = true
+			output.DescribeTable(tbl)
+		}
+	}
+
+	if !isNS && !isTbl {
+		output.Error(fmt.Errorf("%w: table or namespace does not exist: %s",
+			catalog.ErrNoSuchNamespace, ident))
+		os.Exit(1)
+	}
+}
+
+func loadTable(output Output, cat catalog.Catalog, id string) *table.Table {
+	tbl, err := cat.LoadTable(context.Background(), catalog.ToRestIdentifier(id), nil)
+	if err != nil {
+		output.Error(err)
+		os.Exit(1)
+	}
+
+	return tbl
+}
+
+type propCmd struct {
+	get, set, remove bool
+	namespace, table bool
+
+	identifier, propname, value string
+}
+
+func properties(output Output, cat catalog.Catalog, args propCmd) {
+	ctx, ident := context.Background(), catalog.ToRestIdentifier(args.identifier)
+
+	switch {
+	case args.get:
+		var props iceberg.Properties
+		switch {
+		case args.namespace:
+			var err error
+			props, err = cat.LoadNamespaceProperties(ctx, ident)
+			if err != nil {
+				output.Error(err)
+				os.Exit(1)
+			}
+		case args.table:
+			tbl := loadTable(output, cat, args.identifier)
+			props = tbl.Metadata().Properties()
+		}
+
+		if args.propname == "" {
+			output.DescribeProperties(props)
+			return
+		}
+
+		if val, ok := props[args.propname]; ok {
+			output.Text(val)
+		} else {
+			output.Error(errors.New("could not find property " + args.propname + " on namespace " + args.identifier))
+			os.Exit(1)
+		}
+	case args.set:
+		switch {
+		case args.namespace:
+			_, err := cat.UpdateNamespaceProperties(ctx, ident,
+				nil, iceberg.Properties{args.propname: args.value})
+			if err != nil {
+				output.Error(err)
+				os.Exit(1)
+			}
+
+			output.Text("updated " + args.propname + " on " + args.identifier)
+		case args.table:
+			loadTable(output, cat, args.identifier)
+			output.Text("Setting " + args.propname + "=" + args.value + " on " + args.identifier)
+			output.Error(errors.New("not implemented: Writing is WIP"))
+		}
+	case args.remove:
+		switch {
+		case args.namespace:
+			_, err := cat.UpdateNamespaceProperties(ctx, ident,
+				[]string{args.propname}, nil)
+			if err != nil {
+				output.Error(err)
+				os.Exit(1)
+			}
+
+			output.Text("removing " + args.propname + " from " + args.identifier)
+		case args.table:
+			loadTable(output, cat, args.identifier)
+			output.Text("Setting " + args.propname + "=" + args.value + " on " + args.identifier)
+			output.Error(errors.New("not implemented: Writing is WIP"))
+		}
+	}
+}
diff --git a/cmd/iceberg/output.go b/cmd/iceberg/output.go
new file mode 100644
index 0000000..aae59c4
--- /dev/null
+++ b/cmd/iceberg/output.go
@@ -0,0 +1,216 @@
+// 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 main
+
+import (
+	"fmt"
+	"log"
+	"os"
+	"strconv"
+	"strings"
+
+	"github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/table"
+
+	"github.com/google/uuid"
+	"github.com/pterm/pterm"
+	"github.com/pterm/pterm/putils"
+)
+
+type Output interface {
+	Identifiers([]table.Identifier)
+	DescribeTable(*table.Table)
+	Files(tbl *table.Table, history bool)
+	DescribeProperties(iceberg.Properties)
+	Text(string)
+	Schema(*iceberg.Schema)
+	Spec(iceberg.PartitionSpec)
+	Uuid(uuid.UUID)
+	Error(error)
+}
+
+type text struct{}
+
+func (text) Identifiers(idlist []table.Identifier) {
+	data := pterm.TableData{[]string{"IDs"}}
+	for _, ids := range idlist {
+		data = append(data, []string{strings.Join(ids, ".")})
+	}
+
+	pterm.DefaultTable.
+		WithBoxed(true).
+		WithHasHeader(true).
+		WithHeaderRowSeparator("-").
+		WithData(data).Render()
+}
+
+func (t text) DescribeTable(tbl *table.Table) {
+	propData := pterm.TableData{{"key", "value"}}
+	for k, v := range tbl.Metadata().Properties() {
+		propData = append(propData, []string{k, v})
+	}
+	propTable := pterm.DefaultTable.
+		WithHasHeader(true).
+		WithHeaderRowSeparator("-").
+		WithData(propData)
+
+	snapshotList := pterm.LeveledList{}
+	for _, s := range tbl.Metadata().Snapshots() {
+		var manifest string
+		if s.ManifestList != "" {
+			manifest = ": " + s.ManifestList
+		}
+
+		snapshotList = append(snapshotList, pterm.LeveledListItem{
+			Level: 0, Text: fmt.Sprintf("Snapshot %d, schema %d%s",
+				s.SnapshotID, *s.SchemaID, manifest),
+		})
+	}
+
+	snapshotTreeNode := putils.TreeFromLeveledList(snapshotList)
+	snapshotTreeNode.Text = "Snapshots"
+
+	pterm.DefaultTable.
+		WithData(pterm.TableData{
+			{"Table format version", strconv.Itoa(tbl.Metadata().Version())},
+			{"Metadata location", tbl.MetadataLocation()},
+			{"Table UUID", tbl.Metadata().TableUUID().String()},
+			{"Last updated", strconv.Itoa(int(tbl.Metadata().LastUpdatedMillis()))},
+			{"Sort Order", tbl.SortOrder().String()},
+			{"Partition Spec", tbl.Spec().String()},
+		}).Render()
+
+	t.Schema(tbl.Schema())
+	pterm.DefaultTable.
+		WithData(pterm.TableData{
+			{"Current Snapshot", tbl.CurrentSnapshot().String()},
+		}).Render()
+	pterm.DefaultTree.WithRoot(snapshotTreeNode).Render()
+	pterm.Println("Properties")
+	propTable.Render()
+}
+
+func (t text) Files(tbl *table.Table, history bool) {
+	var snapshots []table.Snapshot
+	if history {
+		snapshots = tbl.Metadata().Snapshots()
+	} else {
+		snap := tbl.CurrentSnapshot()
+		if snap != nil {
+			snapshots = []table.Snapshot{*snap}
+		}
+	}
+
+	snapshotTree := pterm.LeveledList{}
+	for _, snap := range snapshots {
+		manifest := snap.ManifestList
+		if manifest != "" {
+			manifest = ": " + manifest
+		}
+
+		snapshotTree = append(snapshotTree, pterm.LeveledListItem{
+			Level: 0,
+			Text: fmt.Sprintf("Snapshot %d, schema %d%s",
+				snap.SnapshotID, *snap.SchemaID, manifest),
+		})
+
+		manifestList, err := snap.Manifests(tbl.FS())
+		if err != nil {
+			t.Error(err)
+			os.Exit(1)
+		}
+
+		for _, m := range manifestList {
+			snapshotTree = append(snapshotTree, pterm.LeveledListItem{
+				Level: 1, Text: "Manifest: " + m.FilePath(),
+			})
+			datafiles, err := m.FetchEntries(tbl.FS(), false)
+			if err != nil {
+				t.Error(err)
+				os.Exit(1)
+			}
+			for _, e := range datafiles {
+				snapshotTree = append(snapshotTree, pterm.LeveledListItem{
+					Level: 2, Text: "Datafile: " + e.DataFile().FilePath(),
+				})
+			}
+		}
+	}
+
+	node := putils.TreeFromLeveledList(snapshotTree)
+	node.Text = "Snapshots: " + strings.Join(tbl.Identifier(), ".")
+	pterm.DefaultTree.WithRoot(node).Render()
+}
+
+func (text) DescribeProperties(props iceberg.Properties) {
+	data := pterm.TableData{[]string{"Key", "Value"}}
+	for k, v := range props {
+		data = append(data, []string{k, v})
+	}
+
+	pterm.DefaultTable.
+		WithBoxed(true).
+		WithHasHeader(true).
+		WithHeaderRowSeparator("-").
+		WithData(data).Render()
+}
+
+func (text) Text(val string) {
+	fmt.Println(val)
+}
+
+func (text) Schema(schema *iceberg.Schema) {
+	schemaTree := pterm.LeveledList{}
+	var addChildren func(iceberg.NestedField, int)
+	addChildren = func(nf iceberg.NestedField, depth int) {
+		if nested, ok := nf.Type.(iceberg.NestedType); ok {
+			for _, n := range nested.Fields() {
+				schemaTree = append(schemaTree, pterm.LeveledListItem{
+					Level: depth, Text: n.String(),
+				})
+				addChildren(n, depth+1)
+			}
+		}
+	}
+
+	for _, f := range schema.Fields() {
+		schemaTree = append(schemaTree, pterm.LeveledListItem{
+			Level: 0, Text: f.String(),
+		})
+		addChildren(f, 1)
+	}
+	schemaTreeNode := putils.TreeFromLeveledList(schemaTree)
+	schemaTreeNode.Text = "Current Schema, id=" + strconv.Itoa(schema.ID)
+	pterm.DefaultTree.WithRoot(schemaTreeNode).Render()
+}
+
+func (text) Spec(spec iceberg.PartitionSpec) {
+	fmt.Println(spec)
+}
+
+func (text) Uuid(u uuid.UUID) {
+	if u.String() != "" {
+		fmt.Println(u.String())
+	} else {
+		fmt.Println("missing")
+	}
+}
+
+func (text) Error(err error) {
+	log.Fatal(err)
+}
diff --git a/dev/docker-compose.yml b/dev/docker-compose.yml
new file mode 100644
index 0000000..2978c42
--- /dev/null
+++ b/dev/docker-compose.yml
@@ -0,0 +1,89 @@
+# 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.
+version: "3"
+
+services:
+  spark-iceberg:
+    image: tabulario/spark-iceberg
+    container_name: spark-iceberg    
+    networks:
+      iceberg_net:
+    depends_on:
+      - rest
+      - minio
+    volumes:
+      - ./warehouse:/home/iceberg/warehouse
+      - ./notebooks:/home/iceberg/notebooks/notebooks
+    environment:
+      - AWS_ACCESS_KEY_ID=admin
+      - AWS_SECRET_ACCESS_KEY=password
+      - AWS_REGION=us-east-1
+    ports:
+      - 8888:8888
+      - 8080:8080
+      - 10000:10000
+      - 10001:10001
+  rest:
+    image: tabulario/iceberg-rest
+    container_name: iceberg-rest
+    networks:
+      iceberg_net:
+    ports:
+      - 8181:8181
+    environment:
+      - AWS_ACCESS_KEY_ID=admin
+      - AWS_SECRET_ACCESS_KEY=password
+      - AWS_REGION=us-east-1
+      - CATALOG_WAREHOUSE=s3://warehouse/
+      - CATALOG_IO__IMPL=org.apache.iceberg.aws.s3.S3FileIO
+      - CATALOG_S3_ENDPOINT=http://minio:9000
+  minio:
+    image: minio/minio
+    container_name: minio
+    environment:
+      - MINIO_ROOT_USER=admin
+      - MINIO_ROOT_PASSWORD=password
+      - MINIO_DOMAIN=minio
+    networks:
+      iceberg_net:
+        aliases:
+          - warehouse.minio
+    ports:
+      - 9001:9001
+      - 9000:9000
+    command: ["server", "/data", "--console-address", ":9001"]
+  mc:
+    depends_on:
+      - minio
+    image: minio/mc
+    container_name: mc
+    networks:
+      iceberg_net:
+    environment:
+      - AWS_ACCESS_KEY_ID=admin
+      - AWS_SECRET_ACCESS_KEY=password
+      - AWS_REGION=us-east-1
+    entrypoint: >
+      /bin/sh -c "
+      until (/usr/bin/mc config host add minio http://minio:9000 admin password) do echo '...waiting...' && sleep 1; done;
+      /usr/bin/mc rm -r --force minio/warehouse;
+      /usr/bin/mc mb minio/warehouse;
+      /usr/bin/mc policy set public minio/warehouse;
+      tail -f /dev/null
+      "
+networks:
+  iceberg_net:
\ No newline at end of file
diff --git a/go.mod b/go.mod
index 2dca7eb..88bb9b5 100644
--- a/go.mod
+++ b/go.mod
@@ -17,7 +17,7 @@
 
 module github.com/apache/iceberg-go
 
-go 1.20
+go 1.21
 
 require (
 	github.com/aws/aws-sdk-go-v2 v1.24.1
@@ -25,14 +25,19 @@ require (
 	github.com/aws/aws-sdk-go-v2/credentials v1.16.14
 	github.com/aws/aws-sdk-go-v2/service/glue v1.73.1
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.48.0
+	github.com/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815
 	github.com/google/uuid v1.3.1
 	github.com/hamba/avro/v2 v2.16.0
+	github.com/pterm/pterm v0.12.78
 	github.com/stretchr/testify v1.8.4
 	github.com/wolfeidau/s3iofs v1.5.0
 	golang.org/x/exp v0.0.0-20231006140011-7918f672742d
 )
 
 require (
+	atomicgo.dev/cursor v0.2.0 // indirect
+	atomicgo.dev/keyboard v0.2.9 // indirect
+	atomicgo.dev/schedule v0.1.0 // indirect
 	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.5.4 // indirect
 	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.14.11 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/configsources v1.2.10 // indirect
@@ -47,14 +52,23 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.21.6 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sts v1.26.7 // indirect
 	github.com/aws/smithy-go v1.19.0 // indirect
+	github.com/containerd/console v1.0.3 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/golang/snappy v0.0.4 // indirect
+	github.com/gookit/color v1.5.4 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/lithammer/fuzzysearch v1.1.8 // indirect
+	github.com/mattn/go-runewidth v0.0.15 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/rivo/uniseg v0.4.4 // indirect
 	github.com/stretchr/objx v0.5.0 // indirect
+	github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e // indirect
 	golang.org/x/net v0.18.0 // indirect
+	golang.org/x/sys v0.16.0 // indirect
+	golang.org/x/term v0.16.0 // indirect
+	golang.org/x/text v0.14.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
diff --git a/go.sum b/go.sum
index 4e0f385..62ea84a 100644
--- a/go.sum
+++ b/go.sum
@@ -1,3 +1,21 @@
+atomicgo.dev/assert v0.0.2 h1:FiKeMiZSgRrZsPo9qn/7vmr7mCsh5SZyXY4YGYiYwrg=
+atomicgo.dev/assert v0.0.2/go.mod h1:ut4NcI3QDdJtlmAxQULOmA13Gz6e2DWbSAS8RUOmNYQ=
+atomicgo.dev/cursor v0.2.0 h1:H6XN5alUJ52FZZUkI7AlJbUc1aW38GWZalpYRPpoPOw=
+atomicgo.dev/cursor v0.2.0/go.mod h1:Lr4ZJB3U7DfPPOkbH7/6TOtJ4vFGHlgj1nc+n900IpU=
+atomicgo.dev/keyboard v0.2.9 h1:tOsIid3nlPLZ3lwgG8KZMp/SFmr7P0ssEN5JUsm78K8=
+atomicgo.dev/keyboard v0.2.9/go.mod h1:BC4w9g00XkxH/f1HXhW2sXmJFOCWbKn9xrOunSFtExQ=
+atomicgo.dev/schedule v0.1.0 h1:nTthAbhZS5YZmgYbb2+DH8uQIZcTlIrd4eYr3UQxEjs=
+atomicgo.dev/schedule v0.1.0/go.mod h1:xeUa3oAkiuHYh8bKiQBRojqAMq3PXXbJujjb0hw8pEU=
+github.com/MarvinJWendt/testza v0.1.0/go.mod h1:7AxNvlfeHP7Z/hDQ5JtE3OKYT3XFUeLCDE2DQninSqs=
+github.com/MarvinJWendt/testza v0.2.1/go.mod h1:God7bhG8n6uQxwdScay+gjm9/LnO4D3kkcZX4hv9Rp8=
+github.com/MarvinJWendt/testza v0.2.8/go.mod h1:nwIcjmr0Zz+Rcwfh3/4UhBp7ePKVhuBExvZqnKYWlII=
+github.com/MarvinJWendt/testza v0.2.10/go.mod h1:pd+VWsoGUiFtq+hRKSU1Bktnn+DMCSrDrXDpX2bG66k=
+github.com/MarvinJWendt/testza v0.2.12/go.mod h1:JOIegYyV7rX+7VZ9r77L/eH6CfJHHzXjB69adAhzZkI=
+github.com/MarvinJWendt/testza v0.3.0/go.mod h1:eFcL4I0idjtIx8P9C6KkAuLgATNKpX4/2oUqKc6bF2c=
+github.com/MarvinJWendt/testza v0.4.2/go.mod h1:mSdhXiKH8sg/gQehJ63bINcCKp7RtYewEjXsvsVUPbE=
+github.com/MarvinJWendt/testza v0.5.2 h1:53KDo64C1z/h/d/stCYCPY69bt/OSwjq5KpFNwi+zB4=
+github.com/MarvinJWendt/testza v0.5.2/go.mod h1:xu53QFE5sCdjtMCKk8YMQ2MnymimEctc4n3EjyIYvEY=
+github.com/atomicgo/cursor v0.0.1/go.mod h1:cBON2QmmrysudxNBFthvMtN32r3jxVRIvzkUiF/RuIk=
 github.com/aws/aws-sdk-go-v2 v1.24.1 h1:xAojnj+ktS95YZlDf0zxWBkbFtymPeDP+rvUQIH3uAU=
 github.com/aws/aws-sdk-go-v2 v1.24.1/go.mod h1:LNh45Br1YAkEKaAqvmE1m8FUx6a5b/V0oAKV7of29b4=
 github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.5.4 h1:OCs21ST2LrepDfD3lwlQiOqIGp6JiEUqG84GzTDoyJs=
@@ -36,19 +54,43 @@ github.com/aws/aws-sdk-go-v2/service/sts v1.26.7 h1:NzO4Vrau795RkUdSHKEwiR01FaGz
 github.com/aws/aws-sdk-go-v2/service/sts v1.26.7/go.mod h1:6h2YuIoxaMSCFf5fi1EgZAwdfkGMgDY+DVfa61uLe4U=
 github.com/aws/smithy-go v1.19.0 h1:KWFKQV80DpP3vJrrA9sVAHQ5gc2z8i4EzrLhLlWXcBM=
 github.com/aws/smithy-go v1.19.0/go.mod h1:NukqUGpCZIILqqiV0NIjeFh24kd/FAa4beRb6nbIUPE=
+github.com/containerd/console v1.0.3 h1:lIr7SlA5PxZyMV30bDW0MGbiOPXwc63yRuCP0ARubLw=
+github.com/containerd/console v1.0.3/go.mod h1:7LqA/THxQ86k76b8c/EMSiaJ3h1eZkMkXar0TQ1gf3U=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815 h1:bWDMxwH3px2JBh6AyO7hdCn/PkvCZXii8TGj7sbtEbQ=
+github.com/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815/go.mod h1:WwZ+bS3ebgob9U8Nd0kOddGdZWjyMGR8Wziv+TBNwSE=
 github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM=
 github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/google/go-cmp v0.5.8 h1:e6P7q2lk1O+qJJb4BtCQXlK8vWEO8V1ZeuEdJNOqZyg=
+github.com/google/go-cmp v0.5.8/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/uuid v1.3.1 h1:KjJaJ9iWZ3jOFZIf1Lqf4laDRCasjl0BCmnEGxkdLb4=
 github.com/google/uuid v1.3.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/gookit/color v1.4.2/go.mod h1:fqRyamkC1W8uxl+lxCQxOT09l/vYfZ+QeiX3rKQHCoQ=
+github.com/gookit/color v1.5.0/go.mod h1:43aQb+Zerm/BWh2GnrgOQm7ffz7tvQXEKV6BFMl7wAo=
+github.com/gookit/color v1.5.4 h1:FZmqs7XOyGgCAxmWyPslpiok1k05wmY3SJTytgvYFs0=
+github.com/gookit/color v1.5.4/go.mod h1:pZJOeOS8DM43rXbp4AZo1n9zCU2qjpcRko0b6/QJi9w=
 github.com/hamba/avro/v2 v2.16.0 h1:0XhyP65Hs8iMLtdSR0v7ZrwRjsbIZdvr7KzYgmx1Mbo=
 github.com/hamba/avro/v2 v2.16.0/go.mod h1:Q9YK+qxAhtVrNqOhwlZTATLgLA8qxG2vtvkhK8fJ7Jo=
 github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
 github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
+github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
+github.com/klauspost/cpuid/v2 v2.0.10/go.mod h1:g2LTdtYhdyuGPqyWyv7qRAmj1WBqxuObKfj5c0PQa7c=
+github.com/klauspost/cpuid/v2 v2.0.12/go.mod h1:g2LTdtYhdyuGPqyWyv7qRAmj1WBqxuObKfj5c0PQa7c=
+github.com/klauspost/cpuid/v2 v2.2.3 h1:sxCkb+qR91z4vsqw4vGGZlDgPz3G7gjaLyK3V8y70BU=
+github.com/klauspost/cpuid/v2 v2.2.3/go.mod h1:RVVoqg1df56z8g3pUjL/3lE5UfnlrJX8tyFgg4nqhuY=
+github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
+github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
+github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
+github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+github.com/lithammer/fuzzysearch v1.1.8 h1:/HIuJnjHuXS8bKaiTMeeDlW2/AyIWk2brx1V8LFgLN4=
+github.com/lithammer/fuzzysearch v1.1.8/go.mod h1:IdqeyBClc3FFqSzYq/MXESsS4S0FsZ5ajtkr5xPLts4=
+github.com/mattn/go-runewidth v0.0.13/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w=
+github.com/mattn/go-runewidth v0.0.15 h1:UNAjwbU9l54TA3KzvqLGxwWjHmMgBUVhBiTjelZgg3U=
+github.com/mattn/go-runewidth v0.0.15/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w=
 github.com/mitchellh/mapstructure v1.5.0 h1:jeMsZIYE/09sWLaz43PL7Gy6RuMjD2eJVyuac5Z2hdY=
 github.com/mitchellh/mapstructure v1.5.0/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
@@ -58,23 +100,90 @@ github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9G
 github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/pterm/pterm v0.12.27/go.mod h1:PhQ89w4i95rhgE+xedAoqous6K9X+r6aSOI2eFF7DZI=
+github.com/pterm/pterm v0.12.29/go.mod h1:WI3qxgvoQFFGKGjGnJR849gU0TsEOvKn5Q8LlY1U7lg=
+github.com/pterm/pterm v0.12.30/go.mod h1:MOqLIyMOgmTDz9yorcYbcw+HsgoZo3BQfg2wtl3HEFE=
+github.com/pterm/pterm v0.12.31/go.mod h1:32ZAWZVXD7ZfG0s8qqHXePte42kdz8ECtRyEejaWgXU=
+github.com/pterm/pterm v0.12.33/go.mod h1:x+h2uL+n7CP/rel9+bImHD5lF3nM9vJj80k9ybiiTTE=
+github.com/pterm/pterm v0.12.36/go.mod h1:NjiL09hFhT/vWjQHSj1athJpx6H8cjpHXNAK5bUw8T8=
+github.com/pterm/pterm v0.12.40/go.mod h1:ffwPLwlbXxP+rxT0GsgDTzS3y3rmpAO1NMjUkGTYf8s=
+github.com/pterm/pterm v0.12.78 h1:QTWKaIAa4B32GKwqVXtu9m1DUMgWw3VRljMkMevX+b8=
+github.com/pterm/pterm v0.12.78/go.mod h1:1v/gzOF1N0FsjbgTHZ1wVycRkKiatFvJSJC4IGaQAAo=
+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/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=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/objx v0.5.0 h1:1zr/of2m5FGMsad5YfcqgdqdWrIhu+EBEJRhR1U7z/c=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
+github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
 github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
 github.com/wolfeidau/s3iofs v1.5.0 h1:l0GsqWYnHwHYkBxDnMVL0fNA2mhRnnXm4rtRG93HsOw=
 github.com/wolfeidau/s3iofs v1.5.0/go.mod h1:fPAKzdWmZ1Z2L9vnqL6d1eb7pVsUgkUstxQUkG1HIHA=
+github.com/xo/terminfo v0.0.0-20210125001918-ca9a967f8778/go.mod h1:2MuV+tbUrU1zIOPMxZ5EncGwgmMJsa+9ucAQZXxsObs=
+github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e h1:JVG44RsyaB9T2KIHavMF/ppJZNG9ZpyihvCd0w101no=
+github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e/go.mod h1:RbqR21r5mrJuqunuUZ/Dhy/avygyECGrLceyNeo4LiM=
+github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/exp v0.0.0-20231006140011-7918f672742d h1:jtJma62tbqLibJ5sFQz8bKtEM8rJBtfilJ2qTU199MI=
 golang.org/x/exp v0.0.0-20231006140011-7918f672742d/go.mod h1:ldy0pHrwJyGW56pPQzzkH36rKxoZW1tw7ZJpeKx+hdo=
+golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
+golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
+golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
+golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/net v0.18.0 h1:mIYleuAkSbHh0tCv7RvjL3F6ZVbLjq4+R7zbOn3Kokg=
 golang.org/x/net v0.18.0/go.mod h1:/czyP5RqHAH4odGYxBJ1qz0+CE5WZ+2j1YgoEo8F2jQ=
-gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
+golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20211013075003-97ac67df715c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220319134239-a9b59b0215f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.16.0 h1:xWw16ngr6ZMtmxDyKyIgsE93KNKz5HKmMa3b8ALHidU=
+golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/term v0.0.0-20210220032956-6a3ed077a48d/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/term v0.0.0-20210615171337-6886f2dfbf5b/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
+golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
+golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
+golang.org/x/term v0.16.0 h1:m+B6fahuftsE9qjo0VWp2FW0mB3MTJvR0BaMQrq0pmE=
+golang.org/x/term v0.16.0/go.mod h1:yn7UURbUtPyrVJPGPq404EukNFxcm/foM+bV/bfcDsY=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
+golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
+golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
+golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
+golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
+golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=
+golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
+gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
diff --git a/io/s3.go b/io/s3.go
index 19f0887..a2bab6c 100644
--- a/io/s3.go
+++ b/io/s3.go
@@ -30,6 +30,7 @@ import (
 	"github.com/aws/aws-sdk-go-v2/config"
 	"github.com/aws/aws-sdk-go-v2/credentials"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/smithy-go/auth/bearer"
 	"github.com/wolfeidau/s3iofs"
 )
 
@@ -65,8 +66,15 @@ func createS3FileIO(parsed *url.URL, props map[string]string) (IO, error) {
 		})))
 	}
 
+	if tok, ok := props["token"]; ok {
+		opts = append(opts, config.WithBearerAuthTokenProvider(
+			&bearer.StaticTokenProvider{Token: bearer.Token{Value: tok}}))
+	}
+
 	if region, ok := props[S3Region]; ok {
 		opts = append(opts, config.WithRegion(region))
+	} else if region, ok := props["client.region"]; ok {
+		opts = append(opts, config.WithRegion(region))
 	}
 
 	accessKey, secretAccessKey := props[S3AccessKeyID], props[S3SecretAccessKey]
diff --git a/manifest.go b/manifest.go
index 1d1113e..3f4b8ca 100644
--- a/manifest.go
+++ b/manifest.go
@@ -23,6 +23,7 @@ import (
 
 	iceio "github.com/apache/iceberg-go/io"
 
+	"github.com/hamba/avro/v2"
 	"github.com/hamba/avro/v2/ocf"
 )
 
@@ -111,6 +112,16 @@ func (b *ManifestV1Builder) Build() ManifestFile {
 	return b.m
 }
 
+type fallbackManifestFileV1 struct {
+	manifestFileV1
+	AddedSnapshotID *int64 `avro:"added_snapshot_id"`
+}
+
+func (f *fallbackManifestFileV1) toManifest() *manifestFileV1 {
+	f.manifestFileV1.AddedSnapshotID = *f.AddedSnapshotID
+	return &f.manifestFileV1
+}
+
 type manifestFileV1 struct {
 	Path               string          `avro:"manifest_path"`
 	Len                int64           `avro:"manifest_length"`
@@ -364,16 +375,34 @@ func fetchManifestEntries(m ManifestFile, fs iceio.IO, discardDeleted bool) ([]M
 	}
 
 	metadata := dec.Metadata()
-	isVer1 := true
+	isVer1, isFallback := true, false
 	if string(metadata["format-version"]) == "2" {
 		isVer1 = false
+	} else {
+		sc, err := avro.ParseBytes(dec.Metadata()["avro.schema"])
+		if err != nil {
+			return nil, err
+		}
+
+		for _, f := range sc.(*avro.RecordSchema).Fields() {
+			if f.Name() == "snapshot_id" {
+				if f.Type().Type() == avro.Union {
+					isFallback = true
+				}
+				break
+			}
+		}
 	}
 
 	results := make([]ManifestEntry, 0)
 	for dec.HasNext() {
 		var tmp ManifestEntry
 		if isVer1 {
-			tmp = &manifestEntryV1{}
+			if isFallback {
+				tmp = &fallbackManifestEntryV1{}
+			} else {
+				tmp = &manifestEntryV1{}
+			}
 		} else {
 			tmp = &manifestEntryV2{}
 		}
@@ -382,6 +411,10 @@ func fetchManifestEntries(m ManifestFile, fs iceio.IO, discardDeleted bool) ([]M
 			return nil, err
 		}
 
+		if isFallback {
+			tmp = tmp.(*fallbackManifestEntryV1).toEntry()
+		}
+
 		if !discardDeleted || tmp.Status() != EntryStatusDELETED {
 			tmp.inheritSeqNum(m)
 			results = append(results, tmp)
@@ -461,19 +494,42 @@ func ReadManifestList(in io.Reader) ([]ManifestFile, error) {
 		return nil, err
 	}
 
-	out := make([]ManifestFile, 0)
+	sc, err := avro.ParseBytes(dec.Metadata()["avro.schema"])
+	if err != nil {
+		return nil, err
+	}
+
+	var fallbackAddedSnapshot bool
+	for _, f := range sc.(*avro.RecordSchema).Fields() {
+		if f.Name() == "added_snapshot_id" {
+			if f.Type().Type() == avro.Union {
+				fallbackAddedSnapshot = true
+			}
+			break
+		}
+	}
 
+	out := make([]ManifestFile, 0)
 	for dec.HasNext() {
 		var file ManifestFile
 		if string(dec.Metadata()["format-version"]) == "2" {
 			file = &manifestFileV2{}
 		} else {
-			file = &manifestFileV1{}
+			if fallbackAddedSnapshot {
+				file = &fallbackManifestFileV1{}
+			} else {
+				file = &manifestFileV1{}
+			}
 		}
 
 		if err := dec.Decode(file); err != nil {
 			return nil, err
 		}
+
+		if fallbackAddedSnapshot {
+			file = file.(*fallbackManifestFileV1).toManifest()
+		}
+
 		out = append(out, file)
 	}
 
@@ -643,6 +699,16 @@ type manifestEntryV1 struct {
 	Data        dataFile `avro:"data_file"`
 }
 
+type fallbackManifestEntryV1 struct {
+	manifestEntryV1
+	Snapshot *int64 `avro:"snapshot_id"`
+}
+
+func (f *fallbackManifestEntryV1) toEntry() *manifestEntryV1 {
+	f.manifestEntryV1.Snapshot = *f.Snapshot
+	return &f.manifestEntryV1
+}
+
 func (m *manifestEntryV1) inheritSeqNum(manifest ManifestFile) {}
 
 func (m *manifestEntryV1) Status() ManifestEntryStatus { return m.EntryStatus }


Test patch:
diff --git a/catalog/rest_internal_test.go b/catalog/rest_internal_test.go
new file mode 100644
index 0000000..8a2966d
--- /dev/null
+++ b/catalog/rest_internal_test.go
@@ -0,0 +1,121 @@
+// 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 catalog
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestAuthHeader(t *testing.T) {
+	mux := http.NewServeMux()
+	srv := httptest.NewServer(mux)
+
+	mux.HandleFunc("/v1/config", func(w http.ResponseWriter, r *http.Request) {
+		json.NewEncoder(w).Encode(map[string]any{
+			"defaults": map[string]any{}, "overrides": map[string]any{}})
+	})
+
+	mux.HandleFunc("/v1/oauth/tokens", func(w http.ResponseWriter, req *http.Request) {
+		assert.Equal(t, http.MethodPost, req.Method)
+
+		assert.Equal(t, req.Header.Get("Content-Type"), "application/x-www-form-urlencoded")
+
+		require.NoError(t, req.ParseForm())
+		values := req.PostForm
+		assert.Equal(t, values.Get("grant_type"), "client_credentials")
+		assert.Equal(t, values.Get("client_id"), "client")
+		assert.Equal(t, values.Get("client_secret"), "secret")
+		assert.Equal(t, values.Get("scope"), "catalog")
+
+		w.WriteHeader(http.StatusOK)
+
+		json.NewEncoder(w).Encode(map[string]any{
+			"access_token":      "some_jwt_token",
+			"token_type":        "Bearer",
+			"expires_in":        86400,
+			"issued_token_type": "urn:ietf:params:oauth:token-type:access_token",
+		})
+	})
+
+	cat, err := NewRestCatalog("rest", srv.URL,
+		WithCredential("client:secret"))
+	require.NoError(t, err)
+	assert.NotNil(t, cat)
+
+	require.IsType(t, (*sessionTransport)(nil), cat.cl.Transport)
+	assert.Equal(t, http.Header{
+		"Authorization":    {"Bearer some_jwt_token"},
+		"Content-Type":     {"application/json"},
+		"User-Agent":       {"GoIceberg/(unknown version)"},
+		"X-Client-Version": {icebergRestSpecVersion},
+	}, cat.cl.Transport.(*sessionTransport).defaultHeaders)
+}
+
+func TestAuthUriHeader(t *testing.T) {
+	mux := http.NewServeMux()
+	srv := httptest.NewServer(mux)
+
+	mux.HandleFunc("/v1/config", func(w http.ResponseWriter, r *http.Request) {
+		json.NewEncoder(w).Encode(map[string]any{
+			"defaults": map[string]any{}, "overrides": map[string]any{}})
+	})
+
+	mux.HandleFunc("/auth-token-url", func(w http.ResponseWriter, req *http.Request) {
+		assert.Equal(t, http.MethodPost, req.Method)
+
+		assert.Equal(t, req.Header.Get("Content-Type"), "application/x-www-form-urlencoded")
+
+		require.NoError(t, req.ParseForm())
+		values := req.PostForm
+		assert.Equal(t, values.Get("grant_type"), "client_credentials")
+		assert.Equal(t, values.Get("client_id"), "client")
+		assert.Equal(t, values.Get("client_secret"), "secret")
+		assert.Equal(t, values.Get("scope"), "catalog")
+
+		w.WriteHeader(http.StatusOK)
+
+		json.NewEncoder(w).Encode(map[string]any{
+			"access_token":      "some_jwt_token",
+			"token_type":        "Bearer",
+			"expires_in":        86400,
+			"issued_token_type": "urn:ietf:params:oauth:token-type:access_token",
+		})
+	})
+
+	authUri, err := url.Parse(srv.URL)
+	require.NoError(t, err)
+	cat, err := NewRestCatalog("rest", srv.URL,
+		WithCredential("client:secret"), WithAuthURI(authUri.JoinPath("auth-token-url")))
+	require.NoError(t, err)
+	assert.NotNil(t, cat)
+
+	require.IsType(t, (*sessionTransport)(nil), cat.cl.Transport)
+	assert.Equal(t, http.Header{
+		"Authorization":    {"Bearer some_jwt_token"},
+		"Content-Type":     {"application/json"},
+		"User-Agent":       {"GoIceberg/(unknown version)"},
+		"X-Client-Version": {icebergRestSpecVersion},
+	}, cat.cl.Transport.(*sessionTransport).defaultHeaders)
+}
diff --git a/catalog/rest_test.go b/catalog/rest_test.go
new file mode 100644
index 0000000..618c5e0
--- /dev/null
+++ b/catalog/rest_test.go
@@ -0,0 +1,817 @@
+// 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 catalog_test
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"testing"
+
+	"github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/catalog"
+	"github.com/apache/iceberg-go/table"
+	"github.com/stretchr/testify/suite"
+)
+
+const (
+	TestCreds = "client:secret"
+	TestToken = "some_jwt_token"
+)
+
+var (
+	TestHeaders = http.Header{
+		"X-Client-Version": {"0.14.1"},
+		"User-Agent":       {"GoIceberg/(unknown version)"},
+		"Authorization":    {"Bearer " + TestToken},
+	}
+	OAuthTestHeaders = http.Header{
+		"Content-Type": {"application/x-www-form-urlencoded"},
+	}
+)
+
+type RestCatalogSuite struct {
+	suite.Suite
+
+	srv *httptest.Server
+	mux *http.ServeMux
+
+	configVals url.Values
+}
+
+func (r *RestCatalogSuite) SetupTest() {
+	r.mux = http.NewServeMux()
+
+	r.mux.HandleFunc("/v1/config", func(w http.ResponseWriter, req *http.Request) {
+		r.Require().Equal(http.MethodGet, req.Method)
+		r.configVals = req.URL.Query()
+
+		json.NewEncoder(w).Encode(map[string]any{
+			"defaults":  map[string]any{},
+			"overrides": map[string]any{},
+		})
+	})
+
+	r.srv = httptest.NewServer(r.mux)
+}
+
+func (r *RestCatalogSuite) TearDownTest() {
+	r.srv.Close()
+	r.srv = nil
+	r.mux = nil
+	r.configVals = nil
+}
+
+func (r *RestCatalogSuite) TestToken200() {
+	r.mux.HandleFunc("/v1/oauth/tokens", func(w http.ResponseWriter, req *http.Request) {
+		r.Equal(http.MethodPost, req.Method)
+
+		r.Equal(req.Header.Get("Content-Type"), "application/x-www-form-urlencoded")
+
+		r.Require().NoError(req.ParseForm())
+		values := req.PostForm
+		r.Equal(values.Get("grant_type"), "client_credentials")
+		r.Equal(values.Get("client_id"), "client")
+		r.Equal(values.Get("client_secret"), "secret")
+		r.Equal(values.Get("scope"), "catalog")
+
+		w.WriteHeader(http.StatusOK)
+
+		json.NewEncoder(w).Encode(map[string]any{
+			"access_token":      TestToken,
+			"token_type":        "Bearer",
+			"expires_in":        86400,
+			"issued_token_type": "urn:ietf:params:oauth:token-type:access_token",
+		})
+	})
+
+	cat, err := catalog.NewRestCatalog("rest", r.srv.URL,
+		catalog.WithWarehouseLocation("s3://some-bucket"),
+		catalog.WithCredential(TestCreds))
+	r.Require().NoError(err)
+
+	r.NotNil(cat)
+	r.Equal(r.configVals.Get("warehouse"), "s3://some-bucket")
+}
+
+func (r *RestCatalogSuite) TestToken400() {
+	r.mux.HandleFunc("/v1/oauth/tokens", func(w http.ResponseWriter, req *http.Request) {
+		r.Equal(http.MethodPost, req.Method)
+
+		r.Equal(req.Header.Get("Content-Type"), "application/x-www-form-urlencoded")
+
+		w.WriteHeader(http.StatusBadRequest)
+
+		json.NewEncoder(w).Encode(map[string]any{
+			"error":             "invalid_client",
+			"error_description": "credentials for key invalid_key do not match",
+		})
+	})
+
+	cat, err := catalog.NewRestCatalog("rest", r.srv.URL, catalog.WithCredential(TestCreds))
+	r.Nil(cat)
+
+	r.ErrorIs(err, catalog.ErrRESTError)
+	r.ErrorIs(err, catalog.ErrOAuthError)
+	r.ErrorContains(err, "invalid_client: credentials for key invalid_key do not match")
+}
+
+func (r *RestCatalogSuite) TestToken200AuthUrl() {
+	r.mux.HandleFunc("/auth-token-url", func(w http.ResponseWriter, req *http.Request) {
+		r.Equal(http.MethodPost, req.Method)
+
+		r.Equal(req.Header.Get("Content-Type"), "application/x-www-form-urlencoded")
+
+		r.Require().NoError(req.ParseForm())
+		values := req.PostForm
+		r.Equal(values.Get("grant_type"), "client_credentials")
+		r.Equal(values.Get("client_id"), "client")
+		r.Equal(values.Get("client_secret"), "secret")
+		r.Equal(values.Get("scope"), "catalog")
+
+		w.WriteHeader(http.StatusOK)
+
+		json.NewEncoder(w).Encode(map[string]any{
+			"access_token":      TestToken,
+			"token_type":        "Bearer",
+			"expires_in":        86400,
+			"issued_token_type": "urn:ietf:params:oauth:token-type:access_token",
+		})
+	})
+
+	authUri, err := url.Parse(r.srv.URL)
+	r.Require().NoError(err)
+	cat, err := catalog.NewRestCatalog("rest", r.srv.URL,
+		catalog.WithWarehouseLocation("s3://some-bucket"),
+		catalog.WithCredential(TestCreds), catalog.WithAuthURI(authUri.JoinPath("auth-token-url")))
+
+	r.Require().NoError(err)
+
+	r.NotNil(cat)
+	r.Equal(r.configVals.Get("warehouse"), "s3://some-bucket")
+}
+
+func (r *RestCatalogSuite) TestToken401() {
+	r.mux.HandleFunc("/v1/oauth/tokens", func(w http.ResponseWriter, req *http.Request) {
+		r.Equal(http.MethodPost, req.Method)
+
+		r.Equal(req.Header.Get("Content-Type"), "application/x-www-form-urlencoded")
+
+		w.WriteHeader(http.StatusUnauthorized)
+
+		json.NewEncoder(w).Encode(map[string]any{
+			"error":             "invalid_client",
+			"error_description": "credentials for key invalid_key do not match",
+		})
+	})
+
+	cat, err := catalog.NewRestCatalog("rest", r.srv.URL, catalog.WithCredential(TestCreds))
+	r.Nil(cat)
+
+	r.ErrorIs(err, catalog.ErrRESTError)
+	r.ErrorIs(err, catalog.ErrOAuthError)
+	r.ErrorContains(err, "invalid_client: credentials for key invalid_key do not match")
+}
+
+func (r *RestCatalogSuite) TestListTables200() {
+	namespace := "examples"
+	r.mux.HandleFunc("/v1/namespaces/"+namespace+"/tables", func(w http.ResponseWriter, req *http.Request) {
+		r.Require().Equal(http.MethodGet, req.Method)
+
+		for k, v := range TestHeaders {
+			r.Equal(v, req.Header.Values(k))
+		}
+
+		json.NewEncoder(w).Encode(map[string]any{
+			"identifiers": []any{
+				map[string]any{
+					"namespace": []string{namespace},
+					"name":      "fooshare",
+				},
+			},
+		})
+	})
+
+	cat, err := catalog.NewRestCatalog("rest", r.srv.URL, catalog.WithOAuthToken(TestToken))
+	r.Require().NoError(err)
+
+	tables, err := cat.ListTables(context.Background(), catalog.ToRestIdentifier(namespace))
+	r.Require().NoError(err)
+	r.Equal([]table.Identifier{{"examples", "fooshare"}}, tables)
+}
+
+func (r *RestCatalogSuite) TestListTablesPrefixed200() {
+	r.mux.HandleFunc("/v1/oauth/tokens", func(w http.ResponseWriter, req *http.Request) {
+		r.Equal(http.MethodPost, req.Method)
+
+		r.Equal(req.Header.Get("Content-Type"), "application/x-www-form-urlencoded")
+
+		r.Require().NoError(req.ParseForm())
+		values := req.PostForm
+		r.Equal(values.Get("grant_type"), "client_credentials")
+		r.Equal(values.Get("client_id"), "client")
+		r.Equal(values.Get("client_secret"), "secret")
+		r.Equal(values.Get("scope"), "catalog")
+
+		w.WriteHeader(http.StatusOK)
+
+		json.NewEncoder(w).Encode(map[string]any{
+			"access_token":      TestToken,
+			"token_type":        "Bearer",
+			"expires_in":        86400,
+			"issued_token_type": "urn:ietf:params:oauth:token-type:access_token",
+		})
+	})
+
+	namespace := "examples"
+	r.mux.HandleFunc("/v1/prefix/namespaces/"+namespace+"/tables", func(w http.ResponseWriter, req *http.Request) {
+		r.Require().Equal(http.MethodGet, req.Method)
+
+		for k, v := range TestHeaders {
+			r.Equal(v, req.Header.Values(k))
+		}
+
+		json.NewEncoder(w).Encode(map[string]any{
+			"identifiers": []any{
+				map[string]any{
+					"namespace": []string{namespace},
+					"name":      "fooshare",
+				},
+			},
+		})
+	})
+
+	cat, err := catalog.NewRestCatalog("rest", r.srv.URL,
+		catalog.WithPrefix("prefix"),
+		catalog.WithWarehouseLocation("s3://some-bucket"),
+		catalog.WithCredential(TestCreds))
+	r.Require().NoError(err)
+
+	r.NotNil(cat)
+	r.Equal(r.configVals.Get("warehouse"), "s3://some-bucket")
+
+	tables, err := cat.ListTables(context.Background(), catalog.ToRestIdentifier(namespace))
+	r.Require().NoError(err)
+	r.Equal([]table.Identifier{{"examples", "fooshare"}}, tables)
+}
+
+func (r *RestCatalogSuite) TestListTables404() {
+	namespace := "examples"
+	r.mux.HandleFunc("/v1/namespaces/"+namespace+"/tables", func(w http.ResponseWriter, req *http.Request) {
+		r.Require().Equal(http.MethodGet, req.Method)
+
+		for k, v := range TestHeaders {
+			r.Equal(v, req.Header.Values(k))
+		}
+
+		w.WriteHeader(http.StatusNotFound)
+		json.NewEncoder(w).Encode(map[string]any{
+			"error": map[string]any{
+				"message": "Namespace does not exist: personal in warehouse 8bcb0838-50fc-472d-9ddb-8feb89ef5f1e",
+				"type":    "NoSuchNamespaceException",
+				"code":    404,
+			},
+		})
+	})
+
+	cat, err := catalog.NewRestCatalog("rest", r.srv.URL, catalog.WithOAuthToken(TestToken))
+	r.Require().NoError(err)
+
+	_, err = cat.ListTables(context.Background(), catalog.ToRestIdentifier(namespace))
+	r.ErrorIs(err, catalog.ErrNoSuchNamespace)
+	r.ErrorContains(err, "Namespace does not exist: personal in warehouse 8bcb0838-50fc-472d-9ddb-8feb89ef5f1e")
+}
+
+func (r *RestCatalogSuite) TestListNamespaces200() {
+	r.mux.HandleFunc("/v1/namespaces", func(w http.ResponseWriter, req *http.Request) {
+		r.Require().Equal(http.MethodGet, req.Method)
+
+		for k, v := range TestHeaders {
+			r.Equal(v, req.Header.Values(k))
+		}
+
+		json.NewEncoder(w).Encode(map[string]any{
+			"namespaces": []table.Identifier{
+				{"default"}, {"examples"}, {"fokko"}, {"system"},
+			},
+		})
+	})
+
+	cat, err := catalog.NewRestCatalog("rest", r.srv.URL, catalog.WithOAuthToken(TestToken))
+	r.Require().NoError(err)
+
+	results, err := cat.ListNamespaces(context.Background(), nil)
+	r.Require().NoError(err)
+
+	r.Equal([]table.Identifier{{"default"}, {"examples"}, {"fokko"}, {"system"}}, results)
+}
+
+func (r *RestCatalogSuite) TestListNamespaceWithParent200() {
+	r.mux.HandleFunc("/v1/namespaces", func(w http.ResponseWriter, req *http.Request) {
+		r.Require().Equal(http.MethodGet, req.Method)
+		r.Require().Equal("accounting", req.URL.Query().Get("parent"))
+
+		for k, v := range TestHeaders {
+			r.Equal(v, req.Header.Values(k))
+		}
+
+		json.NewEncoder(w).Encode(map[string]any{
+			"namespaces": []table.Identifier{
+				{"accounting", "tax"},
+			},
+		})
+	})
+
+	cat, err := catalog.NewRestCatalog("rest", r.srv.URL, catalog.WithOAuthToken(TestToken))
+	r.Require().NoError(err)
+
+	results, err := cat.ListNamespaces(context.Background(), catalog.ToRestIdentifier("accounting"))
+	r.Require().NoError(err)
+
+	r.Equal([]table.Identifier{{"accounting", "tax"}}, results)
+}
+
+func (r *RestCatalogSuite) TestListNamespaces400() {
+	r.mux.HandleFunc("/v1/namespaces", func(w http.ResponseWriter, req *http.Request) {
+		r.Require().Equal(http.MethodGet, req.Method)
+
+		for k, v := range TestHeaders {
+			r.Equal(v, req.Header.Values(k))
+		}
+
+		w.WriteHeader(http.StatusNotFound)
+		json.NewEncoder(w).Encode(map[string]any{
+			"error": map[string]any{
+				"message": "Namespace does not exist: personal in warehouse 8bcb0838-50fc-472d-9ddb-8feb89ef5f1e",
+				"type":    "NoSuchNamespaceException",
+				"code":    404,
+			},
+		})
+	})
+
+	cat, err := catalog.NewRestCatalog("rest", r.srv.URL, catalog.WithOAuthToken(TestToken))
+	r.Require().NoError(err)
+
+	_, err = cat.ListNamespaces(context.Background(), catalog.ToRestIdentifier("accounting"))
+	r.ErrorIs(err, catalog.ErrNoSuchNamespace)
+	r.ErrorContains(err, "Namespace does not exist: personal in warehouse 8bcb0838-50fc-472d-9ddb-8feb89ef5f1e")
+}
+
+func (r *RestCatalogSuite) TestCreateNamespace200() {
+	r.mux.HandleFunc("/v1/namespaces", func(w http.ResponseWriter, req *http.Request) {
+		r.Require().Equal(http.MethodPost, req.Method)
+		r.Require().Equal("application/json", req.Header.Get("Content-Type"))
+
+		for k, v := range TestHeaders {
+			r.Equal(v, req.Header.Values(k))
+		}
+
+		defer req.Body.Close()
+		dec := json.NewDecoder(req.Body)
+		body := struct {
+			Namespace table.Identifier   `json:"namespace"`
+			Props     iceberg.Properties `json:"properties"`
+		}{}
+
+		r.Require().NoError(dec.Decode(&body))
+		r.Equal(table.Identifier{"leden"}, body.Namespace)
+		r.Empty(body.Props)
+
+		json.NewEncoder(w).Encode(map[string]any{
+			"namespace": []string{"leden"}, "properties": map[string]any{},
+		})
+	})
+
+	cat, err := catalog.NewRestCatalog("rest", r.srv.URL, catalog.WithOAuthToken(TestToken))
+	r.Require().NoError(err)
+
+	r.Require().NoError(cat.CreateNamespace(context.Background(), catalog.ToRestIdentifier("leden"), nil))
+}
+
+func (r *RestCatalogSuite) TestCreateNamespaceWithProps200() {
+	r.mux.HandleFunc("/v1/namespaces", func(w http.ResponseWriter, req *http.Request) {
+		r.Require().Equal(http.MethodPost, req.Method)
+		r.Require().Equal("application/json", req.Header.Get("Content-Type"))
+
+		for k, v := range TestHeaders {
+			r.Equal(v, req.Header.Values(k))
+		}
+
+		defer req.Body.Close()
+		dec := json.NewDecoder(req.Body)
+		body := struct {
+			Namespace table.Identifier   `json:"namespace"`
+			Props     iceberg.Properties `json:"properties"`
+		}{}
+
+		r.Require().NoError(dec.Decode(&body))
+		r.Equal(table.Identifier{"leden"}, body.Namespace)
+		r.Equal(iceberg.Properties{"foo": "bar", "super": "duper"}, body.Props)
+
+		json.NewEncoder(w).Encode(map[string]any{
+			"namespace": []string{"leden"}, "properties": body.Props,
+		})
+	})
+
+	cat, err := catalog.NewRestCatalog("rest", r.srv.URL, catalog.WithOAuthToken(TestToken))
+	r.Require().NoError(err)
+
+	r.Require().NoError(cat.CreateNamespace(context.Background(), catalog.ToRestIdentifier("leden"), iceberg.Properties{"foo": "bar", "super": "duper"}))
+}
+
+func (r *RestCatalogSuite) TestCreateNamespace409() {
+	r.mux.HandleFunc("/v1/namespaces", func(w http.ResponseWriter, req *http.Request) {
+		r.Require().Equal(http.MethodPost, req.Method)
+		r.Require().Equal("application/json", req.Header.Get("Content-Type"))
+
+		for k, v := range TestHeaders {
+			r.Equal(v, req.Header.Values(k))
+		}
+
+		defer req.Body.Close()
+		dec := json.NewDecoder(req.Body)
+		body := struct {
+			Namespace table.Identifier   `json:"namespace"`
+			Props     iceberg.Properties `json:"properties"`
+		}{}
+
+		r.Require().NoError(dec.Decode(&body))
+		r.Equal(table.Identifier{"fokko"}, body.Namespace)
+		r.Empty(body.Props)
+
+		w.WriteHeader(http.StatusConflict)
+		json.NewEncoder(w).Encode(map[string]any{
+			"error": map[string]any{
+				"message": "Namespace already exists: fokko in warehouse 8bcb0838-50fc-472d-9ddb-8feb89ef5f1e",
+				"type":    "AlreadyExistsException",
+				"code":    409,
+			},
+		})
+	})
+
+	cat, err := catalog.NewRestCatalog("rest", r.srv.URL, catalog.WithOAuthToken(TestToken))
+	r.Require().NoError(err)
+
+	err = cat.CreateNamespace(context.Background(), catalog.ToRestIdentifier("fokko"), nil)
+	r.ErrorIs(err, catalog.ErrNamespaceAlreadyExists)
+	r.ErrorContains(err, "fokko in warehouse")
+}
+
+func (r *RestCatalogSuite) TestDropNamespace204() {
+	r.mux.HandleFunc("/v1/namespaces/examples", func(w http.ResponseWriter, req *http.Request) {
+		r.Require().Equal(http.MethodDelete, req.Method)
+
+		for k, v := range TestHeaders {
+			r.Equal(v, req.Header.Values(k))
+		}
+
+		w.WriteHeader(http.StatusNoContent)
+	})
+
+	cat, err := catalog.NewRestCatalog("rest", r.srv.URL, catalog.WithOAuthToken(TestToken))
+	r.Require().NoError(err)
+
+	r.NoError(cat.DropNamespace(context.Background(), catalog.ToRestIdentifier("examples")))
+}
+
+func (r *RestCatalogSuite) TestDropNamespace404() {
+	r.mux.HandleFunc("/v1/namespaces/examples", func(w http.ResponseWriter, req *http.Request) {
+		r.Require().Equal(http.MethodDelete, req.Method)
+
+		for k, v := range TestHeaders {
+			r.Equal(v, req.Header.Values(k))
+		}
+
+		w.WriteHeader(http.StatusNotFound)
+		json.NewEncoder(w).Encode(map[string]any{
+			"error": map[string]any{
+				"message": "Namespace does not exist: examples in warehouse",
+				"type":    "NoSuchNamespaceException",
+				"code":    404,
+			},
+		})
+	})
+
+	cat, err := catalog.NewRestCatalog("rest", r.srv.URL, catalog.WithOAuthToken(TestToken))
+	r.Require().NoError(err)
+
+	err = cat.DropNamespace(context.Background(), catalog.ToRestIdentifier("examples"))
+	r.ErrorIs(err, catalog.ErrNoSuchNamespace)
+	r.ErrorContains(err, "examples in warehouse")
+}
+
+func (r *RestCatalogSuite) TestLoadNamespaceProps200() {
+	r.mux.HandleFunc("/v1/namespaces/leden", func(w http.ResponseWriter, req *http.Request) {
+		r.Require().Equal(http.MethodGet, req.Method)
+
+		for k, v := range TestHeaders {
+			r.Equal(v, req.Header.Values(k))
+		}
+
+		w.WriteHeader(http.StatusOK)
+		json.NewEncoder(w).Encode(map[string]any{
+			"namespace":  []string{"fokko"},
+			"properties": map[string]any{"prop": "yes"},
+		})
+	})
+
+	cat, err := catalog.NewRestCatalog("rest", r.srv.URL, catalog.WithOAuthToken(TestToken))
+	r.Require().NoError(err)
+
+	props, err := cat.LoadNamespaceProperties(context.Background(), catalog.ToRestIdentifier("leden"))
+	r.Require().NoError(err)
+	r.Equal(iceberg.Properties{"prop": "yes"}, props)
+}
+
+func (r *RestCatalogSuite) TestLoadNamespaceProps404() {
+	r.mux.HandleFunc("/v1/namespaces/leden", func(w http.ResponseWriter, req *http.Request) {
+		r.Require().Equal(http.MethodGet, req.Method)
+
+		for k, v := range TestHeaders {
+			r.Equal(v, req.Header.Values(k))
+		}
+
+		w.WriteHeader(http.StatusNotFound)
+		json.NewEncoder(w).Encode(map[string]any{
+			"error": map[string]any{
+				"message": "Namespace does not exist: fokko22 in warehouse",
+				"type":    "NoSuchNamespaceException",
+				"code":    404,
+			},
+		})
+	})
+
+	cat, err := catalog.NewRestCatalog("rest", r.srv.URL, catalog.WithOAuthToken(TestToken))
+	r.Require().NoError(err)
+
+	_, err = cat.LoadNamespaceProperties(context.Background(), catalog.ToRestIdentifier("leden"))
+	r.ErrorIs(err, catalog.ErrNoSuchNamespace)
+	r.ErrorContains(err, "Namespace does not exist: fokko22 in warehouse")
+}
+
+func (r *RestCatalogSuite) TestUpdateNamespaceProps200() {
+	r.mux.HandleFunc("/v1/namespaces/fokko/properties", 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))
+		}
+
+		json.NewEncoder(w).Encode(map[string]any{
+			"removed": []string{},
+			"updated": []string{"prop"},
+			"missing": []string{"abc"},
+		})
+	})
+
+	cat, err := catalog.NewRestCatalog("rest", r.srv.URL, catalog.WithOAuthToken(TestToken))
+	r.Require().NoError(err)
+
+	summary, err := cat.UpdateNamespaceProperties(context.Background(), table.Identifier([]string{"fokko"}),
+		[]string{"abc"}, iceberg.Properties{"prop": "yes"})
+	r.Require().NoError(err)
+
+	r.Equal(catalog.PropertiesUpdateSummary{
+		Removed: []string{},
+		Updated: []string{"prop"},
+		Missing: []string{"abc"},
+	}, summary)
+}
+
+func (r *RestCatalogSuite) TestUpdateNamespaceProps404() {
+	r.mux.HandleFunc("/v1/namespaces/fokko/properties", 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.WriteHeader(http.StatusNotFound)
+		json.NewEncoder(w).Encode(map[string]any{
+			"error": map[string]any{
+				"message": "Namespace does not exist: does_not_exist in warehouse",
+				"type":    "NoSuchNamespaceException",
+				"code":    404,
+			},
+		})
+	})
+
+	cat, err := catalog.NewRestCatalog("rest", r.srv.URL, catalog.WithOAuthToken(TestToken))
+	r.Require().NoError(err)
+
+	_, err = cat.UpdateNamespaceProperties(context.Background(),
+		table.Identifier{"fokko"}, []string{"abc"}, iceberg.Properties{"prop": "yes"})
+	r.ErrorIs(err, catalog.ErrNoSuchNamespace)
+	r.ErrorContains(err, "Namespace does not exist: does_not_exist in warehouse")
+}
+
+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)
+
+		for k, v := range TestHeaders {
+			r.Equal(v, req.Header.Values(k))
+		}
+
+		w.Write([]byte(`{			
+			"metadata-location": "s3://warehouse/database/table/metadata/00001-5f2f8166-244c-4eae-ac36-384ecdec81fc.gz.metadata.json",
+			"metadata": {
+				"format-version": 1,
+				"table-uuid": "b55d9dda-6561-423a-8bfc-787980ce421f",
+				"location": "s3://warehouse/database/table",
+				"last-updated-ms": 1646787054459,
+				"last-column-id": 2,
+				"schema": {
+					"type": "struct",
+					"schema-id": 0,
+					"fields": [
+						{"id": 1, "name": "id", "required": false, "type": "int"},
+						{"id": 2, "name": "data", "required": false, "type": "string"}
+					]
+				},
+				"current-schema-id": 0,
+				"schemas": [
+					{
+						"type": "struct",
+						"schema-id": 0,
+						"fields": [
+							{"id": 1, "name": "id", "required": false, "type": "int"},
+							{"id": 2, "name": "data", "required": false, "type": "string"}
+						]
+					}
+				],
+				"partition-spec": [],
+				"default-spec-id": 0,
+				"partition-specs": [{"spec-id": 0, "fields": []}],
+				"last-partition-id": 999,
+				"default-sort-order-id": 0,
+				"sort-orders": [{"order-id": 0, "fields": []}],
+				"properties": {"owner": "bryan", "write.metadata.compression-codec": "gzip"},
+				"current-snapshot-id": 3497810964824022504,
+				"refs": {"main": {"snapshot-id": 3497810964824022504, "type": "branch"}},
+				"snapshots": [
+					{
+						"snapshot-id": 3497810964824022504,
+						"timestamp-ms": 1646787054459,
+						"summary": {
+							"operation": "append",
+							"spark.app.id": "local-1646787004168",
+							"added-data-files": "1",
+							"added-records": "1",
+							"added-files-size": "697",
+							"changed-partition-count": "1",
+							"total-records": "1",
+							"total-files-size": "697",
+							"total-data-files": "1",
+							"total-delete-files": "0",
+							"total-position-deletes": "0",
+							"total-equality-deletes": "0"
+						},
+						"manifest-list": "s3://warehouse/database/table/metadata/snap-3497810964824022504-1-c4f68204-666b-4e50-a9df-b10c34bf6b82.avro",
+						"schema-id": 0
+					}
+				],
+				"snapshot-log": [{"timestamp-ms": 1646787054459, "snapshot-id": 3497810964824022504}],
+				"metadata-log": [
+					{
+						"timestamp-ms": 1646787031514,
+						"metadata-file": "s3://warehouse/database/table/metadata/00000-88484a1c-00e5-4a07-a787-c0e7aeffa805.gz.metadata.json"
+					}
+				]
+			}
+		}`))
+	})
+
+	cat, err := catalog.NewRestCatalog("rest", r.srv.URL, catalog.WithOAuthToken(TestToken))
+	r.Require().NoError(err)
+
+	tbl, err := cat.LoadTable(context.Background(), catalog.ToRestIdentifier("fokko", "table"), nil)
+	r.Require().NoError(err)
+
+	r.Equal(catalog.ToRestIdentifier("rest", "fokko", "table"), tbl.Identifier())
+	r.Equal("s3://warehouse/database/table/metadata/00001-5f2f8166-244c-4eae-ac36-384ecdec81fc.gz.metadata.json", tbl.MetadataLocation())
+	r.EqualValues(1, tbl.Metadata().Version())
+	r.Equal("b55d9dda-6561-423a-8bfc-787980ce421f", tbl.Metadata().TableUUID().String())
+	r.EqualValues(1646787054459, tbl.Metadata().LastUpdatedMillis())
+	r.Equal(2, 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())
+	r.EqualValues(3497810964824022504, tbl.CurrentSnapshot().SnapshotID)
+	zero := 0
+	r.True(tbl.SnapshotByName("main").Equals(table.Snapshot{
+		SnapshotID:   3497810964824022504,
+		TimestampMs:  1646787054459,
+		SchemaID:     &zero,
+		ManifestList: "s3://warehouse/database/table/metadata/snap-3497810964824022504-1-c4f68204-666b-4e50-a9df-b10c34bf6b82.avro",
+		Summary: &table.Summary{
+			Operation: table.OpAppend,
+			Properties: map[string]string{
+				"spark.app.id":            "local-1646787004168",
+				"added-data-files":        "1",
+				"added-records":           "1",
+				"added-files-size":        "697",
+				"changed-partition-count": "1",
+				"total-records":           "1",
+				"total-files-size":        "697",
+				"total-data-files":        "1",
+				"total-delete-files":      "0",
+				"total-position-deletes":  "0",
+				"total-equality-deletes":  "0",
+			},
+		},
+	}))
+}
+
+type RestTLSCatalogSuite struct {
+	suite.Suite
+
+	srv *httptest.Server
+	mux *http.ServeMux
+
+	configVals url.Values
+}
+
+func (r *RestTLSCatalogSuite) SetupTest() {
+	r.mux = http.NewServeMux()
+
+	r.mux.HandleFunc("/v1/config", func(w http.ResponseWriter, req *http.Request) {
+		r.Require().Equal(http.MethodGet, req.Method)
+		r.configVals = req.URL.Query()
+
+		json.NewEncoder(w).Encode(map[string]any{
+			"defaults":  map[string]any{},
+			"overrides": map[string]any{},
+		})
+	})
+
+	r.srv = httptest.NewTLSServer(r.mux)
+}
+
+func (r *RestTLSCatalogSuite) TearDownTest() {
+	r.srv.Close()
+	r.srv = nil
+	r.mux = nil
+	r.configVals = nil
+}
+
+func (r *RestTLSCatalogSuite) TestSSLFail() {
+	cat, err := catalog.NewRestCatalog("rest", r.srv.URL, catalog.WithOAuthToken(TestToken))
+	r.Nil(cat)
+
+	r.ErrorContains(err, "tls: failed to verify certificate")
+}
+
+func (r *RestTLSCatalogSuite) TestSSLConfig() {
+	cat, err := catalog.NewRestCatalog("rest", r.srv.URL, catalog.WithOAuthToken(TestToken),
+		catalog.WithWarehouseLocation("s3://some-bucket"),
+		catalog.WithTLSConfig(&tls.Config{InsecureSkipVerify: true}))
+	r.NoError(err)
+
+	r.NotNil(cat)
+	r.Equal(r.configVals.Get("warehouse"), "s3://some-bucket")
+}
+
+func (r *RestTLSCatalogSuite) TestSSLCerts() {
+	certs := x509.NewCertPool()
+	for _, c := range r.srv.TLS.Certificates {
+		roots, err := x509.ParseCertificates(c.Certificate[len(c.Certificate)-1])
+		r.Require().NoError(err)
+		for _, root := range roots {
+			certs.AddCert(root)
+		}
+	}
+
+	cat, err := catalog.NewRestCatalog("rest", r.srv.URL, catalog.WithOAuthToken(TestToken),
+		catalog.WithWarehouseLocation("s3://some-bucket"),
+		catalog.WithTLSConfig(&tls.Config{RootCAs: certs}))
+	r.NoError(err)
+
+	r.NotNil(cat)
+	r.Equal(r.configVals.Get("warehouse"), "s3://some-bucket")
+}
+
+func TestRestCatalog(t *testing.T) {
+	suite.Run(t, new(RestCatalogSuite))
+	suite.Run(t, new(RestTLSCatalogSuite))
+}