Ensure Oracle index w/ col DESC etc. is reflected

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.

Reworked the "filter out the primary key index" logic in oracle
get_indexes() to be clearer.

This changeset also adds an internal check to ColumnCollection
to accomodate for the case of a column being added twice,
as well as adding a private _table argument to Index such that
reflection can specify the Table explicitly.  The _table
argument can become part of public API in a later revision
or release if needed.

Change-Id: I745711e03b3e450b7f31185fc70e10d3823063fa
Fixes: #4042

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)