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

fix(table-catalog): harden strong backing compatibility (#5941)

Browse Source 
* fix(table-catalog): harden strong backing compatibility

* fix(table-catalog): close strong backing recovery gaps

* fix(table-catalog): harden strong backing recovery

* fix(table-catalog): repair strong backing CI failures

* fix(table-catalog): satisfy test clippy lint

---------

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
This commit is contained in:
Henry Guo
2026-08-12 16:28:56 +08:00
committed by GitHub
parent 493a2cc1ba
commit c7233d6624
15 changed files with 7873 additions and 694 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/admin/handlers/table_catalog/mod.rs
+6 -3
View File
@@ -2067,9 +2067,12 @@ fn job_id_from_params(params: &Params<'_, '_>) -> S3Result<String> {
fn table_catalog_backend_from_extensions(
extensions: &http::Extensions,
) -> S3Result<crate::table_catalog::EcStoreTableCatalogObjectBackend<ECStore>> {
let store = runtime_sources::object_store_from_extensions(extensions)
.ok_or_else(|| table_catalog_internal_error("request object store is not initialized"))?;
Ok(crate::table_catalog::EcStoreTableCatalogObjectBackend::new(store))
let context = runtime_sources::app_context_from_extensions(extensions)
.ok_or_else(|| table_catalog_internal_error("request application context is not initialized"))?;
Ok(crate::table_catalog::EcStoreTableCatalogObjectBackend::new_with_strong_runtime(
context.object_store(),
context.table_catalog_strong_runtime(),
))
}
type EcStoreObjectTableCatalogStore =
rustfs/src/admin/handlers/table_catalog/tests.rs
+2 -2
View File
@@ -2512,7 +2512,7 @@ async fn commit_publication_replays_historical_standard_commit_across_backings()
crate::table_catalog::TableCatalogBackingMode::DurableStrong,
] {
let metadata_backend = TestTableCatalogObjectBackend::default();
let store = crate::table_catalog::ConfiguredTableCatalogStore::new(metadata_backend.clone(), mode);
let store = crate::table_catalog::ConfiguredTableCatalogStore::new_for_test(metadata_backend.clone(), mode);
let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
create_standard_events_table(&store, &metadata_backend, &namespace).await;
let first_request = serde_json::json!({
@@ -2559,7 +2559,7 @@ async fn commit_publication_replays_historical_standard_commit_across_backings()
}
}
let store = crate::table_catalog::ConfiguredTableCatalogStore::new(metadata_backend.clone(), mode);
let store = crate::table_catalog::ConfiguredTableCatalogStore::new_for_test(metadata_backend.clone(), mode);
let second = standard_commit_table_response(
&store,
&trusted_table_commit_backend(&metadata_backend),
rustfs/src/app/context/global.rs
+7
View File
@@ -76,6 +76,7 @@ pub struct AppContext {
buffer_config: Arc<dyn BufferConfigInterface>,
object_data_cache: Arc<ObjectDataCacheAdapter>,
object_traffic_health: Arc<ObjectTrafficHealth>,
table_catalog_strong_runtime: crate::table_catalog::StrongTableCatalogRuntime,
}
impl AppContext {
@@ -125,6 +126,7 @@ impl AppContext {
buffer_config: default_buffer_config_interface(),
object_data_cache,
object_traffic_health: Arc::new(ObjectTrafficHealth::from_env()),
table_catalog_strong_runtime: crate::table_catalog::StrongTableCatalogRuntime::default(),
}
}
@@ -144,6 +146,10 @@ impl AppContext {
Arc::clone(&self.object_traffic_health)
}
pub(crate) fn table_catalog_strong_runtime(&self) -> crate::table_catalog::StrongTableCatalogRuntime {
self.table_catalog_strong_runtime.clone()
}
pub fn iam(&self) -> Arc<dyn IamInterface> {
self.iam.clone()
}
@@ -350,6 +356,7 @@ impl AppContext {
buffer_config: interfaces.buffer_config,
object_data_cache: ObjectDataCacheAdapter::disabled_arc(),
object_traffic_health: Arc::new(ObjectTrafficHealth::from_env()),
table_catalog_strong_runtime: crate::table_catalog::StrongTableCatalogRuntime::default(),
}
}
rustfs/src/storage/access.rs
+9 -2
View File
@@ -1438,8 +1438,15 @@ fn table_data_plane_content_mutation(action: Action) -> bool {
fn table_catalog_backend_for_data_plane<T>(
req: &S3Request<T>,
) -> S3Result<crate::table_catalog::EcStoreTableCatalogObjectBackend<ECStore>> {
let store = request_object_store(req)?;
Ok(crate::table_catalog::EcStoreTableCatalogObjectBackend::new(store))
let context = match req.extensions.get::<Arc<ServerContextSlot>>() {
Some(server_ctx) => server_ctx.installed_app_context(),
None => runtime_sources::current_app_context(),
}
.ok_or_else(object_store_not_initialized_error)?;
Ok(crate::table_catalog::EcStoreTableCatalogObjectBackend::new_with_strong_runtime(
context.object_store(),
context.table_catalog_strong_runtime(),
))
}
fn table_catalog_store_for_data_plane<T>(
rustfs/src/table_catalog/iceberg/commit.rs
+26 -6
View File
@@ -44,11 +44,14 @@ pub(crate) fn table_matches_staged_base(table: &TableEntry, commit_log: &CommitL
pub(crate) struct TableCommitHistoryIndex<'a> {
table_id: &'a str,
reachable_states: BTreeSet<(&'a str, &'a str)>,
ambiguous_states: BTreeSet<(&'a str, &'a str)>,
cycle_detected: bool,
}
impl<'a> TableCommitHistoryIndex<'a> {
pub(crate) fn new(table: &'a TableEntry, commits: impl IntoIterator<Item = &'a CommitLogEntry>) -> Self {
let mut by_new_state = BTreeMap::<(&str, &str), Option<(&str, &str)>>::new();
let mut ambiguous_states = BTreeSet::new();
for commit in commits
.into_iter()
.filter(|commit| commit.table_id == table.table_id && !matches!(commit.status, CommitLogStatus::Failed))
@@ -57,27 +60,39 @@ impl<'a> TableCommitHistoryIndex<'a> {
let previous = (commit.previous_metadata_location.as_str(), commit.expected_version_token.as_str());
by_new_state
.entry(key)
.and_modify(|candidate| *candidate = None)
.and_modify(|candidate| {
*candidate = None;
ambiguous_states.insert(key);
})
.or_insert(Some(previous));
}
let mut reachable_states = BTreeSet::new();
let mut state = (table.metadata_location.as_str(), table.version_token.as_str());
while reachable_states.insert(state) {
let cycle_detected = loop {
if !reachable_states.insert(state) {
break true;
}
let Some(Some(previous)) = by_new_state.get(&state) else {
break;
break false;
};
state = *previous;
}
};
Self {
table_id: &table.table_id,
reachable_states,
ambiguous_states,
cycle_detected,
}
}
pub(crate) fn proves_committed(&self, target: &CommitLogEntry) -> bool {
self.table_id == target.table_id.as_str()
!self.cycle_detected
&& self.table_id == target.table_id.as_str()
&& !matches!(target.status, CommitLogStatus::Failed)
&& !self
.ambiguous_states
.contains(&(target.new_metadata_location.as_str(), target.new_version_token.as_str()))
&& self
.reachable_states
.contains(&(target.new_metadata_location.as_str(), target.new_version_token.as_str()))
@@ -202,7 +217,7 @@ pub(crate) fn table_commit_recovery_entry(
TableCommitRecoveryState::FinalizationRequired,
"a later committed pointer proves this staged commit is part of table history".to_string(),
)
} else if matches!(commit_log.status, CommitLogStatus::Committed) {
} else if matches!(commit_log.status, CommitLogStatus::Committed) && historically_committed {
if idempotency_index_repair_required {
(
TableCommitRecoveryState::IdempotencyIndexRepairRequired,
@@ -214,6 +229,11 @@ pub(crate) fn table_commit_recovery_entry(
"commit is finalized and may be older than the current table pointer".to_string(),
)
}
} else if matches!(commit_log.status, CommitLogStatus::Committed) {
(
TableCommitRecoveryState::ManualReview,
"committed log is not reachable from the current table pointer".to_string(),
)
} else if table_matches_staged_base(table, commit_log) {
(
TableCommitRecoveryState::StagedBeforeTableUpdate,
rustfs/src/table_catalog/iceberg/validation.rs
+6 -8
View File
@@ -286,17 +286,15 @@ where
if table.state != TableCatalogEntryState::Active {
continue;
}
let Ok(warehouse_object_prefix) = table_warehouse_object_prefix(&table) else {
continue;
};
let warehouse_object_prefix = table_warehouse_object_prefix(&table)?;
if !object.starts_with(&warehouse_object_prefix) {
continue;
}
if matched
.as_ref()
.is_some_and(|current| current.warehouse_object_prefix.len() >= warehouse_object_prefix.len())
{
continue;
if let Some(current) = matched.as_ref() {
return Err(TableCatalogStoreError::Invalid(format!(
"object {object} matches overlapping active table warehouse prefixes {} and {warehouse_object_prefix}",
current.warehouse_object_prefix
)));
}
matched = Some(table_data_plane_resource_from_entry(table, warehouse_object_prefix));
}
rustfs/src/table_catalog/mod.rs
+7 -2
View File
@@ -97,6 +97,9 @@ 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 ENV_TABLE_CATALOG_PUBLICATION_FENCE_FLEET_CONFIRMED: &str =
"RUSTFS_TABLE_CATALOG_PUBLICATION_FENCE_FLEET_CONFIRMED";
pub(crate) const ENV_TABLE_CATALOG_STRONG_SNAPSHOT_V2: &str = "RUSTFS_TABLE_CATALOG_STRONG_SNAPSHOT_V2";
pub(crate) const ENV_TABLE_CATALOG_STRONG_SNAPSHOT_V2_FLEET_CONFIRMED: &str =
"RUSTFS_TABLE_CATALOG_STRONG_SNAPSHOT_V2_FLEET_CONFIRMED";
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_DIGEST_REQUIREMENT_TYPE: &str = "assert-rustfs-metadata-sha256";
@@ -159,10 +162,12 @@ const ICEBERG_MAX_REF_AGE_MS_PROPERTY: &str = "history.expire.max-ref-age-ms";
const ICEBERG_REF_MIN_SNAPSHOTS_TO_KEEP_FIELD: &str = "min-snapshots-to-keep";
const ICEBERG_REF_MAX_SNAPSHOT_AGE_MS_FIELD: &str = "max-snapshot-age-ms";
const ICEBERG_REF_MAX_REF_AGE_MS_FIELD: &str = "max-ref-age-ms";
const STRONG_TABLE_CATALOG_SNAPSHOT_VERSION: u16 = 1;
const STRONG_TABLE_CATALOG_SNAPSHOT_MIN_READ_VERSION: u16 = 1;
const STRONG_TABLE_CATALOG_SNAPSHOT_VERSION: u16 = 2;
const STRONG_TABLE_CATALOG_BACKING_ROOT: &str = "strong-backing";
const STRONG_TABLE_CATALOG_SNAPSHOT_FILE: &str = "snapshot.json";
const TABLE_CATALOG_MIGRATION_VERSION: u16 = 1;
const TABLE_CATALOG_MIGRATION_MIN_READ_VERSION: u16 = 1;
const TABLE_CATALOG_MIGRATION_VERSION: u16 = 2;
const TABLE_CATALOG_MIGRATION_ROOT: &str = "backing-migration";
const TABLE_CATALOG_MIGRATION_FENCE_FILE: &str = "durable-strong-fence.json";
const TABLE_CATALOG_MIGRATION_FENCE_LOCK: &str = "durable-strong-fence.lock";
rustfs/src/table_catalog/model.rs
+10 -2
View File
@@ -1067,7 +1067,9 @@ pub(crate) struct TableCatalogBackingProfile {
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
pub(crate) enum TableCatalogBackingKind {
ObjectBacked,
StrongKvWal,
// RUSTFS_COMPAT_TODO(table-catalog-backing-manifest-v1-wire-labels): Keep the version 1 wire label for existing clients. Remove after a versioned manifest with an explicit client migration contract replaces it.
#[serde(rename = "STRONG_KV_WAL")]
DurableStrongSnapshot,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
@@ -1203,7 +1205,9 @@ pub(crate) enum TableCatalogBackingMigrationStep {
ReplayCommitLog,
VerifyCurrentPointer,
EnableSingleWriterFencing,
CutOverLinearizableReads,
// RUSTFS_COMPAT_TODO(table-catalog-backing-manifest-v1-wire-labels): Keep the version 1 wire label for existing clients. Remove after a versioned manifest with an explicit client migration contract replaces it.
#[serde(rename = "CUT_OVER_LINEARIZABLE_READS")]
CutOverDurableSnapshotReads,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
@@ -1213,6 +1217,8 @@ pub(crate) enum TableCatalogBackingMigrationBlocker {
CommitManualReviewRequired,
WarehouseIndexBackfillRequired,
DuplicateWarehousePrefix,
DuplicateTableIdentity,
TableViewIdentifierCollision,
DurableStrongSnapshotChanged,
}
@@ -1222,6 +1228,8 @@ pub(crate) enum TableCatalogBackingMigrationAction {
RunCatalogRecovery,
BackfillWarehouseIndex,
ReviewDuplicateWarehousePrefixes,
ReviewDuplicateTableIdentities,
ReviewTableViewIdentifierCollisions,
SnapshotObjectBackedCatalog,
EnableDurableStrongBacking,
VerifyDurableStrongSnapshot,
rustfs/src/table_catalog/store/migration.rs
+265 -79
View File
@@ -13,11 +13,13 @@
// limitations under the License.
use super::object::{
ObjectTableCatalogStore, validate_namespace_entry_object, validate_table_entry_object, validate_view_entry_object,
ObjectTableCatalogStore, validate_commit_idempotency_entry_object, validate_commit_log_entry_object,
validate_namespace_entry_object, validate_table_bucket_entry_object, validate_table_entry_object, validate_view_entry_object,
};
use super::strong::{
StrongCommitSnapshotRecord, StrongTableCatalogBucketSnapshot, StrongTableCatalogState, TableCatalogBackingMigrationFence,
TableCatalogBackingMigrationFenceStatus, TableCatalogBackingMigrationGlobalFence, table_catalog_bucket_snapshot_fingerprint,
TableCatalogBackingMigrationFenceStatus, TableCatalogBackingMigrationGlobalFence,
TableCatalogBackingMigrationTargetSnapshotState, table_catalog_bucket_snapshot_fingerprint,
};
use super::*;
@@ -82,14 +84,14 @@ pub(super) fn table_catalog_backing_manifest(
},
migration: TableCatalogBackingMigrationPlan {
source_kind: TableCatalogBackingKind::ObjectBacked,
target_kind: TableCatalogBackingKind::StrongKvWal,
target_kind: TableCatalogBackingKind::DurableStrongSnapshot,
status: migration_status,
required_steps: vec![
TableCatalogBackingMigrationStep::SnapshotCatalogExport,
TableCatalogBackingMigrationStep::ReplayCommitLog,
TableCatalogBackingMigrationStep::VerifyCurrentPointer,
TableCatalogBackingMigrationStep::EnableSingleWriterFencing,
TableCatalogBackingMigrationStep::CutOverLinearizableReads,
TableCatalogBackingMigrationStep::CutOverDurableSnapshotReads,
],
blockers,
},
@@ -118,12 +120,100 @@ impl<B> ObjectTableCatalogStore<B>
where
B: TableCatalogObjectBackend,
{
fn migration_target_snapshot_state(
fence: &TableCatalogBackingMigrationFence,
) -> TableCatalogBackingMigrationTargetSnapshotState {
// RUSTFS_COMPAT_TODO(table-catalog-migration-fence-v1): Version 1 PREPARING fences have no durable baseline. Remove after all supported upgrade sources write version 2 fences and all version 1 migrations are completed or cancelled.
if fence.version < TABLE_CATALOG_MIGRATION_VERSION {
TableCatalogBackingMigrationTargetSnapshotState::Unknown
} else if fence.target_snapshot_etag.is_some() {
TableCatalogBackingMigrationTargetSnapshotState::Present
} else {
TableCatalogBackingMigrationTargetSnapshotState::Absent
}
}
fn validate_backing_migration_fence(
table_bucket: &str,
fence: &TableCatalogBackingMigrationFence,
) -> TableCatalogStoreResult<()> {
if !(TABLE_CATALOG_MIGRATION_MIN_READ_VERSION..=TABLE_CATALOG_MIGRATION_VERSION).contains(&fence.version)
|| fence.table_bucket != table_bucket
|| fence.migration_id.is_empty()
{
return Err(TableCatalogStoreError::Invalid(format!(
"invalid durable strong migration fence for table bucket {table_bucket}"
)));
}
let target_snapshot_state = Self::migration_target_snapshot_state(fence);
if fence.target_bucket_existed && target_snapshot_state == TableCatalogBackingMigrationTargetSnapshotState::Absent {
return Err(TableCatalogStoreError::Invalid(format!(
"durable strong migration fence for table bucket {table_bucket} has an inconsistent target snapshot baseline"
)));
}
match fence.status {
TableCatalogBackingMigrationFenceStatus::Preparing if fence.source_fingerprint.is_some() => {
return Err(TableCatalogStoreError::Invalid(format!(
"preparing durable strong migration fence for table bucket {table_bucket} has materialized state"
)));
}
TableCatalogBackingMigrationFenceStatus::Materialized
if fence.source_fingerprint.is_none() || fence.target_snapshot_etag.is_none() =>
{
return Err(TableCatalogStoreError::Invalid(format!(
"materialized durable strong migration fence for table bucket {table_bucket} is incomplete"
)));
}
_ => {}
}
Ok(())
}
fn validate_global_backing_migration_fence(fence: &TableCatalogBackingMigrationGlobalFence) -> TableCatalogStoreResult<()> {
if !(TABLE_CATALOG_MIGRATION_MIN_READ_VERSION..=TABLE_CATALOG_MIGRATION_VERSION).contains(&fence.version)
|| fence.migration_id.is_empty()
{
return Err(TableCatalogStoreError::Invalid(
"invalid durable strong global migration fence".to_string(),
));
}
Ok(())
}
async fn observe_durable_strong_migration_target(
strong_store: &StrongTableCatalogStore<B>,
table_bucket: &str,
fence: Option<&TableCatalogBackingMigrationFence>,
) -> TableCatalogStoreResult<(Option<String>, Option<String>)> {
let target_snapshot_state = fence.map(Self::migration_target_snapshot_state);
let permits_absent_snapshot = fence.is_some_and(|fence| {
fence.status == TableCatalogBackingMigrationFenceStatus::Preparing
&& !fence.target_bucket_existed
&& target_snapshot_state == Some(TableCatalogBackingMigrationTargetSnapshotState::Absent)
});
if permits_absent_snapshot {
strong_store.restore_absent_migration_snapshot_baseline().await?;
}
let observation = strong_store.bucket_snapshot_observation(table_bucket).await?;
if fence.is_some() && observation.1.is_none() && !permits_absent_snapshot {
return Err(TableCatalogStoreError::Conflict(
"durable strong catalog snapshot is missing for an in-progress backing migration".to_string(),
));
}
Ok(observation)
}
async fn read_backing_migration_fence(
&self,
table_bucket: &str,
) -> TableCatalogStoreResult<Option<(TableCatalogBackingMigrationFence, Option<String>)>> {
self.read_entry(self.catalog_bucket(), &self.paths.backing_migration_fence_path(table_bucket))
.await
let fence = self
.read_entry(self.catalog_bucket(), &self.paths.backing_migration_fence_path(table_bucket))
.await?;
if let Some((fence, _)) = fence.as_ref() {
Self::validate_backing_migration_fence(table_bucket, fence)?;
}
Ok(fence)
}
pub(super) async fn acquire_table_bucket_registry_write_permit(&self) -> TableCatalogStoreResult<Box<dyn Send>> {
@@ -164,11 +254,7 @@ where
.read_entry::<TableCatalogBackingMigrationGlobalFence>(self.catalog_bucket(), fence_path)
.await?
{
if fence.version != TABLE_CATALOG_MIGRATION_VERSION {
return Err(TableCatalogStoreError::Invalid(
"invalid durable strong global migration fence".to_string(),
));
}
Self::validate_global_backing_migration_fence(&fence)?;
return Ok(fence);
}
let fence = TableCatalogBackingMigrationGlobalFence {
@@ -181,6 +267,13 @@ where
}
async fn clear_global_backing_migration_fence_if_unused(&self, fence_path: &str) -> TableCatalogStoreResult<()> {
let Some((fence, _)) = self
.read_entry::<TableCatalogBackingMigrationGlobalFence>(self.catalog_bucket(), fence_path)
.await?
else {
return Ok(());
};
Self::validate_global_backing_migration_fence(&fence)?;
let bucket_objects = self
.backend
.list_objects(self.catalog_bucket(), &self.paths.table_bucket_entries_prefix())
@@ -194,19 +287,6 @@ where
self.backend.delete_object(self.catalog_bucket(), fence_path).await
}
pub(super) async fn ensure_object_backed_writes_allowed(&self, table_bucket: &str) -> TableCatalogStoreResult<()> {
if self
.backend
.object_exists(self.catalog_bucket(), &self.paths.backing_migration_fence_path(table_bucket))
.await?
{
return Err(TableCatalogStoreError::Conflict(format!(
"object-backed catalog writes are fenced while table bucket {table_bucket} is prepared for durable strong cutover"
)));
}
Ok(())
}
async fn collect_bucket_snapshot_with_locks(
&self,
table_bucket: &str,
@@ -220,11 +300,7 @@ where
else {
return Err(TableCatalogStoreError::NotFound(format!("table bucket {table_bucket}")));
};
if table_bucket_entry.table_bucket != table_bucket {
return Err(TableCatalogStoreError::Invalid(format!(
"table bucket entry does not match migration target {table_bucket}"
)));
}
validate_table_bucket_entry_object(&self.paths, &bucket_path, &table_bucket_entry)?;
let mut namespaces = Vec::new();
let mut tables = Vec::new();
@@ -286,6 +362,7 @@ where
"commit log changed while preparing durable strong snapshot: {commit_object}"
)));
};
validate_commit_log_entry_object(&self.paths, &commit_object, table_bucket, &table_entry.table_id, &commit)?;
commits.push(StrongCommitSnapshotRecord {
table_bucket: table_bucket.to_string(),
table_id: table_entry.table_id.clone(),
@@ -313,6 +390,13 @@ where
"idempotency index changed while preparing durable strong snapshot: {idempotency_object}"
)));
};
validate_commit_idempotency_entry_object(
&self.paths,
&idempotency_object,
table_bucket,
&table_entry.table_id,
&commit,
)?;
let lookup_key = commit.idempotency_key.clone().ok_or_else(|| {
TableCatalogStoreError::Invalid(format!("idempotency index {idempotency_object} has no idempotency key"))
})?;
@@ -360,6 +444,22 @@ where
fn validate_bucket_snapshot_for_migration(&self, snapshot: &StrongTableCatalogBucketSnapshot) -> TableCatalogStoreResult<()> {
let table_bucket = &snapshot.table_bucket.table_bucket;
let active_table_identifiers = snapshot
.tables
.iter()
.filter(|table| table.state == TableCatalogEntryState::Active)
.map(|table| (&table.namespace, &table.table))
.collect::<BTreeSet<_>>();
if snapshot
.views
.iter()
.filter(|view| view.state == TableCatalogEntryState::Active)
.any(|view| active_table_identifiers.contains(&(&view.namespace, &view.view)))
{
return Err(TableCatalogStoreError::Conflict(format!(
"table bucket {table_bucket} contains an active table/view identifier collision"
)));
}
let tables_by_id = snapshot
.tables
.iter()
@@ -498,6 +598,15 @@ where
if self.get_table_bucket(table_bucket).await?.is_none() {
return Err(TableCatalogStoreError::NotFound(format!("table bucket {table_bucket}")));
}
if let Some((global_fence, _)) = self
.read_entry::<TableCatalogBackingMigrationGlobalFence>(
self.catalog_bucket(),
&self.paths.backing_migration_global_fence_path(),
)
.await?
{
Self::validate_global_backing_migration_fence(&global_fence)?;
}
let namespace_objects = self
.backend
@@ -511,6 +620,10 @@ where
let mut recovery_required_count: usize = 0;
let mut manual_review_count: usize = 0;
let mut warehouse_prefix_owners = BTreeMap::<String, usize>::new();
let mut table_ids = BTreeSet::<String>::new();
let mut duplicate_table_identity = false;
let mut active_table_identifiers = BTreeSet::<(String, String)>::new();
let mut active_view_identifiers = BTreeSet::<(String, String)>::new();
for object in namespace_objects {
if object.ends_with(NAMESPACE_ENTRY_FILE) {
@@ -526,6 +639,9 @@ where
continue;
};
validate_view_entry_object(&self.paths, &object, &entry)?;
if entry.state == TableCatalogEntryState::Active {
active_view_identifiers.insert((entry.namespace.clone(), entry.view.clone()));
}
view_count = view_count.saturating_add(1);
continue;
}
@@ -538,7 +654,11 @@ where
};
validate_table_entry_object(&self.paths, &object, &table)?;
table_count = table_count.saturating_add(1);
if !table_ids.insert(table.table_id.clone()) {
duplicate_table_identity = true;
}
if table.state == TableCatalogEntryState::Active {
active_table_identifiers.insert((table.namespace.clone(), table.table.clone()));
let warehouse_prefix = table_warehouse_object_prefix(&table)?;
warehouse_prefix_owners
.entry(warehouse_prefix)
@@ -548,17 +668,31 @@ where
let recovery = self.table_commit_recovery_report_for_entry(&table, 0).await?;
commit_log_count = commit_log_count.saturating_add(recovery.commits.len());
idempotency_index_count = idempotency_index_count.saturating_add(
self.backend
.list_objects(
self.catalog_bucket(),
&self.paths.commit_idempotency_entries_prefix(table_bucket, &table.table_id),
)
for idempotency_object in self
.backend
.list_objects(
self.catalog_bucket(),
&self.paths.commit_idempotency_entries_prefix(table_bucket, &table.table_id),
)
.await?
.into_iter()
.filter(|object| object.ends_with(".json"))
{
let Some((commit, _)) = self
.read_entry::<CommitLogEntry>(self.catalog_bucket(), &idempotency_object)
.await?
.into_iter()
.filter(|object| object.ends_with(".json"))
.count(),
);
else {
continue;
};
validate_commit_idempotency_entry_object(
&self.paths,
&idempotency_object,
table_bucket,
&table.table_id,
&commit,
)?;
idempotency_index_count = idempotency_index_count.saturating_add(1);
}
recovery_required_count = recovery_required_count
.saturating_add(recovery.staged_before_table_update_count)
.saturating_add(recovery.finalization_required_count)
@@ -568,6 +702,13 @@ where
let warehouse_index_ready = self.warehouse_index_ready(table_bucket).await?;
let duplicate_warehouse_prefix_count = warehouse_prefix_owners.values().filter(|count| **count > 1).count();
let overlapping_warehouse_prefix = warehouse_prefix_owners
.keys()
.collect::<Vec<_>>()
.windows(2)
.any(|window| warehouse_object_prefixes_overlap(window[0], window[1]));
let conflicting_warehouse_prefix = duplicate_warehouse_prefix_count > 0 || overlapping_warehouse_prefix;
let table_view_identifier_collision_count = active_table_identifiers.intersection(&active_view_identifiers).count();
let mut blockers = Vec::new();
let mut recommended_actions = Vec::new();
if recovery_required_count > 0 {
@@ -583,12 +724,24 @@ where
blockers.push(TableCatalogBackingMigrationBlocker::WarehouseIndexBackfillRequired);
recommended_actions.push(TableCatalogBackingMigrationAction::BackfillWarehouseIndex);
}
if duplicate_warehouse_prefix_count > 0 {
if conflicting_warehouse_prefix {
blockers.push(TableCatalogBackingMigrationBlocker::DuplicateWarehousePrefix);
recommended_actions.push(TableCatalogBackingMigrationAction::ReviewDuplicateWarehousePrefixes);
}
if duplicate_table_identity {
blockers.push(TableCatalogBackingMigrationBlocker::DuplicateTableIdentity);
recommended_actions.push(TableCatalogBackingMigrationAction::ReviewDuplicateTableIdentities);
}
if table_view_identifier_collision_count > 0 {
blockers.push(TableCatalogBackingMigrationBlocker::TableViewIdentifierCollision);
recommended_actions.push(TableCatalogBackingMigrationAction::ReviewTableViewIdentifierCollisions);
}
let mut status = if manual_review_count > 0 || duplicate_warehouse_prefix_count > 0 {
let mut status = if manual_review_count > 0
|| conflicting_warehouse_prefix
|| duplicate_table_identity
|| table_view_identifier_collision_count > 0
{
TableCatalogBackingMigrationStatus::ManualReviewRequired
} else if recovery_required_count > 0 || !warehouse_index_ready {
TableCatalogBackingMigrationStatus::RecoveryRequired
@@ -597,6 +750,14 @@ where
};
let strong_store = StrongTableCatalogStore::new(self.backend.clone());
let source_table_buckets = self.object_backed_table_buckets().await?;
let source_table_bucket_names = source_table_buckets.keys().cloned().collect::<BTreeSet<_>>();
let target_table_buckets = strong_store.table_bucket_names().await?;
if !target_table_buckets.is_subset(&source_table_bucket_names) {
status = TableCatalogBackingMigrationStatus::ManualReviewRequired;
blockers.push(TableCatalogBackingMigrationBlocker::DurableStrongSnapshotChanged);
recommended_actions.push(TableCatalogBackingMigrationAction::ReviewDurableStrongSnapshot);
}
let migration_fence = self.read_backing_migration_fence(table_bucket).await?.map(|(fence, _)| fence);
let object_backed_writes_fenced = migration_fence.is_some();
if status == TableCatalogBackingMigrationStatus::ReadyToSnapshot
@@ -633,7 +794,7 @@ where
Ok(TableCatalogBackingMigrationDryRunReport {
table_bucket: table_bucket.to_string(),
source_kind: TableCatalogBackingKind::ObjectBacked,
target_kind: TableCatalogBackingKind::StrongKvWal,
target_kind: TableCatalogBackingKind::DurableStrongSnapshot,
status,
namespace_count,
table_count,
@@ -684,11 +845,17 @@ where
let existing_fence = self.read_backing_migration_fence(table_bucket).await?;
let strong_store = StrongTableCatalogStore::new(self.backend.clone());
if let Some((fence, _)) = existing_fence.as_ref()
&& (fence.version != TABLE_CATALOG_MIGRATION_VERSION || fence.table_bucket != table_bucket)
&& fence.status == TableCatalogBackingMigrationFenceStatus::Preparing
&& !fence.target_bucket_existed
&& Self::migration_target_snapshot_state(fence) == TableCatalogBackingMigrationTargetSnapshotState::Absent
{
return Err(TableCatalogStoreError::Invalid(format!(
"invalid durable strong migration fence for table bucket {table_bucket}"
)));
strong_store.restore_absent_migration_snapshot_baseline().await?;
}
let source_table_bucket_names = self.object_backed_table_buckets().await?.into_keys().collect::<BTreeSet<_>>();
if !strong_store.table_bucket_names().await?.is_subset(&source_table_bucket_names) {
return Err(TableCatalogStoreError::Conflict(
"durable strong snapshot contains table buckets outside the object-backed catalog inventory".to_string(),
));
}
if !self.warehouse_index_ready(table_bucket).await? {
@@ -712,10 +879,16 @@ where
}
self.ensure_global_backing_migration_fence(&global_fence_path).await?;
let (target_fingerprint, target_snapshot_etag) = Self::observe_durable_strong_migration_target(
&strong_store,
table_bucket,
existing_fence.as_ref().map(|(fence, _)| fence),
)
.await?;
let (migration_id, target_bucket_existed) = if let Some((fence, _)) = existing_fence.as_ref() {
(fence.migration_id.clone(), fence.target_bucket_existed)
} else {
let target_bucket_existed = strong_store.bucket_snapshot_fingerprint(table_bucket).await?.is_some();
let target_bucket_existed = target_fingerprint.is_some();
let fence = TableCatalogBackingMigrationFence {
version: TABLE_CATALOG_MIGRATION_VERSION,
table_bucket: table_bucket.to_string(),
@@ -723,7 +896,7 @@ where
status: TableCatalogBackingMigrationFenceStatus::Preparing,
target_bucket_existed,
source_fingerprint: None,
target_snapshot_etag: None,
target_snapshot_etag,
};
self.write_entry(self.catalog_bucket(), &fence_path, &fence, TableCatalogPutPrecondition::IfAbsent)
.await?;
@@ -749,7 +922,7 @@ where
Ok(TableCatalogBackingMigrationExecutionReport {
table_bucket: table_bucket.to_string(),
source_kind: TableCatalogBackingKind::ObjectBacked,
target_kind: TableCatalogBackingKind::StrongKvWal,
target_kind: TableCatalogBackingKind::DurableStrongSnapshot,
status: if created {
TableCatalogBackingMigrationExecutionStatus::SnapshotMaterialized
} else {
@@ -793,11 +966,6 @@ where
});
};
self.ensure_global_backing_migration_fence(&global_fence_path).await?;
if fence.version != TABLE_CATALOG_MIGRATION_VERSION || fence.table_bucket != table_bucket {
return Err(TableCatalogStoreError::Invalid(format!(
"invalid durable strong migration fence for table bucket {table_bucket}"
)));
}
let mut source_guards = Vec::new();
let source = self
@@ -813,12 +981,21 @@ where
}
let strong_store = StrongTableCatalogStore::new(self.backend.clone());
if fence.status == TableCatalogBackingMigrationFenceStatus::Materialized
&& strong_store.bucket_snapshot_fingerprint(table_bucket).await?.as_deref() != Some(&source_fingerprint)
{
return Err(TableCatalogStoreError::Conflict(format!(
"durable strong catalog state changed after materializing table bucket {table_bucket}"
)));
let (target_fingerprint, target_snapshot_etag) =
Self::observe_durable_strong_migration_target(&strong_store, table_bucket, Some(&fence)).await?;
if fence.status == TableCatalogBackingMigrationFenceStatus::Materialized {
let target_matches_source = target_fingerprint.as_deref() == Some(&source_fingerprint);
let target_was_already_removed = !fence.target_bucket_existed && target_fingerprint.is_none();
if !target_matches_source && !target_was_already_removed {
return Err(TableCatalogStoreError::Conflict(format!(
"durable strong catalog state changed after materializing table bucket {table_bucket}"
)));
}
if target_matches_source && target_snapshot_etag != fence.target_snapshot_etag {
return Err(TableCatalogStoreError::Conflict(
"durable strong catalog snapshot advanced after materialization".to_string(),
));
}
}
if !fence.target_bucket_existed {
strong_store
@@ -836,30 +1013,19 @@ where
}
async fn all_table_buckets_materialized(&self, strong_store: &StrongTableCatalogStore<B>) -> TableCatalogStoreResult<bool> {
if self
let Some((global_fence, _)) = self
.read_entry::<TableCatalogBackingMigrationGlobalFence>(
self.catalog_bucket(),
&self.paths.backing_migration_global_fence_path(),
)
.await?
.is_none()
{
else {
return Ok(false);
}
let table_bucket_objects = self
.backend
.list_objects(self.catalog_bucket(), &self.paths.table_bucket_entries_prefix())
.await?;
for table_bucket_object in table_bucket_objects
.iter()
.filter(|object| object.ends_with(TABLE_BUCKET_ENTRY_FILE))
{
let Some((entry, _)) = self
.read_entry::<TableBucketEntry>(self.catalog_bucket(), table_bucket_object)
.await?
else {
return Ok(false);
};
};
Self::validate_global_backing_migration_fence(&global_fence)?;
let source_table_buckets = self.object_backed_table_buckets().await?;
let source_table_bucket_names = source_table_buckets.keys().cloned().collect::<BTreeSet<_>>();
for entry in source_table_buckets.values() {
let Some((fence, _)) = self.read_backing_migration_fence(&entry.table_bucket).await? else {
return Ok(false);
};
@@ -878,6 +1044,26 @@ where
return Ok(false);
}
}
Ok(true)
Ok(strong_store.table_bucket_names().await? == source_table_bucket_names)
}
async fn object_backed_table_buckets(&self) -> TableCatalogStoreResult<BTreeMap<String, TableBucketEntry>> {
let mut table_buckets = BTreeMap::new();
for object in self
.backend
.list_objects(self.catalog_bucket(), &self.paths.table_bucket_entries_prefix())
.await?
.into_iter()
.filter(|object| object.ends_with(TABLE_BUCKET_ENTRY_FILE))
{
let Some((entry, _)) = self.read_entry::<TableBucketEntry>(self.catalog_bucket(), &object).await? else {
return Err(TableCatalogStoreError::Conflict(format!(
"table bucket changed while reading durable strong migration inventory: {object}"
)));
};
validate_table_bucket_entry_object(&self.paths, &object, &entry)?;
table_buckets.insert(entry.table_bucket.clone(), entry);
}
Ok(table_buckets)
}
}
rustfs/src/table_catalog/store/mod.rs
+88 -32
View File
@@ -21,8 +21,39 @@ mod strong;
use migration::table_catalog_backing_manifest;
pub(crate) use object::ObjectTableCatalogStore;
#[cfg(test)]
pub(super) use strong::StrongTableCatalogSnapshot;
pub(crate) use strong::StrongTableCatalogStore;
pub(super) use strong::{
STRONG_TABLE_CATALOG_RELOAD_MAX_ATTEMPTS, STRONG_TABLE_CATALOG_SNAPSHOT_MAX_SIZE, StrongCommitSnapshotRecord,
StrongTableCatalogBucketSnapshot, StrongTableCatalogSnapshot, strong_snapshot_write_version,
table_catalog_bucket_snapshot_fingerprint,
};
pub(crate) use strong::{StrongTableCatalogRuntime, StrongTableCatalogStore};
fn validate_table_bucket_entry(entry: &TableBucketEntry) -> TableCatalogStoreResult<()> {
validate_catalog_entry_version("table bucket", entry.version)?;
if entry.table_bucket.is_empty() {
return Err(TableCatalogStoreError::Invalid("table bucket name cannot be empty".to_string()));
}
if entry.catalog_type != TABLE_BUCKET_CATALOG_TYPE {
return Err(TableCatalogStoreError::Invalid("unsupported table bucket catalog type".to_string()));
}
Ok(())
}
fn validate_table_entry_version_and_id(entry: &TableEntry) -> TableCatalogStoreResult<()> {
validate_catalog_entry_version("table", entry.version)?;
if entry.table_id.is_empty() {
return Err(TableCatalogStoreError::Invalid("table id cannot be empty".to_string()));
}
Ok(())
}
fn validate_view_entry_version_and_id(entry: &ViewEntry) -> TableCatalogStoreResult<()> {
validate_catalog_entry_version("view", entry.version)?;
if entry.view_id.is_empty() {
return Err(TableCatalogStoreError::Invalid("view id cannot be empty".to_string()));
}
Ok(())
}
fn validate_namespace_entry_identity(entry: &NamespaceEntry) -> TableCatalogStoreResult<Namespace> {
validate_catalog_entry_version("namespace", entry.version)?;
@@ -406,8 +437,31 @@ pub(crate) enum TableCatalogPutPrecondition {
IfMatch(String),
}
pub(in crate::table_catalog) fn catalog_list_next_continuation(
seen: &mut BTreeSet<String>,
is_truncated: bool,
next: Option<String>,
) -> TableCatalogStoreResult<Option<String>> {
if !is_truncated {
return Ok(None);
}
let next = next.filter(|next| !next.is_empty()).ok_or_else(|| {
TableCatalogStoreError::Internal("truncated catalog object listing has no continuation token".to_string())
})?;
if !seen.insert(next.clone()) {
return Err(TableCatalogStoreError::Internal(
"catalog object listing continuation token did not advance".to_string(),
));
}
Ok(Some(next))
}
#[async_trait::async_trait]
pub(crate) trait TableCatalogObjectBackend: Clone + Send + Sync + 'static {
fn strong_catalog_runtime(&self) -> Option<StrongTableCatalogRuntime> {
None
}
async fn read_object(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Option<TableCatalogObject>>;
async fn read_object_limited(
@@ -845,10 +899,16 @@ where
B: TableCatalogObjectBackend,
{
pub(crate) fn from_env(backend: B) -> TableCatalogStoreResult<Self> {
Ok(Self::new(backend, TableCatalogBackingMode::from_env()?))
Ok(match TableCatalogBackingMode::from_env()? {
TableCatalogBackingMode::ObjectBacked => Self::ObjectBacked(ObjectTableCatalogStore::new(backend)),
TableCatalogBackingMode::DurableStrong => {
Self::DurableStrong(StrongTableCatalogStore::new_requiring_snapshot(backend))
}
})
}
pub(crate) fn new(backend: B, mode: TableCatalogBackingMode) -> Self {
#[cfg(test)]
pub(crate) fn new_for_test(backend: B, mode: TableCatalogBackingMode) -> Self {
match mode {
TableCatalogBackingMode::ObjectBacked => Self::ObjectBacked(ObjectTableCatalogStore::new(backend)),
TableCatalogBackingMode::DurableStrong => Self::DurableStrong(StrongTableCatalogStore::new(backend)),
@@ -1352,12 +1412,14 @@ where
pub(crate) struct EcStoreTableCatalogObjectBackend<S> {
store: Arc<S>,
strong_runtime: StrongTableCatalogRuntime,
}
impl<S> Clone for EcStoreTableCatalogObjectBackend<S> {
fn clone(&self) -> Self {
Self {
store: self.store.clone(),
strong_runtime: self.strong_runtime.clone(),
}
}
}
@@ -1366,8 +1428,8 @@ impl<S> EcStoreTableCatalogObjectBackend<S>
where
S: TableCatalogStorage,
{
pub fn new(store: Arc<S>) -> Self {
Self { store }
pub fn new_with_strong_runtime(store: Arc<S>, strong_runtime: StrongTableCatalogRuntime) -> Self {
Self { store, strong_runtime }
}
}
@@ -1378,6 +1440,10 @@ impl<S> TableCatalogObjectBackend for EcStoreTableCatalogObjectBackend<S>
where
S: TableCatalogStorage,
{
fn strong_catalog_runtime(&self) -> Option<StrongTableCatalogRuntime> {
Some(self.strong_runtime.clone())
}
async fn read_object(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Option<TableCatalogObject>> {
self.read_object_with_options(bucket, object, ObjectOptions::default(), None)
.await
@@ -1537,6 +1603,7 @@ where
async fn list_objects(&self, bucket: &str, prefix: &str) -> TableCatalogStoreResult<Vec<String>> {
let mut continuation = None;
let mut seen_continuations = BTreeSet::new();
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()))?;
@@ -1553,14 +1620,10 @@ where
objects.insert(object.name);
}
if !result.is_truncated {
break;
}
let Some(next) = result.next_continuation_token else {
break;
match catalog_list_next_continuation(&mut seen_continuations, result.is_truncated, result.next_continuation_token)? {
Some(next) => continuation = Some(next),
None => break,
};
continuation = Some(next);
}
Ok(objects.into_iter().collect())
@@ -1627,20 +1690,6 @@ where
opts: ObjectOptions,
max_size: Option<usize>,
) -> TableCatalogStoreResult<Option<TableCatalogObject>> {
let info = match self.store.get_object_info(bucket, object, &opts).await {
Ok(info) => info,
Err(err) if is_missing_storage_error(&err) => return Ok(None),
Err(err) => return Err(storage_error_to_catalog("read catalog object info", err)),
};
if let Some(max_size) = max_size {
let object_size = usize::try_from(info.size)
.map_err(|_| TableCatalogStoreError::Invalid(format!("catalog object {bucket}/{object} has an invalid size")))?;
if object_size > max_size {
return Err(TableCatalogStoreError::Invalid(format!(
"catalog object {bucket}/{object} exceeds the maximum size of {max_size} bytes"
)));
}
}
let mut reader = match self
.store
.get_object_reader(bucket, object, None, HeaderMap::new(), &opts)
@@ -1650,6 +1699,17 @@ where
Err(err) if is_missing_storage_error(&err) => return Ok(None),
Err(err) => return Err(storage_error_to_catalog("read catalog object", err)),
};
if let Some(max_size) = max_size {
let object_size = usize::try_from(reader.object_info.size)
.map_err(|_| TableCatalogStoreError::Invalid(format!("catalog object {bucket}/{object} has an invalid size")))?;
if object_size > max_size {
return Err(TableCatalogStoreError::Invalid(format!(
"catalog object {bucket}/{object} exceeds the maximum size of {max_size} bytes"
)));
}
}
let etag = reader.object_info.etag.clone();
let mod_time = reader.object_info.mod_time;
let mut data = Vec::new();
if let Some(max_size) = max_size {
let read_limit = u64::try_from(max_size.saturating_add(1)).unwrap_or(u64::MAX);
@@ -1666,11 +1726,7 @@ where
TableCatalogStoreError::Internal(format!("failed to read catalog object {bucket}/{object}: {err}"))
})?;
}
Ok(Some(TableCatalogObject {
data,
etag: info.etag,
mod_time: info.mod_time,
}))
Ok(Some(TableCatalogObject { data, etag, mod_time }))
}
async fn put_object_with_options(
rustfs/src/table_catalog/store/object.rs
+583 -203
View File
File diff suppressed because it is too large Load Diff
rustfs/src/table_catalog/store/strong.rs
+1253 -249
View File
File diff suppressed because it is too large Load Diff
rustfs/src/table_catalog/tests.rs
+5559 -95
View File
File diff suppressed because it is too large Load Diff