feat: add experimental SDP streaming/materialized tables and blob publishing

CHANGELOG.md

@@ -1,5 +1,15 @@
 # Changelog
 
+## v1.13.7
+
+### Features
+
+- Added experimental feature flags for Spark Declarative Pipeline `STREAMING` tables and `MATERIALIZED VIEW`.
+- Added opt-in, message-pattern-based retries for an individual failed Spark SQL
+  statement. Profiles can configure plain substrings or `re:` regular
+  expressions, total attempts, and capped exponential backoff. Defaults contain
+  no patterns, so existing projects do not retry additional statements.
+
 ## v1.13.6
 
 ### Fixes

project.json

@@ -106,6 +106,17 @@
             ],
             "parallel": false
          }
+      },
+      "publish-blob": {
+         "cache": false,
+         "executor": "nx:run-commands",
+         "dependsOn": ["build"],
+         "options": {
+            "commands": [
+               { "command": "tsx tools/scripts/publish-blob.ts", "forwardAllArgs": false }
+            ],
+            "parallel": false
+         }
       }
    }
 }

src/dbt/adapters/fabricspark/__version__.py

@@ -1 +1 @@
-version = "1.13.6"
+version = "1.13.7"

src/dbt/adapters/fabricspark/connections.py

@@ -38,6 +38,7 @@
     LivySessionManager,
     get_lakehouse_properties,
 )
+from dbt.adapters.fabricspark.message_retry import MessageRetryPolicy
 from dbt.adapters.fabricspark.relation import FabricSparkRelation
 from dbt.adapters.sql import SQLConnectionManager
 
@@ -167,6 +168,7 @@ def __init__(self, profile, mp_context) -> None:
         self._hc_pool: Optional[HighConcurrencyReplPool] = None
 
         creds = profile.credentials
+        self._job_retry_policy = MessageRetryPolicy.for_job_retry(creds)
         if (
             creds.method == FabricSparkConnectionMethod.LIVY
             and creds.high_concurrency
@@ -553,68 +555,70 @@ def _execute_query_with_retry(
             sql: str,
             bindings: Optional[Any],
             retryable_exceptions: Tuple[Type[Exception], ...],
-            retry_limit: int,
-            attempt: int,
         ):
-            """
-            A success sees the try exit cleanly and avoid any recursive
-            retries. Failure begins a sleep and retry routine.
-            """
-            retry_limit = connection.credentials.connect_retries or 3
-            try:
-                cursor.execute(sql, bindings)
-            except Exception as e:
-                if self.retries_disabled():
-                    raise e
-
-                is_type_retryable = (
-                    isinstance(e, retryable_exceptions) if retryable_exceptions else False
-                )
-                retryable_message = _is_retryable_error(e)
-
-                # retry_all: fall back to retrying even if the error message
-                # doesn't match a known retryable pattern.  Statement-timeout
-                # errors are still excluded (they are already filtered out by
-                # _is_retryable_error returning "").  Permanent Spark structured
-                # errors (e.g. [SCHEMA_NOT_FOUND]) are also excluded — they will
-                # never succeed on retry and would cause a multi-minute stall.
-                is_retry_all_fallback = (
-                    not is_type_retryable
-                    and not retryable_message
-                    and getattr(connection.credentials, "retry_all", False)
-                    and "increase `statement_timeout` in profiles.yml" not in str(e).lower()
-                    and not _is_permanent_error(e)
-                )
-
-                if not is_type_retryable and not retryable_message and not is_retry_all_fallback:
-                    raise e
+            attempt = 1
+            generic_attempt_limit = connection.credentials.connect_retries or 3
+            while True:
+                try:
+                    cursor.execute(sql, bindings)
+                    return
+                except Exception as e:
+                    if self.retries_disabled():
+                        raise
 
-                # Cease retries and fail when limit is hit.
-                if attempt >= retry_limit:
-                    raise e
+                    message = str(e).lower()
+                    is_statement_timeout = (
+                        "increase `statement_timeout` in profiles.yml" in message
+                    )
+                    is_permanent = _is_permanent_error(e)
+                    job_retry_match = (
+                        None if is_statement_timeout or is_permanent else retry_policy.matches(e)
+                    )
 
-                retry_reason = (
-                    "retry_all"
-                    if is_retry_all_fallback
-                    else ("retryable_type" if is_type_retryable else "retryable_message")
-                )
-                fire_event(
-                    AdapterEventDebug(
-                        name="FabricSpark",
-                        base_msg=f"Got a retryable error ({retry_reason}) {type(e)}. {retry_limit - attempt} retries left. Retrying in {min(5 * (2 ** (attempt - 1)), 60)} seconds.\nError:\n{e}",
-                        args=[],
+                    is_type_retryable = (
+                        isinstance(e, retryable_exceptions) if retryable_exceptions else False
+                    )
+                    retryable_message = _is_retryable_error(e)
+                    is_retry_all_fallback = (
+                        job_retry_match is None
+                        and not is_type_retryable
+                        and not retryable_message
+                        and getattr(connection.credentials, "retry_all", False)
+                        and not is_statement_timeout
+                        and not is_permanent
                     )
-                )
-                time.sleep(min(5 * (2 ** (attempt - 1)), 60))
 
-                return _execute_query_with_retry(
-                    cursor=cursor,
-                    sql=sql,
-                    bindings=bindings,
-                    retryable_exceptions=retryable_exceptions,
-                    retry_limit=retry_limit,
-                    attempt=attempt + 1,
-                )
+                    if job_retry_match is not None:
+                        attempt_limit = retry_policy.max_retries + 1
+                        delay = retry_policy.delay_for_attempt(attempt)
+                        retry_reason = f"job_retry:{job_retry_match}"
+                    elif is_type_retryable or retryable_message or is_retry_all_fallback:
+                        attempt_limit = generic_attempt_limit
+                        delay = min(5 * (2 ** (attempt - 1)), 60)
+                        retry_reason = (
+                            "retry_all"
+                            if is_retry_all_fallback
+                            else ("retryable_type" if is_type_retryable else "retryable_message")
+                        )
+                    else:
+                        raise
+
+                    if attempt >= attempt_limit:
+                        raise
+
+                    fire_event(
+                        AdapterEventDebug(
+                            name="FabricSpark",
+                            base_msg=(
+                                f"Got a retryable error ({retry_reason}) {type(e)}. "
+                                f"{attempt_limit - attempt} retries left. "
+                                f"Retrying in {delay} seconds.\nError:\n{e}"
+                            ),
+                            args=[],
+                        )
+                    )
+                    time.sleep(delay)
+                    attempt += 1
 
         connection = self.get_thread_connection()
         if auto_begin and connection.transaction_open is False:
@@ -633,13 +637,16 @@ def _execute_query_with_retry(
             cursor = connection.handle.cursor()
 
             try:
+                retry_policy = getattr(
+                    self,
+                    "_job_retry_policy",
+                    MessageRetryPolicy.disabled(),
+                )
                 _execute_query_with_retry(
                     cursor=cursor,
                     sql=sql,
                     bindings=bindings,
                     retryable_exceptions=retryable_exceptions,
-                    retry_limit=retry_limit,
-                    attempt=1,
                 )
             except Exception as ex:
                 query_exception = ex

src/dbt/adapters/fabricspark/credentials.py

@@ -1,3 +1,4 @@
+import math
 import os
 import re
 from dataclasses import dataclass, field
@@ -30,6 +31,14 @@
 DEFAULT_SESSION_ID_FILENAME = "livy-session-id.txt"
 
 
+def _validate_experimental_flag(value: Any) -> None:
+    if not isinstance(value, bool):
+        raise DbtRuntimeError(
+            "enable_experimental_unstable must be a boolean in profiles.yml; "
+            "use true or false, not a quoted string or a number."
+        )
+
+
 @dataclass
 class FabricSparkCredentials(Credentials):
     # schema: user-provided from profiles.yml. Defaults to lakehouse name.
@@ -57,6 +66,11 @@ class FabricSparkCredentials(Credentials):
     connect_timeout: int = 10
     create_shortcuts: Optional[bool] = False
     retry_all: bool = False
+    enable_job_retry: bool = True
+    job_retry_on_messages: list[str] = field(default_factory=list)
+    job_retry_max_attempts: int = 3
+    job_retry_initial_wait_seconds: float = 30.0
+    job_retry_max_wait_seconds: float = 300.0
     shortcuts_json_str: Optional[str] = None
     # Auto-detected at connection time via Fabric REST API; not user-configurable.
     # init=False ensures this is never populated from profile YAML.
@@ -124,6 +138,7 @@ class FabricSparkCredentials(Credentials):
     # individual models can override with ``config(auto_optimize=...)`` and the
     # ``DBT_FABRICSPARK_SKIP_OPTIMIZE`` environment variable disables it outright.
     auto_optimize: bool = True
+    enable_experimental_unstable: bool = False
 
     def __repr__(self) -> str:
         """Mask sensitive fields in repr to prevent credential leakage in logs/tracebacks."""
@@ -143,6 +158,7 @@ def __repr__(self) -> str:
 
     @classmethod
     def __pre_deserialize__(cls, data: Any) -> Any:
+        _validate_experimental_flag(data.get("enable_experimental_unstable", False))
         data = super().__pre_deserialize__(data)
         if "lakehouse" not in data:
             data["lakehouse"] = None
@@ -175,6 +191,7 @@ def lakehouse_endpoint(self) -> str:
         return f"{self.endpoint}/workspaces/{self.workspaceid}/lakehouses/{self.lakehouseid}/livyapi/versions/2023-12-01"
 
     def __post_init__(self) -> None:
+        _validate_experimental_flag(self.enable_experimental_unstable)
         if self.method is None:
             raise DbtRuntimeError("Must specify `method` in profile")
 
@@ -261,6 +278,46 @@ def __post_init__(self) -> None:
                 "authentication='token_credential'."
             )
 
+        if self.job_retry_max_attempts < 1:
+            raise DbtRuntimeError(
+                f"job_retry_max_attempts must be >= 1; got {self.job_retry_max_attempts}"
+            )
+        if (
+            not math.isfinite(self.job_retry_initial_wait_seconds)
+            or self.job_retry_initial_wait_seconds <= 0
+        ):
+            raise DbtRuntimeError(
+                "job_retry_initial_wait_seconds must be finite and > 0; "
+                f"got {self.job_retry_initial_wait_seconds}"
+            )
+        if (
+            not math.isfinite(self.job_retry_max_wait_seconds)
+            or self.job_retry_max_wait_seconds <= 0
+        ):
+            raise DbtRuntimeError(
+                "job_retry_max_wait_seconds must be finite and > 0; "
+                f"got {self.job_retry_max_wait_seconds}"
+            )
+        if self.job_retry_initial_wait_seconds > self.job_retry_max_wait_seconds:
+            raise DbtRuntimeError(
+                "job_retry_initial_wait_seconds must be <= "
+                "job_retry_max_wait_seconds; "
+                f"got {self.job_retry_initial_wait_seconds} > "
+                f"{self.job_retry_max_wait_seconds}"
+            )
+        for entry in self.job_retry_on_messages:
+            if not isinstance(entry, str) or not entry:
+                raise DbtRuntimeError(
+                    f"job_retry_on_messages entries must be non-empty strings; got {entry!r}"
+                )
+            if entry.startswith("re:"):
+                try:
+                    re.compile(entry[3:])
+                except re.error as exc:
+                    raise DbtRuntimeError(
+                        f"Invalid job_retry_on_messages regex {entry!r}: {exc}"
+                    ) from exc
+
     def apply_lakehouse_properties(self, lakehouse_properties: dict) -> None:
         """Apply lakehouse properties after fetching them from the Fabric REST API.
 
@@ -356,6 +413,12 @@ def _connection_keys(self) -> Tuple[str, ...]:
             "workspace_name",
             "quote_identifiers",
             "auto_optimize",
+            "enable_job_retry",
+            "job_retry_on_messages",
+            "job_retry_max_attempts",
+            "job_retry_initial_wait_seconds",
+            "job_retry_max_wait_seconds",
+            "enable_experimental_unstable",
             "high_concurrency",
             "spark_config",
         )