Merge "Use index name to determine if an index is for the PK"

3 files changed, 104 insertions, 14 deletions

diff --git a/doc/build/changelog/unreleased_13/5421.rst b/doc/build/changelog/unreleased_13/5421.rst
new file mode 100644
index 000000000..d0e10dd1d
--- /dev/null
+++ b/doc/build/changelog/unreleased_13/5421.rst
@@ -0,0 +1,10 @@
+.. change::
+    :tags: bug, oracle, reflection
+    :tickets: 5421
+
+    Fixed bug in Oracle dialect where indexes that contain the full set of
+    primary key columns would be mistaken as the primary key index itself,
+    which is omitted, even if there were multiples.  The check has been refined
+    to compare the name of the primary key constraint against the index name
+    itself, rather than trying to guess based on the columns present in the
+    index.
\ No newline at end of file
diff --git a/lib/sqlalchemy/dialects/oracle/base.py b/lib/sqlalchemy/dialects/oracle/base.py
index 5e912a0c2..34c665fbe 100644
--- a/lib/sqlalchemy/dialects/oracle/base.py
+++ b/lib/sqlalchemy/dialects/oracle/base.py
@@ -1898,7 +1898,7 @@ class OracleDialect(default.DefaultDialect):
             dblink=dblink,
             info_cache=kw.get("info_cache"),
         )
-        pkeys = pk_constraint["constrained_columns"]
+
         uniqueness = dict(NONUNIQUE=False, UNIQUE=True)
         enabled = dict(DISABLED=False, ENABLED=True)
 
@@ -1906,9 +1906,22 @@ class OracleDialect(default.DefaultDialect):
 
         index = None
         for rset in rp:
+            index_name_normalized = self.normalize_name(rset.index_name)
+
+            # skip primary key index.  This is refined as of
+            # [ticket:5421].  Note that ALL_INDEXES.GENERATED will by "Y"
+            # if the name of this index was generated by Oracle, however
+            # if a named primary key constraint was created then this flag
+            # is false.
+            if (
+                pk_constraint
+                and index_name_normalized == pk_constraint["name"]
+            ):
+                continue
+
             if rset.index_name != last_index_name:
                 index = dict(
-                    name=self.normalize_name(rset.index_name),
+                    name=index_name_normalized,
                     column_names=[],
                     dialect_options={},
                 )
@@ -1930,18 +1943,6 @@ class OracleDialect(default.DefaultDialect):
                 )
             last_index_name = rset.index_name
 
-        def upper_name_set(names):
-            return {i.upper() for i in names}
-
-        pk_names = upper_name_set(pkeys)
-        if pk_names:
-
-            def is_pk_index(index):
-                # don't include the primary key index
-                return upper_name_set(index["column_names"]) == pk_names
-
-            indexes = [idx for idx in indexes if not is_pk_index(idx)]
-
         return indexes
 
     @reflection.cache
diff --git a/test/dialect/oracle/test_reflection.py b/test/dialect/oracle/test_reflection.py
index 3d1361adb..97fb627f9 100644
--- a/test/dialect/oracle/test_reflection.py
+++ b/test/dialect/oracle/test_reflection.py
@@ -533,6 +533,83 @@ class RoundTripIndexTest(fixtures.TestBase):
     __backend__ = True
 
     @testing.provide_metadata
+    def test_no_pk(self):
+        metadata = self.metadata
+
+        Table(
+            "sometable",
+            metadata,
+            Column("id_a", Unicode(255)),
+            Column("id_b", Unicode(255)),
+            Index("pk_idx_1", "id_a", "id_b", unique=True),
+            Index("pk_idx_2", "id_b", "id_a", unique=True),
+        )
+        metadata.create_all()
+
+        insp = inspect(testing.db)
+        eq_(
+            insp.get_indexes("sometable"),
+            [
+                {
+                    "name": "pk_idx_1",
+                    "column_names": ["id_a", "id_b"],
+                    "dialect_options": {},
+                    "unique": True,
+                },
+                {
+                    "name": "pk_idx_2",
+                    "column_names": ["id_b", "id_a"],
+                    "dialect_options": {},
+                    "unique": True,
+                },
+            ],
+        )
+
+    @testing.combinations((True,), (False,))
+    @testing.provide_metadata
+    def test_include_indexes_resembling_pk(self, explicit_pk):
+        metadata = self.metadata
+
+        t = Table(
+            "sometable",
+            metadata,
+            Column("id_a", Unicode(255), primary_key=True),
+            Column("id_b", Unicode(255), primary_key=True),
+            Column("group", Unicode(255), primary_key=True),
+            Column("col", Unicode(255)),
+            # Oracle won't let you do this unless the indexes have
+            # the columns in different order
+            Index("pk_idx_1", "id_b", "id_a", "group", unique=True),
+            Index("pk_idx_2", "id_b", "group", "id_a", unique=True),
+        )
+        if explicit_pk:
+            t.append_constraint(
+                PrimaryKeyConstraint(
+                    "id_a", "id_b", "group", name="some_primary_key"
+                )
+            )
+        metadata.create_all()
+
+        insp = inspect(testing.db)
+        eq_(
+            insp.get_indexes("sometable"),
+            [
+                {
+                    "name": "pk_idx_1",
+                    "column_names": ["id_b", "id_a", "group"],
+                    "dialect_options": {},
+                    "unique": True,
+                },
+                {
+                    "name": "pk_idx_2",
+                    "column_names": ["id_b", "group", "id_a"],
+                    "dialect_options": {},
+                    "unique": True,
+                },
+            ],
+        )
+
+    @testing.provide_metadata
     def test_basic(self):
         metadata = self.metadata
 
@@ -560,8 +637,10 @@ class RoundTripIndexTest(fixtures.TestBase):
         )
 
         metadata.create_all()
+
         mirror = MetaData(testing.db)
         mirror.reflect()
+
         metadata.drop_all()
         mirror.create_all()