diff --git a/crates/paimon/src/catalog/filesystem.rs b/crates/paimon/src/catalog/filesystem.rs index b34d4472..269d7b87 100644 --- a/crates/paimon/src/catalog/filesystem.rs +++ b/crates/paimon/src/catalog/filesystem.rs @@ -189,20 +189,51 @@ 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_dir(&self.database_path(name)).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 { let table_path = self.table_path(identifier); - let schema_manager = SchemaManager::new(self.file_io.clone(), table_path); + 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. + /// + /// 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_manager = SchemaManager::new(self.file_io.clone(), table_path.to_string()); if self.file_io.exists(&schema_manager.schema_path(0)).await? { return Ok(true); } - - Ok(!schema_manager.list_all_ids().await?.is_empty()) + match schema_manager.list_all_ids().await { + Ok(ids) => Ok(!ids.is_empty()), + Err(Error::IoUnexpected { ref source, .. }) + if source.kind() == opendal::ErrorKind::NotFound => + { + Ok(false) + } + Err(e) => Err(e), + } } } @@ -329,13 +360,15 @@ impl Catalog for FileSystemCatalog { }); } + let dirs = self.list_directories(&path).await?; let mut tables = Vec::new(); - for table_name in self.list_directories(&path).await? { - let identifier = Identifier::new(database_name, &table_name); - if self.table_exists(&identifier).await? { - tables.push(table_name); + 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) } @@ -1260,6 +1293,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();