fix: recognize own commit on conflict and avoid deleting committed artifacts

python/python/tests/test_dataset.py

@@ -4368,6 +4368,7 @@ def commit_lock(_version: int):
         lance.write_dataset(
             pa.table({"a": range(100)}), tmp_path / "test2", commit_lock=commit_lock
         )
+    assert lance.dataset(tmp_path / "test2").count_rows() == 100
 
     @contextlib.contextmanager
     def commit_lock(_version: int):

rust/lance-core/src/error.rs

@@ -91,6 +91,41 @@ impl fmt::Display for FieldNotFoundError {
 
 impl std::error::Error for FieldNotFoundError {}
 
+/// A manifest commit returned an error and its final outcome could not be
+/// determined safely.
+///
+/// This is wrapped in [`Error::Wrapped`] so Lance can expose a structured
+/// source without adding a variant to the exhaustive public [`Error`] enum.
+#[derive(Debug)]
+pub struct CommitStatusUnknownError {
+    version: u64,
+    source: BoxedError,
+}
+
+impl CommitStatusUnknownError {
+    /// Return the manifest version whose commit outcome is unknown.
+    pub fn version(&self) -> u64 {
+        self.version
+    }
+}
+
+impl std::fmt::Display for CommitStatusUnknownError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "Commit result for version {} is unknown: the commit may or may not have been \
+             applied; check the table state before retrying: {}",
+            self.version, self.source
+        )
+    }
+}
+
+impl std::error::Error for CommitStatusUnknownError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        Some(self.source.as_ref())
+    }
+}
+
 /// Allocates error on the heap and then places `e` into it.
 #[inline]
 pub fn box_error(e: impl std::error::Error + Send + Sync + 'static) -> BoxedError {
@@ -536,6 +571,11 @@ impl Error {
     pub fn is_not_found(&self) -> bool {
         match self {
             Self::NotFound { .. } => true,
+            Self::Wrapped { error, .. }
+                if error.downcast_ref::<CommitStatusUnknownError>().is_some() =>
+            {
+                false
+            }
             Self::IO { source, .. } | Self::Wrapped { error: source, .. } => {
                 error_source_is_not_found(source.as_ref())
             }
@@ -674,6 +714,21 @@ impl Error {
         RetryableCommitConflictSnafu { version }.into_error(source)
     }
 
+    #[track_caller]
+    pub fn commit_status_unknown_source(version: u64, source: BoxedError) -> Self {
+        Self::wrapped(box_error(CommitStatusUnknownError { version, source }))
+    }
+
+    /// Return whether this error represents a commit whose final outcome could
+    /// not be determined safely.
+    pub fn is_commit_status_unknown(&self) -> bool {
+        matches!(
+            self,
+            Self::Wrapped { error, .. }
+                if error.downcast_ref::<CommitStatusUnknownError>().is_some()
+        )
+    }
+
     #[track_caller]
     pub fn incompatible_transaction_source(source: BoxedError) -> Self {
         IncompatibleTransactionSnafu.into_error(source)
@@ -1154,6 +1209,25 @@ mod test {
         assert!(matches!(converted, Error::IO { .. }));
     }
 
+    #[test]
+    fn test_commit_status_unknown_is_structured_without_masking_as_not_found() {
+        let error = Error::commit_status_unknown_source(
+            42,
+            box_error(Error::not_found("temporarily invisible manifest")),
+        );
+
+        assert!(error.is_commit_status_unknown());
+        assert!(!error.is_not_found());
+        assert!(error.to_string().contains("version 42 is unknown"));
+        let Error::Wrapped { error, .. } = error else {
+            panic!("commit-status-unknown must use the semver-compatible wrapper")
+        };
+        let status = error
+            .downcast_ref::<CommitStatusUnknownError>()
+            .expect("wrapper must retain the typed commit status");
+        assert_eq!(status.version(), 42);
+    }
+
     #[test]
     fn test_external_error_creation() {
         let custom_err = MyCustomError {

rust/lance-core/src/utils/tracing.rs

@@ -66,6 +66,7 @@ pub const AUDIT_TYPE_DELETION: &str = "deletion";
 pub const AUDIT_TYPE_MANIFEST: &str = "manifest";
 pub const AUDIT_TYPE_INDEX: &str = "index";
 pub const AUDIT_TYPE_DATA: &str = "data";
+pub const AUDIT_TYPE_TRANSACTION: &str = "transaction";
 pub const TRACE_FILE_CREATE: &str = "create";
 pub const TRACE_IO_EVENTS: &str = "lance::io_events";
 pub const IO_TYPE_OPEN_SCALAR: &str = "open_scalar_index";

rust/lance-table/src/io/commit.rs

@@ -781,6 +781,27 @@ const DDB_URL_QUERY_KEY: &str = "ddbTableName";
 #[async_trait::async_trait]
 #[allow(clippy::too_many_arguments)]
 pub trait CommitHandler: Debug + Send + Sync {
+    /// Whether a not-found result from [`Self::resolve_version_location`] is
+    /// definitive immediately after a commit attempt.
+    ///
+    /// Handlers backed by an eventually consistent or external source of
+    /// truth should keep the conservative default. This prevents callers from
+    /// deleting files that a newly committed manifest may reference while the
+    /// manifest is not yet visible through the resolver.
+    fn is_version_not_found_definitive(&self) -> bool {
+        false
+    }
+
+    /// Whether an error should still be returned after readback proves that
+    /// the manifest from the current commit attempt landed.
+    ///
+    /// The conservative default preserves errors from custom handlers. Built-in
+    /// object-store handlers override this because their commit errors may be
+    /// ambiguous transport failures whose successful outcome is authoritative.
+    fn propagate_commit_error_after_success(&self) -> bool {
+        true
+    }
+
     async fn resolve_latest_location(
         &self,
         base_path: &Path,
@@ -1201,6 +1222,14 @@ pub struct UnsafeCommitHandler;
 #[async_trait::async_trait]
 #[allow(clippy::too_many_arguments)]
 impl CommitHandler for UnsafeCommitHandler {
+    fn is_version_not_found_definitive(&self) -> bool {
+        true
+    }
+
+    fn propagate_commit_error_after_success(&self) -> bool {
+        false
+    }
+
     async fn commit(
         &self,
         manifest: &mut Manifest,
@@ -1328,6 +1357,10 @@ impl<T: CommitLock + Send + Sync> CommitHandler for T
 where
     T::Lease: 'static,
 {
+    fn is_version_not_found_definitive(&self) -> bool {
+        true
+    }
+
     async fn commit(
         &self,
         manifest: &mut Manifest,
@@ -1387,6 +1420,14 @@ impl<T: CommitLock + Send + Sync> CommitHandler for Arc<T>
 where
     T::Lease: 'static,
 {
+    fn is_version_not_found_definitive(&self) -> bool {
+        self.as_ref().is_version_not_found_definitive()
+    }
+
+    fn propagate_commit_error_after_success(&self) -> bool {
+        self.as_ref().propagate_commit_error_after_success()
+    }
+
     async fn commit(
         &self,
         manifest: &mut Manifest,
@@ -1418,6 +1459,14 @@ pub struct RenameCommitHandler;
 
 #[async_trait::async_trait]
 impl CommitHandler for RenameCommitHandler {
+    fn is_version_not_found_definitive(&self) -> bool {
+        true
+    }
+
+    fn propagate_commit_error_after_success(&self) -> bool {
+        false
+    }
+
     async fn commit(
         &self,
         manifest: &mut Manifest,
@@ -1477,6 +1526,14 @@ pub struct ConditionalPutCommitHandler;
 
 #[async_trait::async_trait]
 impl CommitHandler for ConditionalPutCommitHandler {
+    fn is_version_not_found_definitive(&self) -> bool {
+        true
+    }
+
+    fn propagate_commit_error_after_success(&self) -> bool {
+        false
+    }
+
     async fn commit(
         &self,
         manifest: &mut Manifest,