SinkDelta(Append) fails with 'already exists' on existing Delta tables (Azure Blob non-HNS)

[omarmciver]

Description

LazyFrame.SinkDelta() with DeltaSaveMode.Append fails on existing Delta tables with:

Create table error: A Delta Lake table already exists at that location

This occurs on Azure ADLS Gen2 (HNS-enabled) storage, but the root cause appears to be in the native_shim/src/delta/write.rs logic that applies regardless of storage backend.

Reproduction

var df = DataFrame.From(new[] { new { Id = 1, Value = "test" } });

// First write — succeeds (creates table)
df.Lazy().SinkDelta(tablePath, mode: DeltaSaveMode.Append);

// Second write — fails with "A Delta Lake table already exists"
df.Lazy().SinkDelta(tablePath, mode: DeltaSaveMode.Append);

Expected: Second Append should add rows to the existing table.
Actual: Throws PolarsError::ComputeError("Create table error: A Delta Lake table already exists at that location")

Root Cause Analysis

In native_shim/src/delta/write.rs (L104–L128):

let dt = DeltaTable::try_from_url_with_storage_options(
    table_url.clone(), delta_storage_options.clone()
).await...?;

if dt.version() >= Some(0) {
    match save_mode {
        SaveMode::ErrorIfExists => { return Err(...); },
        SaveMode::Ignore => { skip_write = true; },
        _ => {}   // Append falls through here — OK
    }
}

Then at L210:

if table.version() < Some(0) {
    // Auto-create path — calls table.create() which rejects existing tables
    table = table.create()...

The guard at L112 (dt.version() >= Some(0)) should prevent the auto-create path for existing tables. However, table.version() appears to return None or < Some(0) at L210 despite the table existing and try_from_url_with_storage_options calling .load() internally.

The issue may be that the DeltaTable state is lost or not refreshed after the staging write phase writes .tmp_write_* Parquet files to the table directory — by the time table.version() is checked at L210, the state is stale.

Impact

Any SinkDelta(Append) call to an existing Delta table fails. SinkDelta(Overwrite) works correctly, suggesting the bug is specific to the Append → auto-create code path.

Workaround

Use DataFrame.MergeDelta() instead of SinkDelta(Append). MergeDelta correctly loads the table snapshot via delta-rs's MergeBuilder and doesn't hit the auto-create path:

// Instead of: df.Lazy().SinkDelta(path, mode: DeltaSaveMode.Append, ...);
// Use:
df.MergeDelta(
    path,
    mergeKeys: new[] { "your_key_column" },
    matchedUpdateCond: null,              // Update if matched
    notMatchedInsertCond: null,           // Insert if not matched
    notMatchedBySourceDeleteCond: null,   // Retain target-only rows
    cloudOptions: opts);

Environment

• Polars.NET: v0.3.0 (also confirmed present in v0.3.1 — write.rs unchanged)
• delta-rs: 0.31.0 (pinned by Polars.NET)
• Storage: Azure ADLS Gen2 (HNS-enabled), using abfss:// URI scheme
• .NET: 8.0
• OS: Linux (Docker container on AKS)

[ErrorLSC]

Hi omarmciver,

Thank you for this detailed report and the spot-on root cause analysis, I really appreciate your help, as this specific state-sync issue is difficult to reproduce in my local test docker environment.

The core issue was that the DeltaTable instance state wasn't being properly synced/loaded before sink, causing it to hold a stale version (e.g., -1) and incorrectly fall into the auto-create path during the commit phase.

For delta module in Polars.NET, it is in the middle of refactoring to support UnityCatalog integrations. I have incorporated your findings into this refactor and added defensive state-refresh logic (.load() before validation and .update() before commit) to ensure the table state is always fully synchronized before any write operations.

For example:
pub(crate) async fn phase_init_and_validate_sink(
table_url: Url,
delta_opts: HashMap<String, String>,
save_mode: SaveMode,
partition_cols: Vec,
schema: &SchemaRef,
) -> PolarsResult<(DeltaTable, bool, Vec)> {

let mut table = DeltaTable::try_from_url_with_storage_options(table_url.clone(), delta_opts)
    .await
    .map_err(|e| PolarsError::ComputeError(format!("Delta init error: {}", e).into()))?;

match table.load().await {
    Ok(_) => {},
    Err(deltalake::errors::DeltaTableError::NotATable(_)) => {},
    Err(e) => {
        return Err(PolarsError::ComputeError(format!("Failed to load Delta state: {}", e).into()));
    }
}

let mut skip_write = false;
let mut final_partition_cols = partition_cols.clone();

In init phase deltatable will be refreshed once.

Then,
pub(crate) async fn phase_commit_and_cleanup(
mut table: DeltaTable,
schema: SchemaRef,
staging_dir_name: String,
final_partition_cols: Vec,
save_mode: SaveMode,
can_evolve: bool,
write_id: Uuid,
) -> PolarsResult<()> {

let _ = table.update_state().await;

Before commit action the state will be synced once again.

This fix will be included in the next upcoming release. Please give it a try once it's published. Thanks again for your contribution!

[ErrorLSC]

I added a concurrent write test. It works well with new version update logic:

[Fact]
[Trait("DeltaLake", "ConcurrentWrite")]
public async Task Test_Concurrent_Write_Append_Stress_TestAsync()
{

    var tableName = $"delta_concurrent_write_{Guid.NewGuid():N}";
    var rootUrl = $"s3://{minio.BucketName}/{tableName}";
    
    var rawEndpoint = minio.Endpoint.Replace("http://", "").Replace("https://", "").TrimEnd('/');
    var polarsEndpoint = $"http://{rawEndpoint}";
    
    var options = CloudOptions.Aws(
        region: minio.Region,
        accessKey: minio.AccessKey,
        secretKey: minio.SecretKey,
        endpoint: polarsEndpoint
    );
    options.Credentials!["AWS_ALLOW_HTTP"] = "true";
    options.Credentials!["aws_s3_force_path_style"] = "true";
    options.Credentials!["AWS_S3_ALLOW_UNSAFE_RENAME"] = "true";

    using (var dfInit = DataFrame.FromColumns(new { WorkerId = new[] { 0 }, RowId = new[] { 0 } }))
    {
        dfInit.WriteDelta(rootUrl, mode: DeltaSaveMode.Overwrite, cloudOptions: options);
    }

    int concurrency = 10;
    int rowsPerWorker = 100;
    var tasks = new List<Task>();

    for (int i = 0; i < concurrency; i++)
    {
        int workerId = i + 1;
        tasks.Add(Task.Run(() =>
        {
            try 
            {
                var workerIds = Enumerable.Repeat(workerId, rowsPerWorker).ToArray();
                var rowIds = Enumerable.Range(1, rowsPerWorker).ToArray();

                using var sourceDf = DataFrame.FromColumns(new 
                { 
                    WorkerId = workerIds, 
                    RowId = rowIds 
                });

                Console.WriteLine($"[Worker {workerId}] Starting Append...");
                
                sourceDf.WriteDelta(
                    rootUrl, 
                    mode: DeltaSaveMode.Append, 
                    cloudOptions: options
                );
                
                Console.WriteLine($"[Worker {workerId}] Success!");
            }
            catch (Exception ex)
            {
                Console.WriteLine($"[Worker {workerId}] FAILED: {ex.Message}");
                throw; 
            }
        }));
    }

    await Task.WhenAll(tasks);

    using var result = LazyFrame.ScanDelta(rootUrl, cloudOptions: options).Collect();
    
    result.Show();

    long expectedHeight = 1 + (concurrency * rowsPerWorker);
    Assert.Equal(expectedHeight, result.Height);

    for (int i = 1; i <= concurrency; i++)
    {
        var workerRowCount = result.Filter(Col("WorkerId") == i).Height;
        Assert.Equal(rowsPerWorker, workerRowCount);
    }

}

The console result is

[Worker 5] Starting Append...
[Worker 1] Starting Append...
[Worker 7] Starting Append...
[Worker 4] Starting Append...
[Worker 2] Starting Append...
[Worker 9] Starting Append...
[Worker 3] Starting Append...
[Worker 8] Starting Append...
[Worker 6] Starting Append...
[Worker 10] Starting Append...
[Worker 6] Success!
[Worker 9] Success!
[Worker 10] Success!
[Worker 5] Success!
[Worker 1] Success!
[Worker 7] Success!
[Worker 2] Success!
[Worker 4] Success!
[Worker 8] Success!
[Worker 3] Success!
shape: (1_001, 2)
┌──────────┬───────┐
│ WorkerId ┆ RowId │
│ ---      ┆ ---   │
│ i32      ┆ i32   │
╞══════════╪═══════╡
│ 3        ┆ 1     │
│ 3        ┆ 2     │
│ 3        ┆ 3     │
│ 3        ┆ 4     │
│ 3        ┆ 5     │
│ …        ┆ …     │
│ 6        ┆ 97    │
│ 6        ┆ 98    │
│ 6        ┆ 99    │
│ 6        ┆ 100   │
│ 0        ┆ 0     │
└──────────┴───────┘