fix(io): delete only the named directory in FileIO::delete_dir

crates/integrations/datafusion/tests/rest_format_partition_sql.rs

@@ -562,6 +562,33 @@ async fn test_drop_partition_leaves_a_custom_location_in_place() {
     assert!(temp_dir.path().join("dt=b").is_dir());
 }
 
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_drop_partition_deletes_the_directory_it_unregisters() {
+    let temp_dir = tempfile::tempdir().unwrap();
+    let (server, context) =
+        setup_rest_table(&temp_dir, format_table_schema(&[("dt", varchar())])).await;
+    // The two partitions differ only by a trailing space, which the deleted path must keep.
+    common::exec(
+        &context,
+        &format!("ALTER TABLE {TABLE_NAME} ADD PARTITION (dt = 'a') PARTITION (dt = 'a ')"),
+    )
+    .await;
+    assert!(temp_dir.path().join("dt=a ").is_dir());
+
+    common::exec(
+        &context,
+        &format!("ALTER TABLE {TABLE_NAME} DROP PARTITION (dt = 'a ')"),
+    )
+    .await;
+    assert_eq!(
+        server.table_partition_specs(DATABASE, TABLE),
+        vec![spec(&[("dt", "a")])]
+    );
+    assert!(temp_dir.path().join("dt=a").is_dir());
+    assert!(!temp_dir.path().join("dt=a ").exists());
+}
+
 #[cfg(not(windows))]
 #[tokio::test]
 async fn test_msck_repair_reconciles_registrations_with_directories() {

crates/paimon/src/catalog/filesystem.rs

@@ -989,6 +989,31 @@ mod tests {
         assert!(!catalog.database_exists("db1").await.unwrap());
     }
 
+    #[tokio::test]
+    async fn test_drop_table_keeps_tables_sharing_its_name_prefix() {
+        // The memory backend deletes recursively by key prefix, as S3 and OSS do.
+        let catalog = create_memory_catalog();
+        catalog
+            .create_database("db1", false, HashMap::new())
+            .await
+            .unwrap();
+        for name in ["orders", "orders_bak", "orders2024"] {
+            catalog
+                .create_table(&Identifier::new("db1", name), testing_schema(), false)
+                .await
+                .unwrap();
+        }
+
+        catalog
+            .drop_table(&Identifier::new("db1", "orders"), false)
+            .await
+            .unwrap();
+
+        let mut tables = catalog.list_tables("db1").await.unwrap();
+        tables.sort();
+        assert_eq!(tables, ["orders2024", "orders_bak"]);
+    }
+
     #[tokio::test]
     async fn test_create_database_should_reject_path_traversal_name() {
         let (temp_dir, catalog) = create_test_catalog();

crates/paimon/src/io/file_io.rs

@@ -656,8 +656,12 @@ impl FileIO {
     /// Reference: <https://github.com/apache/paimon/blob/release-0.8.2/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java#L139>
     pub async fn delete_dir(&self, path: &str) -> Result<()> {
         let (op, relative_path, cache_path) = self.create_with_cache_path(path).await?;
+        // Without a trailing `/`, OpenDAL trims the path's surrounding whitespace,
+        // and a recursive delete it emulates by listing removes every key with
+        // this prefix.
+        let dir_path = normalize_root(relative_path.as_ref());
 
-        op.delete_with(relative_path.as_ref())
+        op.delete_with(&dir_path)
             .recursive(true)
             .await
             .map_err(|error| {
@@ -2044,6 +2048,55 @@ mod file_action_test {
             assert!(file_io.exists(&path).await.unwrap());
         });
     }
+
+    #[tokio::test]
+    async fn test_delete_dir_keeps_directories_sharing_its_prefix() {
+        // The memory backend deletes recursively by key prefix, as S3 and OSS do.
+        let file_io = setup_memory_file_io();
+        let partitions = ["month=1", "month=10", "month=2"];
+        for partition in partitions {
+            file_io
+                .new_output(&format!("memory:/delete_dir_prefix/{partition}/data"))
+                .unwrap()
+                .write(Bytes::from("data"))
+                .await
+                .unwrap();
+        }
+
+        file_io
+            .delete_dir("memory:/delete_dir_prefix/month=1")
+            .await
+            .unwrap();
+
+        let mut kept = Vec::new();
+        for partition in partitions {
+            let path = format!("memory:/delete_dir_prefix/{partition}/data");
+            if file_io.exists(&path).await.unwrap() {
+                kept.push(partition);
+            }
+        }
+        assert_eq!(kept, ["month=10", "month=2"]);
+    }
+
+    // Windows drops trailing spaces from file names.
+    #[cfg(not(windows))]
+    #[tokio::test]
+    async fn test_delete_dir_keeps_the_name_without_its_trailing_space() {
+        let dir = tempdir().unwrap();
+        for name in ["t1", "t1 "] {
+            fs::create_dir(dir.path().join(name)).unwrap();
+            fs::write(dir.path().join(name).join("data"), b"data").unwrap();
+        }
+        let file_io = setup_fs_file_io();
+
+        file_io
+            .delete_dir(&local_file_path(&dir.path().join("t1 ")))
+            .await
+            .unwrap();
+
+        assert!(dir.path().join("t1").exists());
+        assert!(!dir.path().join("t1 ").exists());
+    }
 }
 
 #[cfg(all(

crates/paimon/src/table/branch_manager.rs

@@ -602,4 +602,21 @@ mod tests {
         assert!(!bm.branch_exists("branch2").await.unwrap());
         assert!(bm.branch_exists("branch3").await.unwrap());
     }
+
+    #[tokio::test]
+    async fn test_rename_branch_fallback_deletes_only_the_source_directory() {
+        // The memory backend has no native rename, so this takes the copy and
+        // recursive delete fallback that S3 and OSS take.
+        let file_io = test_file_io();
+        let table_path = "memory:/test_rename_fallback_exact".to_string();
+        let schema_manager = SchemaManager::new(file_io.clone(), table_path.clone());
+        write_schema(&file_io, &schema_manager, &test_schema()).await;
+
+        let bm = BranchManager::new(file_io, table_path);
+        bm.create_branch("b1").await.unwrap();
+        bm.create_branch("b10").await.unwrap();
+        bm.rename_branch("b1", "renamed").await.unwrap();
+
+        assert_eq!(bm.list_all().await.unwrap(), vec!["b10", "renamed"]);
+    }
 }