From 013b5b796688c9a6b4ddb514b7187d71b1b24604 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Thu, 4 Jun 2026 11:20:46 +0800 Subject: [PATCH] feat(table-catalog): add catalog store entry models (#3201) Co-authored-by: Henry Guo --- rustfs/src/table_catalog.rs | 420 +++++++++++++++++++++++++++++++++++- 1 file changed, 419 insertions(+), 1 deletion(-) diff --git a/rustfs/src/table_catalog.rs b/rustfs/src/table_catalog.rs index 5b6275bd5..c4a6ea240 100644 --- a/rustfs/src/table_catalog.rs +++ b/rustfs/src/table_catalog.rs @@ -20,7 +20,7 @@ #![allow(dead_code)] -use std::fmt; +use std::{collections::BTreeMap, fmt}; use rustfs_ecstore::bucket::{ metadata::{BUCKET_TABLE_CONFIG, BUCKET_TABLE_RESERVED_PREFIX}, @@ -36,6 +36,7 @@ pub(crate) const DEFAULT_WAREHOUSE_ID: &str = "default"; pub(crate) const TABLE_NAMESPACE_MARKER_VERSION: u16 = 1; pub(crate) const TABLE_RESOURCE_MARKER_VERSION: u16 = 1; pub(crate) const TABLE_METADATA_POINTER_VERSION: u16 = 1; +pub(crate) const TABLE_CATALOG_ENTRY_VERSION: u16 = 1; 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"; @@ -108,6 +109,173 @@ pub(crate) fn table_bucket_marker_json() -> Result, serde_json::Error> { serde_json::to_vec(&TableBucketMarker::default()) } +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub(crate) enum TableCatalogEntryState { + Active, + Deleting, + Deleted, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct TableBucketEntry { + pub version: u16, + pub table_bucket: String, + pub catalog_type: String, + pub warehouse_root: String, + pub state: TableCatalogEntryState, + #[serde(default)] + pub properties: BTreeMap, + pub created_at: Option, + pub updated_at: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct NamespaceEntry { + pub version: u16, + pub table_bucket: String, + pub namespace: String, + pub namespace_id: String, + pub state: TableCatalogEntryState, + #[serde(default)] + pub properties: BTreeMap, + pub created_at: Option, + pub updated_at: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct TableEntry { + pub version: u16, + pub table_bucket: String, + pub namespace: String, + pub table: String, + pub table_id: String, + pub table_uuid: String, + pub format: String, + pub format_version: u16, + pub warehouse_location: String, + pub metadata_location: String, + pub version_token: String, + pub generation: u64, + pub state: TableCatalogEntryState, + #[serde(default)] + pub properties: BTreeMap, + pub created_at: Option, + pub updated_at: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub(crate) enum CommitLogStatus { + Committed, + Failed, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct CommitLogEntry { + pub version: u16, + pub commit_id: String, + pub idempotency_key: Option, + pub table_id: String, + pub operation: String, + pub expected_version_token: String, + pub new_version_token: String, + pub previous_metadata_location: String, + pub new_metadata_location: String, + #[serde(default)] + pub requirements: Vec, + pub status: CommitLogStatus, + pub writer: Option, + pub created_at: Option, + pub updated_at: Option, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct TableCommitRequest { + pub table_bucket: String, + pub namespace: String, + pub table: String, + pub commit_id: String, + pub idempotency_key: Option, + pub operation: String, + pub expected_version_token: String, + pub expected_metadata_location: String, + pub new_metadata_location: String, + #[serde(default)] + pub requirements: Vec, + pub writer: Option, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct TableCommitResult { + pub table: TableEntry, + pub commit_log: CommitLogEntry, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum TableCatalogStoreError { + NotFound(String), + Conflict(String), + Invalid(String), + Internal(String), +} + +impl fmt::Display for TableCatalogStoreError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::NotFound(message) => write!(f, "table catalog entry not found: {message}"), + Self::Conflict(message) => write!(f, "table catalog conflict: {message}"), + Self::Invalid(message) => write!(f, "invalid table catalog entry: {message}"), + Self::Internal(message) => write!(f, "table catalog store error: {message}"), + } + } +} + +impl std::error::Error for TableCatalogStoreError {} + +pub(crate) type TableCatalogStoreResult = Result; + +#[async_trait::async_trait] +pub(crate) trait TableCatalogStore: Send + Sync { + async fn get_table_bucket(&self, table_bucket: &str) -> TableCatalogStoreResult>; + + async fn put_table_bucket(&self, entry: TableBucketEntry) -> TableCatalogStoreResult<()>; + + async fn create_namespace(&self, entry: NamespaceEntry) -> TableCatalogStoreResult<()>; + + async fn list_namespaces(&self, table_bucket: &str) -> TableCatalogStoreResult>; + + async fn get_namespace(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult>; + + async fn drop_namespace(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult<()>; + + async fn create_table(&self, entry: TableEntry) -> TableCatalogStoreResult<()>; + + async fn register_table(&self, entry: TableEntry) -> TableCatalogStoreResult<()>; + + async fn list_tables(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult>; + + async fn load_table(&self, table_bucket: &str, namespace: &str, table: &str) -> TableCatalogStoreResult>; + + async fn commit_table(&self, request: TableCommitRequest) -> TableCatalogStoreResult; + + async fn drop_table(&self, table_bucket: &str, namespace: &str, table: &str) -> TableCatalogStoreResult<()>; + + async fn get_commit_by_id(&self, table_id: &str, commit_id: &str) -> TableCatalogStoreResult>; + + async fn get_commit_by_idempotency_key( + &self, + table_id: &str, + idempotency_key: &str, + ) -> TableCatalogStoreResult>; +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub(crate) struct NamespaceMarker { pub version: u16, @@ -513,6 +681,256 @@ mod tests { assert!(!table_bucket_marker_json().unwrap().is_empty()); } + #[test] + fn catalog_entry_structures_serialize_stable_fields() { + use std::collections::BTreeMap; + + let bucket = TableBucketEntry { + version: TABLE_CATALOG_ENTRY_VERSION, + table_bucket: "analytics".to_string(), + catalog_type: TABLE_BUCKET_CATALOG_TYPE.to_string(), + warehouse_root: "s3://analytics/".to_string(), + state: TableCatalogEntryState::Active, + properties: BTreeMap::from([("owner".to_string(), "platform".to_string())]), + created_at: Some("2026-05-23T00:00:00Z".to_string()), + updated_at: Some("2026-05-23T00:00:00Z".to_string()), + }; + let namespace = NamespaceEntry { + version: TABLE_CATALOG_ENTRY_VERSION, + table_bucket: "analytics".to_string(), + namespace: "sales".to_string(), + namespace_id: "sales".to_string(), + state: TableCatalogEntryState::Active, + properties: BTreeMap::from([("purpose".to_string(), "orders".to_string())]), + created_at: Some("2026-05-23T00:00:00Z".to_string()), + updated_at: Some("2026-05-23T00:00:00Z".to_string()), + }; + let table = TableEntry { + version: TABLE_CATALOG_ENTRY_VERSION, + table_bucket: "analytics".to_string(), + namespace: "sales".to_string(), + table: "orders".to_string(), + table_id: "table-id".to_string(), + table_uuid: "table-uuid".to_string(), + format: "ICEBERG".to_string(), + format_version: 2, + warehouse_location: "s3://analytics/tables/table-id".to_string(), + metadata_location: "s3://analytics/tables/table-id/metadata/v1.metadata.json".to_string(), + version_token: "token-v1".to_string(), + generation: 1, + state: TableCatalogEntryState::Active, + properties: BTreeMap::from([("write.format.default".to_string(), "parquet".to_string())]), + created_at: Some("2026-05-23T00:00:00Z".to_string()), + updated_at: Some("2026-05-23T00:00:00Z".to_string()), + }; + let commit = CommitLogEntry { + version: TABLE_CATALOG_ENTRY_VERSION, + commit_id: "commit-id".to_string(), + idempotency_key: Some("client-request-id".to_string()), + table_id: "table-id".to_string(), + operation: "append".to_string(), + expected_version_token: "token-v1".to_string(), + new_version_token: "token-v2".to_string(), + previous_metadata_location: "s3://analytics/tables/table-id/metadata/v1.metadata.json".to_string(), + new_metadata_location: "s3://analytics/tables/table-id/metadata/v2.metadata.json".to_string(), + requirements: vec![serde_json::json!({"type": "assert-table-uuid", "uuid": "table-uuid"})], + status: CommitLogStatus::Committed, + writer: Some("pyiceberg/test".to_string()), + created_at: Some("2026-05-23T00:01:00Z".to_string()), + updated_at: Some("2026-05-23T00:01:00Z".to_string()), + }; + + let bucket_json = serde_json::to_value(&bucket).unwrap(); + let namespace_json = serde_json::to_value(&namespace).unwrap(); + let table_json = serde_json::to_value(&table).unwrap(); + let commit_json = serde_json::to_value(&commit).unwrap(); + + assert_eq!(bucket_json["state"], "ACTIVE"); + assert_eq!(bucket_json["properties"]["owner"], "platform"); + assert_eq!(namespace_json["namespace_id"], "sales"); + assert_eq!(table_json["version_token"], "token-v1"); + assert_eq!(table_json["generation"], 1); + assert_eq!(table_json["state"], "ACTIVE"); + assert_eq!(commit_json["status"], "COMMITTED"); + assert_eq!(commit_json["requirements"][0]["type"], "assert-table-uuid"); + } + + #[test] + fn catalog_entry_deserialization_rejects_unknown_fields() { + use std::collections::BTreeMap; + + let table = TableEntry { + version: TABLE_CATALOG_ENTRY_VERSION, + table_bucket: "analytics".to_string(), + namespace: "sales".to_string(), + table: "orders".to_string(), + table_id: "table-id".to_string(), + table_uuid: "table-uuid".to_string(), + format: "ICEBERG".to_string(), + format_version: 2, + warehouse_location: "s3://analytics/tables/table-id".to_string(), + metadata_location: "s3://analytics/tables/table-id/metadata/v1.metadata.json".to_string(), + version_token: "token-v1".to_string(), + generation: 1, + state: TableCatalogEntryState::Active, + properties: BTreeMap::new(), + created_at: None, + updated_at: None, + }; + let mut value = serde_json::to_value(table).unwrap(); + value + .as_object_mut() + .unwrap() + .insert("unexpected".to_string(), serde_json::json!(true)); + + assert!(serde_json::from_value::(value).is_err()); + } + + struct NoopTableCatalogStore; + + #[async_trait::async_trait] + impl TableCatalogStore for NoopTableCatalogStore { + async fn get_table_bucket(&self, _table_bucket: &str) -> TableCatalogStoreResult> { + Ok(None) + } + + async fn put_table_bucket(&self, _entry: TableBucketEntry) -> TableCatalogStoreResult<()> { + Ok(()) + } + + async fn create_namespace(&self, _entry: NamespaceEntry) -> TableCatalogStoreResult<()> { + Ok(()) + } + + async fn list_namespaces(&self, _table_bucket: &str) -> TableCatalogStoreResult> { + Ok(Vec::new()) + } + + async fn get_namespace(&self, _table_bucket: &str, _namespace: &str) -> TableCatalogStoreResult> { + Ok(None) + } + + async fn drop_namespace(&self, _table_bucket: &str, _namespace: &str) -> TableCatalogStoreResult<()> { + Ok(()) + } + + async fn create_table(&self, _entry: TableEntry) -> TableCatalogStoreResult<()> { + Ok(()) + } + + async fn register_table(&self, _entry: TableEntry) -> TableCatalogStoreResult<()> { + Ok(()) + } + + async fn list_tables(&self, _table_bucket: &str, _namespace: &str) -> TableCatalogStoreResult> { + Ok(Vec::new()) + } + + async fn load_table( + &self, + _table_bucket: &str, + _namespace: &str, + _table: &str, + ) -> TableCatalogStoreResult> { + Ok(None) + } + + async fn commit_table(&self, request: TableCommitRequest) -> TableCatalogStoreResult { + let table = TableEntry { + version: TABLE_CATALOG_ENTRY_VERSION, + table_bucket: request.table_bucket, + namespace: request.namespace, + table: request.table, + table_id: "table-id".to_string(), + table_uuid: "table-uuid".to_string(), + format: "ICEBERG".to_string(), + format_version: 2, + warehouse_location: "s3://analytics/tables/table-id".to_string(), + metadata_location: request.new_metadata_location.clone(), + version_token: "token-v2".to_string(), + generation: 2, + state: TableCatalogEntryState::Active, + properties: BTreeMap::new(), + created_at: None, + updated_at: None, + }; + let commit_log = CommitLogEntry { + version: TABLE_CATALOG_ENTRY_VERSION, + commit_id: request.commit_id, + idempotency_key: request.idempotency_key, + table_id: table.table_id.clone(), + operation: request.operation, + expected_version_token: request.expected_version_token, + new_version_token: table.version_token.clone(), + previous_metadata_location: request.expected_metadata_location, + new_metadata_location: table.metadata_location.clone(), + requirements: request.requirements, + status: CommitLogStatus::Committed, + writer: request.writer, + created_at: None, + updated_at: None, + }; + + Ok(TableCommitResult { table, commit_log }) + } + + async fn drop_table(&self, _table_bucket: &str, _namespace: &str, _table: &str) -> TableCatalogStoreResult<()> { + Ok(()) + } + + async fn get_commit_by_id(&self, _table_id: &str, _commit_id: &str) -> TableCatalogStoreResult> { + Ok(None) + } + + async fn get_commit_by_idempotency_key( + &self, + _table_id: &str, + _idempotency_key: &str, + ) -> TableCatalogStoreResult> { + Ok(None) + } + } + + #[tokio::test] + async fn table_catalog_store_trait_covers_entry_read_write_shapes() { + let store: &dyn TableCatalogStore = &NoopTableCatalogStore; + + assert!(store.get_table_bucket("analytics").await.unwrap().is_none()); + assert!(store.list_namespaces("analytics").await.unwrap().is_empty()); + assert!(store.get_commit_by_id("table-id", "commit-id").await.unwrap().is_none()); + assert!( + store + .get_commit_by_idempotency_key("table-id", "client-request-id") + .await + .unwrap() + .is_none() + ); + } + + #[tokio::test] + async fn table_catalog_store_trait_has_atomic_commit_shape() { + let store: &dyn TableCatalogStore = &NoopTableCatalogStore; + let request = TableCommitRequest { + table_bucket: "analytics".to_string(), + namespace: "sales".to_string(), + table: "orders".to_string(), + commit_id: "commit-id".to_string(), + idempotency_key: Some("client-request-id".to_string()), + operation: "append".to_string(), + expected_version_token: "token-v1".to_string(), + expected_metadata_location: "s3://analytics/tables/table-id/metadata/v1.metadata.json".to_string(), + new_metadata_location: "s3://analytics/tables/table-id/metadata/v2.metadata.json".to_string(), + requirements: vec![serde_json::json!({"type": "assert-table-uuid", "uuid": "table-uuid"})], + writer: Some("pyiceberg/test".to_string()), + }; + + let result = store.commit_table(request).await.unwrap(); + + assert_eq!(result.table.version_token, "token-v2"); + assert_eq!(result.table.generation, 2); + assert_eq!(result.commit_log.status, CommitLogStatus::Committed); + } + #[test] fn namespace_marker_path_stays_under_default_reserved_boundary() { let namespace = Namespace::parse("analytics.daily_events").unwrap();