Liquid Cluster enabled table - concurrent writes

[raghu2]

I am trying to insert rows into a Liquid cluster enabled delta table using multiple threads. This link, states that liquid clustering is used for : Tables with concurrent write requirements.

I get this error: [DELTA_CONCURRENT_APPEND] ConcurrentAppendException: Files were added to the root of the table by a concurrent update. Please try the operation again.

How do I insert records in parallel?

Thanks,

[Walter_C]

To address the ConcurrentAppendException error when inserting records in parallel into a Liquid cluster enabled delta table, you can consider the following approaches:

1. Isolation Levels and Write Conflicts:

   • Ensure that the isolation level of your table is set appropriately. The default isolation level is WriteSerializable, which ensures that write operations are serializable but allows for some concurrency. If you need stricter isolation, you can set the isolation level to Serializable using the following command:

     ALTER TABLE <table_name> SET TBLPROPERTIES ('delta.isolationLevel' = 'Serializable');

   • Be aware that stricter isolation levels may reduce concurrency and increase the likelihood of conflicts.

2. Row-Level Concurrency:

   • Ensure that row-level concurrency is enabled. This feature is generally available in Databricks Runtime 14.2 and above and helps reduce conflicts between concurrent write operations by detecting changes at the row level.
   • Row-level concurrency is supported for tables with deletion vectors enabled and without partitioning. Ensure that your table meets these conditions.

3. Avoiding Conflicts:

   • To avoid conflicts, you can make the separation explicit in the operation condition. For example, if your table is partitioned by date and country, you can use the following merge operation:

     deltaTable.as("t").merge(
       source.as("s"),
       "s.user_id = t.user_id AND s.date = t.date AND s.country = t.country"
     ).whenMatched().updateAll()
      .whenNotMatched().insertAll()
      .execute()

   • This ensures that the operations are disjoint and do not conflict with each other.

4. Sequential Execution:

   • If possible, avoid running the final pivot task in parallel. Instead, run it once all the parallel staging tasks are finished. This approach can help prevent conflicts that arise from concurrent writes.

5. Optimize Command:

   • Regularly run the OPTIMIZE command to ensure that data is efficiently clustered. For tables experiencing many updates or inserts, schedule an OPTIMIZE job every one or two hours.

[TejeshS]

We encountered a similar issue as well, and the workaround we tried was partitioning those columns, as Liquid clustering can sometimes trigger this error.