YuangGao commented on code in PR #3377:
URL: https://github.com/apache/iceberg-rust/pull/3377#discussion_r4244344046
##########
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:
good catch. checked Java: on a hit it rebuilds the FileIO, but from the
credentials captured at the original load (a 304 has no body). it stays fresh
because VendedCredentialsProvider refreshes them via
client.refresh-credentials-endpoint, and iceberg-rust has no refresh yet
(#2932), so rebuilding per hit wouldn't help here. went with not caching loads
that vend credentials (storage-credentials, or a secret-looking config key via
is_sensitive_prop). we can revisit once refresh lands.
##########
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:
fair point, the ETag is opaque. added invalidation on drop/purge, rename
(source), register, commit and the 404 branch, and softened the doc. went
per-session like Java: cross-session would need invalidate_entries_if, which
has to be enabled up front and scans entries on each call.
##########
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:
moved the ETag extraction into the 200 arm. kept the cache non-optional
though: moka already short-circuits get/insert at max_capacity 0, and with
invalidation on several paths an Option would add a branch to each.
##########
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:
updated, thanks.
##########
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:
updated, thanks.
##########
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:
updated, thanks.
##########
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:
updated, thanks.
##########
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:
updated, thanks.
##########
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:
updated, thanks.
--
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]
