feat: support planning cleanup

rust/lance/src/dataset.rs

@@ -97,7 +97,7 @@ pub mod write;
 pub(crate) use take::row_offsets_to_row_addresses;
 
 use self::builder::DatasetBuilder;
-use self::cleanup::RemovalStats;
+use self::cleanup::{CleanupPlan, RemovalStats};
 use self::fragment::FileFragment;
 use self::refs::Refs;
 use self::scanner::{DatasetRecordBatchStream, Scanner};
@@ -1205,6 +1205,33 @@ impl Dataset {
         cleanup::cleanup_old_versions(self, policy).boxed()
     }
 
+    /// Plan cleanup without removing any files.
+    ///
+    /// The returned plan contains the concrete files that would be removed by
+    /// [`Self::cleanup_with_policy`] for the same dataset snapshot and policy.
+    /// Planning resolves the latest dataset version from storage instead of
+    /// relying on the version cached in this handle.
+    #[instrument(level = "debug", skip(self))]
+    pub fn plan_cleanup_with_policy(
+        &self,
+        policy: CleanupPolicy,
+    ) -> BoxFuture<'_, Result<CleanupPlan>> {
+        cleanup::plan_cleanup(self, policy).boxed()
+    }
+
+    /// Remove the files listed in a cleanup plan.
+    ///
+    /// This validates that the plan targets this dataset and that storage still
+    /// reports the plan's read version as the latest version before deleting.
+    /// There is still a narrow window between that check and object deletion;
+    /// callers must prevent concurrent commits if they require a closed-world
+    /// cleanup execution.
+    #[instrument(level = "debug", skip(self))]
+    pub fn cleanup_with_plan(&self, plan: CleanupPlan) -> BoxFuture<'_, Result<RemovalStats>> {
+        info!(target: TRACE_DATASET_EVENTS, event=DATASET_CLEANING_EVENT, uri=&self.uri);
+        cleanup::cleanup_with_plan(self, plan).boxed()
+    }
+
     #[allow(clippy::too_many_arguments)]
     async fn do_commit(
         base_uri: WriteDestination<'_>,