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
(cherry picked from commit 5e54d5f1ad022781f5d8c6c7da4802613050dde5)

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 40c06ff008032358b2a5d3813999b8e085127a9b..1658f27c70c12ed3b2e57d0536eb9a720a6e822f 100644 (file)
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -3189,14 +3189,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 1fa301e282bd2a5d588a57465590e10f395ac214..781b4ef188f4e79394b6901d41095e65c3591609 100644 (file)
--- a/test/dialect/mssql/test_reflection.py
+++ b/test/dialect/mssql/test_reflection.py
@@ -24,6 +24,7 @@ from sqlalchemy.dialects import mssql
 from sqlalchemy.dialects.mssql import base
 from sqlalchemy.dialects.mssql.information_schema import CoerceUnicode
 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
@@ -34,6 +35,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):
@@ -358,6 +360,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("DROP DATABASE IF EXISTS %s" % temp_db_name)
+        cnxn.exec_driver_sql(
+            "CREATE DATABASE %s COLLATE Danish_Norwegian_CI_AS" % temp_db_name
+        )
+        eng = testing_engine(
+            url=testing.db.url.set(database=temp_db_name),
+            options=dict(poolclass=NullPool, future=True),
+        )
+
+        yield eng
+
+        cnxn.exec_driver_sql("DROP DATABASE IF EXISTS %s" % 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(