forked from thepaul/cassandra-dtest
-
Notifications
You must be signed in to change notification settings - Fork 0
/
compression_test.py
135 lines (115 loc) · 5.92 KB
/
compression_test.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
import os
from scrub_test import TestHelper
from tools import since
class TestCompression(TestHelper):
def _get_compression_type(self, file):
types = {
'0010': 'NONE',
'789c': 'DEFLATE'
}
with open(file, 'rb') as fh:
file_start = fh.read(2)
return types.get(file_start.encode('hex'), 'UNKNOWN')
@since("3.0")
def disable_compression_cql_test(self):
"""
@jira_ticket CASSANDRA-8384
using new cql create table syntax to disable compression
"""
cluster = self.cluster
cluster.populate(1).start(wait_for_binary_proto=True)
[node] = cluster.nodelist()
session = self.patient_cql_connection(node)
self.create_ks(session, 'ks', 1)
session.execute("create table disabled_compression_table (id uuid PRIMARY KEY ) WITH compression = {'enabled': false};")
session.cluster.refresh_schema_metadata()
meta = session.cluster.metadata.keyspaces['ks'].tables['disabled_compression_table']
self.assertEqual('false', meta.options['compression']['enabled'])
for n in range(0, 100):
session.execute("insert into disabled_compression_table (id) values (uuid());")
sstables = self.flush('disabled_compression_table')
sstable_path = self.get_table_path('disabled_compression_table')
self.assertEqual('NONE', self._get_compression_type(os.path.join(sstable_path, sstables['disabled_compression_table'][1])))
@since("3.0")
def compression_cql_options_test(self):
"""
@jira_ticket CASSANDRA-8384
using new cql create table syntax to configure compression
"""
cluster = self.cluster
cluster.populate(1).start(wait_for_binary_proto=True)
[node] = cluster.nodelist()
session = self.patient_cql_connection(node)
self.create_ks(session, 'ks', 1)
session.execute("""
create table compression_opts_table
(id uuid PRIMARY KEY )
WITH compression = {
'class': 'DeflateCompressor',
'chunk_length_in_kb': 256,
'crc_check_chance': 0.25
};
""")
session.cluster.refresh_schema_metadata()
meta = session.cluster.metadata.keyspaces['ks'].tables['compression_opts_table']
self.assertEqual('org.apache.cassandra.io.compress.DeflateCompressor', meta.options['compression']['class'])
self.assertEqual('256', meta.options['compression']['chunk_length_in_kb'])
self.assertEqual('0.25', meta.options['compression']['crc_check_chance'])
for n in range(0, 100):
session.execute("insert into compression_opts_table (id) values (uuid());")
sstables = self.flush('compression_opts_table')
sstable_path = self.get_table_path('compression_opts_table')
self.assertEqual('DEFLATE', self._get_compression_type(os.path.join(sstable_path, sstables['compression_opts_table'][1])))
@since("3.0")
def compression_cql_disabled_with_alter_test(self):
"""
@jira_ticket CASSANDRA-8384
starting with compression enabled then disabling it
"""
cluster = self.cluster
cluster.populate(1).start(wait_for_binary_proto=True)
[node] = cluster.nodelist()
session = self.patient_cql_connection(node)
self.create_ks(session, 'ks', 1)
session.execute("""
create table start_enabled_compression_table
(id uuid PRIMARY KEY )
WITH compression = {
'class': 'SnappyCompressor',
'chunk_length_in_kb': 256,
'crc_check_chance': 0.25
};
""")
meta = session.cluster.metadata.keyspaces['ks'].tables['start_enabled_compression_table']
self.assertEqual('org.apache.cassandra.io.compress.SnappyCompressor', meta.options['compression']['class'])
self.assertEqual('256', meta.options['compression']['chunk_length_in_kb'])
self.assertEqual('0.25', meta.options['compression']['crc_check_chance'])
session.execute("alter table start_enabled_compression_table with compression = {'enabled': false};")
session.cluster.refresh_schema_metadata()
meta = session.cluster.metadata.keyspaces['ks'].tables['start_enabled_compression_table']
self.assertEqual('false', meta.options['compression']['enabled'])
@since("3.0")
def compression_cql_enabled_with_alter_test(self):
"""
@jira_ticket CASSANDRA-8384
starting with compression disabled and enabling it
"""
cluster = self.cluster
cluster.populate(1).start(wait_for_binary_proto=True)
[node] = cluster.nodelist()
session = self.patient_cql_connection(node)
self.create_ks(session, 'ks', 1)
session.execute("create table start_disabled_compression_table (id uuid PRIMARY KEY ) WITH compression = {'enabled': false};")
meta = session.cluster.metadata.keyspaces['ks'].tables['start_disabled_compression_table']
self.assertEqual('false', meta.options['compression']['enabled'])
session.execute("""alter table start_disabled_compression_table
WITH compression = {
'class': 'SnappyCompressor',
'chunk_length_in_kb': 256,
'crc_check_chance': 0.25
};""")
session.cluster.refresh_schema_metadata()
meta = session.cluster.metadata.keyspaces['ks'].tables['start_disabled_compression_table']
self.assertEqual('org.apache.cassandra.io.compress.SnappyCompressor', meta.options['compression']['class'])
self.assertEqual('256', meta.options['compression']['chunk_length_in_kb'])
self.assertEqual('0.25', meta.options['compression']['crc_check_chance'])