Merge pull request #340 from microsoft/v1.9.8

V1.9.8

Author: pradeepsrikakolapu

dbt/adapters/fabric/__version__.py

@@ -1 +1 @@
-version = "1.9.7"
+version = "1.9.8"

dbt/adapters/fabric/fabric_connection_manager.py

@@ -354,7 +354,7 @@ def _run_start_action(credentials: FabricCredentials) -> Dict[str, Any]:
         access_token = AZURE_AUTH_FUNCTIONS[credentials.authentication.lower()](
             credentials, POWER_BI_CREDENTIAL_SCOPE
         ).token
-        _snapshot_manager = wh_snapshot_manager(workspace_id, access_token)
+        _snapshot_manager = wh_snapshot_manager(workspace_id, access_token, credentials.api_url)
 
         if credentials.warehouse_snapshot_name is None:
             logger.info(
@@ -372,15 +372,53 @@ def _run_start_action(credentials: FabricCredentials) -> Dict[str, Any]:
         raise e
 
 
+def get_dbt_run_status() -> str:
+    """
+    Get simple status of dbt run: 'success', 'error', or 'unknown'
+    """
+    import json
+    from pathlib import Path
+
+    try:
+        run_results_path = Path("target/run_results.json")
+
+        if not run_results_path.exists():
+            return "unknown"
+
+        with open(run_results_path, "r") as f:
+            run_results = json.load(f)
+
+        results = run_results.get("results", [])
+
+        if not results:
+            return "unknown"
+
+        # Check if any result has error status
+        has_errors = any(result.get("status") == "error" for result in results)
+
+        return "error" if has_errors else "success"
+
+    except Exception:
+        return "unknown"
+
+
 def _run_end_action(snapshot_result: Optional[Dict[str, Any]] = None):
     """Enhanced run end action with snapshot result."""
     global _snapshot_manager
 
+    # Get simple status
+    status = get_dbt_run_status()
+
+    if status != "success":
+        logger.info(f"Skipping warehouse snapshot update: {status}")
+        return
+
     try:
         if snapshot_result and _snapshot_manager is not None:
             print(
                 "Updating warehouse snapshot timestamp at end of run...",
                 snapshot_result["displayName"],
+                snapshot_result["snapshot_id"],
             )
             _snapshot_manager.update_warehouse_snapshot(snapshot_id=snapshot_result["snapshot_id"])
     except Exception as e:

dbt/adapters/fabric/fabric_credentials.py

@@ -31,6 +31,7 @@ class FabricCredentials(Credentials):
     warehouse_snapshot_name: Optional[str] = None
     warehouse_snapshot_id: Optional[str] = None
     snapshot_timestamp: Optional[str] = None
+    api_url: Optional[str] = "https://api.fabric.microsoft.com/v1"
 
     _ALIASES = {
         "user": "UID",
@@ -47,6 +48,7 @@ class FabricCredentials(Credentials):
         "SQL_ATTR_TRACE": "trace_flag",
         "workspace_id": "workspace_id",
         "warehouse_snapshot_name": "warehouse_snapshot_name",
+        "api_url": "api_url",
     }
 
     @property
@@ -89,6 +91,7 @@ def _connection_keys(self):
             "trace_flag",
             "encrypt",
             "trust_cert",
+            "api_url",
         )
 
     @property

dbt/adapters/fabric/warehouse_snapshots.py

@@ -9,9 +9,14 @@
 class WarehouseSnapshotManager:
     """Manager for Microsoft Fabric warehouse snapshots."""
 
-    def __init__(self, workspace_id: Optional[str], access_token: str):
+    def __init__(
+        self,
+        workspace_id: Optional[str],
+        access_token: str,
+        base_url: Optional[str] = "https://api.fabric.microsoft.com/v1",
+    ):
         self.workspace_id = workspace_id
-        self.base_url = "https://msitapi.fabric.microsoft.com/v1"
+        self.base_url = base_url
         self.headers = {
             "Authorization": f"Bearer {access_token}",
             "Content-Type": "application/json",
@@ -29,7 +34,7 @@ def list_warehouses(self) -> List[Dict[str, Any]]:
             data = response.json()
             warehouses = data.get("value", [])
 
-            logger.info(f"Found {len(warehouses)} warehouses in workspace {self.workspace_id}")
+            logger.debug(f"Found {len(warehouses)} warehouses in workspace {self.workspace_id}")
             return warehouses
 
         except requests.exceptions.RequestException as e:
@@ -71,7 +76,7 @@ def list_warehouse_snapshots(self) -> List[Dict[str, Any]]:
             data = response.json()
             snapshots = data.get("value", [])
 
-            logger.info(
+            logger.debug(
                 f"Found {len(snapshots)} warehouse snapshots in workspace {self.workspace_id}"
             )
             return snapshots
@@ -94,23 +99,23 @@ def _poll_operation_status(
             import time
 
             for attempt in range(max_retries):
-                logger.info(
+                logger.debug(
                     f"Polling operation {operation_id}, attempt {attempt + 1}/{max_retries}"
                 )
 
                 response = requests.get(operation_url, headers=self.headers)
 
                 if response.status_code == 201:
                     # Operation is complete - get the result
-                    logger.info(f"Operation {operation_id} completed (201)")
+                    logger.debug(f"Operation {operation_id} completed (201)")
 
                     # Fetch the actual result from /result endpoint
                     result_url = f"{operation_url}/result"
                     result_response = requests.get(result_url, headers=self.headers)
                     result_response.raise_for_status()
 
                     result = result_response.json()
-                    logger.info(f"Operation {operation_id} result retrieved successfully")
+                    logger.debug(f"Operation {operation_id} result retrieved successfully")
                     return result
 
                 elif response.status_code == 200:

dbt/include/fabric/macros/adapters/columns.sql

@@ -15,7 +15,10 @@
                 row_number() over (partition by object_name(c.object_id) order by c.column_id) as ordinal_position,
                 c.name collate database_default as column_name,
                 t.name as data_type,
-                c.max_length as character_maximum_length,
+                case
+					when (t.name in ('nchar', 'nvarchar', 'sysname') and c.max_length <> -1) then c.max_length / 2
+					else c.max_length
+				end as character_maximum_length,
                 c.precision as numeric_precision,
                 c.scale as numeric_scale
             from sys.columns c {{ information_schema_hints() }}

dbt/include/fabric/macros/materializations/snapshots/helpers.sql

@@ -50,6 +50,9 @@
         {% else %}
             {{ columns.dbt_valid_to }} is null
         {% endif %}
+        {%- if strategy.hard_deletes == 'new_record' -%}
+            and {{ columns.dbt_is_deleted }} = 'False'
+        {% endif -%}
     ),
     insertions_source_data as (
         select *,

tests/conftest.py

@@ -58,6 +58,7 @@ def _profile_ci_azure_base():
             "encrypt": True,
             "trust_cert": True,
             "trace_flag": False,
+            "api_url": "https://msitapi.fabric.microsoft.com/v1",
         },
     }
 
@@ -98,6 +99,7 @@ def _profile_user_azure():
             "encrypt": True,
             "trust_cert": True,
             "database": os.getenv("FABRIC_TEST_DBNAME"),
+            "api_url": "https://msitapi.fabric.microsoft.com/v1",
         },
     }
     return profile