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

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.

Fixes: #5421
Change-Id: I47c2ccdd0b13977cfd9ef249d4de06371c4fb241

diff --git a/doc/build/changelog/unreleased_13/5421.rst b/doc/build/changelog/unreleased_13/5421.rst
new file mode 100644 (file)
index 0000000..d0e10dd
--- /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 5e912a0c2f13edf7ab6a2e7f4193b609f606eae6..34c665fbe56300f96539f8e457ccc7ee01325c29 100644 (file)
--- 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 3d1361adb6b849efd56927eb9fcff997b6a168d4..97fb627f9c7208e9b99647a6792666f9a0720ad6 100644 (file)
--- a/test/dialect/oracle/test_reflection.py
+++ b/test/dialect/oracle/test_reflection.py
@@ -532,6 +532,83 @@ class RoundTripIndexTest(fixtures.TestBase):
     __only_on__ = "oracle"
     __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()