gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-21 20:06:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(table-catalog): paginate Iceberg REST listings (#5466)

Browse Source 
* feat(table-catalog): paginate Iceberg REST listings

* test(table-catalog): remove redundant token clones

---------

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
This commit is contained in:
Henry Guo
2026-08-02 23:27:15 +08:00
committed by GitHub
parent 1fdcbd9225
commit e5cfa8e375
5 changed files with 1546 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/admin/handlers/table_catalog.rs
+632 -46
View File
@@ -36,9 +36,11 @@ use rustfs_policy::{
action::{Action, AdminAction},
},
};
use rustfs_utils::crypto::{base64_decode_url_safe_no_pad, base64_encode_url_safe_no_pad, hex_sha256};
use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, header::CONTENT_TYPE, s3_error};
use serde::{Deserialize, Serialize, de::DeserializeOwned};
use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::num::NonZeroUsize;
use std::time::{Duration as StdDuration, Instant};
use time::{Duration, OffsetDateTime};
use uuid::Uuid;
@@ -60,6 +62,12 @@ const ICEBERG_ERROR_NO_SUCH_RESOURCE: &str = "NoSuchResourceException";
const ICEBERG_ERROR_NO_SUCH_TABLE: &str = "NoSuchTableException";
const ICEBERG_ERROR_NO_SUCH_VIEW: &str = "NoSuchViewException";
const ICEBERG_ERROR_REST: &str = "RESTException";
const REST_PAGE_TOKEN_VERSION: u8 = 1;
const REST_PAGE_TOKEN_MAX_LENGTH: usize = 16 * 1024;
const REST_DEFAULT_PAGE_SIZE: usize = 1000;
const REST_MAX_PAGE_SIZE: usize = 1000;
const REST_PAGE_TOKEN_QUERY_PARAMETER: &str = "pageToken";
const REST_PAGE_SIZE_QUERY_PARAMETER: &str = "pageSize";
const CATALOG_ENDPOINT_PREFIX_CONFIG_KEY: &str = "rustfs.catalog-endpoint-prefix";
const CATALOG_COMPAT_ENDPOINT_PREFIX_CONFIG_KEY: &str = "rustfs.catalog-compat-endpoint-prefix";
const CATALOG_BACKING_CONFIG_KEY: &str = "rustfs.catalog-backing";
@@ -603,6 +611,8 @@ struct RestNamespaceResponse {
#[derive(Debug, Serialize)]
struct RestListNamespacesResponse {
namespaces: Vec<Vec<String>>,
#[serde(rename = "next-page-token")]
next_page_token: Option<String>,
}
#[derive(Debug, Serialize)]
@@ -614,11 +624,56 @@ struct RestTableIdentifier {
#[derive(Debug, Serialize)]
struct RestListTablesResponse {
identifiers: Vec<RestTableIdentifier>,
#[serde(rename = "next-page-token")]
next_page_token: Option<String>,
}
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct RestPageToken {
version: u8,
context: String,
cursor: String,
}
#[derive(Debug)]
enum RestPagination {
Unpaginated,
Paginated {
cursor: Option<String>,
limit: NonZeroUsize,
context: String,
},
}
impl RestPagination {
fn page_request(&self) -> Option<(Option<&str>, NonZeroUsize)> {
match self {
Self::Unpaginated => None,
Self::Paginated { cursor, limit, .. } => Some((cursor.as_deref(), *limit)),
}
}
fn next_page_token(&self, cursor: Option<String>) -> S3Result<Option<String>> {
match (self, cursor) {
(Self::Unpaginated, _) | (_, None) => Ok(None),
(Self::Paginated { context, .. }, Some(cursor)) => encode_rest_page_token(&cursor, context).map(Some),
}
}
}
#[derive(Clone, Copy)]
struct RestPageContext<'a> {
resource: &'static str,
warehouse: &'a str,
namespace: Option<&'a str>,
}
#[derive(Debug, Serialize)]
struct RestListViewsResponse {
identifiers: Vec<RestTableIdentifier>,
#[serde(rename = "next-page-token")]
next_page_token: Option<String>,
}
#[derive(Debug, Serialize)]
@@ -1327,6 +1382,142 @@ fn warehouse_from_config_query(uri: &http::Uri) -> S3Result<Option<String>> {
Ok(warehouse)
}
fn rest_pagination_from_query(uri: &http::Uri, context: RestPageContext<'_>) -> S3Result<RestPagination> {
let mut page_token = None;
let mut page_token_seen = false;
let mut page_size = None;
if let Some(query) = uri.query() {
for (key, value) in url::form_urlencoded::parse(query.as_bytes()) {
match key.as_ref() {
REST_PAGE_TOKEN_QUERY_PARAMETER => {
if page_token_seen {
return Err(iceberg_rest_error(
ICEBERG_ERROR_BAD_REQUEST,
StatusCode::BAD_REQUEST,
"pageToken query parameter must not be repeated",
));
}
page_token_seen = true;
page_token = Some(value.into_owned());
}
REST_PAGE_SIZE_QUERY_PARAMETER => {
if page_size.is_some() {
return Err(iceberg_rest_error(
ICEBERG_ERROR_BAD_REQUEST,
StatusCode::BAD_REQUEST,
"pageSize query parameter must not be repeated",
));
}
let value = value.parse::<usize>().map_err(|_| {
iceberg_rest_error(
ICEBERG_ERROR_BAD_REQUEST,
StatusCode::BAD_REQUEST,
"pageSize query parameter must be a positive integer",
)
})?;
if value == 0 {
return Err(iceberg_rest_error(
ICEBERG_ERROR_BAD_REQUEST,
StatusCode::BAD_REQUEST,
"pageSize query parameter must be greater than zero",
));
}
page_size = Some(value.min(REST_MAX_PAGE_SIZE));
}
_ => {}
}
}
}
if !page_token_seen && page_size.is_none() {
return Ok(RestPagination::Unpaginated);
}
let context = rest_page_context_fingerprint(context);
let cursor = match page_token.as_deref() {
None | Some("") => None,
Some(encoded) => Some(decode_rest_page_token(encoded, &context)?),
};
let limit = NonZeroUsize::new(page_size.unwrap_or(REST_DEFAULT_PAGE_SIZE)).ok_or_else(|| {
iceberg_rest_error(
ICEBERG_ERROR_REST,
StatusCode::INTERNAL_SERVER_ERROR,
"REST page size must be greater than zero",
)
})?;
Ok(RestPagination::Paginated { cursor, limit, context })
}
fn rest_page_context_fingerprint(context: RestPageContext<'_>) -> String {
let mut data =
Vec::with_capacity(context.resource.len() + context.warehouse.len() + context.namespace.map_or(0, str::len) + 2);
data.extend_from_slice(context.resource.as_bytes());
data.push(0);
data.extend_from_slice(context.warehouse.as_bytes());
data.push(0);
if let Some(namespace) = context.namespace {
data.extend_from_slice(namespace.as_bytes());
}
hex_sha256(&data, str::to_string)
}
fn decode_rest_page_token(encoded: &str, expected_context: &str) -> S3Result<String> {
if encoded.len() > REST_PAGE_TOKEN_MAX_LENGTH {
return Err(iceberg_rest_error(
ICEBERG_ERROR_BAD_REQUEST,
StatusCode::BAD_REQUEST,
"pageToken query parameter is too large",
));
}
let data = base64_decode_url_safe_no_pad(encoded.as_bytes()).map_err(|_| {
iceberg_rest_error(
ICEBERG_ERROR_BAD_REQUEST,
StatusCode::BAD_REQUEST,
"pageToken query parameter is malformed",
)
})?;
let token = serde_json::from_slice::<RestPageToken>(&data).map_err(|_| {
iceberg_rest_error(
ICEBERG_ERROR_BAD_REQUEST,
StatusCode::BAD_REQUEST,
"pageToken query parameter is malformed",
)
})?;
if token.version != REST_PAGE_TOKEN_VERSION || token.context != expected_context || token.cursor.is_empty() {
return Err(iceberg_rest_error(
ICEBERG_ERROR_BAD_REQUEST,
StatusCode::BAD_REQUEST,
"pageToken query parameter does not match this list operation",
));
}
Ok(token.cursor)
}
fn encode_rest_page_token(cursor: &str, context: &str) -> S3Result<String> {
let token = RestPageToken {
version: REST_PAGE_TOKEN_VERSION,
context: context.to_string(),
cursor: cursor.to_string(),
};
let data = serde_json::to_vec(&token).map_err(|err| {
iceberg_rest_error(
ICEBERG_ERROR_REST,
StatusCode::INTERNAL_SERVER_ERROR,
format!("failed to serialize REST page token: {err}"),
)
})?;
let encoded = base64_encode_url_safe_no_pad(&data);
if encoded.len() > REST_PAGE_TOKEN_MAX_LENGTH {
return Err(iceberg_rest_error(
ICEBERG_ERROR_REST,
StatusCode::INTERNAL_SERVER_ERROR,
"REST page token exceeds the supported size",
));
}
Ok(encoded)
}
fn namespace_from_params(params: &Params<'_, '_>) -> S3Result<crate::table_catalog::Namespace> {
let namespace = params.get("namespace").unwrap_or("");
crate::table_catalog::Namespace::parse(namespace).map_err(|err| s3_error!(InvalidRequest, "invalid namespace: {}", err))
@@ -1514,6 +1705,7 @@ fn namespace_response_from_entry(entry: crate::table_catalog::NamespaceEntry) ->
fn list_namespaces_response_from_entries(
entries: Vec<crate::table_catalog::NamespaceEntry>,
next_page_token: Option<String>,
) -> S3Result<RestListNamespacesResponse> {
let namespaces = entries
.into_iter()
@@ -1523,10 +1715,16 @@ fn list_namespaces_response_from_entries(
Ok(namespace_segments(&namespace))
})
.collect::<S3Result<Vec<_>>>()?;
Ok(RestListNamespacesResponse { namespaces })
Ok(RestListNamespacesResponse {
namespaces,
next_page_token,
})
}
fn list_tables_response_from_entries(entries: Vec<crate::table_catalog::TableEntry>) -> S3Result<RestListTablesResponse> {
fn list_tables_response_from_entries(
entries: Vec<crate::table_catalog::TableEntry>,
next_page_token: Option<String>,
) -> S3Result<RestListTablesResponse> {
let identifiers = entries
.into_iter()
.map(|entry| {
@@ -1538,10 +1736,16 @@ fn list_tables_response_from_entries(entries: Vec<crate::table_catalog::TableEnt
})
})
.collect::<S3Result<Vec<_>>>()?;
Ok(RestListTablesResponse { identifiers })
Ok(RestListTablesResponse {
identifiers,
next_page_token,
})
}
fn list_views_response_from_entries(entries: Vec<crate::table_catalog::ViewEntry>) -> S3Result<RestListViewsResponse> {
fn list_views_response_from_entries(
entries: Vec<crate::table_catalog::ViewEntry>,
next_page_token: Option<String>,
) -> S3Result<RestListViewsResponse> {
let identifiers = entries
.into_iter()
.map(|entry| {
@@ -1553,7 +1757,10 @@ fn list_views_response_from_entries(entries: Vec<crate::table_catalog::ViewEntry
})
})
.collect::<S3Result<Vec<_>>>()?;
Ok(RestListViewsResponse { identifiers })
Ok(RestListViewsResponse {
identifiers,
next_page_token,
})
}
fn table_credential_vending_enabled() -> bool {
@@ -3605,12 +3812,28 @@ where
namespace_response_from_entry(entry)
}
async fn list_namespaces_response<S>(store: &S, bucket: &str) -> S3Result<RestListNamespacesResponse>
async fn list_namespaces_response<S>(store: &S, bucket: &str, uri: &http::Uri) -> S3Result<RestListNamespacesResponse>
where
S: crate::table_catalog::TableCatalogStore + ?Sized,
{
let entries = store.list_namespaces(bucket).await.map_err(catalog_store_error)?;
list_namespaces_response_from_entries(entries)
let context = RestPageContext {
resource: TABLE_CATALOG_NAMESPACE_RESOURCE_ROOT,
warehouse: bucket,
namespace: None,
};
let pagination = rest_pagination_from_query(uri, context)?;
let page = match pagination.page_request() {
Some((cursor, limit)) => store
.list_namespaces_page(bucket, cursor, limit)
.await
.map_err(catalog_store_error)?,
None => crate::table_catalog::TableCatalogListPage {
entries: store.list_namespaces(bucket).await.map_err(catalog_store_error)?,
next_cursor: None,
},
};
let next_page_token = pagination.next_page_token(page.next_cursor)?;
list_namespaces_response_from_entries(page.entries, next_page_token)
}
async fn get_namespace_response<S>(
@@ -3766,15 +3989,30 @@ async fn list_tables_response<S>(
store: &S,
bucket: &str,
namespace: &crate::table_catalog::Namespace,
uri: &http::Uri,
) -> S3Result<RestListTablesResponse>
where
S: crate::table_catalog::TableCatalogStore + ?Sized,
{
let entries = store
.list_tables(bucket, &namespace.public_name())
.await
.map_err(catalog_store_error)?;
list_tables_response_from_entries(entries)
let namespace = namespace.public_name();
let context = RestPageContext {
resource: TABLE_CATALOG_TABLE_RESOURCE_ROOT,
warehouse: bucket,
namespace: Some(&namespace),
};
let pagination = rest_pagination_from_query(uri, context)?;
let page = match pagination.page_request() {
Some((cursor, limit)) => store
.list_tables_page(bucket, &namespace, cursor, limit)
.await
.map_err(catalog_store_error)?,
None => crate::table_catalog::TableCatalogListPage {
entries: store.list_tables(bucket, &namespace).await.map_err(catalog_store_error)?,
next_cursor: None,
},
};
let next_page_token = pagination.next_page_token(page.next_cursor)?;
list_tables_response_from_entries(page.entries, next_page_token)
}
async fn load_table_response<S>(
@@ -3802,15 +4040,30 @@ async fn list_views_response<S>(
store: &S,
bucket: &str,
namespace: &crate::table_catalog::Namespace,
uri: &http::Uri,
) -> S3Result<RestListViewsResponse>
where
S: crate::table_catalog::TableCatalogStore + ?Sized,
{
let entries = store
.list_views(bucket, &namespace.public_name())
.await
.map_err(catalog_store_error)?;
list_views_response_from_entries(entries)
let namespace = namespace.public_name();
let context = RestPageContext {
resource: TABLE_CATALOG_VIEW_RESOURCE_ROOT,
warehouse: bucket,
namespace: Some(&namespace),
};
let pagination = rest_pagination_from_query(uri, context)?;
let page = match pagination.page_request() {
Some((cursor, limit)) => store
.list_views_page(bucket, &namespace, cursor, limit)
.await
.map_err(catalog_store_error)?,
None => crate::table_catalog::TableCatalogListPage {
entries: store.list_views(bucket, &namespace).await.map_err(catalog_store_error)?,
next_cursor: None,
},
};
let next_page_token = pagination.next_page_token(page.next_cursor)?;
list_views_response_from_entries(page.entries, next_page_token)
}
async fn load_view_response<S>(
@@ -5077,7 +5330,7 @@ impl Operation for RestListNamespacesHandler {
authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableNamespaceAction).await?;
ensure_table_bucket_enabled(&warehouse).await?;
let store = table_catalog_store()?;
let response = list_namespaces_response(&store, &warehouse).await?;
let response = list_namespaces_response(&store, &warehouse, &req.uri).await?;
build_json_response(StatusCode::OK, &response)
}
}
@@ -5156,7 +5409,7 @@ impl Operation for RestListTablesHandler {
authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableAction).await?;
ensure_table_bucket_enabled(&warehouse).await?;
let store = table_catalog_store()?;
let response = list_tables_response(&store, &warehouse, &namespace).await?;
let response = list_tables_response(&store, &warehouse, &namespace, &req.uri).await?;
build_json_response(StatusCode::OK, &response)
}
}
@@ -5210,7 +5463,7 @@ impl Operation for RestListViewsHandler {
authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableMetadataAction).await?;
ensure_table_bucket_enabled(&warehouse).await?;
let store = table_catalog_store()?;
let response = list_views_response(&store, &warehouse, &namespace).await?;
let response = list_views_response(&store, &warehouse, &namespace, &req.uri).await?;
build_json_response(StatusCode::OK, &response)
}
}
@@ -6231,6 +6484,31 @@ mod tests {
}
}
#[test]
fn table_catalog_list_handlers_parse_standard_pagination() {
let src = include_str!("table_catalog.rs");
for (handler, helper_call) in [
(
"RestListNamespacesHandler",
"list_namespaces_response(&store, &warehouse, &req.uri).await?",
),
(
"RestListTablesHandler",
"list_tables_response(&store, &warehouse, &namespace, &req.uri).await?",
),
(
"RestListViewsHandler",
"list_views_response(&store, &warehouse, &namespace, &req.uri).await?",
),
] {
let block = operation_block(src, handler);
assert!(
block.contains(helper_call),
"{handler} should pass the request URI to its paginated list helper"
);
}
}
#[test]
fn table_catalog_handlers_require_enabled_table_bucket_marker_before_catalog_state() {
let src = include_str!("table_catalog.rs");
@@ -6709,26 +6987,29 @@ mod tests {
#[test]
fn list_tables_response_uses_rest_identifier_shape() {
let namespace = crate::table_catalog::Namespace::parse("analytics.daily_events").expect("namespace should parse");
let response = list_tables_response_from_entries(vec![crate::table_catalog::TableEntry {
version: crate::table_catalog::TABLE_CATALOG_ENTRY_VERSION,
table_bucket: "warehouse".to_string(),
namespace: namespace.public_name(),
table: "events".to_string(),
table_id: "table-id".to_string(),
table_uuid: "table-uuid".to_string(),
format: "ICEBERG".to_string(),
format_version: 2,
warehouse_location: "s3://warehouse/tables/table-id".to_string(),
metadata_location:
".rustfs-table/warehouses/default/namespaces/analytics/daily_events/tables/events/metadata/00001.metadata.json"
.to_string(),
version_token: "token-v1".to_string(),
generation: 1,
state: crate::table_catalog::TableCatalogEntryState::Active,
properties: BTreeMap::new(),
created_at: None,
updated_at: None,
}])
let response = list_tables_response_from_entries(
vec![crate::table_catalog::TableEntry {
version: crate::table_catalog::TABLE_CATALOG_ENTRY_VERSION,
table_bucket: "warehouse".to_string(),
namespace: namespace.public_name(),
table: "events".to_string(),
table_id: "table-id".to_string(),
table_uuid: "table-uuid".to_string(),
format: "ICEBERG".to_string(),
format_version: 2,
warehouse_location: "s3://warehouse/tables/table-id".to_string(),
metadata_location:
".rustfs-table/warehouses/default/namespaces/analytics/daily_events/tables/events/metadata/00001.metadata.json"
.to_string(),
version_token: "token-v1".to_string(),
generation: 1,
state: crate::table_catalog::TableCatalogEntryState::Active,
properties: BTreeMap::new(),
created_at: None,
updated_at: None,
}],
None,
)
.expect("table list response should build");
assert_eq!(
@@ -6736,6 +7017,308 @@ mod tests {
vec!["analytics".to_string(), "daily_events".to_string()]
);
assert_eq!(response.identifiers[0].name, "events");
assert!(response.next_page_token.is_none());
}
#[test]
fn rest_pagination_round_trips_context_bound_tokens() {
let context = RestPageContext {
resource: TABLE_CATALOG_TABLE_RESOURCE_ROOT,
warehouse: "warehouse",
namespace: Some("analytics"),
};
let first_request = "/?pageSize=2".parse::<http::Uri>().expect("first page URI should parse");
let first_pagination = rest_pagination_from_query(&first_request, context).expect("pageSize should start pagination");
let (cursor, limit) = first_pagination.page_request().expect("pageSize should enable pagination");
assert_eq!(cursor, None);
assert_eq!(limit.get(), 2);
let next_page_token = first_pagination
.next_page_token(Some("strong:beta".to_string()))
.expect("page token should encode")
.expect("page token should be present");
let second_request = format!("/?pageSize=2&pageToken={next_page_token}")
.parse::<http::Uri>()
.expect("second page URI should parse");
let second_pagination = rest_pagination_from_query(&second_request, context).expect("continuation token should decode");
let (cursor, limit) = second_pagination
.page_request()
.expect("continuation should remain paginated");
assert_eq!(cursor, Some("strong:beta"));
assert_eq!(limit.get(), 2);
assert!(
second_pagination
.next_page_token(None)
.expect("terminal token should build")
.is_none()
);
let default_size_request = format!("/?pageToken={next_page_token}")
.parse::<http::Uri>()
.expect("continuation URI should parse without pageSize");
let default_size_pagination =
rest_pagination_from_query(&default_size_request, context).expect("continuation should use the default page size");
let (cursor, limit) = default_size_pagination
.page_request()
.expect("continuation should remain paginated");
assert_eq!(cursor, Some("strong:beta"));
assert_eq!(limit.get(), REST_DEFAULT_PAGE_SIZE);
for other_context in [
RestPageContext {
resource: TABLE_CATALOG_VIEW_RESOURCE_ROOT,
warehouse: "warehouse",
namespace: Some("analytics"),
},
RestPageContext {
resource: TABLE_CATALOG_TABLE_RESOURCE_ROOT,
warehouse: "other-warehouse",
namespace: Some("analytics"),
},
RestPageContext {
resource: TABLE_CATALOG_TABLE_RESOURCE_ROOT,
warehouse: "warehouse",
namespace: Some("other-namespace"),
},
] {
let expected_context = rest_page_context_fingerprint(other_context);
let error = decode_rest_page_token(&next_page_token, &expected_context).expect_err("cross-context token should fail");
assert_eq!(error.code(), &S3ErrorCode::Custom(ICEBERG_ERROR_BAD_REQUEST.into()));
assert_eq!(error.status_code(), Some(StatusCode::BAD_REQUEST));
}
}
#[test]
fn rest_pagination_rejects_invalid_query_parameters() {
let context = RestPageContext {
resource: TABLE_CATALOG_NAMESPACE_RESOURCE_ROOT,
warehouse: "warehouse",
namespace: None,
};
for uri in [
"/?pageSize=0",
"/?pageSize=one",
"/?pageSize=1&pageSize=2",
"/?pageToken=first&pageToken=second",
] {
let uri = uri.parse::<http::Uri>().expect("invalid pagination URI should still parse");
let error = rest_pagination_from_query(&uri, context).expect_err("invalid pagination query should fail");
assert_eq!(error.code(), &S3ErrorCode::Custom(ICEBERG_ERROR_BAD_REQUEST.into()), "{uri}");
assert_eq!(error.status_code(), Some(StatusCode::BAD_REQUEST), "{uri}");
}
let oversized_token = "a".repeat(REST_PAGE_TOKEN_MAX_LENGTH + 1);
let oversized_uri = format!("/?pageToken={oversized_token}")
.parse::<http::Uri>()
.expect("oversized token URI should parse");
let error = rest_pagination_from_query(&oversized_uri, context).expect_err("oversized token should fail");
assert_eq!(error.code(), &S3ErrorCode::Custom(ICEBERG_ERROR_BAD_REQUEST.into()));
assert_eq!(error.status_code(), Some(StatusCode::BAD_REQUEST));
let empty_token = "/?pageToken=".parse::<http::Uri>().expect("empty token URI should parse");
let pagination = rest_pagination_from_query(&empty_token, context).expect("empty token should start the first page");
let (cursor, limit) = pagination.page_request().expect("empty token should enable pagination");
assert_eq!(cursor, None);
assert_eq!(limit.get(), REST_DEFAULT_PAGE_SIZE);
let capped_size = "/?pageSize=5000"
.parse::<http::Uri>()
.expect("large page size URI should parse");
let pagination = rest_pagination_from_query(&capped_size, context).expect("large page size should be capped");
let (cursor, limit) = pagination.page_request().expect("pageSize alone should enable pagination");
assert_eq!(cursor, None);
assert_eq!(limit.get(), REST_MAX_PAGE_SIZE);
let unpaginated = rest_pagination_from_query(&"/".parse().expect("URI should parse"), context)
.expect("request without pagination should parse");
assert!(unpaginated.page_request().is_none());
}
#[test]
fn rest_pagination_rejects_malformed_token_payloads() {
let context = RestPageContext {
resource: TABLE_CATALOG_NAMESPACE_RESOURCE_ROOT,
warehouse: "warehouse",
namespace: None,
};
let context_fingerprint = rest_page_context_fingerprint(context);
let encoded_json = |value: serde_json::Value| {
base64_encode_url_safe_no_pad(&serde_json::to_vec(&value).expect("test token should encode"))
};
let malformed_tokens = [
"*".to_string(),
base64_encode_url_safe_no_pad(b"not-json"),
encoded_json(serde_json::json!({
"version": REST_PAGE_TOKEN_VERSION,
"context": context_fingerprint,
"cursor": "strong:alpha",
"unknown": true
})),
encoded_json(serde_json::json!({
"version": REST_PAGE_TOKEN_VERSION + 1,
"context": context_fingerprint,
"cursor": "strong:alpha"
})),
encoded_json(serde_json::json!({
"version": REST_PAGE_TOKEN_VERSION,
"context": context_fingerprint,
"cursor": ""
})),
];
for token in malformed_tokens {
let uri = format!(
"/?pageToken={}",
url::form_urlencoded::byte_serialize(token.as_bytes()).collect::<String>()
)
.parse::<http::Uri>()
.expect("malformed token URI should parse");
let error = rest_pagination_from_query(&uri, context).expect_err("malformed token should fail");
assert_eq!(error.code(), &S3ErrorCode::Custom(ICEBERG_ERROR_BAD_REQUEST.into()));
assert_eq!(error.status_code(), Some(StatusCode::BAD_REQUEST));
}
}
#[tokio::test]
async fn rest_list_pagination_covers_namespaces_tables_and_views() {
let store = TestTableCatalogStore::default();
let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
for name in ["beta", "alpha"] {
store.namespaces.lock().await.push(crate::table_catalog::NamespaceEntry {
version: crate::table_catalog::TABLE_CATALOG_ENTRY_VERSION,
table_bucket: "warehouse".to_string(),
namespace: name.to_string(),
namespace_id: name.to_string(),
state: crate::table_catalog::TableCatalogEntryState::Active,
properties: BTreeMap::new(),
created_at: None,
updated_at: None,
});
store.tables.lock().await.push(crate::table_catalog::TableEntry {
version: crate::table_catalog::TABLE_CATALOG_ENTRY_VERSION,
table_bucket: "warehouse".to_string(),
namespace: namespace.public_name(),
table: name.to_string(),
table_id: format!("table-{name}"),
table_uuid: format!("table-uuid-{name}"),
format: "ICEBERG".to_string(),
format_version: 2,
warehouse_location: format!("s3://warehouse/tables/table-{name}"),
metadata_location: format!("s3://warehouse/tables/table-{name}/metadata/00001.metadata.json"),
version_token: "token-v1".to_string(),
generation: 1,
state: crate::table_catalog::TableCatalogEntryState::Active,
properties: BTreeMap::new(),
created_at: None,
updated_at: None,
});
store.views.lock().await.push(crate::table_catalog::ViewEntry {
version: crate::table_catalog::TABLE_CATALOG_ENTRY_VERSION,
table_bucket: "warehouse".to_string(),
namespace: namespace.public_name(),
view: name.to_string(),
view_id: format!("view-{name}"),
view_uuid: format!("view-uuid-{name}"),
format: "ICEBERG_VIEW".to_string(),
format_version: 1,
warehouse_location: format!("s3://warehouse/views/view-{name}"),
metadata_location: format!("s3://warehouse/views/view-{name}/metadata/00001.view.json"),
version_token: "token-v1".to_string(),
generation: 1,
state: crate::table_catalog::TableCatalogEntryState::Active,
properties: BTreeMap::new(),
created_at: None,
updated_at: None,
});
}
let first_uri = "/?pageSize=1".parse::<http::Uri>().expect("first page URI should parse");
let namespaces = list_namespaces_response(&store, "warehouse", &first_uri)
.await
.expect("namespace first page should load");
assert_eq!(namespaces.namespaces, vec![vec!["alpha".to_string()]]);
let namespace_token = namespaces.next_page_token.expect("namespace continuation should exist");
let namespace_uri = format!("/?pageSize=1&pageToken={namespace_token}")
.parse::<http::Uri>()
.expect("namespace continuation URI should parse");
let namespaces = list_namespaces_response(&store, "warehouse", &namespace_uri)
.await
.expect("namespace second page should load");
assert_eq!(namespaces.namespaces, vec![vec!["beta".to_string()]]);
assert!(namespaces.next_page_token.is_none());
let tables = list_tables_response(&store, "warehouse", &namespace, &first_uri)
.await
.expect("table first page should load");
assert_eq!(tables.identifiers.len(), 1);
assert_eq!(tables.identifiers[0].name, "alpha");
let table_token = tables.next_page_token.expect("table continuation should exist");
let table_uri = format!("/?pageSize=1&pageToken={table_token}")
.parse::<http::Uri>()
.expect("table continuation URI should parse");
let tables = list_tables_response(&store, "warehouse", &namespace, &table_uri)
.await
.expect("table second page should load");
assert_eq!(tables.identifiers.len(), 1);
assert_eq!(tables.identifiers[0].name, "beta");
assert!(tables.next_page_token.is_none());
let views = list_views_response(&store, "warehouse", &namespace, &first_uri)
.await
.expect("view first page should load");
assert_eq!(views.identifiers.len(), 1);
assert_eq!(views.identifiers[0].name, "alpha");
let view_token = views.next_page_token.expect("view continuation should exist");
let view_uri = format!("/?pageSize=1&pageToken={view_token}")
.parse::<http::Uri>()
.expect("view continuation URI should parse");
let views = list_views_response(&store, "warehouse", &namespace, &view_uri)
.await
.expect("view second page should load");
assert_eq!(views.identifiers.len(), 1);
assert_eq!(views.identifiers[0].name, "beta");
assert!(views.next_page_token.is_none());
for uri in ["/", "/?pageSize=2"] {
let uri = uri.parse::<http::Uri>().expect("list URI should parse");
let namespaces = list_namespaces_response(&store, "warehouse", &uri)
.await
.expect("namespace exact page should load");
let tables = list_tables_response(&store, "warehouse", &namespace, &uri)
.await
.expect("table exact page should load");
let views = list_views_response(&store, "warehouse", &namespace, &uri)
.await
.expect("view exact page should load");
assert_eq!(namespaces.namespaces.len(), 2);
assert!(namespaces.next_page_token.is_none());
assert_eq!(tables.identifiers.len(), 2);
assert!(tables.next_page_token.is_none());
assert_eq!(views.identifiers.len(), 2);
assert!(views.next_page_token.is_none());
}
}
#[test]
fn list_responses_expose_null_next_page_token_at_end() {
for value in [
serde_json::to_value(RestListNamespacesResponse {
namespaces: vec![vec!["analytics".to_string()]],
next_page_token: None,
})
.expect("namespace response should serialize"),
serde_json::to_value(RestListTablesResponse {
identifiers: Vec::new(),
next_page_token: None,
})
.expect("table response should serialize"),
serde_json::to_value(RestListViewsResponse {
identifiers: Vec::new(),
next_page_token: None,
})
.expect("view response should serialize"),
] {
assert!(value.get("next-page-token").is_some_and(serde_json::Value::is_null));
}
}
#[tokio::test]
@@ -9835,7 +10418,8 @@ mod tests {
.expect("view metadata object lookup should succeed")
);
let listed = list_views_response(&store, "warehouse", &namespace)
let unpaginated_uri = "/".parse::<http::Uri>().expect("list URI should parse");
let listed = list_views_response(&store, "warehouse", &namespace, &unpaginated_uri)
.await
.expect("views should list");
assert_eq!(listed.identifiers.len(), 1);
@@ -9890,7 +10474,7 @@ mod tests {
drop_view_in_store(&store, "warehouse", &namespace, "recent_events")
.await
.expect("view should drop");
let listed = list_views_response(&store, "warehouse", &namespace)
let listed = list_views_response(&store, "warehouse", &namespace, &unpaginated_uri)
.await
.expect("views should list after drop");
assert!(listed.identifiers.is_empty());
@@ -11379,7 +11963,8 @@ mod tests {
assert_eq!(create.namespace, vec!["analytics".to_string()]);
assert_eq!(create.properties.get("owner").map(String::as_str), Some("lakehouse"));
let list = list_namespaces_response(&store, "warehouse")
let unpaginated_uri = "/".parse::<http::Uri>().expect("list URI should parse");
let list = list_namespaces_response(&store, "warehouse", &unpaginated_uri)
.await
.expect("namespace list should load");
assert_eq!(list.namespaces, vec![vec!["analytics".to_string()]]);
@@ -11387,7 +11972,7 @@ mod tests {
drop_namespace_in_store(&store, "warehouse", "analytics")
.await
.expect("namespace should drop");
let list = list_namespaces_response(&store, "warehouse")
let list = list_namespaces_response(&store, "warehouse", &unpaginated_uri)
.await
.expect("namespace list should load after drop");
assert!(list.namespaces.is_empty());
@@ -11445,7 +12030,8 @@ mod tests {
assert_eq!(register.metadata_location, client_metadata_location);
assert_eq!(register.metadata["format-version"], 2);
let list = list_tables_response(&store, "warehouse", &namespace)
let unpaginated_uri = "/".parse::<http::Uri>().expect("list URI should parse");
let list = list_tables_response(&store, "warehouse", &namespace, &unpaginated_uri)
.await
.expect("table list should load");
assert_eq!(list.identifiers[0].name, "events");
rustfs/src/table_catalog.rs
+637 -2
View File
@@ -23,6 +23,8 @@
use std::{
collections::{BTreeMap, BTreeSet},
fmt,
num::NonZeroUsize,
ops::Bound,
sync::Arc,
time::{Duration as StdDuration, Instant},
};
@@ -107,7 +109,9 @@ const MAINTENANCE_LATEST_JOB_FILE: &str = "latest.json";
const MAINTENANCE_CURRENT_JOB_FILE: &str = "current.json";
const MAINTENANCE_JOB_ALIAS_LATEST: &str = "latest";
const MAINTENANCE_JOB_ALIAS_CURRENT: &str = "current";
const TABLE_CATALOG_LIST_MAX_KEYS: i32 = 1000;
const TABLE_CATALOG_LIST_MAX_KEYS: usize = 1000;
const OBJECT_CATALOG_LIST_CURSOR_PREFIX: &str = "object:";
const STRONG_CATALOG_LIST_CURSOR_PREFIX: &str = "strong:";
const TABLE_METADATA_CLEANUP_SAFETY_WINDOW_SECONDS: i64 = 15 * 60;
const TABLE_MAINTENANCE_RETRY_BACKOFF_MAX_SECONDS: u64 = 24 * 60 * 60;
const TABLE_MAINTENANCE_WORKER_LEASE_TIMEOUT_DEFAULT_SECONDS: u64 = 15 * 60;
@@ -197,6 +201,7 @@ impl<T> TableCatalogStorage for T where
pub enum CatalogIdentifierError {
Empty,
TooLong { max: usize },
NamespaceTooLong { max: usize },
InvalidCharacter,
InvalidBoundary,
Ambiguous,
@@ -207,6 +212,7 @@ impl fmt::Display for CatalogIdentifierError {
match self {
Self::Empty => f.write_str("catalog identifier segment is empty"),
Self::TooLong { max } => write!(f, "catalog identifier segment exceeds {max} characters"),
Self::NamespaceTooLong { max } => write!(f, "catalog namespace exceeds {max} characters"),
Self::InvalidCharacter => f.write_str("catalog identifier segment contains invalid characters"),
Self::InvalidBoundary => {
f.write_str("catalog identifier segment must start and end with a lowercase letter or digit")
@@ -1768,6 +1774,58 @@ where
Ok(matched)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct TableCatalogListPage<T> {
pub entries: Vec<T>,
pub next_cursor: Option<String>,
}
fn finish_catalog_list_page<T, F>(
mut entries: Vec<T>,
limit: NonZeroUsize,
cursor_prefix: &str,
key: F,
) -> TableCatalogListPage<T>
where
F: Fn(&T) -> &str,
{
let next_cursor = if entries.len() > limit.get() {
entries.truncate(limit.get());
entries.last().map(|entry| format!("{cursor_prefix}{}", key(entry)))
} else {
None
};
TableCatalogListPage { entries, next_cursor }
}
fn catalog_list_page_from_entries<T, F>(
mut entries: Vec<T>,
cursor: Option<&str>,
limit: NonZeroUsize,
key: F,
) -> TableCatalogListPage<T>
where
F: Fn(&T) -> &str,
{
entries.sort_by(|left, right| key(left).cmp(key(right)));
let start = cursor.map_or(0, |cursor| entries.partition_point(|entry| key(entry) <= cursor));
let entries = entries.into_iter().skip(start).take(limit.get().saturating_add(1)).collect();
finish_catalog_list_page(entries, limit, "", key)
}
fn catalog_list_cursor<'a>(cursor: Option<&'a str>, prefix: &str) -> TableCatalogStoreResult<Option<&'a str>> {
cursor
.map(|cursor| {
cursor
.strip_prefix(prefix)
.filter(|cursor| !cursor.is_empty())
.ok_or_else(|| {
TableCatalogStoreError::Invalid("page cursor does not match the active table catalog backing".to_string())
})
})
.transpose()
}
#[async_trait::async_trait]
pub(crate) trait TableCatalogStore: Send + Sync {
async fn get_table_bucket(&self, table_bucket: &str) -> TableCatalogStoreResult<Option<TableBucketEntry>>;
@@ -1778,6 +1836,20 @@ pub(crate) trait TableCatalogStore: Send + Sync {
async fn list_namespaces(&self, table_bucket: &str) -> TableCatalogStoreResult<Vec<NamespaceEntry>>;
async fn list_namespaces_page(
&self,
table_bucket: &str,
cursor: Option<&str>,
limit: NonZeroUsize,
) -> TableCatalogStoreResult<TableCatalogListPage<NamespaceEntry>> {
Ok(catalog_list_page_from_entries(
self.list_namespaces(table_bucket).await?,
cursor,
limit,
|entry| &entry.namespace,
))
}
async fn get_namespace(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult<Option<NamespaceEntry>>;
async fn drop_namespace(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult<()>;
@@ -1788,6 +1860,21 @@ pub(crate) trait TableCatalogStore: Send + Sync {
async fn list_tables(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult<Vec<TableEntry>>;
async fn list_tables_page(
&self,
table_bucket: &str,
namespace: &str,
cursor: Option<&str>,
limit: NonZeroUsize,
) -> TableCatalogStoreResult<TableCatalogListPage<TableEntry>> {
Ok(catalog_list_page_from_entries(
self.list_tables(table_bucket, namespace).await?,
cursor,
limit,
|entry| &entry.table,
))
}
async fn load_table(&self, table_bucket: &str, namespace: &str, table: &str) -> TableCatalogStoreResult<Option<TableEntry>>;
async fn resolve_table_data_plane_resource(
@@ -1806,6 +1893,21 @@ pub(crate) trait TableCatalogStore: Send + Sync {
async fn list_views(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult<Vec<ViewEntry>>;
async fn list_views_page(
&self,
table_bucket: &str,
namespace: &str,
cursor: Option<&str>,
limit: NonZeroUsize,
) -> TableCatalogStoreResult<TableCatalogListPage<ViewEntry>> {
Ok(catalog_list_page_from_entries(
self.list_views(table_bucket, namespace).await?,
cursor,
limit,
|entry| &entry.view,
))
}
async fn load_view(&self, table_bucket: &str, namespace: &str, view: &str) -> TableCatalogStoreResult<Option<ViewEntry>>;
async fn replace_view(&self, request: ViewCommitRequest) -> TableCatalogStoreResult<ViewCommitResult>;
@@ -1840,6 +1942,12 @@ pub(crate) struct TableCatalogObjectMetadata {
pub mod_time: Option<OffsetDateTime>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct TableCatalogObjectListPage {
pub objects: Vec<String>,
pub is_truncated: bool,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum TableCatalogPutPrecondition {
Any,
@@ -1893,6 +2001,26 @@ pub(crate) trait TableCatalogObjectBackend: Clone + Send + Sync + 'static {
async fn list_objects(&self, bucket: &str, prefix: &str) -> TableCatalogStoreResult<Vec<String>>;
async fn list_objects_page(
&self,
bucket: &str,
prefix: &str,
start_after: Option<&str>,
limit: NonZeroUsize,
) -> TableCatalogStoreResult<TableCatalogObjectListPage> {
let mut objects = self.list_objects(bucket, prefix).await?;
objects.sort();
let start = start_after.map_or(0, |cursor| objects.partition_point(|object| object.as_str() <= cursor));
let mut objects = objects
.into_iter()
.skip(start)
.take(limit.get().saturating_add(1))
.collect::<Vec<_>>();
let is_truncated = objects.len() > limit.get();
objects.truncate(limit.get());
Ok(TableCatalogObjectListPage { objects, is_truncated })
}
async fn acquire_read_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Box<dyn Send>> {
self.acquire_write_lock(bucket, object).await
}
@@ -3116,6 +3244,35 @@ where
Ok(entries)
}
async fn list_namespaces_page(
&self,
table_bucket: &str,
cursor: Option<&str>,
limit: NonZeroUsize,
) -> TableCatalogStoreResult<TableCatalogListPage<NamespaceEntry>> {
self.hydrate_state().await?;
let cursor = catalog_list_cursor(cursor, STRONG_CATALOG_LIST_CURSOR_PREFIX)?;
let cursor = cursor
.map(parse_namespace_for_store)
.transpose()?
.map(|namespace| namespace.public_name());
let start = match cursor {
Some(cursor) => Bound::Excluded((table_bucket.to_string(), cursor)),
None => Bound::Included((table_bucket.to_string(), String::new())),
};
let state = self.state.lock().await;
let entries = state
.namespaces
.range((start, Bound::Unbounded))
.take_while(|((bucket, _), _)| bucket == table_bucket)
.take(limit.get().saturating_add(1))
.map(|(_, entry)| entry.clone())
.collect();
Ok(finish_catalog_list_page(entries, limit, STRONG_CATALOG_LIST_CURSOR_PREFIX, |entry| {
&entry.namespace
}))
}
async fn get_namespace(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult<Option<NamespaceEntry>> {
self.hydrate_state().await?;
let namespace = parse_namespace_for_store(namespace)?;
@@ -3211,6 +3368,37 @@ where
Ok(entries)
}
async fn list_tables_page(
&self,
table_bucket: &str,
namespace: &str,
cursor: Option<&str>,
limit: NonZeroUsize,
) -> TableCatalogStoreResult<TableCatalogListPage<TableEntry>> {
self.hydrate_state().await?;
let namespace = parse_namespace_for_store(namespace)?.public_name();
let cursor = catalog_list_cursor(cursor, STRONG_CATALOG_LIST_CURSOR_PREFIX)?;
let cursor = cursor
.map(parse_table_for_store)
.transpose()?
.map(|table| table.as_str().to_string());
let start = match cursor {
Some(cursor) => Bound::Excluded((table_bucket.to_string(), namespace.clone(), cursor)),
None => Bound::Included((table_bucket.to_string(), namespace.clone(), String::new())),
};
let state = self.state.lock().await;
let entries = state
.tables
.range((start, Bound::Unbounded))
.take_while(|((bucket, entry_namespace, _), _)| bucket == table_bucket && entry_namespace == &namespace)
.take(limit.get().saturating_add(1))
.map(|(_, entry)| entry.clone())
.collect();
Ok(finish_catalog_list_page(entries, limit, STRONG_CATALOG_LIST_CURSOR_PREFIX, |entry| {
&entry.table
}))
}
async fn load_table(&self, table_bucket: &str, namespace: &str, table: &str) -> TableCatalogStoreResult<Option<TableEntry>> {
self.hydrate_state().await?;
let namespace = parse_namespace_for_store(namespace)?;
@@ -3425,6 +3613,37 @@ where
Ok(entries)
}
async fn list_views_page(
&self,
table_bucket: &str,
namespace: &str,
cursor: Option<&str>,
limit: NonZeroUsize,
) -> TableCatalogStoreResult<TableCatalogListPage<ViewEntry>> {
self.hydrate_state().await?;
let namespace = parse_namespace_for_store(namespace)?.public_name();
let cursor = catalog_list_cursor(cursor, STRONG_CATALOG_LIST_CURSOR_PREFIX)?;
let cursor = cursor
.map(parse_table_for_store)
.transpose()?
.map(|view| view.as_str().to_string());
let start = match cursor {
Some(cursor) => Bound::Excluded((table_bucket.to_string(), namespace.clone(), cursor)),
None => Bound::Included((table_bucket.to_string(), namespace.clone(), String::new())),
};
let state = self.state.lock().await;
let entries = state
.views
.range((start, Bound::Unbounded))
.take_while(|((bucket, entry_namespace, _), _)| bucket == table_bucket && entry_namespace == &namespace)
.take(limit.get().saturating_add(1))
.map(|(_, entry)| entry.clone())
.collect();
Ok(finish_catalog_list_page(entries, limit, STRONG_CATALOG_LIST_CURSOR_PREFIX, |entry| {
&entry.view
}))
}
async fn load_view(&self, table_bucket: &str, namespace: &str, view: &str) -> TableCatalogStoreResult<Option<ViewEntry>> {
self.hydrate_state().await?;
let namespace = parse_namespace_for_store(namespace)?;
@@ -3558,6 +3777,71 @@ where
RUSTFS_META_BUCKET
}
async fn list_entry_page<T>(
&self,
prefix: &str,
entry_file: &str,
cursor: Option<&str>,
limit: NonZeroUsize,
) -> TableCatalogStoreResult<TableCatalogListPage<T>>
where
T: DeserializeOwned,
{
let cursor = catalog_list_cursor(cursor, OBJECT_CATALOG_LIST_CURSOR_PREFIX)?;
if cursor.is_some_and(|cursor| !cursor.starts_with(prefix)) {
return Err(TableCatalogStoreError::Invalid(
"page cursor does not match this table catalog list operation".to_string(),
));
}
let scan_limit = NonZeroUsize::new(TABLE_CATALOG_LIST_MAX_KEYS)
.ok_or_else(|| TableCatalogStoreError::Internal("catalog object scan limit must be positive".to_string()))?;
let mut entries = Vec::with_capacity(limit.get());
let mut last_entry_path = None;
let page = self
.backend
.list_objects_page(self.catalog_bucket(), prefix, cursor, scan_limit)
.await?;
let last_scanned_path = page.objects.last().cloned();
if page.is_truncated && last_scanned_path.is_none() {
return Err(TableCatalogStoreError::Internal("catalog object pagination made no progress".to_string()));
}
if cursor
.zip(last_scanned_path.as_deref())
.is_some_and(|(cursor, last)| last <= cursor)
{
return Err(TableCatalogStoreError::Internal("catalog object pagination did not advance".to_string()));
}
for object in page.objects {
if !object.ends_with(entry_file) {
continue;
}
if entries.len() == limit.get() {
let last_entry_path = last_entry_path.ok_or_else(|| {
TableCatalogStoreError::Internal("catalog page cursor is missing its last entry".to_string())
})?;
return Ok(TableCatalogListPage {
entries,
next_cursor: Some(format!("{OBJECT_CATALOG_LIST_CURSOR_PREFIX}{last_entry_path}")),
});
}
let Some((entry, _)) = self.read_entry::<T>(self.catalog_bucket(), &object).await? else {
continue;
};
last_entry_path = Some(object);
entries.push(entry);
}
let next_cursor = if page.is_truncated {
last_scanned_path.map(|path| format!("{OBJECT_CATALOG_LIST_CURSOR_PREFIX}{path}"))
} else {
None
};
Ok(TableCatalogListPage { entries, next_cursor })
}
async fn read_backing_migration_fence(
&self,
table_bucket: &str,
@@ -7400,6 +7684,16 @@ where
Ok(entries)
}
async fn list_namespaces_page(
&self,
table_bucket: &str,
cursor: Option<&str>,
limit: NonZeroUsize,
) -> TableCatalogStoreResult<TableCatalogListPage<NamespaceEntry>> {
self.list_entry_page(&self.paths.namespace_entries_prefix(table_bucket), NAMESPACE_ENTRY_FILE, cursor, limit)
.await
}
async fn get_namespace(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult<Option<NamespaceEntry>> {
let namespace = parse_namespace_for_store(namespace)?;
self.read_entry::<NamespaceEntry>(self.catalog_bucket(), &self.paths.namespace_entry_path(table_bucket, &namespace))
@@ -7475,6 +7769,23 @@ where
Ok(entries)
}
async fn list_tables_page(
&self,
table_bucket: &str,
namespace: &str,
cursor: Option<&str>,
limit: NonZeroUsize,
) -> TableCatalogStoreResult<TableCatalogListPage<TableEntry>> {
let namespace = parse_namespace_for_store(namespace)?;
self.list_entry_page(
&self.paths.table_entries_prefix(table_bucket, &namespace),
TABLE_ENTRY_FILE,
cursor,
limit,
)
.await
}
async fn load_table(&self, table_bucket: &str, namespace: &str, table: &str) -> TableCatalogStoreResult<Option<TableEntry>> {
let namespace = parse_namespace_for_store(namespace)?;
let table = parse_table_for_store(table)?;
@@ -7870,6 +8181,18 @@ where
Ok(entries)
}
async fn list_views_page(
&self,
table_bucket: &str,
namespace: &str,
cursor: Option<&str>,
limit: NonZeroUsize,
) -> TableCatalogStoreResult<TableCatalogListPage<ViewEntry>> {
let namespace = parse_namespace_for_store(namespace)?;
self.list_entry_page(&self.paths.view_entries_prefix(table_bucket, &namespace), VIEW_ENTRY_FILE, cursor, limit)
.await
}
async fn load_view(&self, table_bucket: &str, namespace: &str, view: &str) -> TableCatalogStoreResult<Option<ViewEntry>> {
let namespace = parse_namespace_for_store(namespace)?;
let view = parse_table_for_store(view)?;
@@ -8025,6 +8348,18 @@ where
}
}
async fn list_namespaces_page(
&self,
table_bucket: &str,
cursor: Option<&str>,
limit: NonZeroUsize,
) -> TableCatalogStoreResult<TableCatalogListPage<NamespaceEntry>> {
match self {
Self::ObjectBacked(store) => store.list_namespaces_page(table_bucket, cursor, limit).await,
Self::DurableStrong(store) => store.list_namespaces_page(table_bucket, cursor, limit).await,
}
}
async fn get_namespace(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult<Option<NamespaceEntry>> {
match self {
Self::ObjectBacked(store) => store.get_namespace(table_bucket, namespace).await,
@@ -8060,6 +8395,19 @@ where
}
}
async fn list_tables_page(
&self,
table_bucket: &str,
namespace: &str,
cursor: Option<&str>,
limit: NonZeroUsize,
) -> TableCatalogStoreResult<TableCatalogListPage<TableEntry>> {
match self {
Self::ObjectBacked(store) => store.list_tables_page(table_bucket, namespace, cursor, limit).await,
Self::DurableStrong(store) => store.list_tables_page(table_bucket, namespace, cursor, limit).await,
}
}
async fn load_table(&self, table_bucket: &str, namespace: &str, table: &str) -> TableCatalogStoreResult<Option<TableEntry>> {
match self {
Self::ObjectBacked(store) => store.load_table(table_bucket, namespace, table).await,
@@ -8106,6 +8454,19 @@ where
}
}
async fn list_views_page(
&self,
table_bucket: &str,
namespace: &str,
cursor: Option<&str>,
limit: NonZeroUsize,
) -> TableCatalogStoreResult<TableCatalogListPage<ViewEntry>> {
match self {
Self::ObjectBacked(store) => store.list_views_page(table_bucket, namespace, cursor, limit).await,
Self::DurableStrong(store) => store.list_views_page(table_bucket, namespace, cursor, limit).await,
}
}
async fn load_view(&self, table_bucket: &str, namespace: &str, view: &str) -> TableCatalogStoreResult<Option<ViewEntry>> {
match self {
Self::ObjectBacked(store) => store.load_view(table_bucket, namespace, view).await,
@@ -8474,12 +8835,14 @@ where
async fn list_objects(&self, bucket: &str, prefix: &str) -> TableCatalogStoreResult<Vec<String>> {
let mut continuation = None;
let mut objects = BTreeSet::new();
let max_keys = i32::try_from(TABLE_CATALOG_LIST_MAX_KEYS)
.map_err(|_| TableCatalogStoreError::Internal("catalog list limit exceeds storage API range".to_string()))?;
loop {
let result = self
.store
.clone()
.list_objects_v2(bucket, prefix, continuation, None, TABLE_CATALOG_LIST_MAX_KEYS, false, None, false)
.list_objects_v2(bucket, prefix, continuation, None, max_keys, false, None, false)
.await
.map_err(|err| storage_error_to_catalog("list catalog objects", err))?;
@@ -8500,6 +8863,29 @@ where
Ok(objects.into_iter().collect())
}
async fn list_objects_page(
&self,
bucket: &str,
prefix: &str,
start_after: Option<&str>,
limit: NonZeroUsize,
) -> TableCatalogStoreResult<TableCatalogObjectListPage> {
let max_keys = i32::try_from(limit.get())
.map_err(|_| TableCatalogStoreError::Invalid("catalog page size exceeds storage API range".to_string()))?;
let result = self
.store
.clone()
.list_objects_v2(bucket, prefix, None, None, max_keys, false, start_after.map(str::to_string), false)
.await
.map_err(|err| storage_error_to_catalog("list catalog object page", err))?;
let is_truncated = result.is_truncated;
let objects = result.objects.into_iter().map(|object| object.name).collect::<BTreeSet<_>>();
Ok(TableCatalogObjectListPage {
objects: objects.into_iter().collect(),
is_truncated,
})
}
async fn acquire_write_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Box<dyn Send>> {
let lock = self
.store
@@ -11671,10 +12057,15 @@ pub struct Namespace {
}
impl Namespace {
pub const MAX_LEN: usize = 512;
pub fn parse(value: &str) -> Result<Self, CatalogIdentifierError> {
if value.is_empty() {
return Err(CatalogIdentifierError::Empty);
}
if value.len() > Self::MAX_LEN {
return Err(CatalogIdentifierError::NamespaceTooLong { max: Self::MAX_LEN });
}
let mut segments = Vec::new();
for segment in value.split('.') {
@@ -13099,6 +13490,40 @@ mod tests {
}
}
async fn seed_catalog_list_entries<S>(store: &S, bucket: &str, namespace: &Namespace)
where
S: TableCatalogStore + ?Sized,
{
store
.put_table_bucket(test_bucket_entry(bucket))
.await
.expect("table bucket should be created");
for namespace in [
Namespace::parse("analytics").expect("namespace should parse"),
namespace.clone(),
] {
store
.create_namespace(test_namespace_entry(bucket, &namespace))
.await
.expect("namespace should be created");
}
for name in ["alpha", "beta"] {
let identifier = IdentifierSegment::parse(name).expect("table and view name should parse");
let metadata_location = default_table_metadata_file_path(namespace, &identifier, "00001.metadata.json");
let mut table = test_table_entry(bucket, namespace, &identifier, metadata_location.clone());
table.table_id = format!("table-{name}");
table.table_uuid = format!("table-uuid-{name}");
table.warehouse_location = format!("s3://{bucket}/tables/table-{name}");
store.create_table(table).await.expect("table should be created");
let mut view = test_view_entry(bucket, namespace, &identifier, metadata_location);
view.view_id = format!("view-{name}");
view.view_uuid = format!("view-uuid-{name}");
view.warehouse_location = format!("s3://{bucket}/views/view-{name}");
store.create_view(view).await.expect("view should be created");
}
}
async fn seed_table_for_metadata_maintenance(
store: &ObjectTableCatalogStore<TestCatalogObjectBackend>,
bucket: &str,
@@ -13645,6 +14070,119 @@ mod tests {
);
}
#[tokio::test]
async fn object_catalog_pagination_bounds_reads_and_covers_rest_resources() {
let backend = TestCatalogObjectBackend::default();
let store = ObjectTableCatalogStore::new(backend.clone());
let bucket = "analytics";
let namespace = Namespace::parse("sales").expect("namespace should parse");
let namespace_name = namespace.public_name();
let one = NonZeroUsize::new(1).expect("page size should be non-zero");
seed_catalog_list_entries(&store, bucket, &namespace).await;
let namespace_page = store
.list_namespaces_page(bucket, None, one)
.await
.expect("first namespace page should load");
assert_eq!(namespace_page.entries[0].namespace, "analytics");
assert!(
namespace_page
.next_cursor
.as_deref()
.is_some_and(|cursor| cursor.starts_with(OBJECT_CATALOG_LIST_CURSOR_PREFIX))
);
let namespace_page = store
.list_namespaces_page(bucket, namespace_page.next_cursor.as_deref(), one)
.await
.expect("second namespace page should load");
assert_eq!(namespace_page.entries[0].namespace, "sales");
assert!(namespace_page.next_cursor.is_none());
backend.reset_call_counts().await;
let table_page = store
.list_tables_page(bucket, &namespace_name, None, one)
.await
.expect("first table page should load");
assert_eq!(table_page.entries[0].table, "alpha");
assert_eq!(backend.read_call_count().await, 1);
let table_page = store
.list_tables_page(bucket, &namespace_name, table_page.next_cursor.as_deref(), one)
.await
.expect("second table page should load");
assert_eq!(table_page.entries[0].table, "beta");
assert!(table_page.next_cursor.is_none());
let view_page = store
.list_views_page(bucket, &namespace_name, None, one)
.await
.expect("first view page should load");
assert_eq!(view_page.entries[0].view, "alpha");
let view_page = store
.list_views_page(bucket, &namespace_name, view_page.next_cursor.as_deref(), one)
.await
.expect("second view page should load");
assert_eq!(view_page.entries[0].view, "beta");
assert!(view_page.next_cursor.is_none());
let exact_page = store
.list_tables_page(bucket, &namespace_name, None, NonZeroUsize::new(2).expect("page size should be non-zero"))
.await
.expect("exact table page should load");
assert_eq!(exact_page.entries.len(), 2);
assert!(exact_page.next_cursor.is_none());
assert!(matches!(
store
.list_tables_page(bucket, &namespace_name, Some("strong:alpha"), one)
.await,
Err(TableCatalogStoreError::Invalid(_))
));
}
#[tokio::test]
async fn object_catalog_pagination_bounds_sparse_namespace_scans() {
let backend = TestCatalogObjectBackend::default();
let store = ObjectTableCatalogStore::new(backend.clone());
let bucket = "analytics";
let namespace = Namespace::parse("sales").expect("namespace should parse");
let prefix = store.paths.namespace_entries_prefix(bucket);
store
.put_table_bucket(test_bucket_entry(bucket))
.await
.expect("table bucket should be created");
store
.create_namespace(test_namespace_entry(bucket, &namespace))
.await
.expect("namespace should be created");
for index in 0..TABLE_CATALOG_LIST_MAX_KEYS {
backend
.seed_object(RUSTFS_META_BUCKET, &format!("{prefix}0000-spacer/{index:04}.json"), Vec::new())
.await;
}
backend.reset_call_counts().await;
let one = NonZeroUsize::new(1).expect("page size should be non-zero");
let first = store
.list_namespaces_page(bucket, None, one)
.await
.expect("sparse first page should load");
assert!(first.entries.is_empty());
assert_eq!(backend.list_call_count().await, 1);
let cursor = first.next_cursor.expect("truncated sparse page should have a cursor");
assert!(cursor.starts_with(OBJECT_CATALOG_LIST_CURSOR_PREFIX));
assert!(!cursor.ends_with(NAMESPACE_ENTRY_FILE));
let second = store
.list_namespaces_page(bucket, Some(&cursor), one)
.await
.expect("sparse continuation page should load");
assert_eq!(second.entries.len(), 1);
assert_eq!(second.entries[0].namespace, namespace.public_name());
assert!(second.next_cursor.is_none());
assert_eq!(backend.list_call_count().await, 2);
}
#[tokio::test]
async fn object_table_catalog_store_rolls_back_warehouse_index_when_table_entry_write_fails() {
let backend = TestCatalogObjectBackend::default();
@@ -18559,6 +19097,88 @@ mod tests {
assert_eq!(recovery.manual_review_count, 0);
}
#[tokio::test]
async fn strong_catalog_pagination_uses_ordered_state_and_rejects_object_cursors() {
let backend = TestCatalogObjectBackend::default();
let store = StrongTableCatalogStore::new(backend);
let bucket = "analytics";
let namespace = Namespace::parse("sales").expect("namespace should parse");
let namespace_name = namespace.public_name();
let one = NonZeroUsize::new(1).expect("page size should be non-zero");
seed_catalog_list_entries(&store, bucket, &namespace).await;
let first = store
.list_namespaces_page(bucket, None, one)
.await
.expect("first strong namespace page should load");
assert_eq!(first.entries[0].namespace, "analytics");
let second = store
.list_namespaces_page(bucket, first.next_cursor.as_deref(), one)
.await
.expect("second strong namespace page should load");
assert_eq!(second.entries[0].namespace, "sales");
assert!(second.next_cursor.is_none());
let first = store
.list_tables_page(bucket, &namespace_name, None, one)
.await
.expect("first strong table page should load");
assert_eq!(first.entries[0].table, "alpha");
assert!(first.next_cursor.as_deref().is_some_and(|cursor| cursor == "strong:alpha"));
let second = store
.list_tables_page(bucket, &namespace_name, first.next_cursor.as_deref(), one)
.await
.expect("second strong table page should load");
assert_eq!(second.entries[0].table, "beta");
assert!(second.next_cursor.is_none());
let first = store
.list_views_page(bucket, &namespace_name, None, one)
.await
.expect("first strong view page should load");
assert_eq!(first.entries[0].view, "alpha");
let second = store
.list_views_page(bucket, &namespace_name, first.next_cursor.as_deref(), one)
.await
.expect("second strong view page should load");
assert_eq!(second.entries[0].view, "beta");
assert!(second.next_cursor.is_none());
let exact = NonZeroUsize::new(2).expect("exact page size should be non-zero");
assert!(
store
.list_namespaces_page(bucket, None, exact)
.await
.expect("exact strong namespace page should load")
.next_cursor
.is_none()
);
assert!(
store
.list_tables_page(bucket, &namespace_name, None, exact)
.await
.expect("exact strong table page should load")
.next_cursor
.is_none()
);
assert!(
store
.list_views_page(bucket, &namespace_name, None, exact)
.await
.expect("exact strong view page should load")
.next_cursor
.is_none()
);
assert!(matches!(
store
.list_tables_page(bucket, &namespace_name, Some("object:alpha"), one)
.await,
Err(TableCatalogStoreError::Invalid(_))
));
}
#[tokio::test]
async fn strong_catalog_backing_replays_durable_commit_state_after_restart() {
let backend = TestCatalogObjectBackend::default();
@@ -20307,6 +20927,21 @@ mod tests {
assert_eq!(namespace.storage_id(), "analytics/daily_events");
}
#[test]
fn namespace_length_is_bounded_for_catalog_paths_and_page_tokens() {
let mut segments = vec!["a".repeat(63); 8];
segments[0].push('a');
let max_length_namespace = segments.join(".");
assert_eq!(max_length_namespace.len(), Namespace::MAX_LEN);
Namespace::parse(&max_length_namespace).expect("namespace at the maximum length should parse");
let namespace = format!("{max_length_namespace}.a");
assert_eq!(
Namespace::parse(&namespace),
Err(CatalogIdentifierError::NamespaceTooLong { max: Namespace::MAX_LEN })
);
}
#[test]
fn resolver_builds_paths_under_reserved_table_boundary() {
let table = TableIdentifier::new(