- New Oracle DDL features for tables, indexes: COMPRESS, BITMAP.

Patch courtesy Gabor Gombas.
fixes #3127

5 files changed, 272 insertions, 11 deletions

diff --git a/doc/build/changelog/changelog_10.rst b/doc/build/changelog/changelog_10.rst
index 0256958b2..b71ecc15d 100644
--- a/doc/build/changelog/changelog_10.rst
+++ b/doc/build/changelog/changelog_10.rst
@@ -23,6 +23,12 @@
     on compatibility concerns, see :doc:`/changelog/migration_10`.
 
     .. change::
+        :tags: feature, oracle
+
+        New Oracle DDL features for tables, indexes: COMPRESS, BITMAP.
+        Patch courtesy Gabor Gombas.
+
+    .. change::
         :tags: bug, oracle
 
         An alias name will be properly quoted when referred to using the
diff --git a/doc/build/changelog/migration_10.rst b/doc/build/changelog/migration_10.rst
index 9fbbb889d..27a4fae4c 100644
--- a/doc/build/changelog/migration_10.rst
+++ b/doc/build/changelog/migration_10.rst
@@ -1634,6 +1634,15 @@ CTE support has been fixed up for Oracle, and there is also a new feature
 
 :ticket:`3220`
 
+New Oracle Keywords for DDL
+-----------------------------
+
+Keywords such as COMPRESS, ON COMMIT, BITMAP:
+
+:ref:`oracle_table_options`
+
+:ref:`oracle_index_options`
+
 .. _change_2984:
 
 Drizzle Dialect is now an External Dialect
diff --git a/lib/sqlalchemy/dialects/oracle/base.py b/lib/sqlalchemy/dialects/oracle/base.py
index 524ba8115..9f375da94 100644
--- a/lib/sqlalchemy/dialects/oracle/base.py
+++ b/lib/sqlalchemy/dialects/oracle/base.py
@@ -213,6 +213,8 @@ is reflected and the type is reported as ``DATE``, the time-supporting
    examining the type of column for use in special Python translations or
    for migrating schemas to other database backends.
 
+.. _oracle_table_options:
+
 Oracle Table Options
 -------------------------
 
@@ -228,15 +230,63 @@ in conjunction with the :class:`.Table` construct:
 
 .. versionadded:: 1.0.0
 
+* ``COMPRESS``::
+
+    Table('mytable', metadata, Column('data', String(32)),
+        oracle_compress=True)
+
+    Table('mytable', metadata, Column('data', String(32)),
+        oracle_compress=6)
+
+   The ``oracle_compress`` parameter accepts either an integer compression
+   level, or ``True`` to use the default compression level.
+
+.. versionadded:: 1.0.0
+
+.. _oracle_index_options:
+
+Oracle Specific Index Options
+-----------------------------
+
+Bitmap Indexes
+~~~~~~~~~~~~~~
+
+You can specify the ``oracle_bitmap`` parameter to create a bitmap index
+instead of a B-tree index::
+
+    Index('my_index', my_table.c.data, oracle_bitmap=True)
+
+Bitmap indexes cannot be unique and cannot be compressed. SQLAlchemy will not
+check for such limitations, only the database will.
+
+.. versionadded:: 1.0.0
+
+Index compression
+~~~~~~~~~~~~~~~~~
+
+Oracle has a more efficient storage mode for indexes containing lots of
+repeated values. Use the ``oracle_compress`` parameter to turn on key c
+ompression::
+
+    Index('my_index', my_table.c.data, oracle_compress=True)
+
+    Index('my_index', my_table.c.data1, my_table.c.data2, unique=True,
+           oracle_compress=1)
+
+The ``oracle_compress`` parameter accepts either an integer specifying the
+number of prefix columns to compress, or ``True`` to use the default (all
+columns for non-unique indexes, all but the last column for unique indexes).
+
+.. versionadded:: 1.0.0
+
 """
 
 import re
 
 from sqlalchemy import util, sql
-from sqlalchemy.engine import default, base, reflection
+from sqlalchemy.engine import default, reflection
 from sqlalchemy.sql import compiler, visitors, expression
-from sqlalchemy.sql import (operators as sql_operators,
-                            functions as sql_functions)
+from sqlalchemy.sql import operators as sql_operators
 from sqlalchemy import types as sqltypes, schema as sa_schema
 from sqlalchemy.types import VARCHAR, NVARCHAR, CHAR, \
     BLOB, CLOB, TIMESTAMP, FLOAT
@@ -786,9 +836,32 @@ class OracleDDLCompiler(compiler.DDLCompiler):
 
         return text
 
-    def visit_create_index(self, create, **kw):
-        return super(OracleDDLCompiler, self).\
-            visit_create_index(create, include_schema=True)
+    def visit_create_index(self, create):
+        index = create.element
+        self._verify_index_table(index)
+        preparer = self.preparer
+        text = "CREATE "
+        if index.unique:
+            text += "UNIQUE "
+        if index.dialect_options['oracle']['bitmap']:
+            text += "BITMAP "
+        text += "INDEX %s ON %s (%s)" % (
+            self._prepared_index_name(index, include_schema=True),
+            preparer.format_table(index.table, use_schema=True),
+            ', '.join(
+                self.sql_compiler.process(
+                    expr,
+                    include_table=False, literal_binds=True)
+                for expr in index.expressions)
+        )
+        if index.dialect_options['oracle']['compress'] is not False:
+            if index.dialect_options['oracle']['compress'] is True:
+                text += " COMPRESS"
+            else:
+                text += " COMPRESS %d" % (
+                    index.dialect_options['oracle']['compress']
+                )
+        return text
 
     def post_create_table(self, table):
         table_opts = []
@@ -798,6 +871,14 @@ class OracleDDLCompiler(compiler.DDLCompiler):
             on_commit_options = opts['on_commit'].replace("_", " ").upper()
             table_opts.append('\n ON COMMIT %s' % on_commit_options)
 
+        if opts['compress']:
+            if opts['compress'] is True:
+                table_opts.append("\n COMPRESS")
+            else:
+                table_opts.append("\n COMPRESS FOR %s" % (
+                    opts['compress']
+                ))
+
         return ''.join(table_opts)
 
 
@@ -861,7 +942,12 @@ class OracleDialect(default.DefaultDialect):
     construct_arguments = [
         (sa_schema.Table, {
             "resolve_synonyms": False,
-            "on_commit": None
+            "on_commit": None,
+            "compress": False
+        }),
+        (sa_schema.Index, {
+            "bitmap": False,
+            "compress": False
         })
     ]
 
@@ -893,6 +979,16 @@ class OracleDialect(default.DefaultDialect):
             self.server_version_info < (9, )
 
     @property
+    def _supports_table_compression(self):
+        return self.server_version_info and \
+            self.server_version_info >= (9, 2, )
+
+    @property
+    def _supports_table_compress_for(self):
+        return self.server_version_info and \
+            self.server_version_info >= (11, )
+
+    @property
     def _supports_char_length(self):
         return not self._is_oracle_8
 
@@ -1075,6 +1171,50 @@ class OracleDialect(default.DefaultDialect):
         return [self.normalize_name(row[0]) for row in cursor]
 
     @reflection.cache
+    def get_table_options(self, connection, table_name, schema=None, **kw):
+        options = {}
+
+        resolve_synonyms = kw.get('oracle_resolve_synonyms', False)
+        dblink = kw.get('dblink', '')
+        info_cache = kw.get('info_cache')
+
+        (table_name, schema, dblink, synonym) = \
+            self._prepare_reflection_args(connection, table_name, schema,
+                                          resolve_synonyms, dblink,
+                                          info_cache=info_cache)
+
+        params = {"table_name": table_name}
+
+        columns = ["table_name"]
+        if self._supports_table_compression:
+            columns.append("compression")
+        if self._supports_table_compress_for:
+            columns.append("compress_for")
+
+        text = "SELECT %(columns)s "\
+            "FROM ALL_TABLES%(dblink)s "\
+            "WHERE table_name = :table_name"
+
+        if schema is not None:
+            params['owner'] = schema
+            text += " AND owner = :owner "
+        text = text % {'dblink': dblink, 'columns': ", ".join(columns)}
+
+        result = connection.execute(sql.text(text), **params)
+
+        enabled = dict(DISABLED=False, ENABLED=True)
+
+        row = result.first()
+        if row:
+            if "compression" in row and enabled.get(row.compression, False):
+                if "compress_for" in row:
+                    options['oracle_compress'] = row.compress_for
+                else:
+                    options['oracle_compress'] = True
+
+        return options
+
+    @reflection.cache
     def get_columns(self, connection, table_name, schema=None, **kw):
         """
 
@@ -1159,7 +1299,8 @@ class OracleDialect(default.DefaultDialect):
 
         params = {'table_name': table_name}
         text = \
-            "SELECT a.index_name, a.column_name, b.uniqueness "\
+            "SELECT a.index_name, a.column_name, "\
+            "\nb.index_type, b.uniqueness, b.compression, b.prefix_length "\
             "\nFROM ALL_IND_COLUMNS%(dblink)s a, "\
             "\nALL_INDEXES%(dblink)s b "\
             "\nWHERE "\
@@ -1185,6 +1326,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)
 
         oracle_sys_col = re.compile(r'SYS_NC\d+\$', re.IGNORECASE)
 
@@ -1204,10 +1346,15 @@ class OracleDialect(default.DefaultDialect):
             if rset.index_name != last_index_name:
                 remove_if_primary_key(index)
                 index = dict(name=self.normalize_name(rset.index_name),
-                             column_names=[])
+                             column_names=[], dialect_options={})
                 indexes.append(index)
             index['unique'] = uniqueness.get(rset.uniqueness, False)
 
+            if rset.index_type in ('BITMAP', 'FUNCTION-BASED BITMAP'):
+                index['dialect_options']['oracle_bitmap'] = True
+            if enabled.get(rset.compression, False):
+                index['dialect_options']['oracle_compress'] = rset.prefix_length
+
             # 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):
diff --git a/lib/sqlalchemy/engine/reflection.py b/lib/sqlalchemy/engine/reflection.py
index 2a1def86a..ebc96f5dd 100644
--- a/lib/sqlalchemy/engine/reflection.py
+++ b/lib/sqlalchemy/engine/reflection.py
@@ -394,6 +394,9 @@ class Inspector(object):
         unique
           boolean
 
+        dialect_options
+          dict of dialect-specific index options
+
         :param table_name: string name of the table.  For special quoting,
          use :class:`.quoted_name`.
 
@@ -642,6 +645,8 @@ class Inspector(object):
             columns = index_d['column_names']
             unique = index_d['unique']
             flavor = index_d.get('type', 'index')
+            dialect_options = index_d.get('dialect_options', {})
+
             duplicates = index_d.get('duplicates_constraint')
             if include_columns and \
                     not set(columns).issubset(include_columns):
@@ -667,7 +672,10 @@ class Inspector(object):
                 else:
                     idx_cols.append(idx_col)
 
-            sa_schema.Index(name, *idx_cols, **dict(unique=unique))
+            sa_schema.Index(
+                name, *idx_cols,
+                **dict(list(dialect_options.items()) + [('unique', unique)])
+            )
 
     def _reflect_unique_constraints(
         self, table_name, schema, table, cols_by_orig_name,
diff --git a/test/dialect/test_oracle.py b/test/dialect/test_oracle.py
index b2a490e71..1e50b9070 100644
--- a/test/dialect/test_oracle.py
+++ b/test/dialect/test_oracle.py
@@ -732,6 +732,34 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
         )
 
 
+    def test_create_table_compress(self):
+        m = MetaData()
+        tbl1 = Table('testtbl1', m, Column('data', Integer),
+                     oracle_compress=True)
+        tbl2 = Table('testtbl2', m, Column('data', Integer),
+                     oracle_compress="OLTP")
+
+        self.assert_compile(schema.CreateTable(tbl1),
+                            "CREATE TABLE testtbl1 (data INTEGER) COMPRESS")
+        self.assert_compile(schema.CreateTable(tbl2),
+                            "CREATE TABLE testtbl2 (data INTEGER) "
+                            "COMPRESS FOR OLTP")
+
+    def test_create_index_bitmap_compress(self):
+        m = MetaData()
+        tbl = Table('testtbl', m, Column('data', Integer))
+        idx1 = Index('idx1', tbl.c.data, oracle_compress=True)
+        idx2 = Index('idx2', tbl.c.data, oracle_compress=1)
+        idx3 = Index('idx3', tbl.c.data, oracle_bitmap=True)
+
+        self.assert_compile(schema.CreateIndex(idx1),
+                            "CREATE INDEX idx1 ON testtbl (data) COMPRESS")
+        self.assert_compile(schema.CreateIndex(idx2),
+                            "CREATE INDEX idx2 ON testtbl (data) COMPRESS 1")
+        self.assert_compile(schema.CreateIndex(idx3),
+                            "CREATE BITMAP INDEX idx3 ON testtbl (data)")
+
+
 class CompatFlagsTest(fixtures.TestBase, AssertsCompiledSQL):
 
     def _dialect(self, server_version, **kw):
@@ -1772,6 +1800,58 @@ class UnsupportedIndexReflectTest(fixtures.TestBase):
         m2 = MetaData(testing.db)
         Table('test_index_reflect', m2, autoload=True)
 
+
+def all_tables_compression_missing():
+    try:
+        testing.db.execute('SELECT compression FROM all_tables')
+        return False
+    except:
+        return True
+
+
+def all_tables_compress_for_missing():
+    try:
+        testing.db.execute('SELECT compress_for FROM all_tables')
+        return False
+    except:
+        return True
+
+
+class TableReflectionTest(fixtures.TestBase):
+    __only_on__ = 'oracle'
+
+    @testing.provide_metadata
+    @testing.fails_if(all_tables_compression_missing)
+    def test_reflect_basic_compression(self):
+        metadata = self.metadata
+
+        tbl = Table('test_compress', metadata,
+                    Column('data', Integer, primary_key=True),
+                    oracle_compress=True)
+        metadata.create_all()
+
+        m2 = MetaData(testing.db)
+
+        tbl = Table('test_compress', m2, autoload=True)
+        # Don't hardcode the exact value, but it must be non-empty
+        assert tbl.dialect_options['oracle']['compress']
+
+    @testing.provide_metadata
+    @testing.fails_if(all_tables_compress_for_missing)
+    def test_reflect_oltp_compression(self):
+        metadata = self.metadata
+
+        tbl = Table('test_compress', metadata,
+                    Column('data', Integer, primary_key=True),
+                    oracle_compress="OLTP")
+        metadata.create_all()
+
+        m2 = MetaData(testing.db)
+
+        tbl = Table('test_compress', m2, autoload=True)
+        assert tbl.dialect_options['oracle']['compress'] == "OLTP"
+
+
 class RoundTripIndexTest(fixtures.TestBase):
     __only_on__ = 'oracle'
 
@@ -1789,6 +1869,10 @@ class RoundTripIndexTest(fixtures.TestBase):
 
         # "group" is a keyword, so lower case
         normalind = Index('tableind', table.c.id_b, table.c.group)
+        compress1 = Index('compress1', table.c.id_a, table.c.id_b,
+                          oracle_compress=True)
+        compress2 = Index('compress2', table.c.id_a, table.c.id_b, table.c.col,
+                          oracle_compress=1)
 
         metadata.create_all()
         mirror = MetaData(testing.db)
@@ -1837,8 +1921,15 @@ class RoundTripIndexTest(fixtures.TestBase):
         )
         assert (Index, ('id_b', ), True) in reflected
         assert (Index, ('col', 'group'), True) in reflected
+
+        idx = reflected[(Index, ('id_a', 'id_b', ), False)]
+        assert idx.dialect_options['oracle']['compress'] == 2
+
+        idx = reflected[(Index, ('id_a', 'id_b', 'col', ), False)]
+        assert idx.dialect_options['oracle']['compress'] == 1
+
         eq_(len(reflectedtable.constraints), 1)
-        eq_(len(reflectedtable.indexes), 3)
+        eq_(len(reflectedtable.indexes), 5)
 
 class SequenceTest(fixtures.TestBase, AssertsCompiledSQL):