From e5cfa8e375021c3ec0c0f3023ffaf9b37bda2b32 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Sun, 2 Aug 2026 23:27:15 +0800 Subject: [PATCH] feat(table-catalog): paginate Iceberg REST listings (#5466) * feat(table-catalog): paginate Iceberg REST listings * test(table-catalog): remove redundant token clones --------- Co-authored-by: Henry Guo --- docs/architecture/s3-tables-support-matrix.md | 4 +- rustfs/src/admin/handlers/table_catalog.rs | 678 ++++++++++++++++-- rustfs/src/table_catalog.rs | 639 ++++++++++++++++- scripts/table-catalog/pyiceberg_smoke.py | 147 +++- scripts/table-catalog/test_pyiceberg_smoke.py | 175 ++++- 5 files changed, 1546 insertions(+), 97 deletions(-) diff --git a/docs/architecture/s3-tables-support-matrix.md b/docs/architecture/s3-tables-support-matrix.md index bf77c5815..138b0aca0 100644 --- a/docs/architecture/s3-tables-support-matrix.md +++ b/docs/architecture/s3-tables-support-matrix.md @@ -63,8 +63,8 @@ catalog extension. |---|---|---| | Catalog config | Supported | `GET /v1/config` advertises RustFS catalog defaults and route capabilities. | | Table bucket discovery | Supported | `PUT` and `GET /v1/buckets/{warehouse}` enable and inspect table bucket state. | -| Namespaces | Supported | Create, list, load, existence check, and drop namespace routes are registered on both catalog prefixes. | -| Tables | Supported | Create, register, list, load, existence check, commit, metadata-location get/update, and drop table routes are registered on both catalog prefixes. | +| Namespaces | Supported | Create, list, load, existence check, and drop namespace routes are registered on both catalog prefixes. List responses support Iceberg REST `pageSize`/`pageToken` pagination with context-bound tokens and bounded catalog-store reads. Namespace identifiers are limited to 512 ASCII characters so persisted paths and stateless continuation tokens remain bounded. | +| Tables | Supported | Create, register, list, load, existence check, commit, metadata-location get/update, and drop table routes are registered on both catalog prefixes. Table and view listings support Iceberg REST `pageSize`/`pageToken` pagination with context-bound tokens and bounded catalog-store reads. | | Commit CAS | Supported | Single-table commits validate base metadata, expected version token, referenced object existence, warehouse scope, and Iceberg commit requirements before advancing the current metadata pointer. | | Commit recovery | Supported | Commit log, idempotency lookup, diagnostics, and recovery routes expose staged/finalization gaps and repair safe idempotency gaps without moving the table pointer. | | Snapshot refs | Supported | Refs can be listed, created or replaced, and deleted through catalog commits. `main` is protected and refs with explicit retention require forced delete. | diff --git a/rustfs/src/admin/handlers/table_catalog.rs b/rustfs/src/admin/handlers/table_catalog.rs index 24d0b4e2b..b20b63d61 100644 --- a/rustfs/src/admin/handlers/table_catalog.rs +++ b/rustfs/src/admin/handlers/table_catalog.rs @@ -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>, + #[serde(rename = "next-page-token")] + next_page_token: Option, } #[derive(Debug, Serialize)] @@ -614,11 +624,56 @@ struct RestTableIdentifier { #[derive(Debug, Serialize)] struct RestListTablesResponse { identifiers: Vec, + #[serde(rename = "next-page-token")] + next_page_token: Option, +} + +#[derive(Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct RestPageToken { + version: u8, + context: String, + cursor: String, +} + +#[derive(Debug)] +enum RestPagination { + Unpaginated, + Paginated { + cursor: Option, + 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) -> S3Result> { + 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, + #[serde(rename = "next-page-token")] + next_page_token: Option, } #[derive(Debug, Serialize)] @@ -1327,6 +1382,142 @@ fn warehouse_from_config_query(uri: &http::Uri) -> S3Result> { Ok(warehouse) } +fn rest_pagination_from_query(uri: &http::Uri, context: RestPageContext<'_>) -> S3Result { + 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::().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 { + 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::(&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 { + 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 { 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, + next_page_token: Option, ) -> S3Result { let namespaces = entries .into_iter() @@ -1523,10 +1715,16 @@ fn list_namespaces_response_from_entries( Ok(namespace_segments(&namespace)) }) .collect::>>()?; - Ok(RestListNamespacesResponse { namespaces }) + Ok(RestListNamespacesResponse { + namespaces, + next_page_token, + }) } -fn list_tables_response_from_entries(entries: Vec) -> S3Result { +fn list_tables_response_from_entries( + entries: Vec, + next_page_token: Option, +) -> S3Result { let identifiers = entries .into_iter() .map(|entry| { @@ -1538,10 +1736,16 @@ fn list_tables_response_from_entries(entries: Vec>>()?; - Ok(RestListTablesResponse { identifiers }) + Ok(RestListTablesResponse { + identifiers, + next_page_token, + }) } -fn list_views_response_from_entries(entries: Vec) -> S3Result { +fn list_views_response_from_entries( + entries: Vec, + next_page_token: Option, +) -> S3Result { let identifiers = entries .into_iter() .map(|entry| { @@ -1553,7 +1757,10 @@ fn list_views_response_from_entries(entries: 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(store: &S, bucket: &str) -> S3Result +async fn list_namespaces_response(store: &S, bucket: &str, uri: &http::Uri) -> S3Result 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( @@ -3766,15 +3989,30 @@ async fn list_tables_response( store: &S, bucket: &str, namespace: &crate::table_catalog::Namespace, + uri: &http::Uri, ) -> S3Result 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( @@ -3802,15 +4040,30 @@ async fn list_views_response( store: &S, bucket: &str, namespace: &crate::table_catalog::Namespace, + uri: &http::Uri, ) -> S3Result 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( @@ -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::().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::() + .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::() + .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::().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::() + .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::().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::() + .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::() + ) + .parse::() + .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::().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::() + .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::() + .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::() + .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::().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::().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::().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::().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"); diff --git a/rustfs/src/table_catalog.rs b/rustfs/src/table_catalog.rs index 7a517ace3..e8a8f4ff2 100644 --- a/rustfs/src/table_catalog.rs +++ b/rustfs/src/table_catalog.rs @@ -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 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 { + pub entries: Vec, + pub next_cursor: Option, +} + +fn finish_catalog_list_page( + mut entries: Vec, + limit: NonZeroUsize, + cursor_prefix: &str, + key: F, +) -> TableCatalogListPage +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( + mut entries: Vec, + cursor: Option<&str>, + limit: NonZeroUsize, + key: F, +) -> TableCatalogListPage +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> { + 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>; @@ -1778,6 +1836,20 @@ pub(crate) trait TableCatalogStore: Send + Sync { async fn list_namespaces(&self, table_bucket: &str) -> TableCatalogStoreResult>; + async fn list_namespaces_page( + &self, + table_bucket: &str, + cursor: Option<&str>, + limit: NonZeroUsize, + ) -> TableCatalogStoreResult> { + 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>; 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>; + async fn list_tables_page( + &self, + table_bucket: &str, + namespace: &str, + cursor: Option<&str>, + limit: NonZeroUsize, + ) -> TableCatalogStoreResult> { + 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>; 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>; + async fn list_views_page( + &self, + table_bucket: &str, + namespace: &str, + cursor: Option<&str>, + limit: NonZeroUsize, + ) -> TableCatalogStoreResult> { + 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>; async fn replace_view(&self, request: ViewCommitRequest) -> TableCatalogStoreResult; @@ -1840,6 +1942,12 @@ pub(crate) struct TableCatalogObjectMetadata { pub mod_time: Option, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct TableCatalogObjectListPage { + pub objects: Vec, + 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>; + async fn list_objects_page( + &self, + bucket: &str, + prefix: &str, + start_after: Option<&str>, + limit: NonZeroUsize, + ) -> TableCatalogStoreResult { + 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::>(); + 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> { 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> { + 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> { 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> { + 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> { 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> { + 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> { self.hydrate_state().await?; let namespace = parse_namespace_for_store(namespace)?; @@ -3558,6 +3777,71 @@ where RUSTFS_META_BUCKET } + async fn list_entry_page( + &self, + prefix: &str, + entry_file: &str, + cursor: Option<&str>, + limit: NonZeroUsize, + ) -> TableCatalogStoreResult> + 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::(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> { + 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> { let namespace = parse_namespace_for_store(namespace)?; self.read_entry::(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> { + 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> { 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> { + 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> { 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> { + 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> { 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> { + 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> { 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> { + 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> { 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> { 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 { + 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::>(); + Ok(TableCatalogObjectListPage { + objects: objects.into_iter().collect(), + is_truncated, + }) + } + async fn acquire_write_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult> { 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 { 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(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, 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( diff --git a/scripts/table-catalog/pyiceberg_smoke.py b/scripts/table-catalog/pyiceberg_smoke.py index d743747cc..0fc13c1d5 100755 --- a/scripts/table-catalog/pyiceberg_smoke.py +++ b/scripts/table-catalog/pyiceberg_smoke.py @@ -15,6 +15,7 @@ import time import urllib.error import urllib.parse import urllib.request +import uuid from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path @@ -38,6 +39,7 @@ SENSITIVE_COMMAND_FLAGS = { } TABLE_MAINTENANCE_CONFIG_VERSION = 1 IDENTIFIER_SEGMENT_MAX_LEN = 64 +MAX_PAGINATION_PROBE_PAGES = 16 PROFILE_DEFAULTS: dict[str, dict[str, Any]] = { "rustfs": { @@ -51,7 +53,7 @@ PROFILE_DEFAULTS: dict[str, dict[str, Any]] = { "catalog_uri_shape": "{endpoint}/iceberg", "warehouse_shape": "{bucket}", "namespace_model": "single-level", - "pagination_model": "rustfs", + "pagination_model": "iceberg-rest", "not_claimed": [], }, "rustfs-compat": { @@ -65,7 +67,7 @@ PROFILE_DEFAULTS: dict[str, dict[str, Any]] = { "catalog_uri_shape": "{endpoint}/_iceberg", "warehouse_shape": "{bucket}", "namespace_model": "single-level", - "pagination_model": "rustfs", + "pagination_model": "iceberg-rest", "not_claimed": ["full MinIO AIStor private extension parity"], }, CATALOG_VENDED_PROFILE: { @@ -79,7 +81,7 @@ PROFILE_DEFAULTS: dict[str, dict[str, Any]] = { "catalog_uri_shape": "{endpoint}/iceberg", "warehouse_shape": "{bucket}", "namespace_model": "single-level", - "pagination_model": "rustfs", + "pagination_model": "iceberg-rest", "not_claimed": ["no-long-term-data-credential bootstrap"], }, "aws-s3tables": { @@ -148,7 +150,7 @@ CLIENT_MATRIX: list[dict[str, str]] = [ { "client": "PyIceberg", "status": "automated", - "coverage": "create namespace, create table, append, reload, scan, metadata-location, refs, views, maintenance, diagnostics, optional catalog-vended table credentials with exact-prefix data-plane scope probe", + "coverage": "create namespace, create table, append, reload, scan, metadata-location, refs, paginated views, maintenance, diagnostics, optional catalog-vended table credentials with exact-prefix data-plane scope probe", "entrypoint": "scripts/table-catalog/pyiceberg_smoke.py", }, { @@ -633,15 +635,68 @@ def table_ref_endpoint_path(args: argparse.Namespace, ref_name: str | None = Non return f"{path}/{urllib.parse.quote(ref_name, safe='')}" -def view_endpoint_path(args: argparse.Namespace, view_name: str | None = None) -> str: +def namespace_endpoint_path(args: argparse.Namespace, namespace: str | None = None) -> str: encoded_bucket = urllib.parse.quote(args.bucket, safe="") - encoded_namespace = urllib.parse.quote(args.namespace, safe="") - path = f"{args.rest_path}/v1/{encoded_bucket}/namespaces/{encoded_namespace}/views" + path = f"{args.rest_path}/v1/{encoded_bucket}/namespaces" + if namespace is None: + return path + return f"{path}/{urllib.parse.quote(namespace, safe='')}" + + +def view_endpoint_path( + args: argparse.Namespace, + view_name: str | None = None, + namespace: str | None = None, +) -> str: + encoded_namespace = urllib.parse.quote(namespace or args.namespace, safe="") + path = f"{namespace_endpoint_path(args)}/{encoded_namespace}/views" if view_name is None: return path return f"{path}/{urllib.parse.quote(view_name, safe='')}" +def paginated_identifier_names( + args: argparse.Namespace, + deps: RuntimeDeps, + path: str, + page_size: int, +) -> list[str]: + page_token: str | None = None + seen_names: set[str] = set() + seen_tokens: set[str] = set() + names: list[str] = [] + for _ in range(MAX_PAGINATION_PROBE_PAGES): + query_parameters: dict[str, str | int] = {"pageSize": page_size} + if page_token is not None: + query_parameters["pageToken"] = page_token + query = urllib.parse.urlencode(query_parameters) + response = signed_rest_request(args, deps, "GET", f"{path}?{query}") + identifiers = response.get("identifiers") + if not isinstance(identifiers, list): + raise RuntimeError("paginated list response did not include identifiers") + if len(identifiers) > page_size: + raise RuntimeError("paginated list response exceeded pageSize") + for identifier in identifiers: + name = identifier.get("name") if isinstance(identifier, dict) else None + if not isinstance(name, str) or not name: + raise RuntimeError("paginated list response included an invalid identifier") + if name in seen_names: + raise RuntimeError("paginated list response included a duplicate identifier") + seen_names.add(name) + names.append(name) + + if "next-page-token" not in response: + raise RuntimeError("paginated list response omitted next-page-token") + next_page_token = response["next-page-token"] + if next_page_token is None: + return names + if not isinstance(next_page_token, str) or not next_page_token or next_page_token in seen_tokens: + raise RuntimeError("paginated list response included an invalid next-page-token") + seen_tokens.add(next_page_token) + page_token = next_page_token + raise RuntimeError("paginated list response exceeded the smoke page limit") + + def default_maintenance_config() -> dict[str, Any]: return { "version": TABLE_MAINTENANCE_CONFIG_VERSION, @@ -1165,24 +1220,78 @@ def smoke_view_request(args: argparse.Namespace, view_name: str, version_id: int def run_view_probe(args: argparse.Namespace, deps: RuntimeDeps) -> None: - view_name = f"{args.table}_smoke_view" - create_body = smoke_view_request(args, view_name, 1, f"SELECT id, payload FROM {args.namespace}.{args.table}") - signed_rest_request(args, deps, "POST", view_endpoint_path(args), create_body) + namespace_suffix = f"-views-{uuid.uuid4().hex[:8]}" + namespace_prefix = safe_ref_segment(args.namespace, args.table)[: IDENTIFIER_SEGMENT_MAX_LEN - len(namespace_suffix)].rstrip("-_") + probe_namespace = f"{namespace_prefix}{namespace_suffix}" + view_names = ["view-a", "view-b"] + cleanup_views: list[str] = [] + cleanup_namespace = False try: - views = signed_rest_request(args, deps, "GET", view_endpoint_path(args)) + cleanup_namespace = True + try: + signed_rest_request( + args, + deps, + "POST", + namespace_endpoint_path(args), + {"namespace": [probe_namespace], "properties": {"rustfs.smoke": "true"}}, + ) + except RestRequestError as error: + if error.status_code == 409: + cleanup_namespace = False + raise + + view_path = view_endpoint_path(args, namespace=probe_namespace) + for view_name in view_names: + cleanup_views.append(view_name) + create_body = smoke_view_request( + args, + view_name, + 1, + f"SELECT id, payload FROM {args.namespace}.{args.table}", + ) + signed_rest_request(args, deps, "POST", view_path, create_body) + + views = signed_rest_request(args, deps, "GET", view_path) + if "next-page-token" not in views or views["next-page-token"] is not None: + raise RuntimeError("unpaginated listViews response did not terminate with a null next-page-token") identifiers = views.get("identifiers", []) - if not any(identifier.get("name") == view_name for identifier in identifiers if isinstance(identifier, dict)): - raise RuntimeError("listViews response did not include the smoke view") - loaded = signed_rest_request(args, deps, "GET", view_endpoint_path(args, view_name)) + listed_names = {identifier.get("name") for identifier in identifiers if isinstance(identifier, dict)} + if listed_names != set(view_names): + raise RuntimeError("listViews response did not include exactly the smoke views") + + paginated_names = paginated_identifier_names(args, deps, view_path, page_size=1) + if sorted(paginated_names) != sorted(view_names): + raise RuntimeError("paginated listViews response did not return both smoke views exactly once") + + loaded = signed_rest_request(args, deps, "GET", view_endpoint_path(args, view_names[0], probe_namespace)) metadata_location = loaded.get("metadata-location") if not isinstance(metadata_location, str) or not metadata_location: raise RuntimeError("loadView response did not include metadata-location") finally: - try: - signed_rest_request(args, deps, "DELETE", view_endpoint_path(args, view_name)) - except RestRequestError as error: - if error.status_code != 404: - raise + probe_failed = sys.exc_info()[0] is not None + cleanup_errors: list[Exception] = [] + for view_name in reversed(cleanup_views): + try: + signed_rest_request(args, deps, "DELETE", view_endpoint_path(args, view_name, probe_namespace)) + except RestRequestError as error: + if error.status_code != 404: + cleanup_errors.append(error) + except Exception as error: + cleanup_errors.append(error) + if cleanup_namespace: + try: + signed_rest_request(args, deps, "DELETE", namespace_endpoint_path(args, probe_namespace)) + except RestRequestError as error: + if error.status_code != 404: + cleanup_errors.append(error) + except Exception as error: + cleanup_errors.append(error) + if cleanup_errors: + if not probe_failed: + raise cleanup_errors[0] + for error in cleanup_errors: + print(f"warning: failed to clean up view pagination probe: {error}", file=sys.stderr) def maintenance_job_id(job: object) -> str | None: diff --git a/scripts/table-catalog/test_pyiceberg_smoke.py b/scripts/table-catalog/test_pyiceberg_smoke.py index 573fa3aaf..43cb49136 100644 --- a/scripts/table-catalog/test_pyiceberg_smoke.py +++ b/scripts/table-catalog/test_pyiceberg_smoke.py @@ -67,6 +67,7 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): self.assertIn("minio-aistor", profiles) self.assertIn("cloudflare-r2-data-catalog", profiles) self.assertIn("oss-tables", profiles) + self.assertEqual(profiles["rustfs"]["pagination_model"], "iceberg-rest") self.assertEqual( profiles["rustfs-vended-credentials"]["credential_mode"], "catalog-vended-temporary-credentials", @@ -287,7 +288,9 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): def test_view_probe_drops_smoke_view_after_load_failure(self) -> None: args = self.parse_with_args(["--namespace", "sales", "--table", "orders"]) - view_path = pyiceberg_smoke.view_endpoint_path(args) + probe_namespace = "smoke-sales-orders-views-12345678" + namespace_path = pyiceberg_smoke.namespace_endpoint_path(args) + view_path = pyiceberg_smoke.view_endpoint_path(args, namespace=probe_namespace) calls: list[tuple[str, str, object]] = [] def fake_signed_request( @@ -298,21 +301,144 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): body: object = None, ) -> dict[str, object]: calls.append((method, path, body)) + if (method, path) == ("POST", namespace_path): + return {} if (method, path) == ("POST", view_path): return {} if (method, path) == ("GET", view_path): - return {"identifiers": [{"name": "orders_smoke_view"}]} - if (method, path) == ("GET", f"{view_path}/orders_smoke_view"): + return { + "identifiers": [ + {"name": "view-a"}, + {"name": "view-b"}, + ], + "next-page-token": None, + } + if (method, path) == ("GET", f"{view_path}/view-a"): return {} - if (method, path) == ("DELETE", f"{view_path}/orders_smoke_view"): + if method == "DELETE" and path.startswith(f"{view_path}/view-"): + return {} + if (method, path) == ("DELETE", pyiceberg_smoke.namespace_endpoint_path(args, probe_namespace)): return {} raise AssertionError(f"unexpected REST request: {method} {path}") - with mock.patch.object(pyiceberg_smoke, "signed_rest_request", side_effect=fake_signed_request): - with self.assertRaisesRegex(RuntimeError, "metadata-location"): - pyiceberg_smoke.run_view_probe(args, mock.Mock()) + with mock.patch.object(pyiceberg_smoke.uuid, "uuid4", return_value=SimpleNamespace(hex="1234567890abcdef")): + with mock.patch.object(pyiceberg_smoke, "signed_rest_request", side_effect=fake_signed_request): + with mock.patch.object(pyiceberg_smoke, "paginated_identifier_names", return_value=["view-a", "view-b"]): + with self.assertRaisesRegex(RuntimeError, "metadata-location"): + pyiceberg_smoke.run_view_probe(args, mock.Mock()) - self.assertIn(("DELETE", f"{view_path}/orders_smoke_view", None), calls) + self.assertIn(("DELETE", f"{view_path}/view-a", None), calls) + self.assertIn(("DELETE", f"{view_path}/view-b", None), calls) + self.assertIn(("DELETE", pyiceberg_smoke.namespace_endpoint_path(args, probe_namespace), None), calls) + + def test_view_probe_cleans_candidate_after_create_timeout(self) -> None: + args = self.parse_with_args(["--namespace", "sales", "--table", "orders"]) + probe_namespace = "smoke-sales-orders-views-12345678" + namespace_path = pyiceberg_smoke.namespace_endpoint_path(args) + view_path = pyiceberg_smoke.view_endpoint_path(args, namespace=probe_namespace) + calls: list[tuple[str, str, object]] = [] + + def fake_signed_request( + _args: object, + _deps: object, + method: str, + path: str, + body: object = None, + ) -> dict[str, object]: + calls.append((method, path, body)) + if (method, path) == ("POST", namespace_path): + return {} + if (method, path) == ("POST", view_path): + raise RuntimeError("timed out after commit") + if (method, path) == ("DELETE", f"{view_path}/view-a"): + return {} + if (method, path) == ("DELETE", pyiceberg_smoke.namespace_endpoint_path(args, probe_namespace)): + return {} + raise AssertionError(f"unexpected REST request: {method} {path}") + + with mock.patch.object(pyiceberg_smoke.uuid, "uuid4", return_value=SimpleNamespace(hex="1234567890abcdef")): + with mock.patch.object(pyiceberg_smoke, "signed_rest_request", side_effect=fake_signed_request): + with self.assertRaisesRegex(RuntimeError, "timed out after commit"): + pyiceberg_smoke.run_view_probe(args, mock.Mock()) + + self.assertIn(("DELETE", f"{view_path}/view-a", None), calls) + self.assertIn(("DELETE", pyiceberg_smoke.namespace_endpoint_path(args, probe_namespace), None), calls) + + def test_view_probe_continues_cleanup_after_delete_failure(self) -> None: + args = self.parse_with_args(["--namespace", "sales", "--table", "orders"]) + probe_namespace = "smoke-sales-orders-views-12345678" + namespace_path = pyiceberg_smoke.namespace_endpoint_path(args) + view_path = pyiceberg_smoke.view_endpoint_path(args, namespace=probe_namespace) + calls: list[tuple[str, str, object]] = [] + + def fake_signed_request( + _args: object, + _deps: object, + method: str, + path: str, + body: object = None, + ) -> dict[str, object]: + calls.append((method, path, body)) + if (method, path) == ("POST", namespace_path) or (method, path) == ("POST", view_path): + return {} + if (method, path) == ("GET", view_path): + return {"identifiers": [{"name": "view-a"}, {"name": "view-b"}], "next-page-token": None} + if (method, path) == ("GET", f"{view_path}/view-a"): + return {"metadata-location": "s3://lake/views/view-a/metadata/v1.json"} + if (method, path) == ("DELETE", f"{view_path}/view-b"): + raise pyiceberg_smoke.RestRequestError(method, path, 500, "delete failed") + if (method, path) == ("DELETE", f"{view_path}/view-a"): + return {} + if (method, path) == ("DELETE", pyiceberg_smoke.namespace_endpoint_path(args, probe_namespace)): + return {} + raise AssertionError(f"unexpected REST request: {method} {path}") + + with mock.patch.object(pyiceberg_smoke.uuid, "uuid4", return_value=SimpleNamespace(hex="1234567890abcdef")): + with mock.patch.object(pyiceberg_smoke, "signed_rest_request", side_effect=fake_signed_request): + with mock.patch.object(pyiceberg_smoke, "paginated_identifier_names", return_value=["view-a", "view-b"]): + with self.assertRaisesRegex(RuntimeError, "delete failed"): + pyiceberg_smoke.run_view_probe(args, mock.Mock()) + + self.assertIn(("DELETE", f"{view_path}/view-b", None), calls) + self.assertIn(("DELETE", f"{view_path}/view-a", None), calls) + self.assertIn(("DELETE", pyiceberg_smoke.namespace_endpoint_path(args, probe_namespace), None), calls) + + def test_paginated_identifier_names_follows_tokens_until_null(self) -> None: + args = self.parse_with_args(["--namespace", "sales", "--table", "orders"]) + view_path = pyiceberg_smoke.view_endpoint_path(args) + responses = [ + {"identifiers": [], "next-page-token": "token-0"}, + {"identifiers": [{"name": "orders_a"}], "next-page-token": "token-1"}, + {"identifiers": [{"name": "orders_b"}], "next-page-token": None}, + ] + + with mock.patch.object(pyiceberg_smoke, "signed_rest_request", side_effect=responses) as request: + names = pyiceberg_smoke.paginated_identifier_names(args, mock.Mock(), view_path, page_size=1) + + self.assertEqual(names, ["orders_a", "orders_b"]) + self.assertEqual(request.call_count, 3) + self.assertTrue(request.call_args_list[0].args[3].endswith("?pageSize=1")) + self.assertNotIn("pageToken", request.call_args_list[0].args[3]) + self.assertIn("pageSize=1", request.call_args_list[1].args[3]) + self.assertIn("pageToken=token-0", request.call_args_list[1].args[3]) + self.assertIn("pageSize=1", request.call_args_list[2].args[3]) + self.assertIn("pageToken=token-1", request.call_args_list[2].args[3]) + + def test_paginated_identifier_names_rejects_invalid_sequences(self) -> None: + args = self.parse_with_args(["--namespace", "sales", "--table", "orders"]) + view_path = pyiceberg_smoke.view_endpoint_path(args) + scenarios = [ + ([{"identifiers": [], "next-page-token": "token-1"}, {"identifiers": [], "next-page-token": "token-1"}], "invalid next-page-token"), + ([{"identifiers": [{"name": "orders"}], "next-page-token": "token-1"}, {"identifiers": [{"name": "orders"}], "next-page-token": None}], "duplicate identifier"), + ([{"identifiers": []}], "omitted next-page-token"), + ([{"identifiers": [{"name": "orders_a"}, {"name": "orders_b"}], "next-page-token": None}], "exceeded pageSize"), + ] + + for responses, message in scenarios: + with self.subTest(message=message): + with mock.patch.object(pyiceberg_smoke, "signed_rest_request", side_effect=responses): + with self.assertRaisesRegex(RuntimeError, message): + pyiceberg_smoke.paginated_identifier_names(args, mock.Mock(), view_path, page_size=1) def test_maintenance_probe_rejects_unknown_worker_status(self) -> None: args = self.parse_with_args(["--namespace", "sales", "--table", "orders"]) @@ -364,7 +490,6 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): metadata_location_path = pyiceberg_smoke.table_endpoint_path(args, "/metadata-location") refs_path = pyiceberg_smoke.table_ref_endpoint_path(args) ref_name = pyiceberg_smoke.safe_ref_segment(args.namespace, args.table) - view_path = pyiceberg_smoke.view_endpoint_path(args) config_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/config") maintenance_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/metadata") quarantine_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/jobs/job-1/quarantine") @@ -393,14 +518,6 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): return {"refs": {}} if method == "DELETE" and path.startswith(f"{refs_path}/"): return {} - if (method, path) == ("POST", view_path): - return {} - if (method, path) == ("GET", view_path): - return {"identifiers": [{"name": "orders_smoke_view"}]} - if (method, path) == ("GET", f"{view_path}/orders_smoke_view"): - return {"metadata-location": "s3://lake/views/orders_smoke_view/metadata/v1.json"} - if (method, path) == ("DELETE", f"{view_path}/orders_smoke_view"): - return {} if (method, path) == ("PUT", config_path): return {} if (method, path) == ("GET", config_path): @@ -425,18 +542,20 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): raise AssertionError(f"unexpected REST request: {method} {path}") with mock.patch.object(pyiceberg_smoke, "signed_rest_request", side_effect=fake_signed_request): - with mock.patch.object( - pyiceberg_smoke, - "signed_rest_request_expect_error", - return_value=pyiceberg_smoke.RestRequestError( - "DELETE", - f"{refs_path}/{ref_name}", - 400, - "snapshot ref has retention policy; force is required", - ), - ) as expect_error: - pyiceberg_smoke.run_catalog_api_probes(args, deps) + with mock.patch.object(pyiceberg_smoke, "run_view_probe") as view_probe: + with mock.patch.object( + pyiceberg_smoke, + "signed_rest_request_expect_error", + return_value=pyiceberg_smoke.RestRequestError( + "DELETE", + f"{refs_path}/{ref_name}", + 400, + "snapshot ref has retention policy; force is required", + ), + ) as expect_error: + pyiceberg_smoke.run_catalog_api_probes(args, deps) + view_probe.assert_called_once_with(args, deps) expect_error.assert_called_once_with( args, deps,