Merge pull request #200 from microsoft/dependency_dbtsqlserver_to_dbt…

…fabric

v1.4.1rc1 - switch dependency dbtsqlserver to dbtfabric (no fork)

.github/workflows/integration-tests-azure.yml

@@ -17,6 +17,13 @@ jobs:
         python_version: ["3.7", "3.8", "3.9", "3.10", "3.11"]
         msodbc_version: ["17", "18"]
     runs-on: ubuntu-latest
+    permissions:
+      actions: read
+      contents: read
+      deployments: read
+      packages: none
+      pull-requests: write
+      security-events: write
     container:
       image: ghcr.io/dbt-msft/dbt-sqlserver:CI-${{ matrix.python_version }}-msodbc${{ matrix.msodbc_version }}
     steps:

CHANGELOG.md

@@ -1,4 +1,11 @@
 # Changelog
+## v1.4.1rc1
+
+#### Under the hood
+* Switch dependency from dbt-sqlserver to dbt-fabric [#441](https://github.com/dbt-msft/dbt-sqlserver/issues/441)
+  * for Mac users, before running `make dev`, add `pyodbc==4.0.39 --no-binary :all:` in dev_requirements.txt
+  * [Stackoverflow](https://stackoverflow.com/questions/66731036/unable-to-import-pyodbc-on-apple-silicon-symbol-not-found-sqlallochandle) about pyodbc "Symbol not found: _SQLAllocHandle" error
+
 ## v1.4.0
 
 #### Features

CONTRIBUTING.md

@@ -16,7 +16,7 @@ pyenv activate dbt-synapse
 Install the development dependencies and pre-commit and get information about possible make commands:
 
 ```shell
-make dev
+make dev  # for Mac users, add `pyodbc==4.0.39 --no-binary :all:` in dev_requirements.txt before running `make dev`
 make help
 ```
 

dbt/adapters/synapse/__init__.py

@@ -9,7 +9,7 @@
     adapter=SynapseAdapter,
     credentials=SynapseCredentials,
     include_path=synapse.PACKAGE_PATH,
-    dependencies=["sqlserver"],
+    dependencies=["fabric"],
 )
 
 __all__ = ["Plugin", "SynapseConnectionManager", "SynapseAdapter", "SynapseCredentials"]

dbt/adapters/synapse/__version__.py

@@ -1 +1 @@
-version = "1.4.0"
+version = "1.4.1rc1"

dbt/adapters/synapse/synapse_adapter.py

@@ -1,7 +1,27 @@
-from dbt.adapters.sqlserver import SQLServerAdapter
+from dbt.adapters.base.relation import BaseRelation
+from dbt.adapters.cache import _make_ref_key_msg
+from dbt.adapters.fabric import FabricAdapter
+from dbt.adapters.sql.impl import CREATE_SCHEMA_MACRO_NAME
+from dbt.events.functions import fire_event
+from dbt.events.types import SchemaCreation
 
 from dbt.adapters.synapse.synapse_connection_manager import SynapseConnectionManager
 
 
-class SynapseAdapter(SQLServerAdapter):
+class SynapseAdapter(FabricAdapter):
     ConnectionManager = SynapseConnectionManager
+
+    def create_schema(self, relation: BaseRelation) -> None:
+        relation = relation.without_identifier()
+        fire_event(SchemaCreation(relation=_make_ref_key_msg(relation)))
+        macro_name = CREATE_SCHEMA_MACRO_NAME
+        kwargs = {
+            "relation": relation,
+        }
+
+        if self.config.credentials.schema_authorization:
+            kwargs["schema_authorization"] = self.config.credentials.schema_authorization
+            macro_name = "synapse__create_schema_with_authorization"
+
+        self.execute_macro(macro_name, kwargs=kwargs)
+        self.commit_if_has_connection()

dbt/adapters/synapse/synapse_connection_manager.py

@@ -1,6 +1,6 @@
-from dbt.adapters.sqlserver import SQLServerConnectionManager
+from dbt.adapters.fabric import FabricConnectionManager
 
 
-class SynapseConnectionManager(SQLServerConnectionManager):
+class SynapseConnectionManager(FabricConnectionManager):
     TYPE = "synapse"
     TOKEN = None

dbt/adapters/synapse/synapse_credentials.py

@@ -1,10 +1,10 @@
 from dataclasses import dataclass
 
-from dbt.adapters.sqlserver import SQLServerCredentials
+from dbt.adapters.fabric import FabricCredentials
 
 
 @dataclass
-class SynapseCredentials(SQLServerCredentials):
+class SynapseCredentials(FabricCredentials):
     @property
     def type(self):
         return "synapse"

dbt/include/synapse/macros/adapters/indexes.sql

@@ -126,3 +126,35 @@ declare @drop_remaining_indexes_last nvarchar(max) = (
 {% macro create_nonclustered_index(columns, includes=False) %}
   {{ return(create_nonclustered_index(columns, includes=False)) }}
 {% endmacro %}
+
+
+{% macro drop_fk_indexes_on_table(relation) -%}
+  {% call statement('find_references', fetch_result=true) %}
+      SELECT  obj.name AS FK_NAME,
+      sch.name AS [schema_name],
+      tab1.name AS [table],
+      col1.name AS [column],
+      tab2.name AS [referenced_table],
+      col2.name AS [referenced_column]
+      FROM sys.foreign_key_columns fkc
+      INNER JOIN sys.objects obj
+          ON obj.object_id = fkc.constraint_object_id
+      INNER JOIN sys.tables tab1
+          ON tab1.object_id = fkc.parent_object_id
+      INNER JOIN sys.schemas sch
+          ON tab1.schema_id = sch.schema_id
+      INNER JOIN sys.columns col1
+          ON col1.column_id = parent_column_id AND col1.object_id = tab1.object_id
+      INNER JOIN sys.tables tab2
+          ON tab2.object_id = fkc.referenced_object_id
+      INNER JOIN sys.columns col2
+          ON col2.column_id = referenced_column_id AND col2.object_id = tab2.object_id
+      WHERE sch.name = '{{ relation.schema }}' and tab2.name = '{{ relation.identifier }}'
+  {% endcall %}
+      {% set references = load_result('find_references')['data'] %}
+      {% for reference in references -%}
+        {% call statement('main') -%}
+           alter table [{{reference[1]}}].[{{reference[2]}}] drop constraint [{{reference[0]}}]
+        {%- endcall %}
+      {% endfor %}
+{% endmacro %}

dbt/include/synapse/macros/adapters/relation.sql

@@ -20,7 +20,23 @@
 
 {% macro synapse__rename_relation(from_relation, to_relation) -%}
   {% call statement('rename_relation') -%}
+  -- drop all object types with to_relation.identifier name, to avoid error "new name already in use...duplicate...not permitted"
+  if object_id ('{{ to_relation.include(database=False) }}','V') is not null
+    begin
+    drop view {{ to_relation.include(database=False) }}
+    end
 
-    rename object {{ from_relation.include(database=False) }} to {{ to_relation.identifier }}
+  if object_id ('{{ to_relation.include(database=False) }}','U') is not null
+    begin
+    drop table {{ to_relation.include(database=False) }}
+    end
+
+  rename object {{ from_relation.include(database=False) }} to {{ to_relation.identifier }}
   {%- endcall %}
 {% endmacro %}
+
+{% macro synapse__truncate_relation(relation) %}
+    {% call statement('truncate_relation') -%}
+        truncate table {{ relation }}
+    {%- endcall %}
+{% endmacro %}

dbt/include/synapse/macros/adapters/schema.sql

@@ -6,3 +6,32 @@
     END
   {% endcall %}
 {% endmacro %}
+
+{% macro synapse__create_schema_with_authorization(relation, schema_authorization) -%}
+  {% call statement('create_schema') -%}
+    IF NOT EXISTS (SELECT * FROM sys.schemas WHERE name = '{{ relation.schema }}')
+    BEGIN
+    EXEC('CREATE SCHEMA [{{ relation.schema }}] AUTHORIZATION [{{ schema_authorization }}]')
+    END
+  {% endcall %}
+{% endmacro %}
+
+{% macro synapse__drop_schema(relation) -%}
+  {%- set relations_in_schema = list_relations_without_caching(relation) %}
+
+  {% for row in relations_in_schema %}
+    {%- set schema_relation = api.Relation.create(database=relation.database,
+                                               schema=relation.schema,
+                                               identifier=row[1],
+                                               type=row[3]
+                                               ) -%}
+    {% do drop_relation(schema_relation) %}
+  {%- endfor %}
+
+  {% call statement('drop_schema') -%}
+    IF EXISTS (SELECT * FROM sys.schemas WHERE name = '{{ relation.without_identifier().schema }}')
+    BEGIN
+    EXEC('DROP SCHEMA [{{ relation.without_identifier().schema }}]')
+    END
+  {% endcall %}
+{% endmacro %}

dbt/include/synapse/macros/materializations/snapshots/snapshot.sql

@@ -0,0 +1,61 @@
+{% macro synapse__create_columns(relation, columns) %}
+  {# default__ macro uses "add column"
+     TSQL preferes just "add"
+  #}
+
+  {% set columns %}
+    {% for column in columns %}
+      , CAST(NULL AS {{column.data_type}}) AS {{column_name}}
+    {% endfor %}
+  {% endset %}
+
+  {% set tempTableName %}
+    [{{relation.database}}].[{{ relation.schema }}].[{{ relation.identifier }}_{{ range(1300, 19000) | random }}]
+  {% endset %}
+
+  {%- set index = config.get('index', default="CLUSTERED COLUMNSTORE INDEX") -%}
+  {%- set dist = config.get('dist', default="ROUND_ROBIN") -%}
+  {% set tempTable %}
+      CREATE TABLE {{tempTableName}}
+      WITH(
+      DISTRIBUTION = {{dist}},
+      {{index}}
+      )
+      AS SELECT * {{columns}} FROM [{{relation.database}}].[{{ relation.schema }}].[{{ relation.identifier }}] {{ information_schema_hints() }}
+  {% endset %}
+
+  {% call statement('create_temp_table') -%}
+      {{ tempTable }}
+  {%- endcall %}
+
+  {% set dropTable %}
+      DROP TABLE [{{relation.database}}].[{{ relation.schema }}].[{{ relation.identifier }}]
+  {% endset %}
+
+  {% call statement('drop_table') -%}
+      {{ dropTable }}
+  {%- endcall %}
+
+  {%- set index = config.get('index', default="CLUSTERED COLUMNSTORE INDEX") -%}
+  {%- set dist = config.get('dist', default="ROUND_ROBIN") -%}
+  {% set createTable %}
+      CREATE TABLE {{ relation }}
+      WITH(
+      DISTRIBUTION = {{dist}},
+      {{index}}
+      )
+      AS SELECT * FROM {{tempTableName}} {{ information_schema_hints() }}
+  {% endset %}
+
+  {% call statement('create_Table') -%}
+      {{ createTable }}
+  {%- endcall %}
+
+  {% set dropTempTable %}
+      DROP TABLE {{tempTableName}}
+  {% endset %}
+
+  {% call statement('drop_temp_table') -%}
+      {{ dropTempTable }}
+  {%- endcall %}
+{% endmacro %}

dev_requirements.txt

@@ -4,7 +4,7 @@ wheel==0.40.0
 pre-commit==2.21.0;python_version<"3.8"
 pre-commit==3.3.1;python_version>="3.8"
 pytest-dotenv==0.5.2
-dbt-tests-adapter~=1.4.5
+dbt-tests-adapter~=1.4.9
 aiohttp==3.8.3
 azure-mgmt-synapse==2.0.0
 flaky==3.7.0

setup.py

@@ -16,7 +16,7 @@
     "Sam Debruyn",
 ]
 dbt_version = "1.4"
-dbt_sqlserver_requirement = "dbt-sqlserver~=1.4.0"
+dbt_fabric_requirement = "dbt-fabric~=1.4.0rc3"
 description = """An Azure Synapse adapter plugin for dbt"""
 
 this_directory = os.path.abspath(os.path.dirname(__file__))
@@ -73,7 +73,7 @@ def run(self):
     url="https://github.com/dbt-msft/dbt-synapse",
     packages=find_namespace_packages(include=["dbt", "dbt.*"]),
     include_package_data=True,
-    install_requires=[dbt_sqlserver_requirement],
+    install_requires=[dbt_fabric_requirement],
     cmdclass={
         "verify": VerifyVersionCommand,
     },

tests/functional/adapter/test_basic.py

@@ -32,6 +32,7 @@ class TestEmptySynapse(BaseEmpty):
     pass
 
 
+@pytest.mark.skip(reason="ephemeral not supported")
 class TestEphemeralSynapse(BaseEphemeral):
     pass
 

tests/functional/adapter/test_data_types.py

@@ -46,7 +46,7 @@ def seeds(self):
   - name: expected
     config:
       column_types:
-        timestamp_col: "datetimeoffset"
+        timestamp_col: "datetime2"
         """
 
         return {

tests/functional/adapter/test_docs.py

@@ -22,7 +22,7 @@ def expected_catalog(self, project):
             role="dbo",
             id_type="int",
             text_type="varchar",
-            time_type="datetime",
+            time_type="datetime2",
             view_type="VIEW",
             table_type="BASE TABLE",
             model_stats=no_stats(),
@@ -37,7 +37,7 @@ def expected_catalog(self, project):
             role="dbo",
             id_type="int",
             text_type="varchar",
-            time_type="datetime",
+            time_type="datetime2",
             bigint_type="int",
             view_type="VIEW",
             table_type="BASE TABLE",

tests/functional/adapter/test_grants.py

@@ -1,9 +1,12 @@
-import pytest
 from dbt.tests.adapter.grants.test_incremental_grants import BaseIncrementalGrants
 from dbt.tests.adapter.grants.test_invalid_grants import BaseInvalidGrants
 from dbt.tests.adapter.grants.test_model_grants import BaseModelGrants
 from dbt.tests.adapter.grants.test_seed_grants import BaseSeedGrants
-from dbt.tests.adapter.grants.test_snapshot_grants import BaseSnapshotGrants
+from dbt.tests.adapter.grants.test_snapshot_grants import (
+    BaseSnapshotGrants,
+    user2_snapshot_schema_yml,
+)
+from dbt.tests.util import get_manifest, run_dbt, run_dbt_and_capture, write_file
 
 
 class TestIncrementalGrantsSynapse(BaseIncrementalGrants):
@@ -27,11 +30,33 @@ class TestSeedGrantsSynapse(BaseSeedGrants):
 
 
 class TestSnapshotGrantsSynapse(BaseSnapshotGrants):
-    @pytest.fixture(scope="class")
-    def project_config_update(self):
-        return {
-            # ('42000', '[42000] [Microsoft][ODBC Driver 17 for SQL Server][SQL Server]
-            # Merge statements with a WHEN NOT MATCHED [BY TARGET] clause must
-            # target a hash distributed table. (100087) (SQLExecDirectW)')
-            "snapshots": {"test": {"dist": "HASH(id)", "index": "HEAP"}},
-        }
+    def test_snapshot_grants(self, project, get_test_users):
+        test_users = get_test_users
+        select_privilege_name = self.privilege_grantee_name_overrides()["select"]
+
+        # run the snapshot
+        results = run_dbt(["snapshot"])
+        assert len(results) == 1
+        manifest = get_manifest(project.project_root)
+        snapshot_id = "snapshot.test.my_snapshot"
+        snapshot = manifest.nodes[snapshot_id]
+        expected = {select_privilege_name: [test_users[0]]}
+        assert snapshot.config.grants == expected
+        self.assert_expected_grants_match_actual(project, "my_snapshot", expected)
+
+        # run it again, nothing should have changed
+        # since dbt selects into temporary table, drops existing, selects into original table name,
+        # SELECT needs to be granted again, so "grant " expected in log_output!
+        (results, log_output) = run_dbt_and_capture(["--debug", "snapshot"])
+        assert len(results) == 1
+        assert "revoke " not in log_output
+        assert "grant " in log_output  # grant expected
+        self.assert_expected_grants_match_actual(project, "my_snapshot", expected)
+
+        # change the grantee, assert it updates
+        updated_yaml = self.interpolate_name_overrides(user2_snapshot_schema_yml)
+        write_file(updated_yaml, project.project_root, "snapshots", "schema.yml")
+        (results, log_output) = run_dbt_and_capture(["--debug", "snapshot"])
+        assert len(results) == 1
+        expected = {select_privilege_name: [test_users[1]]}
+        self.assert_expected_grants_match_actual(project, "my_snapshot", expected)