feat: add Microsoft Fabric DWH integration (ELE-5282)

.github/workflows/test-all-warehouses.yml

@@ -48,7 +48,8 @@ jobs:
         dbt-version:
           ${{ inputs.dbt-version && fromJSON(format('["{0}"]', inputs.dbt-version)) ||
           fromJSON('["latest_official", "latest_pre"]') }}
-        warehouse-type: [postgres, clickhouse, trino, dremio, spark, duckdb]
+        warehouse-type:
+          [postgres, clickhouse, trino, dremio, spark, duckdb, sqlserver]
         exclude:
           # latest_pre is only tested on postgres
           - dbt-version: latest_pre
@@ -61,6 +62,8 @@ jobs:
             warehouse-type: spark
           - dbt-version: latest_pre
             warehouse-type: duckdb
+          - dbt-version: latest_pre
+            warehouse-type: sqlserver
     uses: ./.github/workflows/test-warehouse.yml
     with:
       warehouse-type: ${{ matrix.warehouse-type }}
@@ -124,7 +127,7 @@ jobs:
           ${{ inputs.dbt-version && fromJSON(format('["{0}"]', inputs.dbt-version)) ||
           fromJSON('["latest_official"]') }}
         warehouse-type:
-          [snowflake, bigquery, redshift, databricks_catalog, athena]
+          [snowflake, bigquery, redshift, databricks_catalog, athena, fabric]
         # Fusion includes: always run fusion alongside the base version for
         # supported warehouses.  When inputs.dbt-version is already 'fusion' the
         # matrix deduplicates automatically.

.github/workflows/test-warehouse.yml

@@ -18,6 +18,8 @@ on:
           - clickhouse
           - dremio
           - duckdb
+          - sqlserver
+          - fabric
       elementary-ref:
         type: string
         required: false
@@ -55,6 +57,7 @@ env:
 jobs:
   test:
     runs-on: ubuntu-latest
+    timeout-minutes: 60
     concurrency:
       # Serialises runs for the same warehouse × dbt-version × branch.
       # The schema name is derived from a hash of this group (see "Write dbt profiles").
@@ -100,6 +103,23 @@ jobs:
           timeout 180 bash -c 'until [ "$(docker inspect -f {{.State.Health.Status}} dremio 2>/dev/null)" = "healthy" ]; do sleep 5; done'
           echo "Dremio is healthy."
 
+      - name: Start SQL Server
+        if: inputs.warehouse-type == 'sqlserver'
+        working-directory: ${{ env.TESTS_DIR }}
+        run: |
+          docker compose -f docker-compose-sqlserver.yml up -d
+          echo "Waiting for SQL Server to become healthy..."
+          timeout 120 bash -c 'until [ "$(docker inspect -f {{.State.Health.Status}} sqlserver 2>/dev/null)" = "healthy" ]; do sleep 5; done'
+          echo "SQL Server is healthy."
+
+      - name: Install ODBC Driver
+        if: inputs.warehouse-type == 'sqlserver' || inputs.warehouse-type == 'fabric'
+        run: |
+          curl https://packages.microsoft.com/keys/microsoft.asc | sudo tee /etc/apt/trusted.gpg.d/microsoft.asc
+          curl https://packages.microsoft.com/config/ubuntu/$(lsb_release -rs)/prod.list | sudo tee /etc/apt/sources.list.d/mssql-release.list
+          sudo apt-get update
+          sudo ACCEPT_EULA=Y apt-get install -y msodbcsql18 unixodbc-dev
+
       - name: Start Spark
         if: inputs.warehouse-type == 'spark'
         working-directory: ${{ env.TESTS_DIR }}
@@ -206,7 +226,7 @@ jobs:
       - name: Drop test schemas
         if: >-
           always() &&
-          contains(fromJSON('["snowflake","bigquery","redshift","databricks_catalog","athena"]'), inputs.warehouse-type)
+          contains(fromJSON('["snowflake","bigquery","redshift","databricks_catalog","athena","fabric"]'), inputs.warehouse-type)
         working-directory: ${{ env.TESTS_DIR }}
         continue-on-error: true
         run: |

integration_tests/dbt_project/macros/schema_utils/list_schemas.sql

@@ -33,6 +33,14 @@
     {% do return(schemas) %}
 {% endmacro %}
 
+{% macro fabric__edr_list_schemas(database) %}
+    {# Fabric does not support information_schema.schemata; use sys.schemas instead #}
+    {% set results = run_query("SELECT name FROM sys.schemas") %}
+    {% set schemas = [] %}
+    {% for row in results %} {% do schemas.append(row[0]) %} {% endfor %}
+    {% do return(schemas) %}
+{% endmacro %}
+
 {% macro clickhouse__edr_list_schemas(database) %}
     {% set results = run_query("SHOW DATABASES") %}
     {% set schemas = [] %}

integration_tests/dbt_project/macros/schema_utils/schema_exists.sql

@@ -39,6 +39,16 @@
     {% do return(result | length > 0) %}
 {% endmacro %}
 
+{% macro fabric__edr_schema_exists(database, schema_name) %}
+    {% set safe_schema = schema_name | replace("'", "''") %}
+    {% set result = run_query(
+        "SELECT name FROM sys.schemas WHERE lower(name) = lower('"
+        ~ safe_schema
+        ~ "')"
+    ) %}
+    {% do return(result | length > 0) %}
+{% endmacro %}
+
 {% macro clickhouse__edr_schema_exists(database, schema_name) %}
     {% set safe_schema = schema_name | replace("'", "''") %}
     {% set result = run_query(

integration_tests/docker-compose-sqlserver.yml

@@ -0,0 +1,16 @@
+version: "3.8"
+services:
+  sqlserver:
+    image: mcr.microsoft.com/mssql/server:2022-latest
+    container_name: sqlserver
+    ports:
+      - "127.0.0.1:1433:1433"
+    environment:
+      ACCEPT_EULA: "Y"
+      MSSQL_SA_PASSWORD: "Elementary123!"
+      MSSQL_PID: "Developer"
+    healthcheck:
+      test: /opt/mssql-tools18/bin/sqlcmd -S localhost -U sa -P "Elementary123!" -C -Q "SELECT 1" -b
+      interval: 10s
+      timeout: 5s
+      retries: 10

integration_tests/profiles/profiles.yml.j2

@@ -62,6 +62,19 @@ elementary_tests:
       schema: {{ schema_name }}
       threads: 8
 
+    sqlserver: &sqlserver
+      type: sqlserver
+      driver: "ODBC Driver 18 for SQL Server"
+      server: 127.0.0.1
+      port: 1433
+      database: master
+      schema: {{ schema_name }}
+      user: sa
+      password: "Elementary123!"
+      encrypt: false
+      trust_cert: true
+      threads: 4
+
     # ── Cloud targets (secrets substituted at CI time) ─────────────────
 
     snowflake: &snowflake
@@ -106,6 +119,21 @@ elementary_tests:
       client_secret: {{ databricks_client_secret | toyaml }}
       threads: 4
 
+    fabric: &fabric
+      type: fabric
+      driver: "ODBC Driver 18 for SQL Server"
+      server: {{ fabric_server | toyaml }}
+      port: 1433
+      database: {{ fabric_database | toyaml }}
+      schema: {{ schema_name }}
+      authentication: ServicePrincipal
+      tenant_id: {{ fabric_tenant_id | toyaml }}
+      client_id: {{ fabric_client_id | toyaml }}
+      client_secret: {{ fabric_client_secret | toyaml }}
+      encrypt: true
+      trust_cert: false
+      threads: 4
+
     athena: &athena
       type: athena
       s3_staging_dir: {{ athena_s3_staging_dir | toyaml }}
@@ -122,7 +150,7 @@ elementary_tests:
 elementary:
   target: postgres
   outputs:
-{%- set targets = ['postgres', 'clickhouse', 'trino', 'dremio', 'spark', 'duckdb', 'snowflake', 'bigquery', 'redshift', 'databricks_catalog', 'athena'] %}
+{%- set targets = ['postgres', 'clickhouse', 'trino', 'dremio', 'spark', 'duckdb', 'sqlserver', 'snowflake', 'bigquery', 'redshift', 'databricks_catalog', 'athena', 'fabric'] %}
 {%- for t in targets %}
     {{ t }}:
       <<: *{{ t }}

integration_tests/tests/dbt_project.py

@@ -32,6 +32,23 @@
 logger = get_logger(__name__)
 
 
+class SelectLimit:
+    """Cross-adapter TOP / LIMIT helper.
+
+    On T-SQL (Fabric / SQL Server) ``SELECT TOP n ...`` is used instead of
+    ``... LIMIT n``.  Instances expose two string properties so that callers
+    can build dialect-agnostic queries::
+
+        sl = SelectLimit(1, is_tsql=True)
+        query = f"SELECT {sl.top}col FROM t ORDER BY x {sl.limit}"
+        # → "SELECT TOP 1 col FROM t ORDER BY x "
+    """
+
+    def __init__(self, n: int, is_tsql: bool) -> None:
+        self.top = f"TOP {n} " if is_tsql else ""
+        self.limit = "" if is_tsql else f"LIMIT {n}"
+
+
 def get_dbt_runner(
     target: str, project_dir: str, runner_method: Optional[RunnerMethod] = None
 ) -> BaseDbtRunner:
@@ -95,22 +112,61 @@ def _run_query_with_run_operation(self, prerendered_query: str):
             )
         return json.loads(run_operation_results[0])
 
-    @staticmethod
+    @property
+    def is_tsql(self) -> bool:
+        """Return True when the target uses T-SQL dialect (Fabric / SQL Server)."""
+        return self.target in ("fabric", "sqlserver")
+
+    def select_limit(self, n: int = 1) -> "SelectLimit":
+        """Return cross-adapter TOP/LIMIT helpers for use in raw SQL strings.
+
+        Usage::
+
+            sl = dbt_project.select_limit(1)
+            query = f"SELECT {sl.top}col FROM t ORDER BY x {sl.limit}"
+        """
+        return SelectLimit(n, self.is_tsql)
+
+    def samples_query(self, test_id: str, order_by: str = "created_at desc") -> str:
+        """Build a cross-adapter query to fetch test result sample rows.
+
+        This is the shared implementation of the ``SAMPLES_QUERY`` template
+        that was previously duplicated across multiple test files.
+        """
+        sl = self.select_limit(1)
+        return f"""
+            with latest_elementary_test_result as (
+                select {sl.top}id
+                from {{{{ ref("elementary_test_results") }}}}
+                where lower(table_name) = lower('{test_id}')
+                order by {order_by}
+                {sl.limit}
+            )
+
+            select result_row
+            from {{{{ ref("test_result_rows") }}}}
+            where elementary_test_results_id in (select * from latest_elementary_test_result)
+        """
+
     def read_table_query(
+        self,
         table_name: str,
         where: Optional[str] = None,
         group_by: Optional[str] = None,
         order_by: Optional[str] = None,
         limit: Optional[int] = None,
         column_names: Optional[List[str]] = None,
     ):
+        columns = ", ".join(column_names) if column_names else "*"
+        top_clause = f"TOP {limit} " if limit and self.is_tsql else ""
+        limit_clause = f"LIMIT {limit}" if limit and not self.is_tsql else ""
         return f"""
-            SELECT {', '.join(column_names) if column_names else '*'}
+            SELECT {top_clause}{columns}
             FROM {{{{ ref('{table_name}') }}}}
             {f"WHERE {where}" if where else ""}
             {f"GROUP BY {group_by}" if group_by else ""}
             {f"ORDER BY {order_by}" if order_by else ""}
-            {f"LIMIT {limit}" if limit else ""}
+            {limit_clause}
             """
 
     def read_table(

integration_tests/tests/test_anomalies_backfill_logic.py

@@ -29,11 +29,11 @@
 # This returns data points used in the latest anomaly test
 ANOMALY_TEST_POINTS_QUERY = """
     with latest_elementary_test_result as (
-        select id
+        select {top_clause}id
         from {{{{ ref("elementary_test_results") }}}}
         where lower(table_name) = lower('{test_id}')
         order by created_at desc
-        limit 1
+        {limit_clause}
     )
 
     select result_row
@@ -62,7 +62,13 @@ def get_daily_row_count_metrics(dbt_project: DbtProject, test_id: str):
 
 
 def get_latest_anomaly_test_metrics(dbt_project: DbtProject, test_id: str):
-    results = dbt_project.run_query(ANOMALY_TEST_POINTS_QUERY.format(test_id=test_id))
+    sl = dbt_project.select_limit(1)
+    query = ANOMALY_TEST_POINTS_QUERY.format(
+        test_id=test_id,
+        top_clause=sl.top,
+        limit_clause=sl.limit,
+    )
+    results = dbt_project.run_query(query)
     result_rows = [json.loads(result["result_row"]) for result in results]
     return {
         (

integration_tests/tests/test_anomalies_ranges.py

@@ -15,11 +15,11 @@
 
 ANOMALY_TEST_POINTS_QUERY = """
     with latest_elementary_test_result as (
-        select id
+        select {top_clause}id
         from {{{{ ref("elementary_test_results") }}}}
         where lower(table_name) = lower('{test_id}')
         order by created_at desc
-        limit 1
+        {limit_clause}
     )
 
     select result_row
@@ -29,7 +29,13 @@
 
 
 def get_latest_anomaly_test_points(dbt_project: DbtProject, test_id: str):
-    results = dbt_project.run_query(ANOMALY_TEST_POINTS_QUERY.format(test_id=test_id))
+    sl = dbt_project.select_limit(1)
+    query = ANOMALY_TEST_POINTS_QUERY.format(
+        test_id=test_id,
+        top_clause=sl.top,
+        limit_clause=sl.limit,
+    )
+    results = dbt_project.run_query(query)
     return [json.loads(result["result_row"]) for result in results]
 
 

integration_tests/tests/test_anomaly_exclude_metrics.py

@@ -101,9 +101,11 @@ def test_exclude_specific_timestamps(test_id: str, dbt_project: DbtProject):
     )
     assert test_result["status"] == "pass"
 
+    # T-SQL uses datetime2 instead of timestamp.
+    ts_type = "datetime2" if dbt_project.is_tsql else "timestamp"
     excluded_buckets_str = ", ".join(
         [
-            "cast('%s' as timestamp)" % cur_ts.strftime(DATE_FORMAT)
+            "cast('%s' as %s)" % (cur_ts.strftime(DATE_FORMAT), ts_type)
             for cur_ts in excluded_buckets
         ]
     )