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

Patch courtesy Gabor Gombas.
fixes #3127

1 files changed, 92 insertions, 1 deletions

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):