fix: retry advancement of PreparedCommit into PostCommit in case version 0 already exists (created by another writer) (delta-io#3513)

# Description
This PR fixes the `PreparedCommit` to fallback to the retry mechanism in
case version 0 has been created by another writer.

# Related Issue(s)
Resolve delta-io#3505

Signed-off-by: danielgafni <danielgafni16@gmail.com>

Author: danielgafni

crates/core/src/kernel/transaction/mod.rs

@@ -610,25 +610,48 @@ impl<'a> std::future::IntoFuture for PreparedCommit<'a> {
         Box::pin(async move {
             let commit_or_bytes = this.commit_or_bytes;
 
-            if this.table_data.is_none() {
+            let mut attempt_number: usize = 1;
+
+            // Handle the case where table doesn't exist yet (initial table creation)
+            let read_snapshot: EagerSnapshot = if this.table_data.is_none() {
                 debug!("committing initial table version 0");
-                this.log_store
+                match this
+                    .log_store
                     .write_commit_entry(0, commit_or_bytes.clone(), this.operation_id)
-                    .await?;
-                return Ok(PostCommit {
-                    version: 0,
-                    data: this.data,
-                    create_checkpoint: false,
-                    cleanup_expired_logs: None,
-                    log_store: this.log_store,
-                    table_data: None,
-                    custom_execute_handler: this.post_commit_hook_handler,
-                    metrics: CommitMetrics { num_retries: 0 },
-                });
-            }
+                    .await
+                {
+                    Ok(_) => {
+                        return Ok(PostCommit {
+                            version: 0,
+                            data: this.data,
+                            create_checkpoint: false,
+                            cleanup_expired_logs: None,
+                            log_store: this.log_store,
+                            table_data: None,
+                            custom_execute_handler: this.post_commit_hook_handler,
+                            metrics: CommitMetrics { num_retries: 0 },
+                        })
+                    }
+                    Err(TransactionError::VersionAlreadyExists(0)) => {
+                        // Table was created by another writer since the `this.table_data.is_none()` check.
+                        // Load the current table state and continue with the retry loop.
+                        debug!("version 0 already exists, loading table state for retry");
+                        attempt_number = 2;
+                        let latest_version = this.log_store.get_latest_version(0).await?;
+                        EagerSnapshot::try_new(
+                            this.log_store.as_ref(),
+                            Default::default(),
+                            Some(latest_version),
+                        )
+                        .await?
+                    }
+                    Err(e) => return Err(e.into()),
+                }
+            } else {
+                this.table_data.unwrap().eager_snapshot().clone()
+            };
 
-            // unwrap() is safe here due to the above check
-            let mut read_snapshot = this.table_data.unwrap().eager_snapshot().clone();
+            let mut read_snapshot = read_snapshot;
 
             let commit_span = info_span!(
                 "commit_with_retries",
@@ -640,7 +663,6 @@ impl<'a> std::future::IntoFuture for PreparedCommit<'a> {
             );
 
             async move {
-                let mut attempt_number = 1;
                 let total_retries = this.max_retries + 1;
                 while attempt_number <= total_retries {
                     Span::current().record("attempt", attempt_number);

crates/test/src/concurrent.rs

@@ -17,6 +17,63 @@ pub async fn test_concurrent_writes(context: &IntegrationContext) -> TestResult
     Ok(())
 }
 
+pub async fn test_concurrent_table_creation(context: &IntegrationContext) -> TestResult {
+    let table_uri = context.uri_for_table(TestTables::Custom("concurrent_create".into()));
+
+    if table_uri.starts_with("file://") {
+        let path = table_uri.strip_prefix("file://").unwrap();
+        std::fs::create_dir_all(path)?;
+    }
+
+    let schema = StructType::try_new(vec![StructField::new(
+        "Id",
+        DataType::Primitive(PrimitiveType::Integer),
+        true,
+    )])?;
+
+    const NUM_WRITERS: usize = 5;
+
+    // Spawn multiple tasks that all try to create the table simultaneously
+    let mut futures = Vec::new();
+    for i in 0..NUM_WRITERS {
+        let uri = table_uri.clone();
+        let schema = schema.clone();
+        futures.push(tokio::spawn(async move {
+            let table_url = url::Url::parse(&uri).unwrap();
+            let table = DeltaTableBuilder::from_uri(table_url)
+                .unwrap()
+                .with_allow_http(true)
+                .build()
+                .unwrap();
+
+            // Each writer tries to create the table
+            let result = DeltaOps(table)
+                .create()
+                .with_columns(schema.fields().cloned())
+                .await;
+
+            (i, result)
+        }));
+    }
+
+    // Collect results - all writers must succeed
+    let mut versions = Vec::new();
+    for f in futures {
+        let (i, result) = f.await.unwrap();
+        let table = result.unwrap_or_else(|e| panic!("Writer {i} failed: {e}"));
+        versions.push(table.version());
+    }
+
+    // Exactly one should have version 0
+    let version_zero_count = versions.iter().filter(|v| **v == Some(0)).count();
+    assert_eq!(
+        version_zero_count, 1,
+        "Exactly one writer should get version 0"
+    );
+
+    Ok(())
+}
+
 async fn prepare_table(
     context: &IntegrationContext,
 ) -> Result<(DeltaTable, String), Box<dyn std::error::Error + 'static>> {