[WIP] Concurrent schema updates

dlt/destinations/impl/databricks/databricks.py

@@ -486,6 +486,8 @@ def _ensure_arrow_record_batch(self, batch: pyarrow.RecordBatch) -> pyarrow.Reco
 
 
 class DatabricksClient(SqlJobClientWithStagingDataset, SupportsStagingDestination):
+    SCHEMA_UPDATE_MAX_ATTEMPTS: ClassVar[int] = 10
+
     def __init__(
         self,
         schema: Schema,
@@ -672,6 +674,21 @@ def _get_constraints_sql(
 
         return ""
 
+    def _should_retry_schema_update(self, ex: Exception) -> bool:
+        # delta rejects losers of concurrent metadata commits, the statement did not apply
+        msg = str(ex)
+        return "DELTA_METADATA_CHANGED" in msg or "DELTA_CONCURRENT" in msg
+
+    def _is_schema_update_applied(self, ex: Exception) -> bool:
+        # a concurrent load added the same column first
+        msg = str(ex)
+        return "FIELDS_ALREADY_EXISTS" in msg or "COLUMN_ALREADY_EXISTS" in msg
+
+    def _make_create_table(self, qualified_name: str, table: PreparedTableSchema) -> str:
+        if self.capabilities.supports_create_table_if_not_exists:
+            return f"CREATE TABLE IF NOT EXISTS {qualified_name}"
+        return f"CREATE TABLE {qualified_name}"
+
     def _get_table_update_sql(
         self, table_name: str, new_columns: Sequence[TColumnSchema], generate_alter: bool
     ) -> List[str]:
@@ -755,6 +772,9 @@ def _get_table_update_sql(
                         )
         # Note: DELTA is the default format, no explicit USING clause needed
 
+        comment = table.get(TABLE_COMMENT_HINT) or table.get("description")
+        escaped_comment = escape_databricks_literal(comment) if comment else None
+
         # For CREATE TABLE, we need custom generation if we have any custom clauses or non-DELTA format
         if not generate_alter and (
             cluster_clause or partition_clause or tblproperties_clause or using_clause
@@ -778,6 +798,10 @@ def _get_table_update_sql(
             # Add CLUSTER BY clause
             if cluster_clause:
                 sql += f" {cluster_clause}"
+            # comment is inlined: a standalone COMMENT ON TABLE is a separate metadata write
+            # that conflicts with concurrent loads initializing the same schema
+            if escaped_comment:
+                sql += f" COMMENT {escaped_comment}"
             # Add TBLPROPERTIES clause (comes after CLUSTER BY)
             if tblproperties_clause:
                 sql += f" {tblproperties_clause}"
@@ -792,11 +816,12 @@ def _get_table_update_sql(
                 qualified_name = self.sql_client.make_qualified_table_name(table_name)
                 sql_result.append(f"ALTER TABLE {qualified_name} {cluster_clause}")
 
+            if not generate_alter and escaped_comment:
+                sql_result[0] += f" COMMENT {escaped_comment}"
+
         qualified_name = self.sql_client.make_qualified_table_name(table_name)
 
-        if table.get(TABLE_COMMENT_HINT) or table.get("description"):
-            comment = table.get(TABLE_COMMENT_HINT) or table.get("description")
-            escaped_comment = escape_databricks_literal(comment)
+        if generate_alter and escaped_comment:
             sql_result.append(f"COMMENT ON TABLE {qualified_name} IS {escaped_comment}")
 
         if table.get(TABLE_TAGS_HINT):

dlt/destinations/impl/databricks/sql_client.py

@@ -105,6 +105,9 @@ def rollback_transaction(self) -> None:
     def native_connection(self) -> "DatabricksSqlConnection":
         return self._conn
 
+    def create_dataset(self) -> None:
+        self.execute_sql("CREATE SCHEMA IF NOT EXISTS %s" % self.fully_qualified_dataset_name())
+
     def drop_tables(self, *tables: str) -> None:
         # Tables are drop with `IF EXISTS`, but databricks raises when the schema doesn't exist.
         # Multi statement exec is safe and the error can be ignored since all tables are in the same schema.

dlt/destinations/job_client_impl.py

@@ -2,7 +2,9 @@
 from abc import abstractmethod
 import base64
 import contextlib
+import random
 from copy import copy
+from time import sleep
 from types import TracebackType
 from typing import (
     Any,
@@ -68,7 +70,7 @@
     CredentialsConfiguration,
 )
 
-from dlt.destinations.exceptions import DatabaseUndefinedRelation
+from dlt.destinations.exceptions import DatabaseException, DatabaseUndefinedRelation
 from dlt.destinations.job_impl import (
     ReferenceFollowupJobRequest,
 )
@@ -247,6 +249,9 @@ def __init__(
 
 
 class SqlJobClientBase(WithSqlClient, JobClientBase, WithStateSync):
+    SCHEMA_UPDATE_MAX_ATTEMPTS: ClassVar[int] = 1
+    """Attempts to converge the destination schema when concurrent loads modify it, 1 disables retries"""
+
     def __init__(
         self,
         schema: Schema,
@@ -649,20 +654,84 @@ def _get_storage_table_query_columns(self) -> List[str]:
             fields += ["numeric_precision", "numeric_scale"]
         return fields
 
+    def _should_retry_schema_update(self, ex: Exception) -> bool:
+        """Tells if a schema update statement that failed with `ex` did not apply due to a
+        concurrent schema modification and may converge when retried from re-read storage.
+        Destinations where concurrent loads modify table metadata without ddl transactions
+        detect their conflict errors here.
+        """
+        return False
+
+    def _is_schema_update_applied(self, ex: Exception) -> bool:
+        """Tells if a schema update statement that failed with `ex` has its effect already
+        present in the destination i.e. a concurrent load added the same column first.
+        """
+        return False
+
+    def _execute_schema_update_with_tolerance(
+        self, sql_scripts: Sequence[str]
+    ) -> Tuple[List[str], List[Exception]]:
+        """Executes statements one by one so a conflicting statement does not abort the rest.
+        Returns statements that failed on a concurrent schema modification with their errors.
+        """
+        failed: List[str] = []
+        errors: List[Exception] = []
+        for sql in sql_scripts:
+            try:
+                self.sql_client.execute_sql(sql)
+            except DatabaseException as ex:
+                if self._is_schema_update_applied(ex):
+                    continue
+                if not self._should_retry_schema_update(ex):
+                    raise
+                failed.append(sql)
+                errors.append(ex)
+        return failed, errors
+
     def _execute_schema_update_sql(
         self, only_tables: Iterable[str], store_schema: bool = True
     ) -> TSchemaTables:
         # Only `only_tables` are included, or all if None.
-        sql_scripts, schema_update = self._build_schema_update_sql(
-            list(self.get_storage_tables(only_tables or self.schema.tables.keys()))
-        )
-        # Stay within max query size when doing DDL.
-        # Some DB backends use bytes not characters, so decrease the limit by half,
-        # assuming most of the characters in DDL encoded into single bytes.
-        self.sql_client.execute_many(sql_scripts)
-        # skip writing the version row when the schema is already stored (enforced update)
+        table_names = list(only_tables or self.schema.tables.keys())
+        attempts = max(1, self.SCHEMA_UPDATE_MAX_ATTEMPTS)
+        schema_update: TSchemaTables = {}
+        pending: List[str] = []
+        prev_round: Optional[Tuple[Tuple[str, ...], Tuple[str, ...]]] = None
+        for attempt in range(attempts):
+            sql_scripts, update = self._build_schema_update_sql(
+                list(self.get_storage_tables(table_names))
+            )
+            schema_update = schema_update or update
+            if attempts == 1:
+                self.sql_client.execute_many(sql_scripts)
+                break
+            # replay failed statements the rebuilt diff cannot regenerate
+            sql_scripts.extend(sql for sql in pending if sql not in sql_scripts)
+            if not sql_scripts:
+                break
+            pending, errors = self._execute_schema_update_with_tolerance(sql_scripts)
+            if pending:
+                this_round = (tuple(sql_scripts), tuple(str(ex) for ex in errors))
+                # identical statements failing identically twice is a persistent error, not a
+                # concurrent modification
+                if this_round == prev_round or attempt == attempts - 1:
+                    raise errors[0]
+                prev_round = this_round
+                logger.warning(
+                    f"{len(pending)} schema update statements failed on a possible concurrent"
+                    f" schema modification, retrying ({attempt + 1}/{attempts}): {errors[0]}"
+                )
+                # add jitter to prevent lockstep in concurrent loads
+                sleep(random.uniform(0.05, 0.25 * (attempt + 1)))
         if store_schema:
-            self._update_schema_in_storage(self.schema)
+            # under concurrency an identical schema may have been stored by another load
+            if attempts > 1 and self.get_stored_schema_by_hash(self.schema.stored_version_hash):
+                logger.info(
+                    f"Schema with hash {self.schema.stored_version_hash} was stored by a"
+                    " concurrent load, skipping version row"
+                )
+            else:
+                self._update_schema_in_storage(self.schema)
         return schema_update
 
     def _build_schema_update_sql(