laskoviymishka commented on code in PR #3377:
URL: https://github.com/apache/iceberg-rust/pull/3377#discussion_r4242742941
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -1250,18 +1295,33 @@ impl SessionCatalog for RestSessionCatalog {
table_ident: &TableIdent,
) -> Result<Table> {
let client = self.client().await?;
+ let cache_key = (context.session_id().to_string(),
table_ident.clone());
Review Comment:
Worth documenting that the cache is keyed on `session_id`, so it only helps
when callers reuse a `SessionContext` — `SessionContext::empty()` mints a fresh
id each call, so those callers get a 0% hit rate while still paying the key
allocation and the table clone. Java keys the same way, so this is a doc note
rather than a bug — I'd just call it out on the property constants so a
multi-tenant proxy spinning a context per request knows what to expect.
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -4023,6 +4088,136 @@ mod tests {
rename_table_mock.assert_async().await;
}
+ /// Mocks one `loadTable` of `ns1.test1` that expects `if_none_match` and
+ /// answers `status`, carrying `etag` if given.
+ async fn load_table_mock(
+ server: &mut ServerGuard,
+ if_none_match: mockito::Matcher,
+ status: usize,
+ etag: Option<&str>,
+ ) -> Mock {
+ let mut mock = server
+ .mock("GET", "/v1/namespaces/ns1/tables/test1")
+ .match_header("if-none-match", if_none_match)
+ .with_status(status);
+ if let Some(etag) = etag {
+ mock = mock.with_header("etag", etag);
+ }
+ if status == 200 {
+ mock = mock.with_body_from_file(format!(
+ "{}/testdata/load_table_response.json",
+ env!("CARGO_MANIFEST_DIR")
+ ));
+ }
+ mock.create_async().await
+ }
+
+ fn test1() -> TableIdent {
+ TableIdent::from_strs(["ns1", "test1"]).unwrap()
+ }
+
+ #[tokio::test]
+ async fn test_load_table_revalidates_cached_table_with_etag() {
+ let mut server = Server::new_async().await;
+ let _config = create_config_mock(&mut server).await;
+ let v1 = load_table_mock(&mut server, mockito::Matcher::Missing, 200,
Some("\"v1\"")).await;
+ let reuse_v1 = load_table_mock(&mut server, "\"v1\"".into(), 304,
None).await;
+ let catalog =
session_catalog(RestCatalogConfig::builder().uri(server.url()).build());
+ let context = SessionContext::empty();
+
+ let loaded = catalog.load_table(&context, &test1()).await.unwrap();
+ let reused = catalog.load_table(&context, &test1()).await.unwrap();
+ assert_eq!(loaded.metadata_location(), reused.metadata_location());
Review Comment:
Both responses use the same `load_table_response.json`, so comparing
`metadata_location` only proves a table came back — not that the cached table
was the one returned, nor that the v2 200 actually replaced v1. I'd give v1 and
v2 distinct `metadata_location`s and assert the reused table equals v1 after
the 304 and v2 after the second 200.
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -297,6 +309,36 @@ impl RestCatalogConfig {
.unwrap_or(false)
}
+ /// Builds the table cache for freshness-aware loading.
+ fn table_cache(&self) -> Result<TableCache> {
+ let parse = |key: &str, default: u64| -> Result<u64> {
+ self.props.get(key).map_or(Ok(default), |v| {
+ v.parse().map_err(|e| {
+ Error::new(ErrorKind::DataInvalid, format!("Invalid {key}:
{v}")).with_source(e)
+ })
+ })
+ };
+ let expire_after_write = Duration::from_millis(parse(
+ REST_CATALOG_PROP_TABLE_CACHE_EXPIRE_AFTER_WRITE_MS,
+ 300_000,
+ )?);
+ // Zero is ambiguous (disable or never expire), and moka panics beyond
1000 years.
+ if expire_after_write.is_zero()
+ || expire_after_write > Duration::from_secs(1000 * 365 * 24 * 3600)
Review Comment:
`1000 * 365 * 24 * 3600` reads as a magic number and couples us to moka's
internal 1000-year panic bound — I'd lift it to a named `const
MAX_TABLE_CACHE_TTL: Duration`, alongside named consts for the `300_000` and
`100` defaults below. While here, the error drops the offending value (unlike
the `parse` error); including `{expire_after_write:?}` would make a bad config
easier to spot.
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -1250,18 +1295,33 @@ impl SessionCatalog for RestSessionCatalog {
table_ident: &TableIdent,
) -> Result<Table> {
let client = self.client().await?;
+ let cache_key = (context.session_id().to_string(),
table_ident.clone());
+ let cached = client.table_cache.get(&cache_key);
- let request = HttpRequest::build(
- client
- .http_client
- .request(Method::GET,
client.config.table_endpoint(table_ident)),
- )?;
+ let mut request_builder = client
+ .http_client
+ .request(Method::GET, client.config.table_endpoint(table_ident));
+ if let Some((etag, _)) = &cached {
+ request_builder = request_builder.header(header::IF_NONE_MATCH,
etag);
+ }
+ let request = HttpRequest::build(request_builder)?;
let http_response = client.query_catalog(context, request).await?;
+ let etag = http_response
+ .headers()
+ .get(header::ETAG)
+ .and_then(|v| v.to_str().ok())
+ .map(str::to_string);
let response = match http_response.status() {
- StatusCode::OK | StatusCode::NOT_MODIFIED => {
- deserialize_catalog_response::<LoadTableResult>(http_response)?
+ StatusCode::OK =>
deserialize_catalog_response::<LoadTableResult>(http_response)?,
+ StatusCode::NOT_MODIFIED => {
+ return cached.map(|(_, table)| table).ok_or_else(|| {
Review Comment:
A 304 hands back the cached `Table`, and that `Table` carries a `FileIO`
built from the earlier 200's `config` — including any vended `s3.session-token`
/ `gcs.oauth2.token`. The ETag only validates the metadata, and a 304 has no
body to refresh credentials from, so on a hit we return storage credentials
captured up to one TTL ago. With the 5-minute default that's usually under the
STS lifetime, but the TTL is configurable up to 1000 years — any catalog
vending tokens shorter than the TTL gets 403s from object storage on a
"successful" load, and those surface from the FileIO, not the catalog, so
they're painful to trace.
Java's `RESTSessionCatalog` doesn't serve a stale FileIO on a hit
(apache/iceberg#14398) — please confirm the exact mechanism. The cleanest fix
here is to cache the `LoadTableResult` config and rebuild the FileIO per hit;
failing that, skip caching when the 200 carried credential-bearing config, and
document that `expire-after-write-ms` must stay under the credential lifetime.
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -1296,11 +1356,16 @@ impl SessionCatalog for RestSessionCatalog {
table_builder = table_builder.kms_client(kms_client);
}
- if let Some(metadata_location) = response.metadata_location {
+ let table = if let Some(metadata_location) =
response.metadata_location {
table_builder.metadata_location(metadata_location).build()
} else {
table_builder.build()
+ }?;
+
+ if let Some(etag) = etag {
+ client.table_cache.insert(cache_key, (etag, table.clone()));
Review Comment:
When the cache is disabled (`max-entries=0`) we still allocate the key,
clone the `Table`, and call `insert` here only for moka to drop it immediately
— and the ETag string is built on every response, including 304s and errors.
I'd store `Option<TableCache>` (None when max is 0) and skip the whole path,
and move the ETag extraction into the `OK` arm so it's only built when we might
actually cache. Non-blocking perf tidy.
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -4023,6 +4088,136 @@ mod tests {
rename_table_mock.assert_async().await;
}
+ /// Mocks one `loadTable` of `ns1.test1` that expects `if_none_match` and
+ /// answers `status`, carrying `etag` if given.
+ async fn load_table_mock(
+ server: &mut ServerGuard,
+ if_none_match: mockito::Matcher,
+ status: usize,
+ etag: Option<&str>,
+ ) -> Mock {
+ let mut mock = server
+ .mock("GET", "/v1/namespaces/ns1/tables/test1")
+ .match_header("if-none-match", if_none_match)
+ .with_status(status);
+ if let Some(etag) = etag {
+ mock = mock.with_header("etag", etag);
+ }
+ if status == 200 {
+ mock = mock.with_body_from_file(format!(
+ "{}/testdata/load_table_response.json",
+ env!("CARGO_MANIFEST_DIR")
+ ));
+ }
+ mock.create_async().await
+ }
+
+ fn test1() -> TableIdent {
Review Comment:
`test1()` reads like a test rather than a fixture — `test_table_ident()` (or
inlining it) would be clearer. Minor, and while in these tests
`mockito::Matcher` gets spelled out in full repeatedly; a `use` would tidy them
up.
##########
crates/catalog/rest/Cargo.toml:
##########
@@ -35,6 +35,7 @@ chrono = { workspace = true }
http = { workspace = true }
iceberg = { workspace = true }
itertools = { workspace = true }
+moka = { version = "0.12.10", features = ["sync"] }
reqwest = { workspace = true }
Review Comment:
The `iceberg` crate already pins moka (with the `future` feature), so I'd
move the version into `[workspace.dependencies]` and use `moka = { workspace =
true }` here rather than carry two independent pins. Not a blocker.
```toml
moka = { workspace = true, features = ["sync"] }
```
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -65,6 +66,17 @@ pub const REST_CATALOG_PROP_DISABLE_HEADER_REDACTION: &str =
"disable-header-red
/// if a `token`, `credential` or `oauth2-server-uri` is configured, `none`
/// otherwise.
pub const REST_CATALOG_PROP_AUTH_TYPE: &str = "rest.auth.type";
+/// Maximum number of loaded tables kept for freshness-aware loading (defaults
+/// to 100; 0 disables the cache).
+pub const REST_CATALOG_PROP_TABLE_CACHE_MAX_ENTRIES: &str =
"rest-table-cache.max-entries";
+/// How long a loaded table stays cached for freshness-aware loading, in
+/// milliseconds (defaults to 5 minutes; must be positive and at most 1000
years).
+pub const REST_CATALOG_PROP_TABLE_CACHE_EXPIRE_AFTER_WRITE_MS: &str =
+ "rest-table-cache.expire-after-write-ms";
+
+/// Tables loaded with an `ETag`, keyed by session id and identifier. Stale
+/// entries are harmless: the server answers `304` only while the `ETag`
matches.
+type TableCache = moka::sync::Cache<(String, TableIdent), (String, Table)>;
Review Comment:
This "stale entries are harmless" claim leans harder than it looks. We
invalidate nowhere — not on drop, rename, register, commit, or a 404 load — so
a dropped-and-recreated table at the same ident is only caught if the server
never reuses an ETag for recreated content. The spec treats the ETag as opaque,
so a server deriving it from the metadata hash alone would 304 against the
pre-drop entry and hand back a `Table` whose FileIO/credentials predate the
recreate. Correctness otherwise holds (the server still 404s a dropped table),
but the dropped table's `Table` stays pinned in memory until the TTL, and a 404
load leaves the stale `If-None-Match` in place instead of evicting.
Java invalidates by `SessionIdTableId` on drop/purge/rename/commit/register
and on `NoSuchTable`. I'd add `invalidate(&key)` to `delete_table`,
`rename_table` (src), `register_table`, `update_table`, and the 404 branch, and
soften this doc line. One fork worth deciding: invalidate per-session like
Java, or across sessions via `invalidate_entries_if` on the ident? Thoughts?
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -4023,6 +4088,136 @@ mod tests {
rename_table_mock.assert_async().await;
}
+ /// Mocks one `loadTable` of `ns1.test1` that expects `if_none_match` and
+ /// answers `status`, carrying `etag` if given.
+ async fn load_table_mock(
+ server: &mut ServerGuard,
+ if_none_match: mockito::Matcher,
+ status: usize,
+ etag: Option<&str>,
+ ) -> Mock {
+ let mut mock = server
+ .mock("GET", "/v1/namespaces/ns1/tables/test1")
+ .match_header("if-none-match", if_none_match)
+ .with_status(status);
+ if let Some(etag) = etag {
+ mock = mock.with_header("etag", etag);
+ }
+ if status == 200 {
+ mock = mock.with_body_from_file(format!(
+ "{}/testdata/load_table_response.json",
+ env!("CARGO_MANIFEST_DIR")
+ ));
+ }
+ mock.create_async().await
+ }
+
+ fn test1() -> TableIdent {
+ TableIdent::from_strs(["ns1", "test1"]).unwrap()
+ }
+
+ #[tokio::test]
+ async fn test_load_table_revalidates_cached_table_with_etag() {
+ let mut server = Server::new_async().await;
+ let _config = create_config_mock(&mut server).await;
+ let v1 = load_table_mock(&mut server, mockito::Matcher::Missing, 200,
Some("\"v1\"")).await;
+ let reuse_v1 = load_table_mock(&mut server, "\"v1\"".into(), 304,
None).await;
+ let catalog =
session_catalog(RestCatalogConfig::builder().uri(server.url()).build());
+ let context = SessionContext::empty();
+
+ let loaded = catalog.load_table(&context, &test1()).await.unwrap();
+ let reused = catalog.load_table(&context, &test1()).await.unwrap();
+ assert_eq!(loaded.metadata_location(), reused.metadata_location());
+ v1.assert_async().await;
+ reuse_v1.assert_async().await;
+
+ reuse_v1.remove_async().await;
+ let v2 = load_table_mock(&mut server, "\"v1\"".into(), 200,
Some("\"v2\"")).await;
+ let reuse_v2 = load_table_mock(&mut server, "\"v2\"".into(), 304,
None).await;
+ catalog.load_table(&context, &test1()).await.unwrap();
+ catalog.load_table(&context, &test1()).await.unwrap();
+ v2.assert_async().await;
+ reuse_v2.assert_async().await;
+ }
+
+ #[tokio::test]
+ async fn test_load_table_not_modified_without_cached_table() {
+ let mut server = Server::new_async().await;
+ let _config = create_config_mock(&mut server).await;
+ let _load = load_table_mock(&mut server, mockito::Matcher::Missing,
304, None).await;
+ let catalog =
session_catalog(RestCatalogConfig::builder().uri(server.url()).build());
+
+ let err = catalog
+ .load_table(&SessionContext::empty(), &test1())
+ .await
+ .unwrap_err();
+
+ assert_eq!(err.kind(), ErrorKind::Unexpected);
+ }
+
+ #[tokio::test]
+ async fn test_table_cache_is_per_session() {
Review Comment:
This passes even if caching were completely broken — two loads with no
`If-None-Match` is exactly what
`test_table_cache_disabled_with_zero_max_entries` asserts, so the two are
indistinguishable. It also hinges on `SessionContext::empty()` handing out
distinct ids; if that ever became deterministic the second load would send
`If-None-Match`, the `Missing` matcher would miss, and mockito would 501. I'd
make it discriminating: load in session A, load in session B and assert B sent
no header, then load A again and assert it sent `If-None-Match` and got a 304 —
with explicit session ids rather than relying on `empty()`.
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -1296,11 +1356,16 @@ impl SessionCatalog for RestSessionCatalog {
table_builder = table_builder.kms_client(kms_client);
}
- if let Some(metadata_location) = response.metadata_location {
+ let table = if let Some(metadata_location) =
response.metadata_location {
table_builder.metadata_location(metadata_location).build()
} else {
table_builder.build()
+ }?;
+
+ if let Some(etag) = etag {
Review Comment:
On a 200 that comes back without an ETag (a server that stopped sending it,
or a proxy that strips it) we neither replace nor drop the existing entry, so
every subsequent load keeps sending the stale `If-None-Match`. I'd add an
`else` arm that invalidates:
```rust
if let Some(etag) = etag {
client.table_cache.insert(cache_key, (etag, table.clone()));
} else {
client.table_cache.invalidate(&cache_key);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
