STAR-13 Run tests for UnifiedCompactionStrategy (#22)

(cherry picked from commit 47b978d)
(cherry picked from commit 4ecc81b)
(cherry picked from commit a06131c)
(cherry picked from commit 1064342)
(cherry picked from commit ab2abb7)
(cherry picked from commit aae8a6b)

squashes in e74acd8:
STAR-826 Add missing import (#40)

(cherry picked from commit 7c8489f)
(cherry picked from commit 9196eaa)
(cherry picked from commit 28c71d1)
(cherry picked from commit 2880475)
(cherry picked from commit 2a984ff)
(cherry picked from commit aad664b)

squashes in 01b3ee5f:
STAR-836 Fix TestCompaction_with_UnifiedCompactionStrategy.bloomfilter_size_test (#41)

Co-authored-by: Branimir Lambov <[email protected]>
(cherry picked from commit 7777fa9)
(cherry picked from commit d06d4c6)
(cherry picked from commit 12f5fff)
(cherry picked from commit a29f899)
(cherry picked from commit cfc58a3)
(cherry picked from commit 2a2726a)

rebase notes:
will be upstreamed in CASSANDRA-20629

Author: Gerrrr

compaction_test.py

@@ -118,25 +118,30 @@ def test_bloomfilter_size(self, strategy):
         else:
             if strategy == "DateTieredCompactionStrategy":
                 strategy_string = "strategy=DateTieredCompactionStrategy,base_time_seconds=86400"  # we want a single sstable, so make sure we don't have a tiny first window
+            elif self.strategy == "UnifiedCompactionStrategy":
+                strategy_string = "strategy=UnifiedCompactionStrategy,max_sstables_to_compact=4"  # disable layout-preserving compaction which can leave more than one sstable
             else:
                 strategy_string = "strategy={}".format(strategy)
             min_bf_size = 100000
             max_bf_size = 150000
         cluster = self.cluster
         cluster.populate(1).start()
         [node1] = cluster.nodelist()
+        logger.debug("Compaction: " + strategy_string)
 
         for x in range(0, 5):
             node1.stress(['write', 'n=100K', "no-warmup", "cl=ONE", "-rate",
                           "threads=300", "-schema", "replication(factor=1)",
                           "compaction({},enabled=false)".format(strategy_string)])
             node1.flush()
+        logger.debug(node1.nodetool('tablestats keyspace1.standard1').stdout)
 
         node1.nodetool('enableautocompaction')
         node1.wait_for_compactions()
 
         table_name = 'standard1'
-        output = node1.nodetool('tablestats').stdout
+        output = node1.nodetool('tablestats keyspace1.standard1').stdout
+        logger.debug(output)
         output = output[output.find(table_name):]
         output = output[output.find("Bloom filter space used"):]
         bfSize = int(output[output.find(":") + 1:output.find("\n")].strip())
@@ -157,7 +162,12 @@ def test_bloomfilter_size(self, strategy):
 
         logger.debug("bloom filter size is: {}".format(bfSize))
         logger.debug("size factor = {}".format(size_factor))
-        assert bfSize >= size_factor * min_bf_size
+        # In the case where the number of sstables is greater than the number of directories, it's possible this to be
+        # both with unique keys (where the bf size will remain close to the unadjusted limit) or with repetitions
+        # of keys (where the bf size will be a multiple of the expected). Permit both by only using the size factor on
+        # the maximum size. Note that the test is designed to end up with size_factor == 1 and most runs do so, thus
+        # this is not a loosening of the test in the common case, only ensures that we don't end up with flakes.
+        assert bfSize >= min_bf_size
         assert bfSize <= size_factor * max_bf_size
 
     @pytest.mark.parametrize("strategy", strategies)
@@ -313,9 +323,9 @@ def test_compaction_strategy_switching(self, strategy):
         self.skip_if_not_supported(strategy)
 
         if self.cluster.version() >= '5.0':
-            strategies = ['LeveledCompactionStrategy', 'SizeTieredCompactionStrategy']
+            strategies = ['LeveledCompactionStrategy', 'SizeTieredCompactionStrategy', 'UnifiedCompactionStrategy']
         else:
-            strategies = ['LeveledCompactionStrategy', 'SizeTieredCompactionStrategy', 'DateTieredCompactionStrategy']
+            strategies = ['LeveledCompactionStrategy', 'SizeTieredCompactionStrategy', 'DateTieredCompactionStrategy', 'UnifiedCompactionStrategy']
 
         if strategy in strategies:
             strategies.remove(strategy)
@@ -324,6 +334,7 @@ def test_compaction_strategy_switching(self, strategy):
             [node1] = cluster.nodelist()
 
             for strat in strategies:
+                logger.debug("Switching to {}".format(strat))
                 session = self.patient_cql_connection(node1)
                 create_ks(session, 'ks', 1)
 

disk_balance_test.py

@@ -24,6 +24,10 @@ class TestDiskBalance(Tester):
     @jira_ticket CASSANDRA-6696
     """
 
+    STCS_COMPACTION_OPTS = "SizeTieredCompactionStrategy"
+    LCS_COMPACTION_OPTS = "LeveledCompactionStrategy,sstable_size_in_mb=1"
+    UCS_COMPACTION_OPTS = "UnifiedCompactionStrategy"
+
     @pytest.fixture(scope='function', autouse=True)
     def fixture_set_cluster_settings(self, fixture_dtest_setup):
         cluster = fixture_dtest_setup.cluster
@@ -191,16 +195,23 @@ def test_disk_balance_after_boundary_change_stcs(self):
         """
         @jira_ticket CASSANDRA-13948
         """
-        self._disk_balance_after_boundary_change_test(lcs=False)
+        self._disk_balance_after_boundary_change_test(self.STCS_COMPACTION_OPTS)
 
     @since('3.10')
     def test_disk_balance_after_boundary_change_lcs(self):
         """
         @jira_ticket CASSANDRA-13948
         """
-        self._disk_balance_after_boundary_change_test(lcs=True)
+        self._disk_balance_after_boundary_change_test(self.LCS_COMPACTION_OPTS)
+
+    @since('4.0')
+    def test_disk_balance_after_boundary_change_ucs(self):
+        """
+        @jira_ticket CASSANDRA-13948
+        """
+        self._disk_balance_after_boundary_change_test(self.UCS_COMPACTION_OPTS)
 
-    def _disk_balance_after_boundary_change_test(self, lcs):
+    def _disk_balance_after_boundary_change_test(self, compaction_opts):
         """
             @jira_ticket CASSANDRA-13948
 
@@ -231,7 +242,6 @@ def _disk_balance_after_boundary_change_test(self, lcs):
         keys_per_flush = 10000
         keys_to_write = num_flushes * keys_per_flush
 
-        compaction_opts = "LeveledCompactionStrategy,sstable_size_in_mb=1" if lcs else "SizeTieredCompactionStrategy"
         logger.debug("Writing {} keys in {} flushes (compaction_opts={})".format(keys_to_write, num_flushes, compaction_opts))
         total_keys = num_flushes * keys_per_flush
         current_keys = 0
@@ -255,29 +265,36 @@ def _disk_balance_after_boundary_change_test(self, lcs):
         node2.start(wait_for_binary_proto=True, jvm_args=["-Dcassandra.migration_task_wait_in_seconds=10"], set_migration_task=False)
         node2.flush()
 
-        self._assert_balanced_after_boundary_change(node1, total_keys, lcs)
+        self._assert_balanced_after_boundary_change(node1, total_keys, compaction_opts)
 
         logger.debug("Decommissioning node1")
         node1.decommission()
         node1.stop()
 
-        self._assert_balanced_after_boundary_change(node2, total_keys, lcs)
+        self._assert_balanced_after_boundary_change(node2, total_keys, compaction_opts)
 
     @since('3.10')
     def test_disk_balance_after_joining_ring_stcs(self):
         """
         @jira_ticket CASSANDRA-13948
         """
-        self._disk_balance_after_joining_ring_test(lcs=False)
+        self._disk_balance_after_joining_ring_test(self.STCS_COMPACTION_OPTS)
 
     @since('3.10')
     def test_disk_balance_after_joining_ring_lcs(self):
         """
         @jira_ticket CASSANDRA-13948
         """
-        self._disk_balance_after_joining_ring_test(lcs=True)
+        self._disk_balance_after_joining_ring_test(self.LCS_COMPACTION_OPTS)
+
+    @since('4.0')
+    def test_disk_balance_after_joining_ring_ucs(self):
+        """
+        @jira_ticket CASSANDRA-13948
+        """
+        self._disk_balance_after_joining_ring_test(self.UCS_COMPACTION_OPTS)
 
-    def _disk_balance_after_joining_ring_test(self, lcs):
+    def _disk_balance_after_joining_ring_test(self, compaction_opts):
         """
             @jira_ticket CASSANDRA-13948
 
@@ -303,7 +320,6 @@ def _disk_balance_after_joining_ring_test(self, lcs):
         keys_per_flush = 10000
         keys_to_write = num_flushes * keys_per_flush
 
-        compaction_opts = "LeveledCompactionStrategy,sstable_size_in_mb=1" if lcs else "SizeTieredCompactionStrategy"
         logger.debug("Writing {} keys in {} flushes (compaction_opts={})".format(keys_to_write, num_flushes, compaction_opts))
         total_keys = num_flushes * keys_per_flush
         current_keys = 0
@@ -328,9 +344,9 @@ def _disk_balance_after_joining_ring_test(self, lcs):
         node1.nodetool("join")
         node1.nodetool("join")  # Need to run join twice - one to join ring, another to leave write survey mode
 
-        self._assert_balanced_after_boundary_change(node1, total_keys, lcs)
+        self._assert_balanced_after_boundary_change(node1, total_keys, compaction_opts)
 
-    def _assert_balanced_after_boundary_change(self, node, total_keys, lcs):
+    def _assert_balanced_after_boundary_change(self, node, total_keys, compaction_opts):
         logger.debug("Cleanup {}".format(node.name))
         node.cleanup()
 
@@ -352,7 +368,7 @@ def _assert_balanced_after_boundary_change(self, node, total_keys, lcs):
         logger.debug("Reading data back ({} keys)".format(total_keys))
         node.stress(['read', 'n={}'.format(total_keys), "no-warmup", "cl=ALL", "-pop", "seq=1...{}".format(total_keys), "-rate", "threads=1"])
 
-        if lcs:
+        if compaction_opts == self.LCS_COMPACTION_OPTS:
             output = grep_sstables_in_each_level(node, "standard1")
             logger.debug("SSTables in each level: {}".format(output))
 

repair_tests/repair_test.py

@@ -1,6 +1,7 @@
 import os
 import os.path
 import threading
+import tempfile
 import time
 import re
 import pytest
@@ -1168,8 +1169,7 @@ def test_multiple_concurrent_repairs(self):
         _, _, rc = node2.stress(['read', 'n=1M', 'no-warmup', '-rate', 'threads=30'], whitelist=True)
         assert rc == 0
 
-    @since('4.0')
-    def test_wide_row_repair(self):
+    def _test_wide_row_repair(self, compaction_strategy):
         """
         @jira_ticket CASSANDRA-13899
         Make sure compressed vs uncompressed blocks are handled correctly when stream decompressing
@@ -1179,13 +1179,26 @@ def test_wide_row_repair(self):
         cluster.populate(2).start()
         node1, node2 = cluster.nodelist()
         node2.stop(wait_other_notice=True)
-        profile_path = os.path.join(os.getcwd(), 'stress_profiles/repair_wide_rows.yaml')
-        logger.info(("yaml = " + profile_path))
-        node1.stress(['user', 'profile=' + profile_path, 'n=50', 'ops(insert=1)', 'no-warmup', '-rate', 'threads=8',
-                      '-insert', 'visits=FIXED(100K)', 'revisit=FIXED(100K)'])
+        template_path = os.path.join(os.getcwd(), 'stress_profiles/repair_wide_rows.yaml.tmpl')
+        with open(template_path) as profile_template:
+            profile = profile_template.read().replace("{{ compaction_strategy }}", compaction_strategy)
+            with tempfile.NamedTemporaryFile(mode='w+') as stress_profile:
+                stress_profile.write(profile)
+                stress_profile.flush()
+                print("yaml = " + stress_profile.name)
+                node1.stress(['user', 'profile=' + stress_profile.name, 'n=50', 'ops(insert=1)', 'no-warmup', '-rate', 'threads=8',
+                             '-insert', 'visits=FIXED(100K)', 'revisit=FIXED(100K)'])
         node2.start(wait_for_binary_proto=True)
         node2.repair()
 
+    @since('4.0')
+    def test_wide_row_repair_lcs(self):
+        self._test_wide_row_repair('LeveledCompactionStrategy')
+
+    @since('4.0')
+    def test_wide_row_repair_ucs(self):
+        self._test_wide_row_repair('UnifiedCompactionStrategy')
+
     @since('2.1', max_version='4')
     def test_dead_coordinator(self):
         """
@@ -1242,7 +1255,7 @@ def _repair_abort_test(self, options=[], nodes=1, rf=1, no_common_range=False):
         cluster = self.cluster
         logger.debug("Starting cluster..")
         cluster.populate(nodes).start(wait_for_binary_proto=True)
-        
+
         node1 = self.cluster.nodelist()[0]
         session = self.patient_cql_connection(node1)
         create_ks(session, 'ks', rf=rf)

replace_address_test.py

@@ -639,7 +639,14 @@ def test_replace_with_insufficient_replicas(self):
         self.replacement_node.watch_log_for("Unable to find sufficient sources for streaming range")
         assert_not_running(self.replacement_node)
 
-    def test_multi_dc_replace_with_rf1(self):
+    def test_multi_dc_replace_with_rf1_stcs(self):
+        self._test_multi_dc_replace_with_rf1('SizeTieredCompactionStrategy')
+
+    @since("4.0")
+    def test_multi_dc_replace_with_rf1_ucs(self):
+        self._test_multi_dc_replace_with_rf1('UnifiedCompactionStrategy')
+
+    def _test_multi_dc_replace_with_rf1(self, compaction_strategy):
         """
         Test that multi-dc replace works when rf=1 on each dc
         """
@@ -649,7 +656,7 @@ def test_multi_dc_replace_with_rf1(self):
         # Create the keyspace and table
         keyspace: keyspace1
         keyspace_definition: |
-          CREATE KEYSPACE keyspace1 WITH replication = {'class': 'NetworkTopologyStrategy', 'dc1': 1, 'dc2': 1};
+          CREATE KEYSPACE keyspace1 WITH replication = {{'class': 'NetworkTopologyStrategy', 'dc1': 1, 'dc2': 1}};
         table: users
         table_definition:
           CREATE TABLE users (
@@ -658,15 +665,16 @@ def test_multi_dc_replace_with_rf1(self):
             last_name text,
             email text,
             PRIMARY KEY(username)
-          ) WITH compaction = {'class':'SizeTieredCompactionStrategy'};
+          ) WITH compaction = {{'class':'{compaction_strategy}'}};
         insert:
           partitions: fixed(1)
           batchtype: UNLOGGED
         queries:
           read:
             cql: select * from users where username = ?
             fields: samerow
-        """
+        """.format(compaction_strategy=compaction_strategy)
+
         with tempfile.NamedTemporaryFile(mode='w+') as stress_config:
             stress_config.write(yaml_config)
             stress_config.flush()

schema_test.py

@@ -13,7 +13,7 @@
 
 class TestSchema(Tester):
 
-    def test_table_alteration(self):
+    def _test_table_alteration(self, compaction_opts):
         """
         Tests that table alters return as expected with many sstables at different schema points
         """
@@ -24,7 +24,7 @@ def test_table_alteration(self):
         create_ks(session, 'ks', 1)
         session.execute("use ks;")
         session.execute("create table tbl_o_churn (id int primary key, c0 text, c1 text) "
-                        "WITH compaction = {'class': 'SizeTieredCompactionStrategy', 'min_threshold': 1024, 'max_threshold': 1024 };")
+                        "WITH compaction = " + compaction_opts + ";")
 
         stmt1 = session.prepare("insert into tbl_o_churn (id, c0, c1) values (?, ?, ?)")
         rows_to_insert = 50
@@ -54,6 +54,13 @@ def test_table_alteration(self):
                 assert row.c2 == 'ddd'
                 assert not hasattr(row, 'c0')
 
+    def test_table_alteration_stcs(self):
+        self._test_table_alteration("{'class': 'SizeTieredCompactionStrategy', 'min_threshold': 1024, 'max_threshold': 1024 }")
+
+    @since("4.0")
+    def test_table_alteration_ucs(self):
+        self._test_table_alteration("{'class': 'UnifiedCompactionStrategy'}")
+
     @since("2.0", max_version="3.X")  # Compact Storage
     def test_drop_column_compact(self):
         session = self.prepare()

stress_profiles/repair_wide_rows.yaml renamed to stress_profiles/repair_wide_rows.yaml.tmpl

@@ -9,16 +9,16 @@ table_definition: |
         col1 text,
         val blob,
         PRIMARY KEY(key, col1)
-  ) 
-    WITH compaction = { 'class':'LeveledCompactionStrategy' }
+  )
+    WITH compaction = { 'class':'{{ compaction_strategy }}' }
     AND compression = {'chunk_length_in_kb': '1', 'class': 'org.apache.cassandra.io.compress.LZ4Compressor'};
 
 #
 # Optional meta information on the generated columns in the above table
 # The min and max only apply to text and blob types
 # The distribution field represents the total unique population
 # distribution of that column across rows.  Supported types are
-# 
+#
 #      EXP(min..max)                        An exponential distribution over the range [min..max]
 #      EXTREME(min..max,shape)              An extreme value (Weibull) distribution over the range [min..max]
 #      GAUSSIAN(min..max,stdvrng)           A gaussian/normal distribution, where mean=(min+max)/2, and stdev is (mean-min)/stdvrng

update-history/STAR-1892/11-2a2726a4 STAR-836 Fix TestCompaction_with_UnifiedCompactionStrategy.bloomfilter_size_test (#41)

@@ -0,0 +1,36 @@
+--- a/compaction_test.py
++++ b/compaction_test.py
+@@ -134,18 +134,14 @@
+                           "threads=300", "-schema", "replication(factor=1)",
+                           "compaction({},enabled=false)".format(strategy_string)])
+             node1.flush()
+-        logger.debug(node1.nodetool('cfstats keyspace1.standard1').stdout)
++        logger.debug(node1.nodetool('tablestats keyspace1.standard1').stdout)
+ 
+         node1.nodetool('enableautocompaction')
+         node1.wait_for_compactions()
+ 
+         table_name = 'standard1'
+-<<<<<<<
+-        output = node1.nodetool('cfstats keyspace1.standard1').stdout
++        output = node1.nodetool('tablestats keyspace1.standard1').stdout
+         logger.debug(output)
+-=======
+-        output = node1.nodetool('tablestats').stdout
+->>>>>>>
+         output = output[output.find(table_name):]
+         output = output[output.find("Bloom filter space used"):]
+         bfSize = int(output[output.find(":") + 1:output.find("\n")].strip())
+diff --git a/compaction_test.py b/compaction_test.py
+index fa924974..d79f1326 100644
+--- a/compaction_test.py
++++ b/compaction_test.py
+@@ -134,7 +134,7 @@ class TestCompaction(Tester):
+                           "threads=300", "-schema", "replication(factor=1)",
+                           "compaction({},enabled=false)".format(strategy_string)])
+             node1.flush()
+-        logger.debug(node1.nodetool('cfstats keyspace1.standard1').stdout)
++        logger.debug(node1.nodetool('tablestats keyspace1.standard1').stdout)
+ 
+         node1.nodetool('enableautocompaction')
+         node1.wait_for_compactions()