correct for FK reflecting on PK / index col at the same time

Fixed issue in the MSSQL dialect's foreign key reflection query where
duplicate rows could be returned when a foreign key column and its
referenced primary key column have the same name, and both the referencing
and referenced tables have indexes with the same name. This resulted in an
"ForeignKeyConstraint with duplicate source column references are not
supported" error when attempting to reflect such tables. The query has been
corrected to exclude indexes on the child table when looking for unique
indexes referenced by foreign keys.

Fixes: #12907
Change-Id: I435d4cf3bfa9e861cbb5e1e5c8c81bd59c9a9d58
(cherry picked from commit 652f610a35ca0ba5286b2603dd1de84528262ae2)

diff --git a/doc/build/changelog/unreleased_20/12907.rst b/doc/build/changelog/unreleased_20/12907.rst
new file mode 100644 (file)
index 0000000..b89c98f
--- /dev/null
+++ b/doc/build/changelog/unreleased_20/12907.rst
@@ -0,0 +1,12 @@
+.. change::
+    :tags: bug, mssql, reflection
+    :tickets: 12907
+
+    Fixed issue in the MSSQL dialect's foreign key reflection query where
+    duplicate rows could be returned when a foreign key column and its
+    referenced primary key column have the same name, and both the referencing
+    and referenced tables have indexes with the same name. This resulted in an
+    "ForeignKeyConstraint with duplicate source column references are not
+    supported" error when attempting to reflect such tables. The query has been
+    corrected to exclude indexes on the child table when looking for unique
+    indexes referenced by foreign keys.

diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
index 368abaf8ffb71888dbddc4ea268b5b8a3c849102..fed174070352a105918658cba7c3844137630e94 100644 (file)
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -3998,6 +3998,8 @@ index_info AS (
             index_info.index_schema = fk_info.unique_constraint_schema
             AND index_info.index_name = fk_info.unique_constraint_name
             AND index_info.ordinal_position = fk_info.ordinal_position
+            AND NOT (index_info.table_schema = fk_info.table_schema
+                     AND index_info.table_name = fk_info.table_name)
 
     ORDER BY fk_info.constraint_schema, fk_info.constraint_name,
         fk_info.ordinal_position

diff --git a/test/dialect/mssql/test_reflection.py b/test/dialect/mssql/test_reflection.py
index 06e5147fbeefd244cc46b4d91e42bc54467f9b64..733f29fef07014ddceb5720aedef82f9dcb7535d 100644 (file)
--- a/test/dialect/mssql/test_reflection.py
+++ b/test/dialect/mssql/test_reflection.py
@@ -533,6 +533,41 @@ class ReflectionTest(fixtures.TestBase, ComparesTables, AssertsCompiledSQL):
             ],
         )
 
+    def test_fk_with_same_column_name_as_pk_idx(self, metadata, connection):
+        """test #12907"""
+        # Create table A with primary key AId and a unique index IX_A_AId
+        Table(
+            "a",
+            metadata,
+            Column("aid", Integer, nullable=False),
+            Column("name", types.String(50)),
+            PrimaryKeyConstraint("aid", name="PK_A"),
+        ).create(connection)
+
+        # IMPORTANT - create unique index on a *first* before creating
+        # FK on B, this affects how the FK is generated in SQL server
+        connection.exec_driver_sql("CREATE UNIQUE INDEX IX_A_AId ON a (aid)")
+
+        # Create table B with foreign key column AId referencing A(AId)
+        # and an index with the same name IX_A_AId
+        Table(
+            "b",
+            metadata,
+            Column("id", Integer, Identity(), primary_key=True),
+            Column("aid", Integer),
+            ForeignKeyConstraint(["aid"], ["a.aid"], name="FK_B_A"),
+        ).create(connection)
+        connection.exec_driver_sql("CREATE INDEX IX_A_AId ON B(aid)")
+
+        m2 = MetaData()
+        table_b = Table("b", m2, autoload_with=connection)
+
+        fks = list(table_b.foreign_keys)
+        eq_(len(fks), 1)
+        eq_(fks[0].parent.name, "aid")
+        eq_(fks[0].column.table.name, "a")
+        eq_(fks[0].column.name, "aid")
+
     def test_indexes_cols(self, metadata, connection):
         t1 = Table("t", metadata, Column("x", Integer), Column("y", Integer))
         Index("foo", t1.c.x, t1.c.y)