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

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 ff67ee1ef5e4db7d98cbbdced81a6869dbb676cc..96d4b520bc915f2ee67c3e37efb3c7a1f2fe217f 100644 (file)
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -3974,6 +3974,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)