feat: expose uncommitted delete transactions (lance-format#6781)

## Summary
- Add `DeleteBuilder::execute_uncommitted()` so deletes can be staged
and committed later with `CommitBuilder`.
- Return an `UncommittedDelete` wrapper containing the delete
`Transaction`, `affected_rows`, and `num_deleted_rows` so staged commits
can preserve row-level conflict rebasing.
- Factor delete transaction construction into a helper shared by
committed and uncommitted delete paths.

## Test plan
- `cargo fmt --all`
- `cargo test -p lance
dataset::write::delete::tests::test_delete_execute_uncommitted_preserves_affected_rows_for_rebase
--lib`
- `cargo test -p lance
dataset::write::delete::tests::test_delete_false_predicate_still_commits
--lib`
- `cargo test -p lance
dataset::write::delete::tests::test_concurrent_delete_with_retries
--lib`
- `cargo test -p lance
dataset::write::delete::tests::test_delete_concurrency --lib`
- `cargo test -p lance dataset::write::delete::tests --lib`

Fixes lance-format#6658.

---------

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: ragnorc

rust/lance/src/dataset.rs

@@ -137,7 +137,8 @@ pub use write::update::{UpdateBuilder, UpdateJob};
 #[allow(deprecated)]
 pub use write::{
     AutoCleanupParams, CommitBuilder, DeleteBuilder, DeleteResult, ExternalBlobMode, InsertBuilder,
-    WriteDestination, WriteMode, WriteParams, WriteProgressFn, WriteStats, write_fragments,
+    UncommittedDelete, WriteDestination, WriteMode, WriteParams, WriteProgressFn, WriteStats,
+    write_fragments,
 };
 
 pub(crate) const INDICES_DIR: &str = "_indices";

rust/lance/src/dataset/write.rs

@@ -54,7 +54,7 @@ pub mod update;
 
 pub use super::progress::{WriteProgressFn, WriteStats};
 pub use commit::CommitBuilder;
-pub use delete::{DeleteBuilder, DeleteResult};
+pub use delete::{DeleteBuilder, DeleteResult, UncommittedDelete};
 pub use insert::InsertBuilder;
 
 /// The destination to write data to.

rust/lance/src/dataset/write/delete.rs

@@ -31,6 +31,21 @@ pub struct DeleteResult {
     pub num_deleted_rows: u64,
 }
 
+/// Result of a staged delete operation.
+///
+/// The returned transaction can be committed later with [`CommitBuilder`].
+/// Pass `affected_rows` to [`CommitBuilder::with_affected_rows`] when present
+/// to preserve row-level conflict resolution for concurrent deletes and updates.
+#[derive(Debug, Clone)]
+pub struct UncommittedDelete {
+    /// The transaction to commit.
+    pub transaction: Transaction,
+    /// The row addresses affected by the delete, if available.
+    pub affected_rows: Option<RowAddrTreeMap>,
+    /// The number of rows that were deleted.
+    pub num_deleted_rows: u64,
+}
+
 /// Apply deletions to fragments based on a RoaringTreemap of row IDs.
 ///
 /// Returns the set of modified fragments and removed fragments, if any.
@@ -157,6 +172,56 @@ impl DeleteBuilder {
 
         execute_with_retry(job, self.dataset, config).await
     }
+
+    /// Execute the delete operation without committing the transaction.
+    ///
+    /// Use [`CommitBuilder`] to commit the returned transaction.
+    ///
+    /// # Example: Delete rows from a dataset
+    ///
+    /// ```rust
+    /// use lance::dataset::{CommitBuilder, DeleteBuilder};
+    ///
+    /// # use std::sync::Arc;
+    /// # use lance::Result;
+    /// # use lance::dataset::Dataset;
+    /// # async fn example(dataset: Arc<Dataset>) -> Result<()> {
+    /// let staged_delete = DeleteBuilder::new(dataset.clone(), "age > 65")
+    ///     .execute_uncommitted()
+    ///     .await?;
+    /// let mut commit_builder = CommitBuilder::new(dataset);
+    /// if let Some(affected_rows) = staged_delete.affected_rows {
+    ///     commit_builder = commit_builder.with_affected_rows(affected_rows);
+    /// }
+    /// commit_builder
+    ///     .execute(staged_delete.transaction)
+    ///     .await?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub async fn execute_uncommitted(self) -> Result<UncommittedDelete> {
+        let job = DeleteJob {
+            dataset: self.dataset,
+            filter: self.filter,
+        };
+        let data = job.execute_impl().await?;
+        let DeleteData {
+            updated_fragments,
+            deleted_fragment_ids,
+            affected_rows,
+            num_deleted_rows,
+        } = data;
+        let transaction = job.build_transaction(
+            job.dataset.as_ref(),
+            updated_fragments,
+            deleted_fragment_ids,
+        );
+        Ok(UncommittedDelete {
+            transaction,
+            affected_rows,
+            num_deleted_rows,
+        })
+    }
 }
 
 /// Job that executes the delete operation
@@ -174,6 +239,29 @@ struct DeleteData {
     num_deleted_rows: u64,
 }
 
+impl DeleteJob {
+    fn build_transaction(
+        &self,
+        dataset: &Dataset,
+        updated_fragments: Vec<Fragment>,
+        deleted_fragment_ids: Vec<u64>,
+    ) -> Transaction {
+        let predicate = match &self.filter {
+            ExprFilter::Sql(s) => s.clone(),
+            ExprFilter::Datafusion(expr) => expr.to_string(),
+            ExprFilter::Substrait(_) => {
+                unreachable!("Substrait filters are not supported in DeleteBuilder")
+            }
+        };
+        let operation = Operation::Delete {
+            updated_fragments,
+            deleted_fragment_ids,
+            predicate,
+        };
+        Transaction::new(dataset.manifest.version, operation, None)
+    }
+}
+
 impl RetryExecutor for DeleteJob {
     type Data = DeleteData;
     type Result = DeleteResult;
@@ -265,24 +353,18 @@ impl RetryExecutor for DeleteJob {
     }
 
     async fn commit(&self, dataset: Arc<Dataset>, data: Self::Data) -> Result<Self::Result> {
-        let num_deleted_rows = data.num_deleted_rows;
-        let predicate = match &self.filter {
-            ExprFilter::Sql(s) => s.clone(),
-            ExprFilter::Datafusion(expr) => expr.to_string(),
-            ExprFilter::Substrait(_) => {
-                unreachable!("Substrait filters are not supported in DeleteBuilder")
-            }
-        };
-        let operation = Operation::Delete {
-            updated_fragments: data.updated_fragments,
-            deleted_fragment_ids: data.deleted_fragment_ids,
-            predicate,
-        };
-        let transaction = Transaction::new(dataset.manifest.version, operation, None);
+        let DeleteData {
+            updated_fragments,
+            deleted_fragment_ids,
+            affected_rows,
+            num_deleted_rows,
+        } = data;
+        let transaction =
+            self.build_transaction(dataset.as_ref(), updated_fragments, deleted_fragment_ids);
 
         let mut builder = CommitBuilder::new(dataset);
 
-        if let Some(affected_rows) = data.affected_rows {
+        if let Some(affected_rows) = affected_rows {
             builder = builder.with_affected_rows(affected_rows);
         }
 
@@ -615,6 +697,69 @@ mod tests {
         assert!(fragments[0].metadata.deletion_file.is_none());
     }
 
+    #[tokio::test]
+    async fn test_delete_execute_uncommitted_preserves_affected_rows_for_rebase() {
+        fn sequence_data(range: Range<u32>) -> RecordBatch {
+            let schema = Arc::new(ArrowSchema::new(vec![ArrowField::new(
+                "i",
+                DataType::UInt32,
+                false,
+            )]));
+            RecordBatch::try_new(schema, vec![Arc::new(UInt32Array::from_iter_values(range))])
+                .unwrap()
+        }
+
+        let tmp_dir = TempStrDir::default();
+        let tmp_path = tmp_dir.as_str().to_string();
+
+        let dataset = InsertBuilder::new(&tmp_path)
+            .execute(vec![sequence_data(0..100)])
+            .await
+            .unwrap();
+        let initial_version = dataset.version().version;
+
+        let staged_delete = DeleteBuilder::new(Arc::new(dataset.clone()), "i < 10")
+            .execute_uncommitted()
+            .await
+            .unwrap();
+
+        let dataset_before_commit = Dataset::open(&tmp_path).await.unwrap();
+        assert_eq!(dataset_before_commit.version().version, initial_version);
+        assert_eq!(dataset_before_commit.count_rows(None).await.unwrap(), 100);
+
+        assert_eq!(staged_delete.num_deleted_rows, 10);
+        assert!(staged_delete.affected_rows.is_some());
+        assert_eq!(staged_delete.transaction.read_version, initial_version);
+        match &staged_delete.transaction.operation {
+            Operation::Delete {
+                updated_fragments,
+                deleted_fragment_ids,
+                predicate,
+            } => {
+                assert_eq!(predicate, "i < 10");
+                assert_eq!(updated_fragments.len(), 1);
+                assert!(deleted_fragment_ids.is_empty());
+            }
+            other => panic!("expected delete transaction, got {other:?}"),
+        }
+
+        DeleteBuilder::new(Arc::new(dataset.clone()), "i >= 10 AND i < 20")
+            .execute()
+            .await
+            .unwrap();
+
+        let mut commit_builder = CommitBuilder::new(&tmp_path);
+        if let Some(affected_rows) = staged_delete.affected_rows {
+            commit_builder = commit_builder.with_affected_rows(affected_rows);
+        }
+        let committed = commit_builder
+            .execute(staged_delete.transaction)
+            .await
+            .unwrap();
+        assert_eq!(committed.version().version, initial_version + 2);
+        assert_eq!(committed.count_rows(None).await.unwrap(), 80);
+    }
+
     #[tokio::test]
     async fn test_concurrent_delete_with_retries() {
         use futures::future::try_join_all;