From: Mike Bayer <mike_mp@zzzcomputing.com>
Date: Wed, 8 Oct 2025 15:09:59 +0000 (-0400)
Subject: correct for FK reflecting on PK / index col at the same time
X-Git-Url: http://git.ipfire.org/gitweb.cgi?a=commitdiff_plain;h=652f610a35ca0ba5286b2603dd1de84528262ae2;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git

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
index 0000000000..b89c98f358
--- /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 ff67ee1ef5..96d4b520bc 100644
--- 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 06e5147fbe..733f29fef0 100644
--- 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)