This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new c0dc53fd2 feat(rest): freshness-aware table loading with If-None-Match
/ 304 (#2188)
c0dc53fd2 is described below
commit c0dc53fd27620f5cd34ff997901e645f8089fee4
Author: Revanth <[email protected]>
AuthorDate: Mon Oct 19 10:27:32 2026 -0500
feat(rest): freshness-aware table loading with If-None-Match / 304 (#2188)
* feat(rest): freshness-aware table loading with If-None-Match / 304
* fix(rest): copy cached labels per table and key cache hits on snapshot
mode
---
catalog/rest/rest.go | 60 +++++-
catalog/rest/table_cache.go | 169 +++++++++++++++
catalog/rest/table_cache_internal_test.go | 88 ++++++++
catalog/rest/table_cache_test.go | 331 ++++++++++++++++++++++++++++++
4 files changed, 646 insertions(+), 2 deletions(-)
diff --git a/catalog/rest/rest.go b/catalog/rest/rest.go
index d0f727b92..33ffe5c60 100644
--- a/catalog/rest/rest.go
+++ b/catalog/rest/rest.go
@@ -112,6 +112,9 @@ var (
ErrCommitFailed = fmt.Errorf("%w: %w", ErrRESTError,
table.ErrCommitFailed)
ErrCommitStateUnknown = fmt.Errorf("%w: commit failed due to unknown
reason", ErrRESTError)
ErrOAuthError = fmt.Errorf("%w: oauth error", ErrRESTError)
+
+ // errNotModified reports a 304 to a request sent with withIfNoneMatch.
+ errNotModified = errors.New("not modified")
)
// SnapshotMode controls which snapshots are included in a loadTable response.
@@ -381,6 +384,11 @@ type reqConfig struct {
allowNoContent bool
requireBody bool
queryParams url.Values
+ // allowNotModified turns a 304 into errNotModified. withIfNoneMatch
sets it
+ // together with the header, so a 304 to a request that sent no ETag
stays an
+ // error.
+ allowNotModified bool
+ onHeaders func(http.Header)
}
type reqOption func(*reqConfig)
@@ -434,6 +442,20 @@ func requireBody() reqOption {
return func(c *reqConfig) { c.requireBody = true }
}
+// withIfNoneMatch sends If-None-Match with etag and reports a 304 Not
+// Modified response as errNotModified instead of an error response.
+func withIfNoneMatch(etag string) reqOption {
+ return func(c *reqConfig) {
+ withHeaders(map[string]string{"If-None-Match": etag})(c)
+ c.allowNotModified = true
+ }
+}
+
+// withResponseHeaders passes the headers of a 200 response to fn.
+func withResponseHeaders(fn func(http.Header)) reqOption {
+ return func(c *reqConfig) { c.onHeaders = fn }
+}
+
// Appends query parameters to the request.
func withQueryParams(params url.Values) reqOption {
return func(c *reqConfig) {
@@ -511,10 +533,18 @@ func do[T any](ctx context.Context, method string,
baseURI *url.URL, path []stri
return ret, err
}
+ if cfg.allowNotModified && rsp.StatusCode == http.StatusNotModified {
+ return ret, errNotModified
+ }
+
if rsp.StatusCode != http.StatusOK {
return ret, handleNon200(rsp, override, cfg.errorTypeOverride)
}
+ if cfg.onHeaders != nil {
+ cfg.onHeaders(rsp.Header)
+ }
+
if method == http.MethodHead || method == http.MethodDelete {
return ret, err
}
@@ -886,6 +916,9 @@ type Catalog struct {
namespaceSeparator string
snapshotMode SnapshotMode
+ // tables holds ETag-tagged load responses for freshness-aware loading;
nil
+ // when rest-table-cache.max-entries is 0.
+ tables *tableCache
// reporter builds and caches the catalog's metrics reporter once, so
it is
// constructed per-catalog rather than per table load. Released by
Close.
@@ -1213,6 +1246,9 @@ func (r *Catalog) fetchConfig(ctx context.Context, opts
*options) (*options, err
return nil, fmt.Errorf("%w: invalid %s %q (want %q or %q)",
ErrRESTError, keySnapshotLoadingMode, r.snapshotMode,
SnapshotModeAll, SnapshotModeRefs)
}
+ if r.tables, err = newTableCache(cfg); err != nil {
+ return nil, err
+ }
// Negotiate capabilities from the endpoints the server advertises,
falling
// back to a backward-compatible default set when none are provided.
@@ -1252,6 +1288,7 @@ func (r *Catalog) Close() error {
r.metricsDispatcher.close()
}
r.closeErr = r.reporter.Close()
+ r.tables.purge()
})
return r.closeErr
@@ -1946,15 +1983,29 @@ func (r *Catalog) loadTableWithMode(ctx
context.Context, identifier table.Identi
return nil, err
}
- var opts []reqOption
+ var etag string
+ opts := []reqOption{withResponseHeaders(func(h http.Header) { etag =
h.Get("ETag") })}
if mode != "" {
opts = append(opts, withQueryParams(url.Values{"snapshots":
{string(mode)}}))
}
+ cached, hit := r.tables.get(identifier, mode)
+ if hit {
+ opts = append(opts, withIfNoneMatch(cached.etag))
+ }
ret, err := doGet[loadTableResponse](ctx, r.baseURI, path,
r.cl, map[int]error{http.StatusNotFound:
catalog.ErrNoSuchTable}, opts...)
- if err != nil {
+ switch {
+ case errors.Is(err, errNotModified):
+ ret = cached.rsp
+ case errors.Is(err, catalog.ErrNoSuchTable):
+ r.tables.invalidate(identifier)
+
return nil, err
+ case err != nil:
+ return nil, err
+ case etag != "":
+ r.tables.put(identifier, mode, etag, ret)
}
config := maps.Clone(r.props)
@@ -2042,6 +2093,7 @@ func (r *Catalog) DropTable(ctx context.Context,
identifier table.Identifier) er
_, err = doDelete[struct{}](ctx, uri, []string{}, r.cl,
map[int]error{http.StatusNotFound: catalog.ErrNoSuchTable})
+ r.tables.invalidate(identifier)
return err
}
@@ -2068,6 +2120,7 @@ func (r *Catalog) PurgeTable(ctx context.Context,
identifier table.Identifier) e
_, err = doDelete[struct{}](ctx, uri, []string{}, r.cl,
map[int]error{http.StatusNotFound: catalog.ErrNoSuchTable})
+ r.tables.invalidate(identifier)
return err
}
@@ -2111,6 +2164,7 @@ func (r *Catalog) RenameTable(ctx context.Context, from,
to table.Identifier) (*
if err != nil {
return nil, err
}
+ r.tables.invalidate(from)
return r.LoadTable(ctx, to)
}
@@ -2348,6 +2402,8 @@ func (r *Catalog) CheckTableExists(ctx context.Context,
identifier table.Identif
r.cl, map[int]error{http.StatusNotFound:
catalog.ErrNoSuchTable})
if err != nil {
if errors.Is(err, catalog.ErrNoSuchTable) {
+ r.tables.invalidate(identifier)
+
return false, nil
}
diff --git a/catalog/rest/table_cache.go b/catalog/rest/table_cache.go
new file mode 100644
index 000000000..a27a5e930
--- /dev/null
+++ b/catalog/rest/table_cache.go
@@ -0,0 +1,169 @@
+// 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 rest
+
+import (
+ "fmt"
+ "maps"
+ "strconv"
+ "time"
+
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/table"
+ lru "github.com/hashicorp/golang-lru/v2"
+)
+
+// Catalog properties controlling the table cache used for freshness-aware
+// table loading. The names and defaults match Iceberg Java's
+// RESTCatalogProperties so the same configuration works across clients.
+const (
+ keyTableCacheMaxEntries = "rest-table-cache.max-entries"
+ keyTableCacheExpireAfterWriteMs =
"rest-table-cache.expire-after-write-ms"
+
+ defaultTableCacheMaxEntries = 100
+ defaultTableCacheExpireAfterWrite = 5 * time.Minute
+)
+
+// tableCacheEntry is a load response the server tagged with an ETag. The
+// parsed response is kept rather than a *table.Table because Table.Refresh
+// mutates its receiver: every 304 builds a fresh table from the response so
+// callers never share one. The response's Labels are mutable through the
+// table, so the entry holds its own copy and hands out a copy on every hit.
+type tableCacheEntry struct {
+ etag string
+ // mode is the snapshot loading mode the response was loaded with. A
refs
+ // response replayed for an all request would silently drop snapshots,
so
+ // a load with another mode does not reuse the entry.
+ mode SnapshotMode
+ rsp loadTableResponse
+ written time.Time
+}
+
+// tableCache maps a table identifier to its last ETag-tagged load response so
+// LoadTable can send If-None-Match and reuse the response on 304 Not Modified.
+// Entries expire a fixed time after they are written; reads do not extend
+// them. A nil *tableCache is a disabled cache and every method is a no-op.
+type tableCache struct {
+ entries *lru.Cache[string, tableCacheEntry]
+ ttl time.Duration
+ now func() time.Time
+}
+
+// newTableCache builds the cache from the merged catalog properties. It
+// returns nil when max-entries is 0, which disables freshness-aware loading.
+func newTableCache(props iceberg.Properties) (*tableCache, error) {
+ maxEntries := defaultTableCacheMaxEntries
+ if v, ok := props[keyTableCacheMaxEntries]; ok {
+ n, err := strconv.Atoi(v)
+ if err != nil || n < 0 {
+ return nil, fmt.Errorf("%w: invalid %s %q: want a
non-negative integer",
+ ErrRESTError, keyTableCacheMaxEntries, v)
+ }
+ maxEntries = n
+ }
+
+ ttl := defaultTableCacheExpireAfterWrite
+ if v, ok := props[keyTableCacheExpireAfterWriteMs]; ok {
+ ms, err := strconv.ParseInt(v, 10, 64)
+ if err != nil || ms <= 0 {
+ return nil, fmt.Errorf("%w: invalid %s %q: want a
positive integer",
+ ErrRESTError, keyTableCacheExpireAfterWriteMs,
v)
+ }
+ ttl = time.Duration(ms) * time.Millisecond
+ }
+
+ if maxEntries == 0 {
+ return nil, nil
+ }
+
+ entries, err := lru.New[string, tableCacheEntry](maxEntries)
+ if err != nil {
+ return nil, err
+ }
+
+ return &tableCache{entries: entries, ttl: ttl, now: time.Now}, nil
+}
+
+// tableCacheKey is unambiguous for any identifier, including parts that
+// contain separators.
+func tableCacheKey(ident table.Identifier) string {
+ return fmt.Sprintf("%q", []string(ident))
+}
+
+func (c *tableCache) get(ident table.Identifier, mode SnapshotMode)
(tableCacheEntry, bool) {
+ if c == nil {
+ return tableCacheEntry{}, false
+ }
+
+ key := tableCacheKey(ident)
+ e, ok := c.entries.Get(key)
+ if !ok || e.mode != mode {
+ return tableCacheEntry{}, false
+ }
+ if c.now().Sub(e.written) >= c.ttl {
+ c.entries.Remove(key)
+
+ return tableCacheEntry{}, false
+ }
+ e.rsp.Labels = cloneLabels(e.rsp.Labels)
+
+ return e, true
+}
+
+func (c *tableCache) put(ident table.Identifier, mode SnapshotMode, etag
string, rsp loadTableResponse) {
+ if c == nil {
+ return
+ }
+
+ rsp.Labels = cloneLabels(rsp.Labels)
+ c.entries.Add(tableCacheKey(ident), tableCacheEntry{etag: etag, mode:
mode, rsp: rsp, written: c.now()})
+}
+
+// cloneLabels deep-copies l, so a table built from a cached response cannot
+// change the labels of the cache entry or of another table.
+func cloneLabels(l *iceberg.Labels) *iceberg.Labels {
+ if l == nil {
+ return nil
+ }
+
+ out := &iceberg.Labels{ObjectLabels: maps.Clone(l.ObjectLabels)}
+ if l.Fields != nil {
+ out.Fields = make([]iceberg.FieldLabel, len(l.Fields))
+ for i, f := range l.Fields {
+ out.Fields[i] = iceberg.FieldLabel{FieldID: f.FieldID,
Labels: maps.Clone(f.Labels)}
+ }
+ }
+
+ return out
+}
+
+func (c *tableCache) invalidate(ident table.Identifier) {
+ if c == nil {
+ return
+ }
+
+ c.entries.Remove(tableCacheKey(ident))
+}
+
+func (c *tableCache) purge() {
+ if c == nil {
+ return
+ }
+
+ c.entries.Purge()
+}
diff --git a/catalog/rest/table_cache_internal_test.go
b/catalog/rest/table_cache_internal_test.go
new file mode 100644
index 000000000..a2ed48198
--- /dev/null
+++ b/catalog/rest/table_cache_internal_test.go
@@ -0,0 +1,88 @@
+// 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 rest
+
+import (
+ "testing"
+ "time"
+
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/table"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func TestTableCacheDefaults(t *testing.T) {
+ c, err := newTableCache(nil)
+ require.NoError(t, err)
+ require.NotNil(t, c)
+ assert.Equal(t, 5*time.Minute, c.ttl)
+
+ for i := range defaultTableCacheMaxEntries + 1 {
+ c.put(table.Identifier{"ns", string(rune('a' + i%26)),
string(rune('a' + i/26))}, SnapshotModeAll, "e", loadTableResponse{})
+ }
+ assert.Equal(t, defaultTableCacheMaxEntries, c.entries.Len())
+}
+
+func TestTableCacheExpiresAfterWrite(t *testing.T) {
+ c, err :=
newTableCache(iceberg.Properties{keyTableCacheExpireAfterWriteMs: "1000"})
+ require.NoError(t, err)
+
+ now := time.Unix(0, 0)
+ c.now = func() time.Time { return now }
+ ident := table.Identifier{"ns", "tbl"}
+
+ c.put(ident, SnapshotModeAll, `"v1"`, loadTableResponse{})
+
+ // Reads do not extend an entry's lifetime.
+ now = now.Add(999 * time.Millisecond)
+ e, ok := c.get(ident, SnapshotModeAll)
+ require.True(t, ok)
+ assert.Equal(t, `"v1"`, e.etag)
+
+ now = now.Add(time.Millisecond)
+ _, ok = c.get(ident, SnapshotModeAll)
+ assert.False(t, ok)
+ assert.Zero(t, c.entries.Len())
+}
+
+func TestTableCacheKeyIsUnambiguous(t *testing.T) {
+ assert.NotEqual(t,
+ tableCacheKey(table.Identifier{"a.b", "c"}),
+ tableCacheKey(table.Identifier{"a", "b.c"}))
+ assert.NotEqual(t,
+ tableCacheKey(table.Identifier{`a" "b`, "c"}),
+ tableCacheKey(table.Identifier{"a", "b", "c"}))
+}
+
+// TestTableCacheIgnoresEntryForOtherSnapshotMode checks that a response loaded
+// with one snapshot mode is not reused for a load with another: a refs
response
+// replayed for an all request would silently drop snapshots.
+func TestTableCacheIgnoresEntryForOtherSnapshotMode(t *testing.T) {
+ c, err := newTableCache(nil)
+ require.NoError(t, err)
+ ident := table.Identifier{"ns", "tbl"}
+
+ c.put(ident, SnapshotModeRefs, `"v1"`, loadTableResponse{})
+
+ _, ok := c.get(ident, SnapshotModeAll)
+ assert.False(t, ok)
+ e, ok := c.get(ident, SnapshotModeRefs)
+ require.True(t, ok)
+ assert.Equal(t, `"v1"`, e.etag)
+}
diff --git a/catalog/rest/table_cache_test.go b/catalog/rest/table_cache_test.go
new file mode 100644
index 000000000..55a5c88bc
--- /dev/null
+++ b/catalog/rest/table_cache_test.go
@@ -0,0 +1,331 @@
+// 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 rest_test
+
+import (
+ "context"
+ "encoding/json"
+ "net/http"
+ "net/http/httptest"
+ "sync"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/catalog"
+ "github.com/apache/iceberg-go/catalog/rest"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// etagServer serves one table and answers loads conditionally: 304 when the
+// request's If-None-Match equals the current ETag, otherwise 200 with the
+// ETag. It records the If-None-Match of every load.
+type etagServer struct {
+ mu sync.Mutex
+ etag string
+ exists bool
+ body string
+ ifNoneMatch []string
+}
+
+func (s *etagServer) lastIfNoneMatch() string {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+
+ return s.ifNoneMatch[len(s.ifNoneMatch)-1]
+}
+
+func newETagCatalog(t *testing.T, props iceberg.Properties) (*rest.Catalog,
*etagServer) {
+ t.Helper()
+
+ s := &etagServer{etag: `"v1"`, exists: true, body:
createTableRestExample}
+ mux := http.NewServeMux()
+ mux.HandleFunc("/v1/config", func(w http.ResponseWriter, _
*http.Request) {
+ assert.NoError(t, json.NewEncoder(w).Encode(map[string]any{
+ "defaults": map[string]any{}, "overrides":
map[string]any{},
+ "endpoints": rest.AllEndpointStrings,
+ }))
+ })
+ mux.HandleFunc("/v1/namespaces/fokko/tables/table", func(w
http.ResponseWriter, req *http.Request) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+
+ switch req.Method {
+ case http.MethodGet:
+ s.ifNoneMatch = append(s.ifNoneMatch,
req.Header.Get("If-None-Match"))
+ case http.MethodHead:
+ case http.MethodDelete:
+ s.exists = false
+ w.WriteHeader(http.StatusNoContent)
+
+ return
+ }
+ if !s.exists {
+ w.WriteHeader(http.StatusNotFound)
+
+ return
+ }
+ if req.Method == http.MethodHead {
+ w.WriteHeader(http.StatusNoContent)
+
+ return
+ }
+ if s.etag != "" && req.Header.Get("If-None-Match") == s.etag {
+ w.WriteHeader(http.StatusNotModified)
+
+ return
+ }
+ if s.etag != "" {
+ w.Header().Set("ETag", s.etag)
+ }
+ _, err := w.Write([]byte(s.body))
+ assert.NoError(t, err)
+ })
+ mux.HandleFunc("/v1/tables/rename", func(w http.ResponseWriter, _
*http.Request) {
+ w.WriteHeader(http.StatusNoContent)
+ })
+ mux.HandleFunc("/v1/namespaces/fokko/tables/renamed", func(w
http.ResponseWriter, _ *http.Request) {
+ _, err := w.Write([]byte(createTableRestExample))
+ assert.NoError(t, err)
+ })
+ srv := httptest.NewServer(mux)
+ t.Cleanup(srv.Close)
+
+ cat, err := rest.NewCatalog(context.Background(), "rest", srv.URL,
+ rest.WithOAuthToken(TestToken), rest.WithAdditionalProps(props))
+ require.NoError(t, err)
+ t.Cleanup(func() { _ = cat.Close() })
+
+ return cat, s
+}
+
+var etagIdent = catalog.ToIdentifier("fokko", "table")
+
+func TestLoadTableNotModifiedReusesCachedResponse(t *testing.T) {
+ ctx := context.Background()
+ cat, srv := newETagCatalog(t, nil)
+
+ first, err := cat.LoadTable(ctx, etagIdent)
+ require.NoError(t, err)
+ assert.Empty(t, srv.lastIfNoneMatch())
+
+ second, err := cat.LoadTable(ctx, etagIdent)
+ require.NoError(t, err)
+ assert.Equal(t, `"v1"`, srv.lastIfNoneMatch())
+
+ // A 304 builds a new table from the cached response: Table.Refresh
mutates
+ // its receiver, so two loads must never share one.
+ assert.NotSame(t, first, second)
+ assert.True(t, first.Equals(*second))
+ assert.Equal(t, first.MetadataLocation(), second.MetadataLocation())
+ assert.Equal(t, first.Properties(), second.Properties())
+}
+
+// TestLoadTableNotModifiedDoesNotShareLabels checks that tables built from one
+// cached response get independent labels. Labels are mutable through the
table,
+// so a change made through one table must not reach the cache or another
table.
+func TestLoadTableNotModifiedDoesNotShareLabels(t *testing.T) {
+ ctx := context.Background()
+ cat, srv := newETagCatalog(t, nil)
+
+ var body map[string]any
+ require.NoError(t, json.Unmarshal([]byte(createTableRestExample),
&body))
+ body["labels"] = map[string]any{
+ "object-labels": map[string]string{"owner": "data-eng"},
+ "fields": []any{map[string]any{"field-id": 1, "labels":
map[string]string{"pii": "no"}}},
+ }
+ raw, err := json.Marshal(body)
+ require.NoError(t, err)
+ srv.mu.Lock()
+ srv.body = string(raw)
+ srv.mu.Unlock()
+
+ first, err := cat.LoadTable(ctx, etagIdent)
+ require.NoError(t, err)
+ first.Labels().Object()["owner"] = "changed"
+ first.Labels().Field(1)["pii"] = "yes"
+
+ for range 2 {
+ next, err := cat.LoadTable(ctx, etagIdent)
+ require.NoError(t, err)
+ require.Equal(t, `"v1"`, srv.lastIfNoneMatch(), "the load must
be served from the cache")
+ assert.Equal(t, "data-eng", next.Labels().Object()["owner"])
+ assert.Equal(t, "no", next.Labels().Field(1)["pii"])
+ next.Labels().Object()["owner"] = "changed-again"
+ }
+}
+
+func TestLoadTableModifiedReplacesCachedETag(t *testing.T) {
+ ctx := context.Background()
+ cat, srv := newETagCatalog(t, nil)
+
+ _, err := cat.LoadTable(ctx, etagIdent)
+ require.NoError(t, err)
+
+ srv.mu.Lock()
+ srv.etag = `"v2"`
+ srv.mu.Unlock()
+
+ _, err = cat.LoadTable(ctx, etagIdent)
+ require.NoError(t, err)
+ assert.Equal(t, `"v1"`, srv.lastIfNoneMatch())
+
+ _, err = cat.LoadTable(ctx, etagIdent)
+ require.NoError(t, err)
+ assert.Equal(t, `"v2"`, srv.lastIfNoneMatch())
+}
+
+func TestLoadTableWithoutETagIsNotCached(t *testing.T) {
+ ctx := context.Background()
+ cat, srv := newETagCatalog(t, nil)
+ srv.etag = ""
+
+ for range 2 {
+ _, err := cat.LoadTable(ctx, etagIdent)
+ require.NoError(t, err)
+ assert.Empty(t, srv.lastIfNoneMatch())
+ }
+}
+
+func TestLoadTableCacheDisabled(t *testing.T) {
+ ctx := context.Background()
+ cat, srv := newETagCatalog(t,
iceberg.Properties{"rest-table-cache.max-entries": "0"})
+
+ for range 2 {
+ _, err := cat.LoadTable(ctx, etagIdent)
+ require.NoError(t, err)
+ assert.Empty(t, srv.lastIfNoneMatch())
+ }
+}
+
+func TestLoadTableUnsolicitedNotModified(t *testing.T) {
+ mux := http.NewServeMux()
+ mux.HandleFunc("/v1/config", func(w http.ResponseWriter, _
*http.Request) {
+ assert.NoError(t, json.NewEncoder(w).Encode(map[string]any{
+ "defaults": map[string]any{}, "overrides":
map[string]any{},
+ "endpoints": rest.AllEndpointStrings,
+ }))
+ })
+ mux.HandleFunc("/v1/namespaces/fokko/tables/table", func(w
http.ResponseWriter, _ *http.Request) {
+ w.WriteHeader(http.StatusNotModified)
+ })
+ srv := httptest.NewServer(mux)
+ defer srv.Close()
+
+ cat, err := rest.NewCatalog(context.Background(), "rest", srv.URL,
rest.WithOAuthToken(TestToken))
+ require.NoError(t, err)
+
+ _, err = cat.LoadTable(context.Background(), etagIdent)
+ require.ErrorIs(t, err, rest.ErrRESTError)
+}
+
+func TestTableCacheInvalidation(t *testing.T) {
+ tests := []struct {
+ name string
+ op func(ctx context.Context, cat *rest.Catalog, srv
*etagServer) error
+ }{
+ {"drop", func(ctx context.Context, cat *rest.Catalog, srv
*etagServer) error {
+ if err := cat.DropTable(ctx, etagIdent); err != nil {
+ return err
+ }
+ srv.mu.Lock()
+ srv.exists = true
+ srv.mu.Unlock()
+
+ return nil
+ }},
+ {"purge", func(ctx context.Context, cat *rest.Catalog, srv
*etagServer) error {
+ if err := cat.PurgeTable(ctx, etagIdent); err != nil {
+ return err
+ }
+ srv.mu.Lock()
+ srv.exists = true
+ srv.mu.Unlock()
+
+ return nil
+ }},
+ {"rename", func(ctx context.Context, cat *rest.Catalog, _
*etagServer) error {
+ _, err := cat.RenameTable(ctx, etagIdent,
catalog.ToIdentifier("fokko", "renamed"))
+
+ return err
+ }},
+ {"exists reports missing", func(ctx context.Context, cat
*rest.Catalog, srv *etagServer) error {
+ srv.mu.Lock()
+ srv.exists = false
+ srv.mu.Unlock()
+ if ok, err := cat.CheckTableExists(ctx, etagIdent); err
!= nil || ok {
+ return assert.AnError
+ }
+ srv.mu.Lock()
+ srv.exists = true
+ srv.mu.Unlock()
+
+ return nil
+ }},
+ {"load reports missing", func(ctx context.Context, cat
*rest.Catalog, srv *etagServer) error {
+ srv.mu.Lock()
+ srv.exists = false
+ srv.mu.Unlock()
+ if _, err := cat.LoadTable(ctx, etagIdent);
!assert.ErrorIs(t, err, catalog.ErrNoSuchTable) {
+ return assert.AnError
+ }
+ srv.mu.Lock()
+ srv.exists = true
+ srv.mu.Unlock()
+
+ return nil
+ }},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ ctx := context.Background()
+ cat, srv := newETagCatalog(t, nil)
+
+ _, err := cat.LoadTable(ctx, etagIdent)
+ require.NoError(t, err)
+ require.NoError(t, tt.op(ctx, cat, srv))
+
+ _, err = cat.LoadTable(ctx, etagIdent)
+ require.NoError(t, err)
+ assert.Empty(t, srv.lastIfNoneMatch())
+ })
+ }
+}
+
+func TestTableCacheInvalidProperties(t *testing.T) {
+ for _, props := range []iceberg.Properties{
+ {"rest-table-cache.max-entries": "-1"},
+ {"rest-table-cache.max-entries": "many"},
+ {"rest-table-cache.expire-after-write-ms": "0"},
+ {"rest-table-cache.expire-after-write-ms": "-5"},
+ } {
+ mux := http.NewServeMux()
+ mux.HandleFunc("/v1/config", func(w http.ResponseWriter, _
*http.Request) {
+ assert.NoError(t,
json.NewEncoder(w).Encode(map[string]any{
+ "defaults": map[string]any{}, "overrides":
map[string]any{},
+ }))
+ })
+ srv := httptest.NewServer(mux)
+
+ _, err := rest.NewCatalog(context.Background(), "rest", srv.URL,
+ rest.WithOAuthToken(TestToken),
rest.WithAdditionalProps(props))
+ assert.ErrorIs(t, err, rest.ErrRESTError, "%v", props)
+ srv.Close()
+ }
+}
