Use collation in reflection in MSSQL

Explicitly specify the collation when reflecting table columns using
MSSQL to prevent "collation conflict" errors.

Fixes: #8035
Change-Id: I4239a5ca8b041f56d7b3bba67b3357c176db31ee

diff --git a/doc/build/changelog/unreleased_14/8035.rst b/doc/build/changelog/unreleased_14/8035.rst
new file mode 100644 (file)
index 0000000..ea6ece0
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/8035.rst
@@ -0,0 +1,6 @@
+.. change::
+    :tags: bug, mssql, reflection
+    :tickets: 8035
+
+    Explicitly specify the collation when reflecting table columns using
+    MSSQL to prevent "collation conflict" errors.

diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
index 2e0238dcf993f1754a9d966429f98f236b2feb72..b4c620f91ddd500306bb674c2b29dd5eb42703c8 100644 (file)
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -3198,14 +3198,16 @@ class MSDialect(default.DefaultDialect):
             computed_cols,
             onclause=sql.and_(
                 computed_cols.c.object_id == func.object_id(full_name),
-                computed_cols.c.name == columns.c.column_name,
+                computed_cols.c.name
+                == columns.c.column_name.collate("DATABASE_DEFAULT"),
             ),
             isouter=True,
         ).join(
             identity_cols,
             onclause=sql.and_(
                 identity_cols.c.object_id == func.object_id(full_name),
-                identity_cols.c.name == columns.c.column_name,
+                identity_cols.c.name
+                == columns.c.column_name.collate("DATABASE_DEFAULT"),
             ),
             isouter=True,
         )

diff --git a/test/dialect/mssql/test_reflection.py b/test/dialect/mssql/test_reflection.py
index 1489f9a6d9f7c6fbd9fbd789f81e08d8f131c40e..7ee380477bda07649670c0a2774c63836e33eb37 100644 (file)
--- a/test/dialect/mssql/test_reflection.py
+++ b/test/dialect/mssql/test_reflection.py
@@ -22,6 +22,7 @@ from sqlalchemy import types as sqltypes
 from sqlalchemy.dialects import mssql
 from sqlalchemy.dialects.mssql import base
 from sqlalchemy.dialects.mssql.information_schema import tables
+from sqlalchemy.pool import NullPool
 from sqlalchemy.schema import CreateIndex
 from sqlalchemy.testing import AssertsCompiledSQL
 from sqlalchemy.testing import ComparesTables
@@ -32,6 +33,7 @@ from sqlalchemy.testing import in_
 from sqlalchemy.testing import is_
 from sqlalchemy.testing import is_true
 from sqlalchemy.testing import mock
+from sqlalchemy.testing import provision
 
 
 class ReflectionTest(fixtures.TestBase, ComparesTables, AssertsCompiledSQL):
@@ -357,6 +359,52 @@ class ReflectionTest(fixtures.TestBase, ComparesTables, AssertsCompiledSQL):
                         "drop table #myveryveryuniquetemptablename"
                     )
 
+    @testing.fixture
+    def temp_db_alt_collation_fixture(
+        self, connection_no_trans, testing_engine
+    ):
+        temp_db_name = "%s_different_collation" % (
+            provision.FOLLOWER_IDENT or "default"
+        )
+        cnxn = connection_no_trans.execution_options(
+            isolation_level="AUTOCOMMIT"
+        )
+        cnxn.exec_driver_sql(f"DROP DATABASE IF EXISTS {temp_db_name}")
+        cnxn.exec_driver_sql(
+            f"CREATE DATABASE {temp_db_name} COLLATE Danish_Norwegian_CI_AS"
+        )
+        eng = testing_engine(
+            url=testing.db.url.set(database=temp_db_name),
+            options=dict(poolclass=NullPool),
+        )
+
+        yield eng
+
+        cnxn.exec_driver_sql(f"DROP DATABASE IF EXISTS {temp_db_name}")
+
+    def test_global_temp_different_collation(
+        self, temp_db_alt_collation_fixture
+    ):
+        """test #8035"""
+
+        with temp_db_alt_collation_fixture.connect() as conn:
+            conn.exec_driver_sql("CREATE TABLE ##foo (id int primary key)")
+            conn.commit()
+
+            eq_(
+                inspect(conn).get_columns("##foo"),
+                [
+                    {
+                        "name": "id",
+                        "type": testing.eq_type_affinity(sqltypes.INTEGER),
+                        "nullable": False,
+                        "default": None,
+                        "autoincrement": False,
+                    }
+                ],
+            )
+            Table("##foo", MetaData(), autoload_with=conn)
+
     def test_db_qualified_items(self, metadata, connection):
         Table("foo", metadata, Column("id", Integer, primary_key=True))
         Table(