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

feat(table-catalog): add external catalog bridge sync (#3569)

Browse Source 
Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
This commit is contained in:
Henry Guo
2026-06-18 15:30:41 +08:00
committed by GitHub
parent 8cf11af649
commit 51409c40ef
7 changed files with 1014 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/admin/handlers/table_catalog.rs
+853 -38
View File
File diff suppressed because it is too large Load Diff
rustfs/src/admin/route_policy.rs
+35 -1
View File
@@ -822,6 +822,18 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
GET_TABLE_METADATA,
RouteRiskLevel::Sensitive,
),
admin(
HttpMethod::Put,
"/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/external",
REGISTER_TABLE,
RouteRiskLevel::High,
),
admin(
HttpMethod::Post,
"/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/external/sync",
SET_TABLE_METADATA_LOCATION,
RouteRiskLevel::High,
),
admin(
HttpMethod::Get,
"/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/diagnostics",
@@ -1051,6 +1063,18 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
GET_TABLE_METADATA,
RouteRiskLevel::Sensitive,
),
admin(
HttpMethod::Put,
"/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/external",
REGISTER_TABLE,
RouteRiskLevel::High,
),
admin(
HttpMethod::Post,
"/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/external/sync",
SET_TABLE_METADATA_LOCATION,
RouteRiskLevel::High,
),
admin(
HttpMethod::Get,
"/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/diagnostics",
@@ -1251,7 +1275,7 @@ mod tests {
let table_specs = ADMIN_ROUTE_POLICY_SPECS
.iter()
.filter(|spec| spec.path().starts_with("/iceberg/v1") || spec.path().starts_with("/_iceberg/v1"));
assert_eq!(table_specs.count(), 78);
assert_eq!(table_specs.count(), 82);
assert_action(HttpMethod::Put, "/iceberg/v1/buckets/{warehouse}", SET_TABLE_BUCKET);
assert_action(HttpMethod::Get, "/_iceberg/v1/buckets/{warehouse}", GET_TABLE_BUCKET);
assert_action(HttpMethod::Get, "/iceberg/v1/{warehouse}/namespaces", GET_TABLE_NAMESPACE);
@@ -1396,6 +1420,16 @@ mod tests {
"/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/external",
GET_TABLE_METADATA,
);
assert_action(
HttpMethod::Put,
"/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/external",
REGISTER_TABLE,
);
assert_action(
HttpMethod::Post,
"/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/external/sync",
SET_TABLE_METADATA_LOCATION,
);
assert_action(
HttpMethod::Post,
"/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/recovery",
rustfs/src/admin/route_registration_test.rs
+40
View File
@@ -437,6 +437,16 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
"/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/external",
"/analytics/namespaces/sales/tables/orders/catalog/external",
),
table_route_sample(
Method::PUT,
"/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/external",
"/analytics/namespaces/sales/tables/orders/catalog/external",
),
table_route_sample(
Method::POST,
"/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/external/sync",
"/analytics/namespaces/sales/tables/orders/catalog/external/sync",
),
table_route_sample(
Method::GET,
"/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/diagnostics",
@@ -600,6 +610,16 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
"/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/external",
"/analytics/namespaces/sales/tables/orders/catalog/external",
),
compat_table_route_sample(
Method::PUT,
"/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/external",
"/analytics/namespaces/sales/tables/orders/catalog/external",
),
compat_table_route_sample(
Method::POST,
"/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/external/sync",
"/analytics/namespaces/sales/tables/orders/catalog/external/sync",
),
compat_table_route_sample(
Method::GET,
"/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/diagnostics",
@@ -834,6 +854,16 @@ fn test_register_routes_cover_representative_admin_paths() {
Method::GET,
&table_catalog_path("/analytics/namespaces/sales/tables/orders/catalog/external"),
);
assert_route(
&router,
Method::PUT,
&table_catalog_path("/analytics/namespaces/sales/tables/orders/catalog/external"),
);
assert_route(
&router,
Method::POST,
&table_catalog_path("/analytics/namespaces/sales/tables/orders/catalog/external/sync"),
);
assert_route(
&router,
Method::GET,
@@ -971,6 +1001,16 @@ fn test_register_routes_cover_representative_admin_paths() {
Method::GET,
&compat_table_catalog_path("/analytics/namespaces/sales/tables/orders/catalog/external"),
);
assert_route(
&router,
Method::PUT,
&compat_table_catalog_path("/analytics/namespaces/sales/tables/orders/catalog/external"),
);
assert_route(
&router,
Method::POST,
&compat_table_catalog_path("/analytics/namespaces/sales/tables/orders/catalog/external/sync"),
);
assert_route(
&router,
Method::GET,
rustfs/src/table_catalog.rs
+81
View File
@@ -71,6 +71,7 @@ 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_MAINTENANCE_CONFIG_VERSION: u16 = 1;
pub(crate) const TABLE_EXTERNAL_CATALOG_BRIDGE_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";
@@ -92,6 +93,8 @@ const INTERNAL_CATALOG_ROOT: &str = BUCKET_TABLE_CATALOG_META_PREFIX;
const TABLE_BUCKET_ROOT: &str = BUCKET_TABLE_CATALOG_TABLE_BUCKETS_PREFIX;
const COMMIT_LOG_ROOT: &str = "commits";
const COMMIT_IDEMPOTENCY_ROOT: &str = "commit-idempotency";
const EXTERNAL_CATALOG_ROOT: &str = "external-catalog";
const EXTERNAL_CATALOG_BRIDGE_FILE: &str = "bridge.json";
const MAINTENANCE_ROOT: &str = "maintenance";
const MAINTENANCE_CONFIG_FILE: &str = "config.json";
const MAINTENANCE_JOB_ROOT: &str = "jobs";
@@ -367,6 +370,32 @@ pub(crate) struct TableCommitResult {
pub commit_log: CommitLogEntry,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct ExternalCatalogBridgeEntry {
pub version: u16,
pub table_bucket: String,
pub namespace: String,
pub table: String,
pub catalog: String,
pub external_catalog_id: Option<String>,
pub external_namespace: String,
pub external_table: String,
pub external_table_uuid: Option<String>,
pub metadata_location: Option<String>,
pub external_version_token: Option<String>,
pub policy_mode: String,
pub credential_mode: String,
pub sync_mode: String,
pub rollback_strategy: String,
pub last_sync_status: Option<String>,
pub last_synced_metadata_location: Option<String>,
#[serde(default)]
pub properties: BTreeMap<String, String>,
pub created_at: Option<String>,
pub updated_at: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct ViewCommitRequest {
@@ -1225,6 +1254,14 @@ impl TableCatalogObjectPaths {
)
}
pub fn external_catalog_bridge_path(&self, table_bucket: &str, namespace: &Namespace, table: &IdentifierSegment) -> String {
format!(
"{}{}/{EXTERNAL_CATALOG_ROOT}/{EXTERNAL_CATALOG_BRIDGE_FILE}",
self.table_entries_prefix(table_bucket, namespace),
table.as_str()
)
}
pub fn view_entries_prefix(&self, table_bucket: &str, namespace: &Namespace) -> String {
format!("{}{}/{}/", self.namespace_entries_prefix(table_bucket), namespace.storage_id(), VIEW_ROOT)
}
@@ -1527,6 +1564,50 @@ where
.await
}
pub(crate) async fn get_external_catalog_bridge(
&self,
table_bucket: &str,
namespace: &str,
table: &str,
) -> TableCatalogStoreResult<Option<ExternalCatalogBridgeEntry>> {
self.require_table_bucket(table_bucket).await?;
let namespace = parse_namespace_for_store(namespace)?;
let table = parse_table_for_store(table)?;
if self.get_namespace(table_bucket, &namespace.public_name()).await?.is_none() {
return Err(TableCatalogStoreError::NotFound(format!(
"namespace {}/{}",
table_bucket,
namespace.public_name()
)));
}
let bridge_path = self.paths.external_catalog_bridge_path(table_bucket, &namespace, &table);
self.read_entry::<ExternalCatalogBridgeEntry>(self.catalog_bucket(), &bridge_path)
.await
.map(|entry| entry.map(|(bridge, _)| bridge))
}
pub(crate) async fn put_external_catalog_bridge(
&self,
entry: ExternalCatalogBridgeEntry,
) -> TableCatalogStoreResult<ExternalCatalogBridgeEntry> {
validate_catalog_entry_version("external catalog bridge", entry.version)?;
self.require_table_bucket(&entry.table_bucket).await?;
let namespace = parse_namespace_for_store(&entry.namespace)?;
let table = parse_table_for_store(&entry.table)?;
if self.get_namespace(&entry.table_bucket, &entry.namespace).await?.is_none() {
return Err(TableCatalogStoreError::NotFound(format!(
"namespace {}/{}",
entry.table_bucket, entry.namespace
)));
}
let bridge_path = self
.paths
.external_catalog_bridge_path(&entry.table_bucket, &namespace, &table);
self.write_entry(self.catalog_bucket(), &bridge_path, &entry, TableCatalogPutPrecondition::Any)
.await?;
Ok(entry)
}
async fn read_commit_by_path(&self, object: &str) -> TableCatalogStoreResult<Option<CommitLogEntry>> {
self.read_entry::<CommitLogEntry>(self.catalog_bucket(), object)
.await
scripts/table-catalog/README.md
+1 -1
View File
@@ -179,7 +179,7 @@ current unsupported inventory is:
- automatic maintenance scheduling: external scheduler hook supported through the worker run endpoint; built-in periodic scheduling is not claimed
- compaction rewrite: controlled run-once support for unpartitioned Parquet binpack through metadata maintenance; built-in periodic scheduling, sort compaction, delete-file rewrite, and row-level compaction are not claimed
- row-level delete/update/merge commits: standard catalog commit validates append, overwrite, delete, and replace snapshot manifests for table-warehouse scope, referenced object existence, current-live-file deletes, and stale add/delete conflicts; end-to-end SQL DML client coverage remains a compatibility validation item
- external catalog bridges: metadata import/register is supported, but Polaris/Glue/DLF/Hive synchronization is unsupported
- external catalog bridges: metadata import/register and operator-supplied metadata pointer sync are supported for Polaris/Glue/DLF/Hive identity boundaries; online vendor SDK polling and policy mirroring are not claimed
- multi-table transactions: not a short-term production claim
## Credential Boundary
scripts/table-catalog/pyiceberg_smoke.py
+2 -2
View File
@@ -166,9 +166,9 @@ UNSUPPORTED_INVENTORY: list[dict[str, str]] = [
},
{
"capability": "external-catalog-bridge",
"status": "metadata-import-only",
"status": "operator-sync-supported",
"roadmap_area": "external-catalog",
"expected_behavior": "catalog import/register supports an existing Iceberg metadata location, while Polaris, Glue, DLF, and Hive synchronization remain unsupported",
"expected_behavior": "catalog import/register and operator-supplied external metadata pointer sync are supported for Polaris, Glue, DLF, and Hive identity boundaries; online vendor SDK polling and policy mirroring are not claimed",
},
{
"capability": "multi-table-transactions",
scripts/table-catalog/test_pyiceberg_smoke.py
+2
View File
@@ -747,6 +747,8 @@ class PyIcebergSmokeConfigTest(unittest.TestCase):
self.assertIn("background-maintenance-worker", capabilities)
self.assertIn("external-catalog-bridge", capabilities)
self.assertIn("multi-table-transactions", capabilities)
external_bridge = next(entry for entry in inventory if entry["capability"] == "external-catalog-bridge")
self.assertEqual(external_bridge["status"], "operator-sync-supported")
for entry in inventory:
self.assertIn("status", entry)
self.assertIn("roadmap_area", entry)