- An adjustment to table/index reflection such that if an index

reports a column that isn't found to be present in the table,
a warning is emitted and the column is skipped.  This can occur
for some special system column situations as has been observed
with Oracle. fixes #3180

diff --git a/doc/build/changelog/changelog_09.rst b/doc/build/changelog/changelog_09.rst
index 2931916e3cc4042c72aa9a5b8bf3660189709489..44a2add716aa1ef1c4dc6062c4268715ade49c2c 100644 (file)
--- a/doc/build/changelog/changelog_09.rst
+++ b/doc/build/changelog/changelog_09.rst
@@ -13,6 +13,17 @@
 .. changelog::
     :version: 0.9.8
 
+    .. change::
+        :tags: bug, sql
+        :versions: 1.0.0
+        :tickets: 3180
+
+        An adjustment to table/index reflection such that if an index
+        reports a column that isn't found to be present in the table,
+        a warning is emitted and the column is skipped.  This can occur
+        for some special system column situations as has been observed
+        with Oracle.
+
     .. change::
         :tags: bug, ext
         :versions: 1.0.0

diff --git a/lib/sqlalchemy/engine/reflection.py b/lib/sqlalchemy/engine/reflection.py
index 012d1d35ddb60dc8b4ecd8ad9df0dd728ea842a9..cf1f2d3dd1302937353bc988a935b79cd58a240e 100644 (file)
--- a/lib/sqlalchemy/engine/reflection.py
+++ b/lib/sqlalchemy/engine/reflection.py
@@ -578,18 +578,27 @@ class Inspector(object):
             name = index_d['name']
             columns = index_d['column_names']
             unique = index_d['unique']
-            flavor = index_d.get('type', 'unknown type')
+            flavor = index_d.get('type', 'index')
             if include_columns and \
                     not set(columns).issubset(include_columns):
                 util.warn(
-                    "Omitting %s KEY for (%s), key covers omitted columns." %
+                    "Omitting %s key for (%s), key covers omitted columns." %
                     (flavor, ', '.join(columns)))
                 continue
             # look for columns by orig name in cols_by_orig_name,
             # but support columns that are in-Python only as fallback
-            sa_schema.Index(name, *[
-                cols_by_orig_name[c] if c in cols_by_orig_name
-                else table.c[c]
-                for c in columns
-            ],
-                **dict(unique=unique))
+            idx_cols = []
+            for c in columns:
+                try:
+                    idx_col = cols_by_orig_name[c] \
+                        if c in cols_by_orig_name else table.c[c]
+                except KeyError:
+                    util.warn(
+                        "%s key '%s' was not located in "
+                        "columns for table '%s'" % (
+                            flavor, c, table_name
+                        ))
+                else:
+                    idx_cols.append(idx_col)
+
+            sa_schema.Index(name, *idx_cols, **dict(unique=unique))

diff --git a/test/engine/test_reflection.py b/test/engine/test_reflection.py
index 807abc84f79c8455405a57b0791b2d120c432f26..c18b8b94462b2f633831ba7c35352fa3e1f004fc 100644 (file)
--- a/test/engine/test_reflection.py
+++ b/test/engine/test_reflection.py
@@ -983,6 +983,26 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
         assert set([t2.c.name, t2.c.id]) == set(r2.columns)
         assert set([t2.c.name]) == set(r3.columns)
 
+    @testing.provide_metadata
+    def test_index_reflection_cols_busted(self):
+        t = Table('x', self.metadata,
+                  Column('a', Integer), Column('b', Integer))
+        sa.Index('x_ix', t.c.a, t.c.b)
+        self.metadata.create_all()
+
+        def mock_get_columns(self, connection, table_name, **kw):
+            return [
+                {"name": "b", "type": Integer, "primary_key": False}
+            ]
+
+        with testing.mock.patch.object(
+                testing.db.dialect, "get_columns", mock_get_columns):
+            m = MetaData()
+            with testing.expect_warnings(
+                    "index key 'a' was not located in columns"):
+                t = Table('x', m, autoload=True, autoload_with=testing.db)
+
+        eq_(list(t.indexes)[0].columns, [t.c.b])
 
     @testing.requires.views
     @testing.provide_metadata