From e8ce57399d15dd3a58cd346734f7703e5921c51f Mon Sep 17 00:00:00 2001 From: mlboy Date: Fri, 17 Jul 2026 03:57:43 +0800 Subject: [PATCH 1/5] fix(catalog): align FileSystemCatalog with Java for object store compatibility - database_exists: append trailing slash so OpenDAL correctly identifies directory paths on OSS/S3 (Hadoop does this via getFileStatus fallback) - table_exists: check schema file presence instead of bare directory existence, mirroring Java's AbstractCatalog.tableExistsInFileSystem - list_tables: filter directories by schema validity and sort results, matching Java's listTablesInFileSystem behavior --- crates/paimon/src/catalog/filesystem.rs | 53 +++++++++++++++++++++++-- 1 file changed, 49 insertions(+), 4 deletions(-) diff --git a/crates/paimon/src/catalog/filesystem.rs b/crates/paimon/src/catalog/filesystem.rs index 2fe7bb78..fc2e6aa3 100644 --- a/crates/paimon/src/catalog/filesystem.rs +++ b/crates/paimon/src/catalog/filesystem.rs @@ -189,13 +189,47 @@ impl FileSystemCatalog { } /// Check if a database exists. + /// + /// Uses a trailing slash so that object stores (OSS, S3, etc.) correctly + /// identify the path as a directory rather than a file. + /// Without trailing slash, `op.exists("dir")` may return false on object + /// stores even when `dir/` prefix contains objects. + /// See: async fn database_exists(&self, name: &str) -> Result { - self.file_io.exists(&self.database_path(name)).await + let raw = self.database_path(name); + let path = format!("{}/", raw.trim_end_matches('/')); + self.file_io.exists(&path).await } - /// Check if a table exists. + /// Check if a table exists by verifying that at least one schema file is + /// present under `{table_path}/schema/`. + /// + /// This mirrors Java's `AbstractCatalog.tableExistsInFileSystem` which + /// checks `schema-0` first, then falls back to listing all schema IDs. + /// A bare directory without schema files is NOT considered a valid table. async fn table_exists(&self, identifier: &Identifier) -> Result { - self.file_io.exists(&self.table_path(identifier)).await + let table_path = self.table_path(identifier); + self.table_exists_in_filesystem(&table_path).await + } + + /// Verify a table path contains valid schema metadata. + /// + /// Mirrors Java's `AbstractCatalog.tableExistsInFileSystem`: + /// 1. Fast path — check if `schema/schema-0` exists. + /// 2. Slow path — list schema directory for any `schema-{N}` file. + /// + /// If the schema directory is missing entirely (e.g. on local filesystem), + /// the list operation may error; this is treated as "table does not exist". + async fn table_exists_in_filesystem(&self, table_path: &str) -> Result { + let schema_0_path = self.schema_file_path(table_path, 0); + if self.file_io.exists(&schema_0_path).await? { + return Ok(true); + } + let manager = SchemaManager::new(self.file_io.clone(), table_path.to_string()); + match manager.list_all_ids().await { + Ok(ids) => Ok(!ids.is_empty()), + Err(_) => Ok(false), + } } } @@ -322,7 +356,18 @@ impl Catalog for FileSystemCatalog { }); } - self.list_directories(&path).await + // Mirror Java's listTablesInFileSystem: only include directories that + // contain at least one schema file (i.e. are valid Paimon tables). + let dirs = self.list_directories(&path).await?; + let mut tables = Vec::new(); + for dir in dirs { + let table_path = make_path(&path, &dir); + if self.table_exists_in_filesystem(&table_path).await? { + tables.push(dir); + } + } + tables.sort(); + Ok(tables) } async fn create_table( From af68738c9b3493854563d3199fe3e9ea2820a910 Mon Sep 17 00:00:00 2001 From: mlboy Date: Fri, 17 Jul 2026 14:56:25 +0800 Subject: [PATCH 2/5] fix(catalog): only treat NotFound as table-not-exist, propagate other errors The previous `Err(_) => Ok(false)` swallowed all list_all_ids errors including permission denied and transient storage failures, which could cause create_table to overwrite an existing table's schema. Now only opendal::ErrorKind::NotFound (schema directory missing) maps to Ok(false). All other errors propagate to the caller. --- crates/paimon/src/catalog/filesystem.rs | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/crates/paimon/src/catalog/filesystem.rs b/crates/paimon/src/catalog/filesystem.rs index fc2e6aa3..1cbf16cd 100644 --- a/crates/paimon/src/catalog/filesystem.rs +++ b/crates/paimon/src/catalog/filesystem.rs @@ -218,8 +218,9 @@ impl FileSystemCatalog { /// 1. Fast path — check if `schema/schema-0` exists. /// 2. Slow path — list schema directory for any `schema-{N}` file. /// - /// If the schema directory is missing entirely (e.g. on local filesystem), - /// the list operation may error; this is treated as "table does not exist". + /// Only a `NotFound` error from the schema directory is treated as + /// "table does not exist." All other errors (permission, network, etc.) + /// are propagated to avoid masking real failures. async fn table_exists_in_filesystem(&self, table_path: &str) -> Result { let schema_0_path = self.schema_file_path(table_path, 0); if self.file_io.exists(&schema_0_path).await? { @@ -228,7 +229,12 @@ impl FileSystemCatalog { let manager = SchemaManager::new(self.file_io.clone(), table_path.to_string()); match manager.list_all_ids().await { Ok(ids) => Ok(!ids.is_empty()), - Err(_) => Ok(false), + Err(Error::IoUnexpected { ref source, .. }) + if source.kind() == opendal::ErrorKind::NotFound => + { + Ok(false) + } + Err(e) => Err(e), } } } From 3fe09d32a349cbbf5906ba62f8e2973f72bbd24b Mon Sep 17 00:00:00 2001 From: mlboy Date: Fri, 17 Jul 2026 15:20:36 +0800 Subject: [PATCH 3/5] relax unicode-segmentation pin to allow >=1.13.2 The exact pin =1.13.2 was for upstream Vortex CI stability but conflicts with datafusion 53.x which resolves to 1.13.3. --- crates/paimon/Cargo.toml | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/crates/paimon/Cargo.toml b/crates/paimon/Cargo.toml index d33debf6..002c560f 100644 --- a/crates/paimon/Cargo.toml +++ b/crates/paimon/Cargo.toml @@ -107,9 +107,8 @@ paimon-mosaic-core = "0.2.0" paimon-vindex-core = "0.2.0" vortex = { version = "0.75.0", features = ["tokio"], optional = true } libloading = "0.9" -# Keep CI on the dependency set that passed before unicode-segmentation 1.13.3. -# The 1.13.3 resolver update correlates with Linux Vortex tests hanging. -unicode-segmentation = "=1.13.2" +# Relaxed from =1.13.2 (upstream pinned for Vortex CI issue, not relevant here) +unicode-segmentation = ">=1.13.2" [dev-dependencies] axum = { version = "0.7", features = ["macros", "tokio", "http1", "http2"] } From a7204d9e76cedeb094486ad1d298735b1924a421 Mon Sep 17 00:00:00 2001 From: mlboy Date: Fri, 17 Jul 2026 16:49:13 +0800 Subject: [PATCH 4/5] revert: restore unicode-segmentation =1.13.2 pin The fix should be applied downstream in lakeserve instead. --- crates/paimon/Cargo.toml | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/crates/paimon/Cargo.toml b/crates/paimon/Cargo.toml index 002c560f..d33debf6 100644 --- a/crates/paimon/Cargo.toml +++ b/crates/paimon/Cargo.toml @@ -107,8 +107,9 @@ paimon-mosaic-core = "0.2.0" paimon-vindex-core = "0.2.0" vortex = { version = "0.75.0", features = ["tokio"], optional = true } libloading = "0.9" -# Relaxed from =1.13.2 (upstream pinned for Vortex CI issue, not relevant here) -unicode-segmentation = ">=1.13.2" +# Keep CI on the dependency set that passed before unicode-segmentation 1.13.3. +# The 1.13.3 resolver update correlates with Linux Vortex tests hanging. +unicode-segmentation = "=1.13.2" [dev-dependencies] axum = { version = "0.7", features = ["macros", "tokio", "http1", "http2"] } From ae6d83b022ac26fad896de560d8b3536f459530a Mon Sep 17 00:00:00 2001 From: mlboy Date: Fri, 17 Jul 2026 19:35:56 +0800 Subject: [PATCH 5/5] test(catalog): add regression tests for table_exists_in_filesystem Cover the previously unverified behaviors: - schema fallback when schema-0 is absent but schema-N exists - markerless directory (no schema/ subdir) returns false - empty schema directory returns false - list_tables filters out invalid directories - non-NotFound errors (e.g. permission denied) are propagated --- crates/paimon/src/catalog/filesystem.rs | 158 ++++++++++++++++++++++++ 1 file changed, 158 insertions(+) diff --git a/crates/paimon/src/catalog/filesystem.rs b/crates/paimon/src/catalog/filesystem.rs index 1cbf16cd..bf69c201 100644 --- a/crates/paimon/src/catalog/filesystem.rs +++ b/crates/paimon/src/catalog/filesystem.rs @@ -1211,6 +1211,164 @@ mod tests { .unwrap(); } + // ─── table_exists_in_filesystem regression tests ───────────────────────── + + /// When `schema-0` is absent but another schema file (e.g. `schema-5`) exists, + /// the slow-path fallback via `list_all_ids` should detect the table. + #[tokio::test] + async fn test_table_exists_schema_fallback_without_schema_0() { + let (temp_dir, catalog) = create_test_catalog(); + catalog + .create_database("db", false, HashMap::new()) + .await + .unwrap(); + + let table_path = format!( + "{}/{}/mytable", + temp_dir.path().to_str().unwrap(), + "db.db" + ); + let schema_dir = format!("{table_path}/schema"); + std::fs::create_dir_all(&schema_dir).unwrap(); + + // Write only schema-5 (skip schema-0) to trigger the slow path. + let schema = testing_schema(); + let table_schema = TableSchema::new(5, &schema); + let json = serde_json::to_string(&table_schema).unwrap(); + std::fs::write(format!("{schema_dir}/schema-5"), json).unwrap(); + + assert!( + catalog + .table_exists_in_filesystem(&table_path) + .await + .unwrap(), + "table with schema-5 (but no schema-0) should be detected via fallback" + ); + } + + /// A directory without any `schema-{N}` file (markerless prefix) should NOT + /// be recognized as a valid table. + #[tokio::test] + async fn test_table_exists_returns_false_for_markerless_directory() { + let (temp_dir, catalog) = create_test_catalog(); + catalog + .create_database("db", false, HashMap::new()) + .await + .unwrap(); + + let table_path = format!("{}/{}/bare_dir", temp_dir.path().to_str().unwrap(), "db.db"); + // Create the directory but no schema/ subdirectory at all. + std::fs::create_dir_all(&table_path).unwrap(); + + assert!( + !catalog + .table_exists_in_filesystem(&table_path) + .await + .unwrap(), + "directory without schema/ should not be treated as a table" + ); + } + + /// A directory with an empty `schema/` subdirectory (no schema-N files) + /// should NOT be recognized as a valid table. + #[tokio::test] + async fn test_table_exists_returns_false_for_empty_schema_dir() { + let (temp_dir, catalog) = create_test_catalog(); + catalog + .create_database("db", false, HashMap::new()) + .await + .unwrap(); + + let table_path = format!( + "{}/{}/empty_schema", + temp_dir.path().to_str().unwrap(), + "db.db" + ); + let schema_dir = format!("{table_path}/schema"); + std::fs::create_dir_all(&schema_dir).unwrap(); + + assert!( + !catalog + .table_exists_in_filesystem(&table_path) + .await + .unwrap(), + "directory with empty schema/ should not be treated as a table" + ); + } + + /// `list_tables` must filter out directories that exist but lack valid schema + /// metadata (invalid-directory filtering). + #[tokio::test] + async fn test_list_tables_filters_invalid_directories() { + let (temp_dir, catalog) = create_test_catalog(); + catalog + .create_database("db", false, HashMap::new()) + .await + .unwrap(); + + // Create a valid table via the catalog API. + let id = Identifier::new("db", "valid_table"); + catalog + .create_table(&id, testing_schema(), false) + .await + .unwrap(); + + // Manually create bare directories (markerless prefixes) that should be + // excluded from the listing. + let db_path = format!("{}/{}", temp_dir.path().to_str().unwrap(), "db.db"); + std::fs::create_dir_all(format!("{db_path}/no_schema_at_all")).unwrap(); + std::fs::create_dir_all(format!("{db_path}/has_empty_schema/schema")).unwrap(); + + let tables = catalog.list_tables("db").await.unwrap(); + assert_eq!( + tables, + vec!["valid_table".to_string()], + "only directories with valid schema files should be listed" + ); + } + + /// Non-`NotFound` errors (e.g. permission denied) from the schema directory + /// must be propagated, not swallowed as "table does not exist". + #[cfg(unix)] + #[tokio::test] + async fn test_table_exists_propagates_non_not_found_error() { + use std::os::unix::fs::PermissionsExt; + + let (temp_dir, catalog) = create_test_catalog(); + catalog + .create_database("db", false, HashMap::new()) + .await + .unwrap(); + + let table_path = format!( + "{}/{}/forbidden", + temp_dir.path().to_str().unwrap(), + "db.db" + ); + let schema_dir = format!("{table_path}/schema"); + std::fs::create_dir_all(&schema_dir).unwrap(); + + // Remove all permissions so listing will fail with PermissionDenied. + std::fs::set_permissions(&schema_dir, std::fs::Permissions::from_mode(0o000)).unwrap(); + + // Probe whether the OS actually enforces the restriction (root bypasses). + let enforced = std::fs::read_dir(&schema_dir).is_err(); + + let result = catalog.table_exists_in_filesystem(&table_path).await; + + // Restore permissions before asserting so cleanup can succeed. + std::fs::set_permissions(&schema_dir, std::fs::Permissions::from_mode(0o755)).unwrap(); + + if enforced { + assert!( + result.is_err(), + "permission error should be propagated, not treated as 'table not found'; got {result:?}" + ); + } + } + + // ─── end table_exists_in_filesystem regression tests ──────────────────────── + #[tokio::test] async fn test_alter_table_rename_primary_key_column_propagates() { let (_tmp, catalog) = create_test_catalog();