gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 08:18:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(table-catalog): wire durable strong backing config (#3971)

Browse Source 
* feat(table-catalog): wire durable strong backing config

* fix(table-catalog): guard strong backing warehouse prefixes

* fix(checksums): remove unused base64 decode helper

---------

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
This commit is contained in:
Henry Guo
2026-06-29 07:58:14 +08:00
committed by GitHub
parent 51d44f20c7
commit cf7af17b43
5 changed files with 652 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/checksums/src/base64.rs
-20
View File
@@ -14,26 +14,6 @@
#![allow(dead_code)]
use base64_simd::STANDARD;
use std::error::Error;
#[derive(Debug)]
pub(crate) struct DecodeError(base64_simd::Error);
impl Error for DecodeError {
fn source(&self) -> Option<&(dyn Error + 'static)> {
Some(&self.0)
}
}
impl std::fmt::Display for DecodeError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "failed to decode base64")
}
}
pub(crate) fn decode(input: impl AsRef<str>) -> Result<Vec<u8>, DecodeError> {
STANDARD.decode_to_vec(input.as_ref()).map_err(DecodeError)
}
pub(crate) fn encode(input: impl AsRef<[u8]>) -> String {
STANDARD.encode_to_string(input.as_ref())
crates/checksums/src/lib.rs
+4 -3
View File
@@ -338,8 +338,7 @@ mod tests {
use crate::ChecksumAlgorithm;
use crate::http::HttpChecksum;
use crate::base64;
use base64_simd::STANDARD;
use http::HeaderValue;
use pretty_assertions::assert_eq;
use std::fmt::Write;
@@ -347,7 +346,9 @@ mod tests {
const TEST_DATA: &str = r#"test data"#;
fn base64_encoded_checksum_to_hex_string(header_value: &HeaderValue) -> String {
let decoded_checksum = base64::decode(header_value.to_str().unwrap()).unwrap();
let decoded_checksum = STANDARD
.decode_to_vec(header_value.to_str().expect("checksum header value should be ASCII"))
.expect("checksum header value should be valid base64");
let decoded_checksum = decoded_checksum.into_iter().fold(String::new(), |mut acc, byte| {
write!(acc, "{byte:02X?}").expect("string will always be writable");
acc
rustfs/src/admin/handlers/table_catalog.rs
+79 -26
View File
@@ -52,6 +52,7 @@ const MAX_TABLE_CATALOG_CREDENTIAL_TTL_SECONDS: i64 = 60 * 60;
const WAREHOUSE_PROPERTY: &str = "warehouse";
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";
const CREDENTIAL_VENDING_CONFIG_KEY: &str = "rustfs.credential-vending";
const CREDENTIAL_VENDING_REASON_CONFIG_KEY: &str = "rustfs.credential-vending-reason";
const CREDENTIAL_SCOPE_CONFIG_KEY: &str = "rustfs.credential-scope";
@@ -988,22 +989,28 @@ fn register_table_catalog_prefix_routes(r: &mut S3Router<AdminOperation>, prefix
Ok(())
}
fn catalog_config_response() -> CatalogConfigResponse {
fn catalog_config_response() -> S3Result<CatalogConfigResponse> {
let usecase = default_admin_usecase();
CatalogConfigResponse {
let backing_mode = crate::table_catalog::TableCatalogBackingMode::from_env().map_err(catalog_store_error)?;
let mut overrides = BTreeMap::new();
if backing_mode != crate::table_catalog::TableCatalogBackingMode::ObjectBacked {
overrides.insert(CATALOG_BACKING_CONFIG_KEY, backing_mode.as_str());
}
Ok(CatalogConfigResponse {
defaults: BTreeMap::from([
(WAREHOUSE_PROPERTY, DEFAULT_WAREHOUSE_ID),
(CATALOG_ENDPOINT_PREFIX_CONFIG_KEY, TABLE_CATALOG_PREFIX),
(CATALOG_COMPAT_ENDPOINT_PREFIX_CONFIG_KEY, TABLE_CATALOG_COMPAT_PREFIX),
(CATALOG_BACKING_CONFIG_KEY, crate::table_catalog::TABLE_CATALOG_BACKING_OBJECT),
]),
overrides: BTreeMap::new(),
overrides,
endpoints: TABLE_CATALOG_ENDPOINTS.to_vec(),
admin_discovery: CatalogAdminDiscovery {
runtime_capabilities: usecase.runtime_capabilities_route(),
cluster_snapshot: usecase.cluster_snapshot_route(),
extensions_catalog: usecase.extensions_catalog_route(),
},
}
})
}
fn build_json_response<T: Serialize>(status: StatusCode, body: &T) -> S3Result<S3Response<(StatusCode, Body)>> {
@@ -1262,9 +1269,32 @@ fn table_catalog_backend() -> S3Result<crate::table_catalog::EcStoreTableCatalog
Ok(crate::table_catalog::EcStoreTableCatalogObjectBackend::new(store))
}
type EcStoreObjectTableCatalogStore =
crate::table_catalog::ObjectTableCatalogStore<crate::table_catalog::EcStoreTableCatalogObjectBackend<ECStore>>;
fn table_catalog_store_from_backend(
backend: crate::table_catalog::EcStoreTableCatalogObjectBackend<ECStore>,
) -> S3Result<crate::table_catalog::EcStoreTableCatalogStore<ECStore>> {
crate::table_catalog::ConfiguredTableCatalogStore::from_env(backend).map_err(catalog_store_error)
}
fn table_catalog_store() -> S3Result<crate::table_catalog::EcStoreTableCatalogStore<ECStore>> {
let backend = table_catalog_backend()?;
Ok(crate::table_catalog::ObjectTableCatalogStore::new(backend))
table_catalog_store_from_backend(backend)
}
fn table_catalog_object_store() -> S3Result<EcStoreObjectTableCatalogStore> {
match crate::table_catalog::TableCatalogBackingMode::from_env().map_err(catalog_store_error)? {
crate::table_catalog::TableCatalogBackingMode::ObjectBacked => {
let backend = table_catalog_backend()?;
Ok(crate::table_catalog::ObjectTableCatalogStore::new(backend))
}
crate::table_catalog::TableCatalogBackingMode::DurableStrong => Err(s3_error!(
InvalidRequest,
"operation is not supported with {} table catalog backing",
crate::table_catalog::TABLE_CATALOG_BACKING_DURABLE_STRONG
)),
}
}
async fn table_bucket_enabled_from_metadata(bucket: &str) -> S3Result<bool> {
@@ -4501,8 +4531,8 @@ where
})
}
async fn catalog_import_response<B>(
store: &crate::table_catalog::ObjectTableCatalogStore<B>,
async fn catalog_import_response<S, B>(
store: &S,
metadata_backend: &B,
bucket: &str,
namespace: &crate::table_catalog::Namespace,
@@ -4511,6 +4541,7 @@ async fn catalog_import_response<B>(
table_bucket_enabled: bool,
) -> S3Result<RestLoadTableResponse>
where
S: crate::table_catalog::TableCatalogStore + ?Sized,
B: crate::table_catalog::TableCatalogObjectBackend,
{
let started = Instant::now();
@@ -4601,7 +4632,7 @@ pub struct GetCatalogConfigHandler {}
impl Operation for GetCatalogConfigHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
authorize_table_catalog_request(&req, AdminAction::GetTableCatalogAction).await?;
build_json_response(StatusCode::OK, &catalog_config_response())
build_json_response(StatusCode::OK, &catalog_config_response()?)
}
}
@@ -4739,7 +4770,7 @@ impl Operation for RestCreateTableHandler {
authorize_table_catalog_resource_request(&req, &resource, AdminAction::CreateTableAction).await?;
let request = read_json_body::<CreateTableRequest>(req.input).await?;
let metadata_backend = table_catalog_backend()?;
let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone());
let store = table_catalog_store_from_backend(metadata_backend.clone())?;
let table_bucket_enabled = table_bucket_enabled_from_metadata(&warehouse).await?;
let response =
create_table_response(&store, &metadata_backend, &warehouse, &namespace, request, table_bucket_enabled).await?;
@@ -4758,7 +4789,7 @@ impl Operation for RestRegisterTableHandler {
authorize_table_catalog_resource_request(&req, &resource, AdminAction::RegisterTableAction).await?;
let request = read_json_body::<RegisterTableRequest>(req.input).await?;
let metadata_backend = table_catalog_backend()?;
let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone());
let store = table_catalog_store_from_backend(metadata_backend.clone())?;
let table_bucket_enabled = table_bucket_enabled_from_metadata(&warehouse).await?;
let response =
register_table_response(&store, &metadata_backend, &warehouse, &namespace, request, table_bucket_enabled).await?;
@@ -4793,7 +4824,7 @@ impl Operation for RestCreateViewHandler {
authorize_table_catalog_resource_request(&req, &resource, AdminAction::CreateTableAction).await?;
let request = read_json_body::<CreateViewRequest>(req.input).await?;
let metadata_backend = table_catalog_backend()?;
let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone());
let store = table_catalog_store_from_backend(metadata_backend.clone())?;
let table_bucket_enabled = table_bucket_enabled_from_metadata(&warehouse).await?;
let response =
create_view_response(&store, &metadata_backend, &warehouse, &namespace, request, table_bucket_enabled).await?;
@@ -4813,7 +4844,7 @@ impl Operation for RestLoadTableHandler {
authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableMetadataAction).await?;
ensure_table_bucket_enabled(&warehouse).await?;
let metadata_backend = table_catalog_backend()?;
let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone());
let store = table_catalog_store_from_backend(metadata_backend.clone())?;
let response = load_table_response(&store, &metadata_backend, &warehouse, &namespace, &table).await?;
build_json_response(StatusCode::OK, &response)
}
@@ -4867,7 +4898,7 @@ impl Operation for RestCommitTableHandler {
ensure_table_bucket_enabled(&warehouse).await?;
let request = read_json_body::<RestCommitTableRequest>(req.input).await?;
let metadata_backend = table_catalog_backend()?;
let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone());
let store = table_catalog_store_from_backend(metadata_backend.clone())?;
let response = commit_table_response(&store, &metadata_backend, &warehouse, &namespace, &table, request).await?;
build_json_response(StatusCode::OK, &response)
}
@@ -4902,7 +4933,7 @@ impl Operation for RestLoadViewHandler {
authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableMetadataAction).await?;
ensure_table_bucket_enabled(&warehouse).await?;
let metadata_backend = table_catalog_backend()?;
let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone());
let store = table_catalog_store_from_backend(metadata_backend.clone())?;
let response = load_view_response(&store, &metadata_backend, &warehouse, &namespace, &view).await?;
build_json_response(StatusCode::OK, &response)
}
@@ -4937,7 +4968,7 @@ impl Operation for RestReplaceViewHandler {
ensure_table_bucket_enabled(&warehouse).await?;
let request = read_json_body::<RestCommitViewRequest>(req.input).await?;
let metadata_backend = table_catalog_backend()?;
let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone());
let store = table_catalog_store_from_backend(metadata_backend.clone())?;
let response = replace_view_response(&store, &metadata_backend, &warehouse, &namespace, &view, request).await?;
build_json_response(StatusCode::OK, &response)
}
@@ -4972,7 +5003,7 @@ impl Operation for ListTableRefsHandler {
authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableMetadataAction).await?;
ensure_table_bucket_enabled(&warehouse).await?;
let metadata_backend = table_catalog_backend()?;
let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone());
let store = table_catalog_store_from_backend(metadata_backend.clone())?;
let response = table_refs_response(&store, &metadata_backend, &warehouse, &namespace, &table).await?;
build_json_response(StatusCode::OK, &response)
}
@@ -4992,7 +5023,7 @@ impl Operation for PutTableRefHandler {
ensure_table_bucket_enabled(&warehouse).await?;
let request = read_json_body::<PutTableRefRequest>(req.input).await?;
let metadata_backend = table_catalog_backend()?;
let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone());
let store = table_catalog_store_from_backend(metadata_backend.clone())?;
let response =
put_table_ref_response(&store, &metadata_backend, &warehouse, &namespace, &table, &ref_name, request).await?;
build_json_response(StatusCode::OK, &response)
@@ -5013,7 +5044,7 @@ impl Operation for DeleteTableRefHandler {
ensure_table_bucket_enabled(&warehouse).await?;
let request = read_json_body_or_default::<DeleteTableRefRequest>(req.input).await?;
let metadata_backend = table_catalog_backend()?;
let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone());
let store = table_catalog_store_from_backend(metadata_backend.clone())?;
let response =
delete_table_ref_response(&store, &metadata_backend, &warehouse, &namespace, &table, &ref_name, request).await?;
build_json_response(StatusCode::OK, &response)
@@ -5050,7 +5081,7 @@ impl Operation for UpdateTableMetadataLocationHandler {
ensure_table_bucket_enabled(&warehouse).await?;
let request = read_json_body::<UpdateTableMetadataLocationRequest>(req.input).await?;
let metadata_backend = table_catalog_backend()?;
let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone());
let store = table_catalog_store_from_backend(metadata_backend.clone())?;
let response =
update_table_metadata_location_response(&store, &metadata_backend, &warehouse, &namespace, &table, request).await?;
build_json_response(StatusCode::OK, &response)
@@ -5070,7 +5101,7 @@ impl Operation for RestTableMetadataMaintenanceHandler {
ensure_table_bucket_enabled(&warehouse).await?;
let request = read_json_body::<TableMetadataMaintenanceRequest>(req.input).await?;
let metadata_backend = table_catalog_backend()?;
let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone());
let store = table_catalog_object_store()?;
let response =
table_metadata_maintenance_response(&store, &metadata_backend, &warehouse, &namespace, &table, request).await?;
build_json_response(StatusCode::OK, &response)
@@ -5232,7 +5263,7 @@ impl Operation for ImportTableCatalogHandler {
authorize_table_catalog_resource_request(&req, &resource, AdminAction::RegisterTableAction).await?;
let request = read_json_body::<CatalogImportRequest>(req.input).await?;
let metadata_backend = table_catalog_backend()?;
let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone());
let store = table_catalog_store_from_backend(metadata_backend.clone())?;
let table_bucket_enabled = table_bucket_enabled_from_metadata(&warehouse).await?;
let response =
catalog_import_response(&store, &metadata_backend, &warehouse, &namespace, &table, request, table_bucket_enabled)
@@ -5252,7 +5283,7 @@ impl Operation for ExternalCatalogBridgeHandler {
let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableMetadataAction).await?;
ensure_table_bucket_enabled(&warehouse).await?;
let store = table_catalog_store()?;
let store = table_catalog_object_store()?;
let response = external_catalog_bridge_response(&store, &warehouse, &namespace, &table).await?;
build_json_response(StatusCode::OK, &response)
}
@@ -5270,7 +5301,7 @@ impl Operation for PutExternalCatalogBridgeHandler {
authorize_table_catalog_resource_request(&req, &resource, AdminAction::RegisterTableAction).await?;
ensure_table_bucket_enabled(&warehouse).await?;
let request = read_json_body::<ExternalCatalogBridgeRequest>(req.input).await?;
let store = table_catalog_store()?;
let store = table_catalog_object_store()?;
let response = put_external_catalog_bridge_response(&store, &warehouse, &namespace, &table, request).await?;
build_json_response(StatusCode::OK, &response)
}
@@ -5288,7 +5319,7 @@ impl Operation for SyncExternalCatalogBridgeHandler {
authorize_table_catalog_resource_request(&req, &resource, AdminAction::SetTableMetadataLocationAction).await?;
ensure_table_bucket_enabled(&warehouse).await?;
let metadata_backend = table_catalog_backend()?;
let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone());
let store = table_catalog_object_store()?;
if store
.load_table(&warehouse, &namespace.public_name(), &table)
.await
@@ -5383,7 +5414,7 @@ impl Operation for RollbackTableCatalogHandler {
ensure_table_bucket_enabled(&warehouse).await?;
let request = read_json_body::<RollbackTableRequest>(req.input).await?;
let metadata_backend = table_catalog_backend()?;
let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone());
let store = table_catalog_store_from_backend(metadata_backend.clone())?;
let response = rollback_table_response(&store, &metadata_backend, &warehouse, &namespace, &table, request).await?;
build_json_response(StatusCode::OK, &response)
}
@@ -5396,8 +5427,10 @@ mod tests {
use std::sync::Arc;
#[test]
#[serial_test::serial]
fn catalog_config_response_lists_standard_rest_endpoints() {
let response = catalog_config_response();
let response = temp_env::with_var_unset(crate::table_catalog::ENV_TABLE_CATALOG_BACKING, catalog_config_response)
.expect("catalog config should build");
assert_eq!(response.defaults.get(WAREHOUSE_PROPERTY), Some(&DEFAULT_WAREHOUSE_ID));
assert_eq!(response.defaults.get(CATALOG_ENDPOINT_PREFIX_CONFIG_KEY), Some(&TABLE_CATALOG_PREFIX));
@@ -5405,6 +5438,10 @@ mod tests {
response.defaults.get(CATALOG_COMPAT_ENDPOINT_PREFIX_CONFIG_KEY),
Some(&TABLE_CATALOG_COMPAT_PREFIX)
);
assert_eq!(
response.defaults.get(CATALOG_BACKING_CONFIG_KEY),
Some(&crate::table_catalog::TABLE_CATALOG_BACKING_OBJECT)
);
assert!(response.overrides.is_empty());
assert_eq!(response.admin_discovery.runtime_capabilities, "/rustfs/admin/v4/runtime/capabilities");
assert_eq!(response.admin_discovery.cluster_snapshot, "/rustfs/admin/v4/cluster/snapshot");
@@ -5508,6 +5545,22 @@ mod tests {
);
}
#[test]
#[serial_test::serial]
fn catalog_config_response_reports_durable_strong_backing_override() {
let response = temp_env::with_var(
crate::table_catalog::ENV_TABLE_CATALOG_BACKING,
Some(crate::table_catalog::TABLE_CATALOG_BACKING_DURABLE_STRONG),
catalog_config_response,
)
.expect("catalog config should build");
assert_eq!(
response.overrides.get(CATALOG_BACKING_CONFIG_KEY),
Some(&crate::table_catalog::TABLE_CATALOG_BACKING_DURABLE_STRONG)
);
}
#[test]
fn table_catalog_admin_operation_result_labels_are_stable() {
let success: Result<(), ()> = Ok(());
rustfs/src/storage/access.rs
+2 -1
View File
@@ -787,7 +787,8 @@ fn table_catalog_store_for_data_plane() -> S3Result<crate::table_catalog::EcStor
let store =
runtime_sources::current_object_store_handle().ok_or_else(|| s3_error!(InternalError, "object store not initialized"))?;
let backend = crate::table_catalog::EcStoreTableCatalogObjectBackend::new(store);
Ok(crate::table_catalog::ObjectTableCatalogStore::new(backend))
crate::table_catalog::ConfiguredTableCatalogStore::from_env(backend)
.map_err(|err| s3_error!(InternalError, "failed to configure table catalog backing: {}", err))
}
async fn table_data_plane_resource_for_request(
rustfs/src/table_catalog.rs
+567 -1
View File
@@ -70,6 +70,9 @@ pub(crate) const TABLE_CATALOG_ENTRY_VERSION: u16 = 1;
pub(crate) const TABLE_MAINTENANCE_CONFIG_VERSION: u16 = 1;
pub(crate) const TABLE_EXTERNAL_CATALOG_BRIDGE_VERSION: u16 = 1;
pub(crate) const TABLE_CATALOG_BACKING_MANIFEST_VERSION: u16 = 1;
pub(crate) const ENV_TABLE_CATALOG_BACKING: &str = "RUSTFS_TABLE_CATALOG_BACKING";
pub(crate) const TABLE_CATALOG_BACKING_OBJECT: &str = "object";
pub(crate) const TABLE_CATALOG_BACKING_DURABLE_STRONG: &str = "durable-strong";
pub(crate) const TABLE_METADATA_FILE_NAME_MAX_LEN: usize = 128;
pub const TABLE_RESERVED_PREFIX: &str = BUCKET_TABLE_RESERVED_PREFIX;
const WAREHOUSE_ROOT: &str = "warehouses";
@@ -1187,6 +1190,41 @@ impl std::error::Error for TableCatalogStoreError {}
pub(crate) type TableCatalogStoreResult<T> = Result<T, TableCatalogStoreError>;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum TableCatalogBackingMode {
ObjectBacked,
DurableStrong,
}
impl TableCatalogBackingMode {
pub(crate) fn from_env() -> TableCatalogStoreResult<Self> {
match std::env::var(ENV_TABLE_CATALOG_BACKING) {
Ok(value) => Self::parse(&value),
Err(std::env::VarError::NotPresent) => Ok(Self::ObjectBacked),
Err(std::env::VarError::NotUnicode(_)) => Err(TableCatalogStoreError::Invalid(format!(
"{ENV_TABLE_CATALOG_BACKING} must be valid UTF-8"
))),
}
}
fn parse(value: &str) -> TableCatalogStoreResult<Self> {
match value.trim() {
"" | TABLE_CATALOG_BACKING_OBJECT => Ok(Self::ObjectBacked),
TABLE_CATALOG_BACKING_DURABLE_STRONG => Ok(Self::DurableStrong),
value => Err(TableCatalogStoreError::Invalid(format!(
"unsupported table catalog backing {value}; expected {TABLE_CATALOG_BACKING_OBJECT} or {TABLE_CATALOG_BACKING_DURABLE_STRONG}"
))),
}
}
pub(crate) fn as_str(self) -> &'static str {
match self {
Self::ObjectBacked => TABLE_CATALOG_BACKING_OBJECT,
Self::DurableStrong => TABLE_CATALOG_BACKING_DURABLE_STRONG,
}
}
}
fn normalize_warehouse_object_prefix(object_prefix: &str, max_prefix_depth: Option<usize>) -> TableCatalogStoreResult<String> {
let object_prefix = object_prefix.strip_suffix('/').unwrap_or(object_prefix);
if object_prefix.is_empty() {
@@ -2041,6 +2079,34 @@ where
Ok(())
}
fn ensure_table_warehouse_prefix_available_locked(
state: &StrongTableCatalogState,
candidate: &TableEntry,
candidate_key: &StrongResourceKey,
) -> TableCatalogStoreResult<()> {
if candidate.state != TableCatalogEntryState::Active {
return Ok(());
}
let candidate_prefix = table_warehouse_object_prefix(candidate)?;
for (existing_key, existing) in &state.tables {
if existing_key == candidate_key
|| existing.table_bucket != candidate.table_bucket
|| existing.state != TableCatalogEntryState::Active
{
continue;
}
let Ok(existing_prefix) = table_warehouse_object_prefix(existing) else {
continue;
};
if existing_prefix == candidate_prefix {
return Err(TableCatalogStoreError::Conflict(format!(
"table warehouse location is already registered: {candidate_prefix}"
)));
}
}
Ok(())
}
fn table_commit_recovery_report_for_entry_locked(
state: &StrongTableCatalogState,
entry: &TableEntry,
@@ -2225,6 +2291,7 @@ where
if let Some(warehouse_location) = next_warehouse_location {
next.warehouse_location = warehouse_location;
}
Self::ensure_table_warehouse_prefix_available_locked(state, &next, &key)?;
next.version_token = commit_log.new_version_token.clone();
next.generation = next.generation.saturating_add(1);
@@ -2264,6 +2331,44 @@ where
}
}
#[derive(Clone)]
pub(crate) enum ConfiguredTableCatalogStore<B>
where
B: TableCatalogObjectBackend,
{
ObjectBacked(ObjectTableCatalogStore<B>),
DurableStrong(StrongTableCatalogStore<B>),
}
impl<B> ConfiguredTableCatalogStore<B>
where
B: TableCatalogObjectBackend,
{
pub(crate) fn from_env(backend: B) -> TableCatalogStoreResult<Self> {
Ok(Self::new(backend, TableCatalogBackingMode::from_env()?))
}
pub(crate) fn new(backend: B, mode: TableCatalogBackingMode) -> Self {
match mode {
TableCatalogBackingMode::ObjectBacked => Self::ObjectBacked(ObjectTableCatalogStore::new(backend)),
TableCatalogBackingMode::DurableStrong => Self::DurableStrong(StrongTableCatalogStore::new(backend)),
}
}
pub(crate) fn backing_mode(&self) -> TableCatalogBackingMode {
match self {
Self::ObjectBacked(_) => TableCatalogBackingMode::ObjectBacked,
Self::DurableStrong(_) => TableCatalogBackingMode::DurableStrong,
}
}
fn unsupported_for_durable_strong(operation: &str) -> TableCatalogStoreError {
TableCatalogStoreError::Invalid(format!(
"{operation} is not supported with {TABLE_CATALOG_BACKING_DURABLE_STRONG} table catalog backing"
))
}
}
#[async_trait::async_trait]
impl<B> TableCatalogStore for StrongTableCatalogStore<B>
where
@@ -2403,6 +2508,7 @@ where
entry.table_bucket, entry.namespace, entry.table
)));
}
Self::ensure_table_warehouse_prefix_available_locked(&state, &entry, &key)?;
let (precondition, rollback_state) = Self::snapshot_write_context_locked(&state);
state.tables.insert(key, entry);
(Self::snapshot_from_state_locked(&state), precondition, rollback_state)
@@ -5454,6 +5560,324 @@ where
}
}
#[async_trait::async_trait]
impl<B> TableCatalogStore for ConfiguredTableCatalogStore<B>
where
B: TableCatalogObjectBackend,
{
async fn get_table_bucket(&self, table_bucket: &str) -> TableCatalogStoreResult<Option<TableBucketEntry>> {
match self {
Self::ObjectBacked(store) => store.get_table_bucket(table_bucket).await,
Self::DurableStrong(store) => store.get_table_bucket(table_bucket).await,
}
}
async fn put_table_bucket(&self, entry: TableBucketEntry) -> TableCatalogStoreResult<()> {
match self {
Self::ObjectBacked(store) => store.put_table_bucket(entry).await,
Self::DurableStrong(store) => store.put_table_bucket(entry).await,
}
}
async fn create_namespace(&self, entry: NamespaceEntry) -> TableCatalogStoreResult<()> {
match self {
Self::ObjectBacked(store) => store.create_namespace(entry).await,
Self::DurableStrong(store) => store.create_namespace(entry).await,
}
}
async fn list_namespaces(&self, table_bucket: &str) -> TableCatalogStoreResult<Vec<NamespaceEntry>> {
match self {
Self::ObjectBacked(store) => store.list_namespaces(table_bucket).await,
Self::DurableStrong(store) => store.list_namespaces(table_bucket).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,
Self::DurableStrong(store) => store.get_namespace(table_bucket, namespace).await,
}
}
async fn drop_namespace(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult<()> {
match self {
Self::ObjectBacked(store) => store.drop_namespace(table_bucket, namespace).await,
Self::DurableStrong(store) => store.drop_namespace(table_bucket, namespace).await,
}
}
async fn create_table(&self, entry: TableEntry) -> TableCatalogStoreResult<()> {
match self {
Self::ObjectBacked(store) => store.create_table(entry).await,
Self::DurableStrong(store) => store.create_table(entry).await,
}
}
async fn register_table(&self, entry: TableEntry) -> TableCatalogStoreResult<()> {
match self {
Self::ObjectBacked(store) => store.register_table(entry).await,
Self::DurableStrong(store) => store.register_table(entry).await,
}
}
async fn list_tables(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult<Vec<TableEntry>> {
match self {
Self::ObjectBacked(store) => store.list_tables(table_bucket, namespace).await,
Self::DurableStrong(store) => store.list_tables(table_bucket, namespace).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,
Self::DurableStrong(store) => store.load_table(table_bucket, namespace, table).await,
}
}
async fn resolve_table_data_plane_resource(
&self,
table_bucket: &str,
object: &str,
) -> TableCatalogStoreResult<Option<TableDataPlaneResource>> {
match self {
Self::ObjectBacked(store) => store.resolve_table_data_plane_resource(table_bucket, object).await,
Self::DurableStrong(store) => store.resolve_table_data_plane_resource(table_bucket, object).await,
}
}
async fn commit_table(&self, request: TableCommitRequest) -> TableCatalogStoreResult<TableCommitResult> {
match self {
Self::ObjectBacked(store) => store.commit_table(request).await,
Self::DurableStrong(store) => store.commit_table(request).await,
}
}
async fn drop_table(&self, table_bucket: &str, namespace: &str, table: &str) -> TableCatalogStoreResult<()> {
match self {
Self::ObjectBacked(store) => store.drop_table(table_bucket, namespace, table).await,
Self::DurableStrong(store) => store.drop_table(table_bucket, namespace, table).await,
}
}
async fn create_view(&self, entry: ViewEntry) -> TableCatalogStoreResult<()> {
match self {
Self::ObjectBacked(store) => store.create_view(entry).await,
Self::DurableStrong(store) => store.create_view(entry).await,
}
}
async fn list_views(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult<Vec<ViewEntry>> {
match self {
Self::ObjectBacked(store) => store.list_views(table_bucket, namespace).await,
Self::DurableStrong(store) => store.list_views(table_bucket, namespace).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,
Self::DurableStrong(store) => store.load_view(table_bucket, namespace, view).await,
}
}
async fn replace_view(&self, request: ViewCommitRequest) -> TableCatalogStoreResult<ViewCommitResult> {
match self {
Self::ObjectBacked(store) => store.replace_view(request).await,
Self::DurableStrong(store) => store.replace_view(request).await,
}
}
async fn drop_view(&self, table_bucket: &str, namespace: &str, view: &str) -> TableCatalogStoreResult<()> {
match self {
Self::ObjectBacked(store) => store.drop_view(table_bucket, namespace, view).await,
Self::DurableStrong(store) => store.drop_view(table_bucket, namespace, view).await,
}
}
async fn get_commit_by_id(
&self,
table_bucket: &str,
table_id: &str,
commit_id: &str,
) -> TableCatalogStoreResult<Option<CommitLogEntry>> {
match self {
Self::ObjectBacked(store) => store.get_commit_by_id(table_bucket, table_id, commit_id).await,
Self::DurableStrong(store) => store.get_commit_by_id(table_bucket, table_id, commit_id).await,
}
}
async fn get_commit_by_idempotency_key(
&self,
table_bucket: &str,
table_id: &str,
idempotency_key: &str,
) -> TableCatalogStoreResult<Option<CommitLogEntry>> {
match self {
Self::ObjectBacked(store) => {
store
.get_commit_by_idempotency_key(table_bucket, table_id, idempotency_key)
.await
}
Self::DurableStrong(store) => {
store
.get_commit_by_idempotency_key(table_bucket, table_id, idempotency_key)
.await
}
}
}
}
impl<B> ConfiguredTableCatalogStore<B>
where
B: TableCatalogObjectBackend,
{
pub(crate) async fn get_table_maintenance_config(
&self,
table_bucket: &str,
namespace: &str,
table: &str,
) -> TableCatalogStoreResult<TableMaintenanceConfig> {
match self {
Self::ObjectBacked(store) => store.get_table_maintenance_config(table_bucket, namespace, table).await,
Self::DurableStrong(_) => Err(Self::unsupported_for_durable_strong("table maintenance config")),
}
}
pub(crate) async fn put_table_maintenance_config(
&self,
table_bucket: &str,
namespace: &str,
table: &str,
config: TableMaintenanceConfig,
) -> TableCatalogStoreResult<TableMaintenanceConfig> {
match self {
Self::ObjectBacked(store) => {
store
.put_table_maintenance_config(table_bucket, namespace, table, config)
.await
}
Self::DurableStrong(_) => Err(Self::unsupported_for_durable_strong("table maintenance config")),
}
}
pub(crate) async fn get_table_metadata_maintenance_report(
&self,
table_bucket: &str,
namespace: &str,
table: &str,
job_id: &str,
) -> TableCatalogStoreResult<Option<TableMetadataMaintenanceReport>> {
match self {
Self::ObjectBacked(store) => {
store
.get_table_metadata_maintenance_report(table_bucket, namespace, table, job_id)
.await
}
Self::DurableStrong(_) => Err(Self::unsupported_for_durable_strong("table maintenance report")),
}
}
pub(crate) async fn run_table_metadata_maintenance_worker_once(
&self,
table_bucket: &str,
namespace: &str,
table: &str,
worker_id: String,
) -> TableCatalogStoreResult<TableMetadataMaintenanceReport> {
match self {
Self::ObjectBacked(store) => {
store
.run_table_metadata_maintenance_worker_once(table_bucket, namespace, table, worker_id)
.await
}
Self::DurableStrong(_) => Err(Self::unsupported_for_durable_strong("table maintenance worker")),
}
}
pub(crate) async fn heartbeat_table_metadata_maintenance_job(
&self,
table_bucket: &str,
namespace: &str,
table: &str,
job_id: &str,
lease_id: &str,
worker_id: &str,
) -> TableCatalogStoreResult<TableMetadataMaintenanceReport> {
match self {
Self::ObjectBacked(store) => {
store
.heartbeat_table_metadata_maintenance_job(table_bucket, namespace, table, job_id, lease_id, worker_id)
.await
}
Self::DurableStrong(_) => Err(Self::unsupported_for_durable_strong("table maintenance heartbeat")),
}
}
pub(crate) async fn export_table_catalog_entry(
&self,
table_bucket: &str,
namespace: &str,
table: &str,
) -> TableCatalogStoreResult<TableCatalogExport> {
match self {
Self::ObjectBacked(store) => store.export_table_catalog_entry(table_bucket, namespace, table).await,
Self::DurableStrong(_) => Err(Self::unsupported_for_durable_strong("catalog export")),
}
}
pub(crate) async fn diagnose_table_catalog(
&self,
table_bucket: &str,
namespace: &str,
table: &str,
retain_recent_metadata_files: usize,
) -> TableCatalogStoreResult<TableCatalogDiagnosticsReport> {
match self {
Self::ObjectBacked(store) => {
store
.diagnose_table_catalog(table_bucket, namespace, table, retain_recent_metadata_files)
.await
}
Self::DurableStrong(_) => Err(Self::unsupported_for_durable_strong("catalog diagnostics")),
}
}
pub(crate) async fn recover_table_commits(
&self,
table_bucket: &str,
namespace: &str,
table: &str,
) -> TableCatalogStoreResult<TableCommitRecoveryReport> {
match self {
Self::ObjectBacked(store) => store.recover_table_commits(table_bucket, namespace, table).await,
Self::DurableStrong(store) => store.plan_table_commit_recovery(table_bucket, namespace, table).await,
}
}
pub(crate) async fn get_external_catalog_bridge(
&self,
table_bucket: &str,
namespace: &str,
table: &str,
) -> TableCatalogStoreResult<Option<ExternalCatalogBridgeEntry>> {
match self {
Self::ObjectBacked(store) => store.get_external_catalog_bridge(table_bucket, namespace, table).await,
Self::DurableStrong(_) => Err(Self::unsupported_for_durable_strong("external catalog bridge")),
}
}
pub(crate) async fn put_external_catalog_bridge(
&self,
entry: ExternalCatalogBridgeEntry,
) -> TableCatalogStoreResult<ExternalCatalogBridgeEntry> {
match self {
Self::ObjectBacked(store) => store.put_external_catalog_bridge(entry).await,
Self::DurableStrong(_) => Err(Self::unsupported_for_durable_strong("external catalog bridge")),
}
}
}
pub(crate) struct EcStoreTableCatalogObjectBackend<S> {
store: Arc<S>,
}
@@ -5475,7 +5899,7 @@ where
}
}
pub(crate) type EcStoreTableCatalogStore<S> = ObjectTableCatalogStore<EcStoreTableCatalogObjectBackend<S>>;
pub(crate) type EcStoreTableCatalogStore<S> = ConfiguredTableCatalogStore<EcStoreTableCatalogObjectBackend<S>>;
#[async_trait::async_trait]
impl<S> TableCatalogObjectBackend for EcStoreTableCatalogObjectBackend<S>
@@ -8947,6 +9371,33 @@ mod tests {
assert!(serde_json::from_value::<TableEntry>(value).is_err());
}
#[test]
#[serial_test::serial]
fn table_catalog_backing_mode_defaults_to_object() {
temp_env::with_var_unset(ENV_TABLE_CATALOG_BACKING, || {
assert_eq!(TableCatalogBackingMode::from_env().unwrap(), TableCatalogBackingMode::ObjectBacked);
});
}
#[test]
#[serial_test::serial]
fn table_catalog_backing_mode_accepts_durable_strong_value() {
temp_env::with_var(ENV_TABLE_CATALOG_BACKING, Some(TABLE_CATALOG_BACKING_DURABLE_STRONG), || {
assert_eq!(TableCatalogBackingMode::from_env().unwrap(), TableCatalogBackingMode::DurableStrong);
});
}
#[test]
#[serial_test::serial]
fn table_catalog_backing_mode_rejects_unknown_value() {
temp_env::with_var(ENV_TABLE_CATALOG_BACKING, Some("memory"), || {
assert!(matches!(
TableCatalogBackingMode::from_env().unwrap_err(),
TableCatalogStoreError::Invalid(_)
));
});
}
struct NoopTableCatalogStore;
#[async_trait::async_trait]
@@ -9767,6 +10218,22 @@ mod tests {
assert_eq!(object_buckets, BTreeSet::from([RUSTFS_META_BUCKET]));
}
#[tokio::test]
async fn configured_table_catalog_store_uses_durable_strong_snapshot() {
let backend = TestCatalogObjectBackend::default();
let store = ConfiguredTableCatalogStore::new(backend.clone(), TableCatalogBackingMode::DurableStrong);
let bucket = "analytics";
assert_eq!(store.backing_mode(), TableCatalogBackingMode::DurableStrong);
store.put_table_bucket(test_bucket_entry(bucket)).await.unwrap();
let snapshot_path = StrongTableCatalogStore::<TestCatalogObjectBackend>::snapshot_object_path();
assert!(backend.object_exists(RUSTFS_META_BUCKET, &snapshot_path).await.unwrap());
let reloaded = ConfiguredTableCatalogStore::new(backend.clone(), TableCatalogBackingMode::DurableStrong);
assert!(reloaded.get_table_bucket(bucket).await.unwrap().is_some());
}
#[tokio::test]
async fn object_table_catalog_store_persists_view_entries_and_blocks_non_empty_namespace_drop() {
let backend = TestCatalogObjectBackend::default();
@@ -13607,6 +14074,105 @@ mod tests {
assert!(store.load_table(bucket, "sales", "orders").await.unwrap().is_none());
}
#[tokio::test]
async fn strong_catalog_backing_rejects_duplicate_table_warehouse_location_on_register() {
let backend = TestCatalogObjectBackend::default();
let store = StrongTableCatalogStore::new(backend);
let bucket = "analytics";
let namespace = Namespace::parse("sales").unwrap();
let orders = IdentifierSegment::parse("orders").unwrap();
let customers = IdentifierSegment::parse("customers").unwrap();
let orders_metadata = default_table_metadata_file_path(&namespace, &orders, "00001.metadata.json");
let customers_metadata = default_table_metadata_file_path(&namespace, &customers, "00001.metadata.json");
store.put_table_bucket(test_bucket_entry(bucket)).await.unwrap();
store
.create_namespace(test_namespace_entry(bucket, &namespace))
.await
.unwrap();
store
.create_table(test_table_entry(bucket, &namespace, &orders, orders_metadata))
.await
.unwrap();
let mut duplicate = test_table_entry(bucket, &namespace, &customers, customers_metadata);
duplicate.table_id = "table-id-2".to_string();
duplicate.table_uuid = "table-uuid-2".to_string();
duplicate.warehouse_location = format!("s3://{bucket}/tables/table-id");
let err = store.register_table(duplicate).await.unwrap_err();
assert_matches!(err, TableCatalogStoreError::Conflict(_));
assert!(store.load_table(bucket, "sales", "customers").await.unwrap().is_none());
}
#[tokio::test]
async fn strong_catalog_backing_rejects_duplicate_table_warehouse_location_on_commit_relocation() {
let backend = TestCatalogObjectBackend::default();
let store = StrongTableCatalogStore::new(backend.clone());
let bucket = "analytics";
let namespace = Namespace::parse("sales").unwrap();
let orders = IdentifierSegment::parse("orders").unwrap();
let customers = IdentifierSegment::parse("customers").unwrap();
let orders_metadata = default_table_metadata_file_path(&namespace, &orders, "00001.metadata.json");
let relocated_metadata = default_table_metadata_file_path(&namespace, &orders, "00002.metadata.json");
let customers_metadata = default_table_metadata_file_path(&namespace, &customers, "00001.metadata.json");
store.put_table_bucket(test_bucket_entry(bucket)).await.unwrap();
store
.create_namespace(test_namespace_entry(bucket, &namespace))
.await
.unwrap();
store
.create_table(test_table_entry(bucket, &namespace, &orders, orders_metadata.clone()))
.await
.unwrap();
let mut customers_entry = test_table_entry(bucket, &namespace, &customers, customers_metadata);
customers_entry.table_id = "table-id-2".to_string();
customers_entry.table_uuid = "table-uuid-2".to_string();
customers_entry.warehouse_location = format!("s3://{bucket}/tables/customer-id");
store.create_table(customers_entry).await.unwrap();
backend
.seed_object(
bucket,
&relocated_metadata,
serde_json::to_vec(&serde_json::json!({
"location": "s3://analytics/tables/customer-id",
"table-uuid": "table-uuid"
}))
.unwrap(),
)
.await;
let err = store
.commit_table(TableCommitRequest {
table_bucket: bucket.to_string(),
namespace: namespace.public_name(),
table: orders.as_str().to_string(),
commit_id: "commit-1".to_string(),
idempotency_key: Some("client-request".to_string()),
operation: "set-location".to_string(),
expected_version_token: "token-v1".to_string(),
expected_metadata_location: orders_metadata.clone(),
new_metadata_location: relocated_metadata,
requirements: Vec::new(),
writer: Some("pyiceberg/test".to_string()),
})
.await
.unwrap_err();
assert_matches!(err, TableCatalogStoreError::Conflict(_));
let loaded = store.load_table(bucket, "sales", "orders").await.unwrap().unwrap();
assert_eq!(loaded.metadata_location, orders_metadata);
assert_eq!(loaded.warehouse_location, format!("s3://{bucket}/tables/table-id"));
assert!(
store
.get_commit_by_id(bucket, "table-id", "commit-1")
.await
.unwrap()
.is_none()
);
}
#[tokio::test]
async fn diagnostics_backing_manifest_requires_recovery_before_migration() {
let backend = TestCatalogObjectBackend::default();