fix: recognize own commit on conflict and avoid deleting committed artifacts - #7722
Merged
Xuanwo merged 5 commits intoJul 31, 2026
Merged
Conversation
…tifacts A commit whose conditional PUT lands but returns an ambiguous error (e.g. a GCS 500 whose internal retry then sees "already exists") was surfaced as a CommitConflict with the writer's own manifest. The failure handling then deleted the transaction file (commit_transaction) and, for compaction, the newly written data files (commit_compaction) — both referenced by the manifest that had actually committed, leaving the table head pointing at missing objects. The same false conflict also made Append commits rebase and re-commit the same fragments (duplicate rows). - On any failed commit attempt, verify who owns the manifest at the target version by comparing its recorded transaction file with the attempt's own: ours means the commit succeeded (return Ok); foreign/absent means the loss is confirmed and existing cleanup stays safe. Applied to commit_transaction, do_commit_new_dataset and do_commit_detached_transaction. The uncontended path is unchanged and the conflict path piggybacks one small manifest read on an already-heavy path. - Introduce Error::CommitStatusUnknown for the residue where verification itself is unavailable; destructive cleanup never runs on unknown status (commit_compaction keeps the rewritten files as GC-able orphans instead). - Route reserve_fragment_ids failures through the same compaction cleanup gate (previously leaked all rewrite outputs with no cleanup). - Emit file_audit events for transaction-file and data-file deletions so cleanup is never silent.
Contributor
📝 WalkthroughWalkthroughCommit handling now verifies whether conflict or error responses correspond to landed commits, preserves artifacts when status is unknown, centralizes success bookkeeping, and updates compaction cleanup and tests for definite, ambiguous, and unknown outcomes. ChangesCommit outcome handling
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant CommitHandler
participant commit_transaction
participant verify_commit_outcome
participant Manifest
participant cleanup_transaction_file
CommitHandler->>commit_transaction: commit transaction
commit_transaction->>verify_commit_outcome: inspect conflict or error
verify_commit_outcome->>Manifest: read target manifest
Manifest-->>verify_commit_outcome: transaction_file identity
verify_commit_outcome-->>commit_transaction: Ours, Foreign, Absent, or Unknown
alt Ours
commit_transaction-->>CommitHandler: return success
else Foreign or Absent
commit_transaction->>cleanup_transaction_file: remove orphan transaction file
else Unknown
commit_transaction-->>CommitHandler: return CommitStatusUnknown
end
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@rust/lance/src/io/commit.rs`:
- Around line 2217-2447: Move the repeated inline imports of
AmbiguousCommitHandler and AmbiguousFailure from the new tests to the
module-level use block near the existing imports. Remove each function-local use
statement while keeping all test references unchanged.
- Line 178: Add a descriptive failure message to the debug_assert! validating
transaction_file in the commit-related function, clearly stating that the
transaction file must not be empty.
In `@rust/lance/src/utils/test.rs`:
- Line 798: Move the inline imports of CommitError and
ConditionalPutCommitHandler out of the method bodies and add them to the
file-level use declarations in rust/lance/src/utils/test.rs, consolidating any
duplicate ConditionalPutCommitHandler import while preserving all references.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: ee923c07-edb8-42d2-bdd9-a9fbde78279f
📒 Files selected for processing (6)
rust/lance-core/src/error.rsrust/lance-core/src/utils/tracing.rsrust/lance/src/dataset/optimize.rsrust/lance/src/dataset/write.rsrust/lance/src/io/commit.rsrust/lance/src/utils/test.rs
| version: u64, | ||
| transaction_file: &str, | ||
| ) -> CommitOutcome { | ||
| debug_assert!(!transaction_file.is_empty()); |
Contributor
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Add a descriptive message to this debug_assert!.
As per coding guidelines: "Prefer debug_assert! over assert! for non-safety invariants; ... and always include descriptive messages."
Proposed change
- debug_assert!(!transaction_file.is_empty());
+ debug_assert!(
+ !transaction_file.is_empty(),
+ "verify_commit_outcome requires a non-empty transaction file for identity comparison at version {version}"
+ );📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| debug_assert!(!transaction_file.is_empty()); | |
| debug_assert!( | |
| !transaction_file.is_empty(), | |
| "verify_commit_outcome requires a non-empty transaction file for identity comparison at version {version}" | |
| ); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@rust/lance/src/io/commit.rs` at line 178, Add a descriptive failure message
to the debug_assert! validating transaction_file in the commit-related function,
clearly stating that the transaction file must not be empty.
Source: Coding guidelines
Comment on lines
+2217
to
+2447
| async fn test_commit_succeeds_when_conflict_is_own_commit() { | ||
| use crate::utils::test::{AmbiguousCommitHandler, AmbiguousFailure}; | ||
|
|
||
| let tmp = TempStrDir::default(); | ||
| let uri = tmp.as_str(); | ||
| let schema = simple_schema(); | ||
| let handler = Arc::new(AmbiguousCommitHandler::default()); | ||
|
|
||
| let params = WriteParams { | ||
| commit_handler: Some(handler.clone()), | ||
| ..Default::default() | ||
| }; | ||
| let reader = RecordBatchIterator::new( | ||
| vec![Ok(simple_batch(&schema, vec![1, 2, 3]))], | ||
| schema.clone(), | ||
| ); | ||
| Dataset::write(reader, uri, Some(params)).await.unwrap(); | ||
| assert_eq!( | ||
| handler.resolve_calls(), | ||
| 0, | ||
| "an uncontended commit must not perform verification reads" | ||
| ); | ||
| let txn_files_before = count_txn_files(uri); | ||
|
|
||
| handler.fail_next(AmbiguousFailure::LandAndConflict); | ||
| let params = WriteParams { | ||
| mode: WriteMode::Append, | ||
| commit_handler: Some(handler.clone()), | ||
| ..Default::default() | ||
| }; | ||
| let reader = RecordBatchIterator::new( | ||
| vec![Ok(simple_batch(&schema, vec![4, 5, 6]))], | ||
| schema.clone(), | ||
| ); | ||
| let ds = Dataset::write(reader, uri, Some(params)) | ||
| .await | ||
| .expect("a conflict with our own landed commit must be reported as success"); | ||
|
|
||
| assert_eq!(ds.version().version, 2); | ||
| assert_eq!( | ||
| ds.count_rows(None).await.unwrap(), | ||
| 6, | ||
| "rows must appear exactly once (no duplicate re-commit)" | ||
| ); | ||
| assert_eq!( | ||
| count_txn_files(uri), | ||
| txn_files_before + 1, | ||
| "the landed commit's transaction file is referenced by the manifest and must survive" | ||
| ); | ||
|
|
||
| // A fresh reader sees the committed version. | ||
| let ds2 = Dataset::open(uri).await.unwrap(); | ||
| assert_eq!(ds2.version().version, 2); | ||
| assert_eq!(ds2.count_rows(None).await.unwrap(), 6); | ||
| } | ||
|
|
||
| /// Same as above, but the landed commit is reported as a plain I/O error | ||
| /// (e.g. the store's retries all returned 5xx while the first attempt had | ||
| /// landed). Verification must still recognize the commit as ours. | ||
| #[tokio::test] | ||
| async fn test_commit_succeeds_when_landed_with_other_error() { | ||
| use crate::utils::test::{AmbiguousCommitHandler, AmbiguousFailure}; | ||
|
|
||
| let tmp = TempStrDir::default(); | ||
| let uri = tmp.as_str(); | ||
| let schema = simple_schema(); | ||
| let handler = Arc::new(AmbiguousCommitHandler::default()); | ||
|
|
||
| let params = WriteParams { | ||
| commit_handler: Some(handler.clone()), | ||
| ..Default::default() | ||
| }; | ||
| let reader = RecordBatchIterator::new( | ||
| vec![Ok(simple_batch(&schema, vec![1, 2, 3]))], | ||
| schema.clone(), | ||
| ); | ||
| Dataset::write(reader, uri, Some(params)).await.unwrap(); | ||
| let txn_files_before = count_txn_files(uri); | ||
|
|
||
| handler.fail_next(AmbiguousFailure::LandAndError); | ||
| let params = WriteParams { | ||
| mode: WriteMode::Append, | ||
| commit_handler: Some(handler.clone()), | ||
| ..Default::default() | ||
| }; | ||
| let reader = RecordBatchIterator::new( | ||
| vec![Ok(simple_batch(&schema, vec![4, 5, 6]))], | ||
| schema.clone(), | ||
| ); | ||
| let ds = Dataset::write(reader, uri, Some(params)) | ||
| .await | ||
| .expect("an errored commit that actually landed must be reported as success"); | ||
|
|
||
| assert_eq!(ds.version().version, 2); | ||
| assert_eq!(ds.count_rows(None).await.unwrap(), 6); | ||
| assert_eq!(count_txn_files(uri), txn_files_before + 1); | ||
| } | ||
|
|
||
| /// A commit that errors without landing keeps today's behavior: | ||
| /// verification finds no manifest at the target version, the original | ||
| /// error propagates (not status-unknown), and the orphaned transaction | ||
| /// file is cleaned up. | ||
| #[tokio::test] | ||
| async fn test_commit_definite_failure_cleans_up_and_keeps_error() { | ||
| use crate::utils::test::{AmbiguousCommitHandler, AmbiguousFailure}; | ||
|
|
||
| let tmp = TempStrDir::default(); | ||
| let uri = tmp.as_str(); | ||
| let schema = simple_schema(); | ||
| let handler = Arc::new(AmbiguousCommitHandler::default()); | ||
|
|
||
| let params = WriteParams { | ||
| commit_handler: Some(handler.clone()), | ||
| ..Default::default() | ||
| }; | ||
| let reader = RecordBatchIterator::new( | ||
| vec![Ok(simple_batch(&schema, vec![1, 2, 3]))], | ||
| schema.clone(), | ||
| ); | ||
| Dataset::write(reader, uri, Some(params)).await.unwrap(); | ||
| let txn_files_before = count_txn_files(uri); | ||
|
|
||
| handler.fail_next(AmbiguousFailure::FailOutright); | ||
| let params = WriteParams { | ||
| mode: WriteMode::Append, | ||
| commit_handler: Some(handler.clone()), | ||
| ..Default::default() | ||
| }; | ||
| let reader = RecordBatchIterator::new( | ||
| vec![Ok(simple_batch(&schema, vec![4, 5, 6]))], | ||
| schema.clone(), | ||
| ); | ||
| let result = Dataset::write(reader, uri, Some(params)).await; | ||
| let err = result.expect_err("commit that did not land must fail"); | ||
| assert!( | ||
| !matches!(err, Error::CommitStatusUnknown { .. }), | ||
| "a verified-absent commit is a definite failure, got: {:?}", | ||
| err | ||
| ); | ||
| assert_eq!( | ||
| count_txn_files(uri), | ||
| txn_files_before, | ||
| "orphaned transaction file of a definitely-failed commit must be cleaned up" | ||
| ); | ||
| } | ||
|
|
||
| /// When the commit errors AND verification itself is unavailable, the | ||
| /// commit status is unknown: surface `CommitStatusUnknown` and delete | ||
| /// nothing. | ||
| #[tokio::test] | ||
| async fn test_commit_status_unknown_when_verification_unavailable() { | ||
| use crate::utils::test::{AmbiguousCommitHandler, AmbiguousFailure}; | ||
|
|
||
| let tmp = TempStrDir::default(); | ||
| let uri = tmp.as_str(); | ||
| let schema = simple_schema(); | ||
| let handler = Arc::new(AmbiguousCommitHandler::default()); | ||
|
|
||
| let params = WriteParams { | ||
| commit_handler: Some(handler.clone()), | ||
| ..Default::default() | ||
| }; | ||
| let reader = RecordBatchIterator::new( | ||
| vec![Ok(simple_batch(&schema, vec![1, 2, 3]))], | ||
| schema.clone(), | ||
| ); | ||
| Dataset::write(reader, uri, Some(params)).await.unwrap(); | ||
| let txn_files_before = count_txn_files(uri); | ||
|
|
||
| // The commit lands but errors, and verification reads fail too. | ||
| handler.fail_next(AmbiguousFailure::LandAndError); | ||
| handler | ||
| .fail_resolve | ||
| .store(true, std::sync::atomic::Ordering::SeqCst); | ||
| let params = WriteParams { | ||
| mode: WriteMode::Append, | ||
| commit_handler: Some(handler.clone()), | ||
| ..Default::default() | ||
| }; | ||
| let reader = RecordBatchIterator::new( | ||
| vec![Ok(simple_batch(&schema, vec![4, 5, 6]))], | ||
| schema.clone(), | ||
| ); | ||
| let result = Dataset::write(reader, uri, Some(params)).await; | ||
| let err = result.expect_err("unknown status must not be reported as success"); | ||
| assert!( | ||
| matches!(err, Error::CommitStatusUnknown { .. }), | ||
| "expected CommitStatusUnknown, got: {:?}", | ||
| err | ||
| ); | ||
| assert_eq!( | ||
| count_txn_files(uri), | ||
| txn_files_before + 1, | ||
| "nothing may be deleted while the commit status is unknown" | ||
| ); | ||
|
|
||
| // The commit did land: a fresh reader must see a consistent v2. | ||
| handler | ||
| .fail_resolve | ||
| .store(false, std::sync::atomic::Ordering::SeqCst); | ||
| let ds = Dataset::open(uri).await.unwrap(); | ||
| assert_eq!(ds.version().version, 2); | ||
| assert_eq!(ds.count_rows(None).await.unwrap(), 6); | ||
| } | ||
|
|
||
| /// Dataset creation whose manifest lands but is reported as a conflict | ||
| /// must succeed instead of returning "dataset already exists". | ||
| #[tokio::test] | ||
| async fn test_create_dataset_succeeds_when_conflict_is_own_commit() { | ||
| use crate::utils::test::{AmbiguousCommitHandler, AmbiguousFailure}; | ||
|
|
||
| let tmp = TempStrDir::default(); | ||
| let uri = tmp.as_str(); | ||
| let schema = simple_schema(); | ||
| let handler = Arc::new(AmbiguousCommitHandler::default()); | ||
| handler.fail_next(AmbiguousFailure::LandAndConflict); | ||
|
|
||
| let params = WriteParams { | ||
| commit_handler: Some(handler.clone()), | ||
| ..Default::default() | ||
| }; | ||
| let reader = RecordBatchIterator::new( | ||
| vec![Ok(simple_batch(&schema, vec![1, 2, 3]))], | ||
| schema.clone(), | ||
| ); | ||
| let ds = Dataset::write(reader, uri, Some(params)) | ||
| .await | ||
| .expect("creation whose commit landed must succeed"); | ||
| assert_eq!(ds.version().version, 1); | ||
| assert_eq!(ds.count_rows(None).await.unwrap(), 3); | ||
| } |
Contributor
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Move the repeated use crate::utils::test::{...} to the top of the module.
Each new test (Lines 2218, 2278, 2321, 2368, 2426) re-imports AmbiguousCommitHandler/AmbiguousFailure inline. As per coding guidelines: "Place use imports at the top of the file, not inline within function bodies." Consolidate into the module-level use block (near Line 1524).
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@rust/lance/src/io/commit.rs` around lines 2217 - 2447, Move the repeated
inline imports of AmbiguousCommitHandler and AmbiguousFailure from the new tests
to the module-level use block near the existing imports. Remove each
function-local use statement while keeping all test references unchanged.
Source: Coding guidelines
| lance_table::io::commit::ManifestLocation, | ||
| lance_table::io::commit::CommitError, | ||
| > { | ||
| use lance_table::io::commit::{CommitError, ConditionalPutCommitHandler}; |
Contributor
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Hoist these use statements to the top of the file.
use lance_table::io::commit::{CommitError, ConditionalPutCommitHandler}; (Line 798) and use ...::ConditionalPutCommitHandler; (Line 873) are declared inside the method bodies. As per coding guidelines: "Place use imports at the top of the file, not inline within function bodies."
Also applies to: 873-873
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@rust/lance/src/utils/test.rs` at line 798, Move the inline imports of
CommitError and ConditionalPutCommitHandler out of the method bodies and add
them to the file-level use declarations in rust/lance/src/utils/test.rs,
consolidating any duplicate ConditionalPutCommitHandler import while preserving
all references.
Source: Coding guidelines
jackye1995
marked this pull request as draft
July 10, 2026 07:08
Xuanwo
marked this pull request as ready for review
July 31, 2026 17:39
Xuanwo
approved these changes
Jul 31, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Problem
A commit whose conditional PUT physically lands but returns an ambiguous error — e.g. a GCS 500 whose internal
object_storeretry then sees "already exists" — is surfaced as aCommitConflictwith the writer's own manifest. The failure handling then destroys artifacts that the committed manifest references:commit_transactiondeletes the attempt's transaction file on conflict;commit_compactiondeletes the newly written data files on anyapply_commiterror (violatingcleanup_data_fragments' documented pre-commit-only contract).The result is a table whose head manifest points at deleted objects: every read of the affected fragment fails with object-not-found and every merge-insert dies at its read phase. This corrupted a production table for 7 days (compaction rewrite at 17:36 UTC, GCS 500 on the commit response, both the rewritten data file and the txn file deleted seconds later by the failure path). The same false conflict also makes a falsely-conflicted
Appendrebase and re-commit the same fragments (duplicate rows), and reports genuinely-succeeded commits as failures.Fix
Verify commit outcome on every failed attempt. Transaction file names embed a per-attempt UUID and the manifest records the path, so ownership is an exact identity check. On
CommitConflictor commit error, read the manifest at the target version:Ok(with the manifest read back from storage);Error::CommitStatusUnknown; destructive cleanup never runs on unknown status.commit_compactionkeeps the rewritten files as GC-able orphans instead — leaking an orphan is recoverable, deleting a referenced file is not.Applied to
commit_transaction,do_commit_new_dataset(its own "already exists" could be its own landed create) anddo_commit_detached_transaction. The uncontended commit path is completely unchanged (a test asserts zero verification reads); the conflict path adds one small manifest read to an already-heavy reload-and-rebase path.Also:
reserve_fragment_idsfailures incommit_compactionpreviously propagated with?and orphaned all rewrite outputs with no cleanup at all; they now route through the same cleanup gate.cleanup_transaction_fileandcleanup_data_fragmentsnow emitlance::file_auditevents (the incident's deletions were invisible because both are silent today).merge_insert's conflict-gated cleanup ("provably uncommitted") becomes truly safe as a side effect, since a surfaced conflict now implies verified-foreign.Notes
disable_transaction_filecannot self-verify; the conflict arm keeps legacy behavior there.Summary by CodeRabbit