- filter out SYS_NC\d+$ columns [ticket:1513]

- remove explicit INNER JOIN from index query to support oracle 8

diff --git a/lib/sqlalchemy/dialects/oracle/base.py b/lib/sqlalchemy/dialects/oracle/base.py
index 689b518f1d6aa3c76c35ce5a8e3c5227af0976d8..d02eb4984aeda6812b4fd28546c8172a8dfe5e54 100644 (file)
--- a/lib/sqlalchemy/dialects/oracle/base.py
+++ b/lib/sqlalchemy/dialects/oracle/base.py
@@ -725,12 +725,14 @@ class OracleDialect(default.DefaultDialect):
         indexes = []
         q = sql.text("""
         SELECT a.index_name, a.column_name, b.uniqueness
-        FROM ALL_IND_COLUMNS%(dblink)s a
-        INNER JOIN ALL_INDEXES%(dblink)s b
-            ON a.index_name = b.index_name
+        FROM ALL_IND_COLUMNS%(dblink)s a, 
+        ALL_INDEXES%(dblink)s b 
+        WHERE
+            a.index_name = b.index_name
             AND a.table_owner = b.table_owner
             AND a.table_name = b.table_name
-        WHERE a.table_name = :table_name
+        
+        AND a.table_name = :table_name
         AND a.table_owner = :schema
         ORDER BY a.index_name, a.column_position""" % {'dblink': dblink})
         rp = connection.execute(q, table_name=self.denormalize_name(table_name),
@@ -742,6 +744,8 @@ class OracleDialect(default.DefaultDialect):
                                       dblink=dblink,
                                       info_cache=kw.get('info_cache'))
         uniqueness = dict(NONUNIQUE=False, UNIQUE=True)
+        
+        oracle_sys_col = re.compile(r'SYS_NC\d+\$', re.IGNORECASE)
         for rset in rp:
             # don't include the primary key columns
             if rset.column_name in [s.upper() for s in pkeys]:
@@ -750,7 +754,11 @@ class OracleDialect(default.DefaultDialect):
                 index = dict(name=self.normalize_name(rset.index_name), column_names=[])
                 indexes.append(index)
             index['unique'] = uniqueness.get(rset.uniqueness, False)
-            index['column_names'].append(self.normalize_name(rset.column_name))
+
+            # filter out Oracle SYS_NC names.  could also do an outer join
+            # to the all_tab_columns table and check for real col names there.
+            if not oracle_sys_col.match(rset.column_name):
+                index['column_names'].append(self.normalize_name(rset.column_name))
             last_index_name = rset.index_name
         return indexes
 

diff --git a/test/dialect/test_oracle.py b/test/dialect/test_oracle.py
index 2153c322fb9ccea3f1b5679613a76b75ce7bf922..2630698174029df570b79103866814e442e90cfb 100644 (file)
--- a/test/dialect/test_oracle.py
+++ b/test/dialect/test_oracle.py
@@ -560,6 +560,22 @@ class BufferedColumnTest(TestBase, AssertsCompiledSQL):
         result = eng.execute(binary_table.select()).fetchall()
         eq_(result, [(i, stream) for i in range(1, 11)])
 
+class UnsupportedIndexReflectTest(TestBase):
+    def setup(self):
+        global metadata
+        metadata = MetaData(testing.db)
+        t1 = Table('test_index_reflect', metadata, Column('data', String(20), primary_key=True))
+        metadata.create_all()
+    
+    def teardown(self):
+        metadata.drop_all()
+        
+    def test_reflect_functional_index(self):
+        testing.db.execute("CREATE INDEX DATA_IDX ON TEST_INDEX_REFLECT (UPPER(DATA))")
+        m2 = MetaData(testing.db)
+        t2 = Table('test_index_reflect', m2, autoload=True)
+        
+        
 class SequenceTest(TestBase, AssertsCompiledSQL):
     def test_basic(self):
         seq = Sequence("my_seq_no_schema")