Merge "Ensure Oracle index w/ col DESC etc. is reflected"

8 files changed, 142 insertions, 24 deletions

diff --git a/doc/build/changelog/unreleased_12/4042.rst b/doc/build/changelog/unreleased_12/4042.rst
new file mode 100644
index 000000000..8ce04a9a7
--- /dev/null
+++ b/doc/build/changelog/unreleased_12/4042.rst
@@ -0,0 +1,8 @@
+.. change::
+    :tags: bug, oracle
+    :tickets: 4042
+
+    Fixed bug where an index reflected under Oracle with an expression like
+    "column DESC" would not be returned, if the table also had no primary
+    key, as a result of logic that attempts to filter out the
+    index implicitly added by Oracle onto the primary key columns.
diff --git a/lib/sqlalchemy/dialects/oracle/base.py b/lib/sqlalchemy/dialects/oracle/base.py
index d9fa80df1..9478f6531 100644
--- a/lib/sqlalchemy/dialects/oracle/base.py
+++ b/lib/sqlalchemy/dialects/oracle/base.py
@@ -1469,21 +1469,9 @@ class OracleDialect(default.DefaultDialect):
 
         oracle_sys_col = re.compile(r'SYS_NC\d+\$', re.IGNORECASE)
 
-        def upper_name_set(names):
-            return {i.upper() for i in names}
-
-        pk_names = upper_name_set(pkeys)
-
-        def remove_if_primary_key(index):
-            # don't include the primary key index
-            if index is not None and \
-               upper_name_set(index['column_names']) == pk_names:
-                indexes.pop()
-
         index = None
         for rset in rp:
             if rset.index_name != last_index_name:
-                remove_if_primary_key(index)
                 index = dict(name=self.normalize_name(rset.index_name),
                              column_names=[], dialect_options={})
                 indexes.append(index)
@@ -1500,7 +1488,17 @@ class OracleDialect(default.DefaultDialect):
                 index['column_names'].append(
                     self.normalize_name(rset.column_name))
             last_index_name = rset.index_name
-        remove_if_primary_key(index)
+
+        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/lib/sqlalchemy/engine/reflection.py b/lib/sqlalchemy/engine/reflection.py
index 531be3939..aff9de063 100644
--- a/lib/sqlalchemy/engine/reflection.py
+++ b/lib/sqlalchemy/engine/reflection.py
@@ -810,6 +810,7 @@ class Inspector(object):
 
             sa_schema.Index(
                 name, *idx_cols,
+                _table=table,
                 **dict(list(dialect_options.items()) + [('unique', unique)])
             )
 
diff --git a/lib/sqlalchemy/sql/base.py b/lib/sqlalchemy/sql/base.py
index 7a04beb57..4d638ea1d 100644
--- a/lib/sqlalchemy/sql/base.py
+++ b/lib/sqlalchemy/sql/base.py
@@ -517,6 +517,10 @@ class ColumnCollection(util.OrderedProperties):
             # columns collection
 
             existing = self[key]
+
+            if existing is value:
+                return
+
             if not existing.shares_lineage(value):
                 util.warn('Column %r on table %r being replaced by '
                           '%r, which has the same key.  Consider '
diff --git a/lib/sqlalchemy/sql/schema.py b/lib/sqlalchemy/sql/schema.py
index 9b73eca63..7a78a715f 100644
--- a/lib/sqlalchemy/sql/schema.py
+++ b/lib/sqlalchemy/sql/schema.py
@@ -3402,7 +3402,7 @@ class Index(DialectKWArgs, ColumnCollectionMixin, SchemaItem):
             documented arguments.
 
         """
-        self.table = None
+        self.table = table = None
 
         columns = []
         processed_expressions = []
@@ -3417,12 +3417,21 @@ class Index(DialectKWArgs, ColumnCollectionMixin, SchemaItem):
         self.unique = kw.pop('unique', False)
         if 'info' in kw:
             self.info = kw.pop('info')
+
+        # TODO: consider "table" argument being public, but for
+        # the purpose of the fix here, it starts as private.
+        if '_table' in kw:
+            table = kw.pop('_table')
+
         self._validate_dialect_kwargs(kw)
 
         # will call _set_parent() if table-bound column
         # objects are present
         ColumnCollectionMixin.__init__(self, *columns)
 
+        if table is not None:
+            self._set_parent(table)
+
     def _set_parent(self, table):
         ColumnCollectionMixin._set_parent(self, table)
 
diff --git a/lib/sqlalchemy/testing/suite/test_reflection.py b/lib/sqlalchemy/testing/suite/test_reflection.py
index 1c4fd7d4d..54b59a432 100644
--- a/lib/sqlalchemy/testing/suite/test_reflection.py
+++ b/lib/sqlalchemy/testing/suite/test_reflection.py
@@ -4,11 +4,11 @@ import sqlalchemy as sa
 from sqlalchemy import exc as sa_exc
 from sqlalchemy import types as sql_types
 from sqlalchemy import inspect
-from sqlalchemy import MetaData, Integer, String
+from sqlalchemy import MetaData, Integer, String, func
 from sqlalchemy.engine.reflection import Inspector
 from sqlalchemy.testing import engines, fixtures
 from sqlalchemy.testing.schema import Table, Column
-from sqlalchemy.testing import eq_, assert_raises_message
+from sqlalchemy.testing import eq_, is_, assert_raises_message
 from sqlalchemy import testing
 from .. import config
 import operator
@@ -110,6 +110,20 @@ class ComponentReflectionTest(fixtures.TablesTest):
 
         if testing.requires.index_reflection.enabled:
             cls.define_index(metadata, users)
+
+            if not schema:
+                noncol_idx_test_nopk = Table(
+                    'noncol_idx_test_nopk', metadata,
+                    Column('q', sa.String(5)),
+                )
+                noncol_idx_test_pk = Table(
+                    'noncol_idx_test_pk', metadata,
+                    Column('id', sa.Integer, primary_key=True),
+                    Column('q', sa.String(5)),
+                )
+                Index('noncol_idx_nopk', noncol_idx_test_nopk.c.q.desc())
+                Index('noncol_idx_pk', noncol_idx_test_pk.c.q.desc())
+
         if testing.requires.view_column_reflection.enabled:
             cls.define_views(metadata, schema)
         if not schema and testing.requires.temp_table_reflection.enabled:
@@ -197,6 +211,9 @@ class ComponentReflectionTest(fixtures.TablesTest):
     @testing.provide_metadata
     def _test_get_table_names(self, schema=None, table_type='table',
                               order_by=None):
+        _ignore_tables = [
+            'comment_test', 'noncol_idx_test_pk', 'noncol_idx_test_nopk'
+        ]
         meta = self.metadata
         users, addresses, dingalings = self.tables.users, \
             self.tables.email_addresses, self.tables.dingalings
@@ -211,7 +228,7 @@ class ComponentReflectionTest(fixtures.TablesTest):
             table_names = [
                 t for t in insp.get_table_names(
                     schema,
-                    order_by=order_by) if t not in ('comment_test', )]
+                    order_by=order_by) if t not in _ignore_tables]
 
             if order_by == 'foreign_key':
                 answer = ['users', 'email_addresses', 'dingalings']
@@ -569,6 +586,14 @@ class ComponentReflectionTest(fixtures.TablesTest):
             {'onupdate': 'SET NULL', 'ondelete': 'CASCADE'}
         )
 
+    def _assert_insp_indexes(self, indexes, expected_indexes):
+        index_names = [d['name'] for d in indexes]
+        for e_index in expected_indexes:
+            assert e_index['name'] in index_names
+            index = indexes[index_names.index(e_index['name'])]
+            for key in e_index:
+                eq_(e_index[key], index[key])
+
     @testing.provide_metadata
     def _test_get_indexes(self, schema=None):
         meta = self.metadata
@@ -586,12 +611,7 @@ class ComponentReflectionTest(fixtures.TablesTest):
              'column_names': ['user_id', 'test2', 'test1'],
              'name': 'users_all_idx'}
         ]
-        index_names = [d['name'] for d in indexes]
-        for e_index in expected_indexes:
-            assert e_index['name'] in index_names
-            index = indexes[index_names.index(e_index['name'])]
-            for key in e_index:
-                eq_(e_index[key], index[key])
+        self._assert_insp_indexes(indexes, expected_indexes)
 
     @testing.requires.index_reflection
     def test_get_indexes(self):
@@ -602,6 +622,34 @@ class ComponentReflectionTest(fixtures.TablesTest):
     def test_get_indexes_with_schema(self):
         self._test_get_indexes(schema=testing.config.test_schema)
 
+    @testing.provide_metadata
+    def _test_get_noncol_index(self, tname, ixname):
+        meta = self.metadata
+        insp = inspect(meta.bind)
+        indexes = insp.get_indexes(tname)
+
+        # reflecting an index that has "x DESC" in it as the column.
+        # the DB may or may not give us "x", but make sure we get the index
+        # back, it has a name, it's connected to the table.
+        expected_indexes = [
+            {'unique': False,
+             'name': ixname}
+        ]
+        self._assert_insp_indexes(indexes, expected_indexes)
+
+        t = Table(tname, meta, autoload_with=meta.bind)
+        eq_(len(t.indexes), 1)
+        is_(list(t.indexes)[0].table, t)
+        eq_(list(t.indexes)[0].name, ixname)
+
+    @testing.requires.index_reflection
+    def test_get_noncol_index_no_pk(self):
+        self._test_get_noncol_index("noncol_idx_test_nopk", "noncol_idx_nopk")
+
+    @testing.requires.index_reflection
+    def test_get_noncol_index_pk(self):
+        self._test_get_noncol_index("noncol_idx_test_pk", "noncol_idx_pk")
+
     @testing.requires.unique_constraint_reflection
     def test_get_unique_constraints(self):
         self._test_get_unique_constraints()
diff --git a/test/base/test_utils.py b/test/base/test_utils.py
index b181a6fdb..9c36aeb9f 100644
--- a/test/base/test_utils.py
+++ b/test/base/test_utils.py
@@ -439,7 +439,7 @@ class ToListTest(fixtures.TestBase):
         )
 
 
-class ColumnCollectionTest(fixtures.TestBase):
+class ColumnCollectionTest(testing.AssertsCompiledSQL, fixtures.TestBase):
 
     def test_in(self):
         cc = sql.ColumnCollection()
@@ -496,6 +496,41 @@ class ColumnCollectionTest(fixtures.TestBase):
         eq_(ci._all_columns, [c1, c2a, c3, c2b])
         eq_(list(ci), [c1, c2b, c3])
 
+    def test_identical_dupe_add(self):
+        cc = sql.ColumnCollection()
+
+        c1, c2, c3 = (column('c1'),
+                      column('c2'),
+                      column('c3'))
+
+        cc.add(c1)
+        cc.add(c2)
+        cc.add(c3)
+        cc.add(c2)
+
+        eq_(cc._all_columns, [c1, c2, c3])
+
+        self.assert_compile(
+            cc == [c1, c2, c3],
+            "c1 = c1 AND c2 = c2 AND c3 = c3"
+        )
+
+        # for iter, c2a is replaced by c2b, ordering
+        # is maintained in that way.  ideally, iter would be
+        # the same as the "_all_columns" collection.
+        eq_(list(cc), [c1, c2, c3])
+
+        assert cc.contains_column(c2)
+
+        ci = cc.as_immutable()
+        eq_(ci._all_columns, [c1, c2, c3])
+        eq_(list(ci), [c1, c2, c3])
+
+        self.assert_compile(
+            ci == [c1, c2, c3],
+            "c1 = c1 AND c2 = c2 AND c3 = c3"
+        )
+
     def test_replace(self):
         cc = sql.ColumnCollection()
 
diff --git a/test/sql/test_metadata.py b/test/sql/test_metadata.py
index 61fbbc57b..e204375f4 100644
--- a/test/sql/test_metadata.py
+++ b/test/sql/test_metadata.py
@@ -2255,6 +2255,21 @@ class ConstraintTest(fixtures.TestBase):
         i = Index('i', func.foo(t.c.x))
         self._assert_index_col_x(t, i)
 
+    def test_index_no_cols_private_table_arg(self):
+        m = MetaData()
+        t = Table('t', m, Column('x', Integer))
+        i = Index('i', _table=t)
+        is_(i.table, t)
+        eq_(list(i.columns), [])
+
+    def test_index_w_cols_private_table_arg(self):
+        m = MetaData()
+        t = Table('t', m, Column('x', Integer))
+        i = Index('i', t.c.x, _table=t)
+        is_(i.table, t)
+
+        eq_(i.columns, [t.c.x])
+
     def test_inline_decl_columns(self):
         m = MetaData()
         c = Column('x', Integer)