Merge branch 'main' into 1.10.latest

Author: benc-db

.github/workflows/integration.yml

@@ -1,6 +1,20 @@
+# Integration Tests for dbt-databricks
+#
+# This workflow runs integration tests that require Databricks secrets.
+#
+# For testing external contributions (PRs from forks):
+# 1. Go to Actions tab -> Integration Tests -> Run workflow
+# 2. Enter the PR number in the 'pr_number' field
+# 3. Click "Run workflow"
+#
+# This approach is secure because:
+# - The workflow runs in the databricks repository context (access to secrets)
+# - The code to test is explicitly specified by maintainers
+# - No automatic execution of untrusted code with secrets
 name: Integration Tests
 on:
-  push:
+  pull_request:
+    # Run on PRs to the same repository (internal contributors)
     paths-ignore:
       - "**.MD"
       - "**.md"
@@ -9,6 +23,18 @@ on:
       - ".github/workflows/main.yml"
       - ".github/workflows/stale.yml"
 
+  workflow_dispatch:
+    # Manual triggering for external contributions and ad-hoc testing
+    inputs:
+      pr_number:
+        description: "PR number to test (for external contributions)"
+        required: false
+        type: string
+      git_ref:
+        description: "Git ref (branch/tag/commit) to test"
+        required: false
+        type: string
+
 concurrency:
   group: ${{ github.workflow }}-${{ github.ref }}
   cancel-in-progress: true
@@ -17,6 +43,8 @@ jobs:
   run-uc-cluster-e2e-tests:
     runs-on: ubuntu-latest
     environment: azure-prod
+    # Only run on internal PRs or manual dispatch - skip external forks to avoid secret access failures
+    if: github.event_name == 'workflow_dispatch' || github.event.pull_request.head.repo.full_name == github.repository
     env:
       DBT_DATABRICKS_HOST_NAME: ${{ secrets.DATABRICKS_HOST }}
       DBT_DATABRICKS_CLIENT_ID: ${{ secrets.TEST_PECO_SP_ID }}
@@ -27,6 +55,14 @@ jobs:
     steps:
       - name: Check out repository
         uses: actions/checkout@v4
+        with:
+          # For pull_request: checkout the PR head commit
+          # For workflow_dispatch with pr_number: checkout that PR's head
+          # For workflow_dispatch with git_ref: checkout that ref
+          # Otherwise: checkout current branch
+          ref: ${{ github.event.pull_request.head.sha || (github.event.inputs.pr_number && format('refs/pull/{0}/head', github.event.inputs.pr_number)) || github.event.inputs.git_ref || github.ref }}
+          # Fetch enough history for PR testing
+          fetch-depth: 0
 
       - name: Set up python
         id: setup-python
@@ -56,6 +92,8 @@ jobs:
   run-sqlwarehouse-e2e-tests:
     runs-on: ubuntu-latest
     environment: azure-prod
+    # Only run on internal PRs or manual dispatch - skip external forks to avoid secret access failures
+    if: github.event_name == 'workflow_dispatch' || github.event.pull_request.head.repo.full_name == github.repository
     env:
       DBT_DATABRICKS_HOST_NAME: ${{ secrets.DATABRICKS_HOST }}
       DBT_DATABRICKS_CLIENT_ID: ${{ secrets.TEST_PECO_SP_ID }}
@@ -67,6 +105,14 @@ jobs:
     steps:
       - name: Check out repository
         uses: actions/checkout@v4
+        with:
+          # For pull_request: checkout the PR head commit
+          # For workflow_dispatch with pr_number: checkout that PR's head
+          # For workflow_dispatch with git_ref: checkout that ref
+          # Otherwise: checkout current branch
+          ref: ${{ github.event.pull_request.head.sha || (github.event.inputs.pr_number && format('refs/pull/{0}/head', github.event.inputs.pr_number)) || github.event.inputs.git_ref || github.ref }}
+          # Fetch enough history for PR testing
+          fetch-depth: 0
 
       - name: Set up python
         id: setup-python
@@ -96,6 +142,8 @@ jobs:
   run-cluster-e2e-tests:
     runs-on: ubuntu-latest
     environment: azure-prod
+    # Only run on internal PRs or manual dispatch - skip external forks to avoid secret access failures
+    if: github.event_name == 'workflow_dispatch' || github.event.pull_request.head.repo.full_name == github.repository
     env:
       DBT_DATABRICKS_HOST_NAME: ${{ secrets.DATABRICKS_HOST }}
       DBT_DATABRICKS_TOKEN: ${{ secrets.DATABRICKS_TOKEN }}
@@ -104,6 +152,14 @@ jobs:
     steps:
       - name: Check out repository
         uses: actions/checkout@v4
+        with:
+          # For pull_request: checkout the PR head commit
+          # For workflow_dispatch with pr_number: checkout that PR's head
+          # For workflow_dispatch with git_ref: checkout that ref
+          # Otherwise: checkout current branch
+          ref: ${{ github.event.pull_request.head.sha || (github.event.inputs.pr_number && format('refs/pull/{0}/head', github.event.inputs.pr_number)) || github.event.inputs.git_ref || github.ref }}
+          # Fetch enough history for PR testing
+          fetch-depth: 0
 
       - name: Set up python
         id: setup-python

.gitignore

@@ -21,3 +21,5 @@ logs/
 .python-version
 .hatch
 .coverage*
+CLAUDE.md
+.claude/

CHANGELOG.md

@@ -1,4 +1,23 @@
-## dbt-databricks 1.10.11 (TBD)
+## dbt-databricks 1.10.13 (TBD)
+
+## dbt-databricks 1.10.12 (September 8, 2025)
+
+### Under the hood
+
+- Update dependency versions ([1178](https://github.com/databricks/dbt-databricks/pull/1178))
+
+## dbt-databricks 1.10.11 (September 2, 2025)
+
+### Fixes
+
+- Improve ANSI mode error handling for Python models and add debug instrumentation ([1157](https://github.com/databricks/dbt-databricks/pull/1157))
+- Remove external path on intermediate tables for incremental models (with Materialization V2) ([1161](https://github.com/databricks/dbt-databricks/pull/1161))
+- Fix get_columns_in_relation branching logic for streaming tables to prevent it from running `AS JSON`
+- Fix model-level compute override connection logic that was causing invalid spark configs to be set on SQL warehouses
+
+### Under the hood
+
+- Improve performance of schema enumeration/validation ([1168](https://github.com/databricks/dbt-databricks/pull/1168))
 
 ## dbt-databricks 1.10.10 (August 20, 2025)
 

README.md

@@ -99,3 +99,20 @@ def model(dbt, session):
       http_path="sql/protocolv1/..."
     )
 ```
+
+## Python models and ANSI mode
+
+When ANSI mode is enabled (`spark.sql.ansi.enabled=true`), there are limitations when using pandas DataFrames in Python models:
+
+1. **Regular pandas DataFrames**: dbt-databricks will automatically handle conversion even when ANSI mode is enabled, falling back to `spark.createDataFrame()` if needed.
+
+2. **pandas-on-Spark DataFrames**: If you create pandas-on-Spark DataFrames directly in your model (using `pyspark.pandas` or `databricks.koalas`), you may encounter errors with ANSI mode enabled. In this case, you have two options:
+   - Disable ANSI mode for your session: Set `spark.sql.ansi.enabled=false` in your cluster or SQL warehouse configuration
+   - Set the pandas-on-Spark option in your model code:
+     ```python
+     import pyspark.pandas as ps
+     ps.set_option('compute.fail_on_ansi_mode', False)
+     ```
+     Note: This may cause unexpected behavior as pandas-on-Spark follows pandas semantics (returning null/NaN for invalid operations) rather than ANSI SQL semantics (raising errors).
+
+For more information about ANSI mode and its implications, see the [Spark documentation on ANSI compliance](https://spark.apache.org/docs/latest/sql-ref-ansi-compliance.html).

dbt/adapters/databricks/__version__.py

@@ -1 +1 @@
-version = "1.10.10"
+version = "1.10.12"

dbt/adapters/databricks/api_client.py

@@ -452,31 +452,68 @@ def _get_exception(self, response: Response) -> None:
         result_state = state.get("result_state")
         life_cycle_state = state["life_cycle_state"]
 
+        # Add detailed logging for debugging
+        logger.debug(f"[Python Model Debug] Full response state: {state}")
+        logger.debug(f"[Python Model Debug] Life cycle state: {life_cycle_state}")
+        logger.debug(f"[Python Model Debug] Result state: {result_state}")
+
         if result_state == "CANCELED":
             raise DbtRuntimeError(f"Python model run ended in result_state {result_state}")
 
         if life_cycle_state != "TERMINATED":
             try:
+                # Log task information for debugging
+                tasks = response_json.get("tasks", [])
+                logger.debug(f"[Python Model Debug] Tasks in response: {len(tasks)}")
+                for i, task in enumerate(tasks):
+                    logger.debug(f"[Python Model Debug] Task {i}: {task}")
+
                 task_id = response_json["tasks"][0]["run_id"]
+                logger.debug(f"[Python Model Debug] Getting output for task_id: {task_id}")
+
                 # get end state to return to user
                 run_output = self.session.get("/get-output", params={"run_id": task_id})
                 json_run_output = run_output.json()
+
+                # Log the full output for debugging
+                logger.debug(f"[Python Model Debug] Run output status: {run_output.status_code}")
+                logger.debug(
+                    f"[Python Model Debug] Run output keys: {list(json_run_output.keys())}"
+                )
+
+                # Extract more detailed error information
+                error_msg = json_run_output.get("error", "No error message available")
+                error_trace = utils.remove_ansi(json_run_output.get("error_trace", ""))
+
+                # Check for specific Python model issues
+                if "error_trace" in json_run_output:
+                    logger.debug(f"[Python Model Debug] Error trace found: {error_trace[:500]}...")
+
+                # Include run ID and task information in error
+                run_id = response_json.get("run_id")
                 raise DbtRuntimeError(
-                    "Python model failed with traceback as:\n"
+                    f"Python model failed (run_id: {run_id}, task_id: {task_id})\n"
+                    "Traceback:\n"
                     "(Note that the line number here does not "
                     "match the line number in your code due to dbt templating)\n"
-                    f"{json_run_output['error']}\n"
-                    f"{utils.remove_ansi(json_run_output.get('error_trace', ''))}"
+                    f"{error_msg}\n"
+                    f"{error_trace}"
                 )
 
             except Exception as e:
                 if isinstance(e, DbtRuntimeError):
                     raise e
                 else:
+                    # Log the exception for debugging
+                    logger.debug(f"[Python Model Debug] Exception during error extraction: {e}")
                     state_message = response.json()["state"]["state_message"]
+
+                    # Include more context in error
                     raise DbtRuntimeError(
-                        f"Python model run ended in state {life_cycle_state}"
-                        f"with state_message\n{state_message}"
+                        f"Python model run ended in state {life_cycle_state} "
+                        f"(run_id: {response_json.get('run_id')})\n"
+                        f"State message: {state_message}\n"
+                        f"Result state: {result_state}"
                     )
 
     def cancel(self, run_id: str) -> None:

dbt/adapters/databricks/catalogs/_unity.py

@@ -34,6 +34,7 @@ def build_relation(self, model: RelationConfig) -> DatabricksCatalogRelation:
         Args:
             model: `config.model` (not `model`) from the jinja context
         """
+
         return DatabricksCatalogRelation(
             catalog_type=self.catalog_type,
             catalog_name=self.catalog_name

dbt/adapters/databricks/connections.py

@@ -36,7 +36,7 @@
 from dbt.adapters.databricks.handle import CursorWrapper, DatabricksHandle, SqlUtils
 from dbt.adapters.databricks.logging import logger
 from dbt.adapters.databricks.python_models.run_tracking import PythonRunTracker
-from dbt.adapters.databricks.utils import redact_credentials
+from dbt.adapters.databricks.utils import is_cluster_http_path, redact_credentials
 from dbt.adapters.events.types import (
     ConnectionClosedInCleanup,
     ConnectionReused,
@@ -130,12 +130,8 @@ def api_client(self) -> DatabricksApiClient:
 
     def is_cluster(self) -> bool:
         conn = self.get_thread_connection()
-        return (
-            conn.credentials.cluster_id is not None
-            # Credentials field is not updated when overriding the compute at model level.
-            # This secondary check is a workaround for that case
-            or "/warehouses/" not in cast(DatabricksDBTConnection, conn).http_path
-        )
+        databricks_conn = cast(DatabricksDBTConnection, conn)
+        return is_cluster_http_path(databricks_conn.http_path, conn.credentials.cluster_id)
 
     def cancel_open(self) -> list[str]:
         cancelled = super().cancel_open()
@@ -402,7 +398,8 @@ def connect() -> DatabricksHandle:
             try:
                 # TODO: what is the error when a user specifies a catalog they don't have access to
                 conn = DatabricksHandle.from_connection_args(
-                    conn_args, creds.cluster_id is not None
+                    conn_args,
+                    is_cluster_http_path(databricks_connection.http_path, creds.cluster_id),
                 )
                 if conn:
                     databricks_connection.session_id = conn.session_id

dbt/adapters/databricks/impl.py

@@ -300,21 +300,14 @@ def compare_dbr_version(self, major: int, minor: int) -> int:
         return self.connections.compare_dbr_version(major, minor)
 
     def list_schemas(self, database: Optional[str]) -> list[str]:
-        """
-        Get a list of existing schemas in database.
-
-        If `database` is `None`, fallback to executing `show databases` because
-        `list_schemas` tries to collect schemas from all catalogs when `database` is `None`.
-        """
-        if database is not None:
-            results = self.connections.list_schemas(database=database)
-        else:
-            results = self.execute_macro(LIST_SCHEMAS_MACRO_NAME, kwargs={"database": database})
+        results = self.execute_macro(LIST_SCHEMAS_MACRO_NAME, kwargs={"database": database})
         return [row[0] for row in results]
 
     def check_schema_exists(self, database: Optional[str], schema: str) -> bool:
         """Check if a schema exists."""
-        return schema.lower() in set(s.lower() for s in self.list_schemas(database=database))
+        return schema.lower() in set(
+            s.lower() for s in self.connections.list_schemas(database or "hive_metastore", schema)
+        )
 
     def execute(
         self,
@@ -485,10 +478,9 @@ def get_columns_in_relation(  # type: ignore[override]
             relation.is_hive_metastore()
             or self.compare_dbr_version(16, 2) < 0
             or relation.type == DatabricksRelationType.MaterializedView
-            or (
-                relation.type == DatabricksRelationType.StreamingTable
-                and self.compare_dbr_version(17, 1) < 0
-            )
+            # TODO: Replace with self.compare_dbr_version(17, 1) < 0 when 17.1 is current version
+            #       for SQL warehouses
+            or relation.type == DatabricksRelationType.StreamingTable
         )
         return self.get_column_behavior.get_columns_in_relation(self, relation, use_legacy_logic)
 

dbt/adapters/databricks/parse_model.py

@@ -40,7 +40,8 @@ def table_format(model: RelationConfig) -> Optional[str]:
 def _get(
     model: RelationConfig, setting: str, case_sensitive: Optional[bool] = False
 ) -> Optional[str]:
-    if not model.config:
+    # dbt-core can sometimes pass in non-model configs that don't have "get" defined
+    if not model.config or not hasattr(model.config, "get"):
         return None
 
     if value := model.config.get(setting):