This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 7c56827 fix(catalog): handle filesystem catalog existence on object
stores (#528)
7c56827 is described below
commit 7c568274a4c5df6bd00e8d60edea21395412202b
Author: Pandas <[email protected]>
AuthorDate: Sat Jul 18 10:22:05 2026 +0800
fix(catalog): handle filesystem catalog existence on object stores (#528)
---
crates/paimon/src/catalog/filesystem.rs | 108 ++++++++++++++++++++++++++++++--
crates/paimon/src/io/file_io.rs | 28 +++++++++
2 files changed, 132 insertions(+), 4 deletions(-)
diff --git a/crates/paimon/src/catalog/filesystem.rs
b/crates/paimon/src/catalog/filesystem.rs
index 2fe7bb7..b34d447 100644
--- a/crates/paimon/src/catalog/filesystem.rs
+++ b/crates/paimon/src/catalog/filesystem.rs
@@ -190,12 +190,19 @@ impl FileSystemCatalog {
/// Check if a database exists.
async fn database_exists(&self, name: &str) -> Result<bool> {
- self.file_io.exists(&self.database_path(name)).await
+ self.file_io.exists_dir(&self.database_path(name)).await
}
/// Check if a table exists.
async fn table_exists(&self, identifier: &Identifier) -> Result<bool> {
- self.file_io.exists(&self.table_path(identifier)).await
+ let table_path = self.table_path(identifier);
+ let schema_manager = SchemaManager::new(self.file_io.clone(),
table_path);
+
+ if self.file_io.exists(&schema_manager.schema_path(0)).await? {
+ return Ok(true);
+ }
+
+ Ok(!schema_manager.list_all_ids().await?.is_empty())
}
}
@@ -273,7 +280,7 @@ impl Catalog for FileSystemCatalog {
return Ok(());
}
- let tables = self.list_directories(&path).await?;
+ let tables = self.list_tables(name).await?;
if !tables.is_empty() && !cascade {
return Err(Error::DatabaseNotEmpty {
database: name.to_string(),
@@ -322,7 +329,14 @@ impl Catalog for FileSystemCatalog {
});
}
- 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);
+ }
+ }
+ Ok(tables)
}
async fn create_table(
@@ -482,6 +496,12 @@ mod tests {
(temp_dir, catalog)
}
+ fn create_memory_catalog() -> FileSystemCatalog {
+ let mut options = Options::new();
+ options.set(CatalogOptions::WAREHOUSE, "memory:/warehouse");
+ FileSystemCatalog::new(options).unwrap()
+ }
+
fn testing_schema() -> Schema {
Schema::builder()
.column(
@@ -552,6 +572,54 @@ mod tests {
assert!(!catalog.database_exists("db1").await.unwrap());
}
+ #[tokio::test]
+ async fn test_memory_database_directory_existence() {
+ let catalog = create_memory_catalog();
+
+ catalog
+ .create_database("empty", false, HashMap::new())
+ .await
+ .unwrap();
+ catalog.get_database("empty").await.unwrap();
+
+ let result = catalog
+ .create_database("empty", false, HashMap::new())
+ .await;
+ assert!(matches!(result, Err(Error::DatabaseAlreadyExist { .. })));
+
+ catalog
+ .file_io()
+ .new_output("memory:/warehouse/markerless.db/table/data/file")
+ .unwrap()
+ .write(Bytes::from("data"))
+ .await
+ .unwrap();
+ catalog.get_database("markerless").await.unwrap();
+ assert!(catalog
+ .list_databases()
+ .await
+ .unwrap()
+ .contains(&"markerless".to_string()));
+ }
+
+ #[tokio::test]
+ async fn test_drop_database_ignores_incomplete_table_directories() {
+ let catalog = create_memory_catalog();
+ catalog
+ .create_database("db1", false, HashMap::new())
+ .await
+ .unwrap();
+ catalog
+ .file_io()
+ .mkdirs("memory:/warehouse/db1.db/incomplete")
+ .await
+ .unwrap();
+
+ assert!(catalog.list_tables("db1").await.unwrap().is_empty());
+ catalog.drop_database("db1", false, false).await.unwrap();
+ assert!(!catalog.database_exists("db1").await.unwrap());
+ }
+
#[tokio::test]
async fn test_create_database_should_reject_path_traversal_name() {
let (temp_dir, catalog) = create_test_catalog();
@@ -634,6 +702,38 @@ mod tests {
assert!(!tables.contains(&"table1".to_string()));
}
+ #[tokio::test]
+ async fn test_memory_table_existence_requires_schema() {
+ let catalog = create_memory_catalog();
+ let valid_table = Identifier::new("db1", "valid");
+ let invalid_table = Identifier::new("db1", "invalid");
+ let table_schema = TableSchema::new(1, &testing_schema());
+
+ catalog
+ .file_io()
+ .new_output("memory:/warehouse/db1.db/valid/schema/schema-1")
+ .unwrap()
+ .write(Bytes::from(serde_json::to_string(&table_schema).unwrap()))
+ .await
+ .unwrap();
+ catalog
+ .file_io()
+ .mkdirs("memory:/warehouse/db1.db/invalid")
+ .await
+ .unwrap();
+
+ assert!(catalog.table_exists(&valid_table).await.unwrap());
+ let table = catalog.get_table(&valid_table).await.unwrap();
+ assert_eq!(table.schema().id(), 1);
+ let tables = catalog.list_tables("db1").await.unwrap();
+ assert!(tables.contains(&"valid".to_string()));
+ assert!(!tables.contains(&"invalid".to_string()));
+
+ assert!(!catalog.table_exists(&invalid_table).await.unwrap());
+ let result = catalog.get_table(&invalid_table).await;
+ assert!(matches!(result, Err(Error::TableNotExist { .. })));
+ }
+
#[tokio::test]
async fn test_create_table_should_reject_path_traversal_object_name() {
let (temp_dir, catalog) = create_test_catalog();
diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs
index 4a62c73..1e7517b 100644
--- a/crates/paimon/src/io/file_io.rs
+++ b/crates/paimon/src/io/file_io.rs
@@ -215,6 +215,16 @@ impl FileIO {
})
}
+ /// Check if a directory exists.
+ pub async fn exists_dir(&self, path: &str) -> Result<bool> {
+ let (op, relative_path) = self.storage.create(path)?;
+ let dir_path = normalize_root(relative_path.as_ref());
+
+ op.exists(&dir_path).await.context(IoUnexpectedSnafu {
+ message: format!("Failed to check existence of directory
'{path}'"),
+ })
+ }
+
/// Delete a file.
///
/// Reference:
<https://github.com/apache/paimon/blob/release-0.8.2/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java#L139>
@@ -622,6 +632,24 @@ mod file_action_test {
assert!(matches!(result, Err(Error::ConfigInvalid { .. })));
}
+ #[tokio::test]
+ async fn test_exists_dir_memory() {
+ let file_io = setup_memory_file_io();
+
+ file_io.mkdirs("memory:/empty").await.unwrap();
+ assert!(file_io.exists_dir("memory:/empty").await.unwrap());
+
+ file_io
+ .new_output("memory:/markerless/child")
+ .unwrap()
+ .write(Bytes::from("data"))
+ .await
+ .unwrap();
+ assert!(file_io.exists_dir("memory:/markerless").await.unwrap());
+
+ assert!(!file_io.exists_dir("memory:/missing").await.unwrap());
+ }
+
#[tokio::test]
async fn test_memory_operator_reuse_across_file_io_calls() {
let file_io = setup_memory_file_io();
