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, fabric]
         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: fabric
     uses: ./.github/workflows/test-warehouse.yml
     with:
       warehouse-type: ${{ matrix.warehouse-type }}

.github/workflows/test-warehouse.yml

@@ -18,6 +18,7 @@ on:
           - clickhouse
           - dremio
           - duckdb
+          - fabric
       elementary-ref:
         type: string
         required: false
@@ -51,6 +52,7 @@ on:
 env:
   BRANCH_NAME: ${{ github.head_ref || github.ref_name }}
   TESTS_DIR: ${{ github.workspace }}/dbt-data-reliability/integration_tests
+  MSSQL_SA_PASSWORD: ${{ secrets.MSSQL_SA_PASSWORD || 'Elementary123!' }}
 
 jobs:
   test:
@@ -100,6 +102,25 @@ 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 == 'fabric'
+        working-directory: ${{ env.TESTS_DIR }}
+        env:
+          MSSQL_SA_PASSWORD: ${{ env.MSSQL_SA_PASSWORD }}
+        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 == '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 }}
@@ -136,19 +157,20 @@ jobs:
         run:
           pip install${{ (inputs.dbt-version == 'latest_pre' && ' --pre') || '' }}
           "dbt-core${{ (!startsWith(inputs.dbt-version, 'latest') && format('=={0}', inputs.dbt-version)) || '' }}"
-          "dbt-${{ (inputs.warehouse-type == 'databricks_catalog' && 'databricks') || (inputs.warehouse-type == 'spark' && 'spark[PyHive]') || (inputs.warehouse-type == 'athena' && 'athena-community') || inputs.warehouse-type }}${{ (!startsWith(inputs.dbt-version, 'latest') && format('~={0}', inputs.dbt-version)) || '' }}"
+          "dbt-${{ (inputs.warehouse-type == 'databricks_catalog' && 'databricks') || (inputs.warehouse-type == 'spark' && 'spark[PyHive]') || (inputs.warehouse-type == 'athena' && 'athena-community') || (inputs.warehouse-type == 'fabric' && 'sqlserver') || inputs.warehouse-type }}${{ (!startsWith(inputs.dbt-version, 'latest') && format('~={0}', inputs.dbt-version)) || '' }}"
 
       - name: Install dbt-fusion
         if: inputs.dbt-version == 'fusion'
         run: |
           curl -fsSL https://public.cdn.getdbt.com/fs/install/install.sh | sh -s --
 
       - name: Install Elementary
-        run: pip install "./elementary[${{ (inputs.warehouse-type == 'databricks_catalog' && 'databricks') || inputs.warehouse-type }}]"
+        run: pip install "./elementary[${{ (inputs.warehouse-type == 'databricks_catalog' && 'databricks') || (inputs.warehouse-type == 'fabric' && 'sqlserver') || inputs.warehouse-type }}]"
 
       - name: Write dbt profiles
         env:
           CI_WAREHOUSE_SECRETS: ${{ secrets.CI_WAREHOUSE_SECRETS || '' }}
+          MSSQL_SA_PASSWORD: ${{ env.MSSQL_SA_PASSWORD }}
         run: |
           # Schema name = dbt_<YYMMDD_HHMMSS>_<branch≤18>_<8-char hash>
           # The hash prevents collisions across concurrent jobs; the branch

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: "${MSSQL_SA_PASSWORD}"
+      MSSQL_PID: "Developer"
+    healthcheck:
+      test: /opt/mssql-tools18/bin/sqlcmd -S localhost -U sa -P "${MSSQL_SA_PASSWORD}" -C -Q "SELECT 1" -b
+      interval: 10s
+      timeout: 5s
+      retries: 10

integration_tests/profiles/generate_profiles.py

@@ -123,6 +123,7 @@ def main(
         keep_trailing_newline=True,
     )
     env.filters["toyaml"] = _yaml_inline
+    env.globals["env_var"] = lambda key, default="": os.environ.get(key, default)
     tmpl = env.from_string(template.read_text())
     rendered = tmpl.render(**context)
     output.write_text(rendered)

integration_tests/profiles/profiles.yml.j2

@@ -62,6 +62,19 @@ elementary_tests:
       schema: {{ schema_name }}
       threads: 8
 
+    fabric: &fabric
+      type: sqlserver
+      driver: "ODBC Driver 18 for SQL Server"
+      server: 127.0.0.1
+      port: 1433
+      database: master
+      schema: {{ schema_name }}
+      user: sa
+      password: "{{ env_var('MSSQL_SA_PASSWORD') }}"
+      encrypt: false
+      trust_cert: true
+      threads: 4
+
     # ── Cloud targets (secrets substituted at CI time) ─────────────────
 
     snowflake: &snowflake
@@ -122,7 +135,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', 'fabric', 'snowflake', 'bigquery', 'redshift', 'databricks_catalog', 'athena'] %}
 {%- for t in targets %}
     {{ t }}:
       <<: *{{ t }}

integration_tests/tests/dbt_project.py

@@ -95,22 +95,26 @@ def _run_query_with_run_operation(self, prerendered_query: str):
             )
         return json.loads(run_operation_results[0])
 
-    @staticmethod
     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,
     ):
+        is_tsql = self.target in ("fabric", "sqlserver")
+        columns = ", ".join(column_names) if column_names else "*"
+        top_clause = f"TOP {limit} " if limit and is_tsql else ""
+        limit_clause = f"LIMIT {limit}" if limit and not 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))
+    is_tsql = dbt_project.target in ("fabric", "sqlserver")
+    query = ANOMALY_TEST_POINTS_QUERY.format(
+        test_id=test_id,
+        top_clause="TOP 1 " if is_tsql else "",
+        limit_clause="" if is_tsql else "limit 1",
+    )
+    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))
+    is_tsql = dbt_project.target in ("fabric", "sqlserver")
+    query = ANOMALY_TEST_POINTS_QUERY.format(
+        test_id=test_id,
+        top_clause="TOP 1 " if is_tsql else "",
+        limit_clause="" if is_tsql else "limit 1",
+    )
+    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,12 @@ def test_exclude_specific_timestamps(test_id: str, dbt_project: DbtProject):
     )
     assert test_result["status"] == "pass"
 
+    ts_type = (
+        "datetime2" if dbt_project.target in ("fabric", "sqlserver") 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
         ]
     )

integration_tests/tests/test_column_pii_sampling.py

@@ -7,18 +7,28 @@
 
 SAMPLES_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
     from {{{{ ref("test_result_rows") }}}}
     where elementary_test_results_id in (select * from latest_elementary_test_result)
 """
 
+
+def _fmt_samples_query(dbt_project: DbtProject, test_id: str) -> str:
+    is_tsql = dbt_project.target in ("fabric", "sqlserver")
+    return SAMPLES_QUERY.format(
+        test_id=test_id,
+        top_clause="TOP 1 " if is_tsql else "",
+        limit_clause="" if is_tsql else "limit 1",
+    )
+
+
 TEST_SAMPLE_ROW_COUNT = 5
 
 
@@ -47,7 +57,7 @@ def test_column_pii_sampling_enabled(test_id: str, dbt_project: DbtProject):
 
     samples = [
         json.loads(row["result_row"])
-        for row in dbt_project.run_query(SAMPLES_QUERY.format(test_id=test_id))
+        for row in dbt_project.run_query(_fmt_samples_query(dbt_project, test_id))
     ]
 
     assert len(samples) == 0
@@ -77,7 +87,7 @@ def test_column_pii_sampling_disabled(test_id: str, dbt_project: DbtProject):
 
     samples = [
         json.loads(row["result_row"])
-        for row in dbt_project.run_query(SAMPLES_QUERY.format(test_id=test_id))
+        for row in dbt_project.run_query(_fmt_samples_query(dbt_project, test_id))
     ]
 
     # sample should be {'unique_field': 'user@example.com', 'n_records': 10}
@@ -113,7 +123,7 @@ def test_column_pii_default_tag_override(test_id: str, dbt_project: DbtProject):
 
     samples = [
         json.loads(row["result_row"])
-        for row in dbt_project.run_query(SAMPLES_QUERY.format(test_id=test_id))
+        for row in dbt_project.run_query(_fmt_samples_query(dbt_project, test_id))
     ]
 
     # sample should be {'unique_field': 'user@example.com', 'n_records': 10}
@@ -150,7 +160,7 @@ def test_column_pii_sampling_tags_exist_but_flag_disabled(
 
     samples = [
         json.loads(row["result_row"])
-        for row in dbt_project.run_query(SAMPLES_QUERY.format(test_id=test_id))
+        for row in dbt_project.run_query(_fmt_samples_query(dbt_project, test_id))
     ]
 
     # When flag is disabled, we get the full sample (not limited by PII filtering)
@@ -188,7 +198,7 @@ def test_column_pii_sampling_all_columns_pii(test_id: str, dbt_project: DbtProje
 
     samples = [
         json.loads(row["result_row"])
-        for row in dbt_project.run_query(SAMPLES_QUERY.format(test_id=test_id))
+        for row in dbt_project.run_query(_fmt_samples_query(dbt_project, test_id))
     ]
 
     # When all columns are PII, no samples should be collected
@@ -219,7 +229,7 @@ def test_unique_test_custom_tag(test_id: str, dbt_project: DbtProject):
 
     samples = [
         json.loads(row["result_row"])
-        for row in dbt_project.run_query(SAMPLES_QUERY.format(test_id=test_id))
+        for row in dbt_project.run_query(_fmt_samples_query(dbt_project, test_id))
     ]
 
     assert len(samples) == 0
@@ -249,7 +259,7 @@ def test_accepted_values_multi_tags(test_id: str, dbt_project: DbtProject):
 
     samples = [
         json.loads(row["result_row"])
-        for row in dbt_project.run_query(SAMPLES_QUERY.format(test_id=test_id))
+        for row in dbt_project.run_query(_fmt_samples_query(dbt_project, test_id))
     ]
 
     assert len(samples) == 0
@@ -279,7 +289,7 @@ def test_not_null_test_multi_matched_tags(test_id: str, dbt_project: DbtProject)
 
     samples = [
         json.loads(row["result_row"])
-        for row in dbt_project.run_query(SAMPLES_QUERY.format(test_id=test_id))
+        for row in dbt_project.run_query(_fmt_samples_query(dbt_project, test_id))
     ]
 
     assert len(samples) == 0
@@ -312,7 +322,7 @@ def test_multiple_pii_columns_mapping(test_id: str, dbt_project: DbtProject):
 
     samples = [
         json.loads(row["result_row"])
-        for row in dbt_project.run_query(SAMPLES_QUERY.format(test_id=test_id))
+        for row in dbt_project.run_query(_fmt_samples_query(dbt_project, test_id))
     ]
 
     assert len(samples) == 0
@@ -342,7 +352,7 @@ def test_custom_sql_test_with_pii_column_simple(test_id: str, dbt_project: DbtPr
 
     samples = [
         json.loads(row["result_row"])
-        for row in dbt_project.run_query(SAMPLES_QUERY.format(test_id=test_id))
+        for row in dbt_project.run_query(_fmt_samples_query(dbt_project, test_id))
     ]
 
     assert len(samples) == 0
@@ -372,7 +382,7 @@ def test_meta_tags_and_accepted_values(test_id: str, dbt_project: DbtProject):
 
     samples = [
         json.loads(row["result_row"])
-        for row in dbt_project.run_query(SAMPLES_QUERY.format(test_id=test_id))
+        for row in dbt_project.run_query(_fmt_samples_query(dbt_project, test_id))
     ]
 
     assert len(samples) == 0