fix: add handling for unmanaged files to vacuum command

crates/deltalake-core/src/operations/vacuum.rs

@@ -180,7 +180,6 @@ impl VacuumBuilder {
         };
 
         let expired_tombstones = get_stale_files(&self.snapshot, retention_period, now_millis);
-        let valid_files = self.snapshot.file_paths_iter().collect::<HashSet<Path>>();
 
         let mut files_to_delete = vec![];
         let mut file_sizes = vec![];
@@ -195,12 +194,31 @@ impl VacuumBuilder {
             .ok_or(DeltaTableError::NoMetadata)?
             .partition_columns;
 
+        let managed_files = self
+            .snapshot
+            .files()
+            .iter()
+            .map(|a| a.path.as_str())
+            .chain(
+                self.snapshot
+                    .all_tombstones()
+                    .iter()
+                    .map(|r| r.path.as_str()),
+            )
+            .collect::<HashSet<&str>>();
+
         while let Some(obj_meta) = all_files.next().await {
             // TODO should we allow NotFound here in case we have a temporary commit file in the list
             let obj_meta = obj_meta.map_err(DeltaTableError::from)?;
-            if valid_files.contains(&obj_meta.location) // file is still being tracked in table
-            || !expired_tombstones.contains(obj_meta.location.as_ref()) // file is not an expired tombstone
-            || is_hidden_directory(partition_columns, &obj_meta.location)?
+            let is_hidden = is_hidden_directory(partition_columns, &obj_meta.location)?;
+
+            if managed_files.contains(obj_meta.location.as_ref()) {
+                if !expired_tombstones.contains(obj_meta.location.as_ref()) || is_hidden {
+                    continue;
+                }
+            } else if now_millis - retention_period.num_milliseconds()
+                < obj_meta.last_modified.timestamp_millis()
+                || is_hidden
             {
                 continue;
             }
@@ -363,8 +381,12 @@ impl VacuumPlan {
 /// deleted even if they'd normally be hidden. The _db_index directory contains (bloom filter)
 /// indexes and these must be deleted when the data they are tied to is deleted.
 fn is_hidden_directory(partition_columns: &[String], path: &Path) -> Result<bool, DeltaTableError> {
-    let path_name = path.to_string();
-    Ok((path_name.starts_with('.') || path_name.starts_with('_'))
+    let is_hidden = path
+        .parts()
+        .any(|p| p.as_ref().starts_with('.') || p.as_ref().starts_with('_'));
+
+    let path_name = path.as_ref();
+    Ok(is_hidden
         && !path_name.starts_with("_delta_index")
         && !path_name.starts_with("_change_data")
         && !partition_columns

crates/deltalake-core/tests/command_vacuum.rs

@@ -216,7 +216,6 @@ async fn test_partitions_included() {
     );
 }
 
-#[ignore]
 #[tokio::test]
 // files that are not managed by the delta log and have a last_modified greater
 // than the retention period should be deleted. Unmanaged files and directories
@@ -276,7 +275,7 @@ async fn test_non_managed_files() {
 
     // Validate unmanaged files are deleted after the retention period
     let res = {
-        clock.tick(Duration::hours(1));
+        clock.tick(Duration::days(7));
         let (_, metrics) = DeltaOps(table)
             .vacuum()
             .with_clock(Arc::new(clock.clone()))