This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-dtest.git
The following commit(s) were added to refs/heads/trunk by this push:
new c1ff79f7 More dtests against UnifiedCompactionStrategy
c1ff79f7 is described below
commit c1ff79f78ec9101b88d0f8775d00f1b64296636f
Author: Aleksandr Sorokoumov <[email protected]>
AuthorDate: Mon Aug 24 11:51:53 2026 +0200
More dtests against UnifiedCompactionStrategy
CASSANDRA-20148 added UCS to the compaction_test strategy list. This adds
the (forgotten) UCS variants:
- disk_balance_test: test_disk_balance_after_boundary_change_ucs and
test_disk_balance_after_joining_ring_ucs, with the lcs boolean replaced by
compaction-option constants
- repair_test: test_wide_row_repair_lcs and test_wide_row_repair_ucs,
rendering stress_profiles/repair_wide_rows.yaml.tmpl into a temporary file
- replace_address_test: test_multi_dc_replace_with_rf1_stcs and
test_multi_dc_replace_with_rf1_ucs
- schema_test: test_table_alteration_stcs and test_table_alteration_ucs
- compaction_test: cap max_sstables_to_compact for the bloom filter test,
and add UCS to the 5.0 strategy-switching list
UCS reached Cassandra in 5.0, so every new variant gates on @since('5.0');
the fork gates on 4.0. The bloom filter lower bound drops the size factor:
with more sstables than data directories the filter size can stay near the
unadjusted limit or reach a multiple of it, and both are correct.
This renames four existing tests: test_wide_row_repair,
test_multi_dc_replace_with_rf1 and test_table_alteration (CI/Butler loses their
result history).
Upstreamed from datastax/cassandra-dtest
f7320168137f1620d3b659c0171a71593c79fbdc,
249d932c228b02af0b6e989faad21aa44cb8ab7b and
453027efeef7adfc9f5eadc8497b2e8f917f0501
patch by Aleksandr Sorokoumov, Branimir Lambov; reviewed by Brandon
Williams, Mick Semb Wever for CASSANDRA-20629
Assisted-by: Claude Code:claude-opus-5
---
compaction_test.py | 17 +++++++--
disk_balance_test.py | 42 +++++++++++++++-------
repair_tests/repair_test.py | 25 +++++++++----
replace_address_test.py | 16 ++++++---
schema_test.py | 11 ++++--
...r_wide_rows.yaml => repair_wide_rows.yaml.tmpl} | 6 ++--
6 files changed, 86 insertions(+), 31 deletions(-)
diff --git a/compaction_test.py b/compaction_test.py
index e71fc5e7..f81712cd 100644
--- a/compaction_test.py
+++ b/compaction_test.py
@@ -118,6 +118,8 @@ class TestCompaction(Tester):
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 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
@@ -125,18 +127,21 @@ class TestCompaction(Tester):
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 @@ class TestCompaction(Tester):
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,7 +323,7 @@ class TestCompaction(Tester):
self.skip_if_not_supported(strategy)
if self.cluster.version() >= '5.0':
- strategies = ['LeveledCompactionStrategy',
'SizeTieredCompactionStrategy']
+ strategies = ['LeveledCompactionStrategy',
'SizeTieredCompactionStrategy', 'UnifiedCompactionStrategy']
else:
strategies = ['LeveledCompactionStrategy',
'SizeTieredCompactionStrategy', 'DateTieredCompactionStrategy']
@@ -324,6 +334,7 @@ class TestCompaction(Tester):
[node1] = cluster.nodelist()
for strat in strategies:
+ logger.debug("Switching to {}".format(strat))
session = self.patient_cql_connection(node1)
create_ks(session, 'ks', 1)
diff --git a/disk_balance_test.py b/disk_balance_test.py
index 15311cc9..5efc3671 100644
--- a/disk_balance_test.py
+++ b/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
@@ -200,16 +204,23 @@ class TestDiskBalance(Tester):
"""
@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('5.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
@@ -240,7 +251,6 @@ class TestDiskBalance(Tester):
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
@@ -264,29 +274,36 @@ class TestDiskBalance(Tester):
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('5.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
@@ -312,7 +329,6 @@ class TestDiskBalance(Tester):
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
@@ -337,9 +353,9 @@ class TestDiskBalance(Tester):
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()
@@ -361,7 +377,7 @@ class TestDiskBalance(Tester):
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))
diff --git a/repair_tests/repair_test.py b/repair_tests/repair_test.py
index 87e265a3..fad82475 100644
--- a/repair_tests/repair_test.py
+++ b/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 @@ class TestRepair(BaseRepairTest):
_, _, 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 @@ class TestRepair(BaseRepairTest):
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()
+ logger.info(("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('5.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):
"""
diff --git a/replace_address_test.py b/replace_address_test.py
index d63112d2..4c8666f2 100644
--- a/replace_address_test.py
+++ b/replace_address_test.py
@@ -644,7 +644,14 @@ class TestReplaceAddress(BaseReplaceAddressTest):
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("5.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
"""
@@ -654,7 +661,7 @@ class TestReplaceAddress(BaseReplaceAddressTest):
# 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 (
@@ -663,7 +670,7 @@ class TestReplaceAddress(BaseReplaceAddressTest):
last_name text,
email text,
PRIMARY KEY(username)
- ) WITH compaction = {'class':'SizeTieredCompactionStrategy'};
+ ) WITH compaction = {{'class':'{compaction_strategy}'}};
insert:
partitions: fixed(1)
batchtype: UNLOGGED
@@ -671,7 +678,8 @@ class TestReplaceAddress(BaseReplaceAddressTest):
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()
diff --git a/schema_test.py b/schema_test.py
index 6c9f8a1a..382fe5b8 100644
--- a/schema_test.py
+++ b/schema_test.py
@@ -13,7 +13,7 @@ logger = logging.getLogger(__name__)
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 @@ class TestSchema(Tester):
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 @@ class TestSchema(Tester):
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("5.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()
diff --git a/stress_profiles/repair_wide_rows.yaml
b/stress_profiles/repair_wide_rows.yaml.tmpl
similarity index 97%
rename from stress_profiles/repair_wide_rows.yaml
rename to stress_profiles/repair_wide_rows.yaml.tmpl
index 87f46f02..c35ebbe9 100644
--- a/stress_profiles/repair_wide_rows.yaml
+++ b/stress_profiles/repair_wide_rows.yaml.tmpl
@@ -9,8 +9,8 @@ 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'};
#
@@ -18,7 +18,7 @@ table_definition: |
# 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
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
