natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add compaction strategy override via system properties 

Introduce the ability to override compaction strategy for specific keyspaces
and tables at startup via two new system properties:
- cassandra.override_compaction.entities: comma-separated list of keyspaces
  and keyspace.table pairs (e.g. "ks1,ks2.tbl1,ks3.tbl2")
- cassandra.override_compaction.params: JSON compaction parameters to apply

Patch by Paulo Motta; Reviewed by Jaydeepkumar Chovatia for CASSANDRA-21169
This commit is contained in:
Paulo Motta 2026-02-12 14:12:45 -05:00
parent b7eb323c5a
commit a9ee34b62d
6 changed files with 310 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
5.1
* Allow overriding compaction strategy parameters during startup (CASSANDRA-21169)
* Introduce created_at column to system_distributed.compression_dictionaries (CASSANDRA-21178)
* Be able to detect and remove orphaned compression dictionaries (CASSANDRA-21157)
* Fix BigTableVerifier to only read a data file during extended verification (CASSANDRA-21150)

6
conf/jvm-server.options
View File

@ -113,6 +113,12 @@
# Imposes an upper bound on hint lifetime below the normal min gc_grace_seconds
#-Dcassandra.maxHintTTL=max_hint_ttl_in_seconds
# Override compaction strategy for specific keyspaces or tables at startup.
# Entities is a comma-separated list of keyspaces and keyspace.table pairs.
# Params is a JSON string with the compaction parameters to apply.
#-Dcassandra.override_compaction.entities=ks1,ks2.tbl1
#-Dcassandra.override_compaction.params={"class":"SizeTieredCompactionStrategy"}
########################
# GENERAL JVM SETTINGS #
########################

2
src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
View File

@ -431,6 +431,8 @@ public enum CassandraRelevantProperties
OTCP_LARGE_MESSAGE_THRESHOLD("cassandra.otcp_large_message_threshold", convertToString(1024 * 64)),
/** Enabled/disable TCP_NODELAY for intradc connections. Defaults is enabled. */
OTC_INTRADC_TCP_NODELAY("cassandra.otc_intradc_tcp_nodelay", "true"),
OVERRIDE_COMPACTION_ENTITIES("cassandra.override_compaction.entities"),
OVERRIDE_COMPACTION_PARAMS("cassandra.override_compaction.params"),
OVERRIDE_DECOMMISSION("cassandra.override_decommission"),
PARENT_REPAIR_STATUS_CACHE_SIZE("cassandra.parent_repair_status_cache_size", "100000"),
PARENT_REPAIR_STATUS_EXPIRY_SECONDS("cassandra.parent_repair_status_expiry_seconds", convertToString(TimeUnit.SECONDS.convert(1, TimeUnit.DAYS))),

78
src/java/org/apache/cassandra/service/CassandraDaemon.java
View File

@ -26,8 +26,11 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ExecutionException;
@ -44,6 +47,7 @@ import com.codahale.metrics.SharedMetricRegistries;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -116,6 +120,8 @@ import static org.apache.cassandra.config.CassandraRelevantProperties.JAVA_CLASS
import static org.apache.cassandra.config.CassandraRelevantProperties.JAVA_RMI_SERVER_RANDOM_ID;
import static org.apache.cassandra.config.CassandraRelevantProperties.JAVA_VERSION;
import static org.apache.cassandra.config.CassandraRelevantProperties.JAVA_VM_NAME;
import static org.apache.cassandra.config.CassandraRelevantProperties.OVERRIDE_COMPACTION_ENTITIES;
import static org.apache.cassandra.config.CassandraRelevantProperties.OVERRIDE_COMPACTION_PARAMS;
import static org.apache.cassandra.config.CassandraRelevantProperties.SIZE_RECORDER_INTERVAL;
import static org.apache.cassandra.config.CassandraRelevantProperties.START_NATIVE_TRANSPORT;
import static org.apache.cassandra.metrics.CassandraMetricsRegistry.createMetricsKeyspaceTables;
@ -196,8 +202,8 @@ public class CassandraDaemon
}
@VisibleForTesting
public static Runnable SPECULATION_THRESHOLD_UPDATER =
() ->
public static Runnable SPECULATION_THRESHOLD_UPDATER =
() ->
{
try
{
@ -209,7 +215,7 @@ public class CassandraDaemon
JVMStabilityInspector.inspectThrowable(t);
}
};
static final CassandraDaemon instance = new CassandraDaemon();
private volatile NativeTransportService nativeTransportService;
@ -415,6 +421,8 @@ public class CassandraDaemon
ScheduledExecutors.optionalTasks.schedule(viewRebuild, StorageService.RING_DELAY_MILLIS, TimeUnit.MILLISECONDS);
StorageService.instance.doAuthSetup();
// Apply overrides before re-enabling auto-compaction
setCompactionStrategyOverrides(Schema.instance.getKeyspaces());
// re-enable auto-compaction after replay, so correct disk boundaries are used
enableAutoCompaction(Schema.instance.getKeyspaces());
@ -427,7 +435,7 @@ public class CassandraDaemon
ScheduledExecutors.optionalTasks.scheduleWithFixedDelay(ColumnFamilyStore.getBackgroundCompactionTaskSubmitter(), 5, 1, TimeUnit.MINUTES);
// schedule periodic recomputation of speculative retry thresholds
ScheduledExecutors.optionalTasks.scheduleWithFixedDelay(SPECULATION_THRESHOLD_UPDATER,
ScheduledExecutors.optionalTasks.scheduleWithFixedDelay(SPECULATION_THRESHOLD_UPDATER,
DatabaseDescriptor.getReadRpcTimeout(NANOSECONDS),
DatabaseDescriptor.getReadRpcTimeout(NANOSECONDS),
NANOSECONDS);
@ -564,6 +572,68 @@ public class CassandraDaemon
}
}
public static void setCompactionStrategyOverrides(Collection<String> keyspaces)
{
if (StringUtils.isBlank(OVERRIDE_COMPACTION_ENTITIES.getString()) || StringUtils.isBlank(OVERRIDE_COMPACTION_PARAMS.getString()))
{
return;
}
Map<String, List<String>> entitiesToChangeCompaction = parseEntititesToOverrideCompaction();
logger.info("Compaction strategy override is enabled via 'cassandra.override_compaction.params' for the following 'cassandra.override_compaction.entities': {}",
entitiesToChangeCompaction);
String overrideParams = OVERRIDE_COMPACTION_PARAMS.getString();
for (String ksNme : keyspaces)
{
Keyspace keyspace = Keyspace.open(ksNme);
for (ColumnFamilyStore cfs : keyspace.getColumnFamilyStores())
{
for (final ColumnFamilyStore store : cfs.concatWithIndexes())
{
List<String> tablesToOverrideCompaction = entitiesToChangeCompaction.get(ksNme);
if (tablesToOverrideCompaction != null && (tablesToOverrideCompaction.isEmpty() || tablesToOverrideCompaction.contains(store.name)))
{
logger.info("Overriding compaction parameters for {}.{} with {}", store.getKeyspaceName(), store.name, overrideParams);
cfs.setCompactionParametersJson(overrideParams);
}
}
}
}
}
@VisibleForTesting
static Map<String, List<String>> parseEntititesToOverrideCompaction()
{
String entitiesCsv = OVERRIDE_COMPACTION_ENTITIES.getString();
if (StringUtils.isBlank(entitiesCsv))
return Collections.emptyMap();
// entititesCSV can be like "ks1,ks2,k3.tbl3,ks4.tbl1"
Map<String, List<String>> entitiesToChangeCompaction = new HashMap<>();
for (String entity : entitiesCsv.split(","))
{
String[] ksTable = entity.split("\\.");
String keyspace = ksTable[0].trim();
if (ksTable.length == 1)
{
entitiesToChangeCompaction.put(keyspace, new java.util.ArrayList<>());
}
else if (ksTable.length == 2)
{
// Empty list for a keyspace means all tables in that keyspace should be changed, so if we already have an entry for the keyspace with an empty list,
// we can skip adding specific tables for that keyspace as they are redundant.
List<String> existing = entitiesToChangeCompaction.get(keyspace);
if (existing == null || !existing.isEmpty())
{
String table = ksTable[1].trim();
entitiesToChangeCompaction.computeIfAbsent(keyspace, k -> new java.util.ArrayList<>()).add(table);
}
}
}
return entitiesToChangeCompaction;
}
public void setupVirtualKeyspaces()
{
VirtualKeyspaceRegistry.instance.register(VirtualSchemaKeyspace.instance);

95
test/distributed/org/apache/cassandra/distributed/test/CompactionStrategyOverrideTest.java Normal file
View File

@ -0,0 +1,95 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.distributed.test;
import java.util.Map;
import org.junit.Assert;
import org.junit.Test;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.Constants;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.Feature;
import static org.apache.cassandra.config.CassandraRelevantProperties.OVERRIDE_COMPACTION_ENTITIES;
import static org.apache.cassandra.config.CassandraRelevantProperties.OVERRIDE_COMPACTION_PARAMS;
public class CompactionStrategyOverrideTest extends TestBaseImpl
{
private static final String OVERRIDE_PARAMS = "{\"class\":\"org.apache.cassandra.db.compaction.LeveledCompactionStrategy\",\"sstable_size_in_mb\":\"512\"}";
@Test
public void testCompactionStrategyOverrideOnRestart() throws Exception
{
try (Cluster cluster = init(builder().withNodes(1)
.withConfig(config -> config.with(Feature.NETWORK, Feature.GOSSIP)
.set(Constants.KEY_DTEST_FULL_STARTUP, true))
.start()))
{
cluster.coordinator(1).execute("CREATE KEYSPACE ks1 WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1}", ConsistencyLevel.ALL);
cluster.coordinator(1).execute("CREATE KEYSPACE ks2 WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1}", ConsistencyLevel.ALL);
cluster.coordinator(1).execute("CREATE TABLE ks1.tbl1 (id int PRIMARY KEY, value text)", ConsistencyLevel.ALL);
cluster.coordinator(1).execute("CREATE TABLE ks1.tbl2 (id int PRIMARY KEY, value text)", ConsistencyLevel.ALL);
cluster.coordinator(1).execute("CREATE TABLE ks2.tbl1 (id int PRIMARY KEY, value text)", ConsistencyLevel.ALL);
cluster.coordinator(1).execute("CREATE TABLE ks2.tbl2 (id int PRIMARY KEY, value text)", ConsistencyLevel.ALL);
// Verify all tables start with the default SizeTieredCompactionStrategy
cluster.get(1).runOnInstance(() -> {
for (String ks : new String[]{ "ks1", "ks2" })
for (String tbl : new String[]{ "tbl1", "tbl2" })
Assert.assertEquals("org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy",
Keyspace.open(ks).getColumnFamilyStore(tbl).getCompactionParameters().get("class"));
});
// Shut down, set override properties, and restart so CassandraDaemon.setup() applies them
cluster.get(1).shutdown().get();
OVERRIDE_COMPACTION_ENTITIES.setString("ks1.tbl1,ks1,ks2.tbl2");
OVERRIDE_COMPACTION_PARAMS.setString(OVERRIDE_PARAMS);
cluster.get(1).startup();
cluster.get(1).runOnInstance(() -> {
// ks1 was listed as a whole keyspace (ks1.tbl1,ks1 -> ks1 overrides), so both tables should be overridden
Map<String, String> ks1tbl1 = Keyspace.open("ks1").getColumnFamilyStore("tbl1").getCompactionParameters();
Assert.assertEquals("org.apache.cassandra.db.compaction.LeveledCompactionStrategy", ks1tbl1.get("class"));
Assert.assertEquals("512", ks1tbl1.get("sstable_size_in_mb"));
Map<String, String> ks1tbl2 = Keyspace.open("ks1").getColumnFamilyStore("tbl2").getCompactionParameters();
Assert.assertEquals("org.apache.cassandra.db.compaction.LeveledCompactionStrategy", ks1tbl2.get("class"));
Assert.assertEquals("512", ks1tbl2.get("sstable_size_in_mb"));
// ks2.tbl2 was explicitly listed, so it should be overridden
Map<String, String> ks2tbl2 = Keyspace.open("ks2").getColumnFamilyStore("tbl2").getCompactionParameters();
Assert.assertEquals("org.apache.cassandra.db.compaction.LeveledCompactionStrategy", ks2tbl2.get("class"));
Assert.assertEquals("512", ks2tbl2.get("sstable_size_in_mb"));
// ks2.tbl1 was not listed, so it should retain the default SizeTieredCompactionStrategy
Map<String, String> ks2tbl1 = Keyspace.open("ks2").getColumnFamilyStore("tbl1").getCompactionParameters();
Assert.assertEquals("org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy", ks2tbl1.get("class"));
Assert.assertNull(ks2tbl1.get("sstable_size_in_mb"));
});
System.clearProperty(OVERRIDE_COMPACTION_ENTITIES.getKey());
System.clearProperty(OVERRIDE_COMPACTION_PARAMS.getKey());
}
}
}

132
test/unit/org/apache/cassandra/service/CassandraDaemonTest.java Normal file
View File

@ -0,0 +1,132 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.service;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import org.junit.After;
import org.junit.Test;
import static org.apache.cassandra.config.CassandraRelevantProperties.OVERRIDE_COMPACTION_ENTITIES;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
public class CassandraDaemonTest
{
@After
public void tearDown()
{
System.clearProperty(OVERRIDE_COMPACTION_ENTITIES.getKey());
}
@Test
public void testParseEntitiesBlankReturnsEmptyMap()
{
for (String blank : Arrays.asList(null, "", " "))
{
if (blank == null)
System.clearProperty(OVERRIDE_COMPACTION_ENTITIES.getKey());
else
OVERRIDE_COMPACTION_ENTITIES.setString(blank);
Map<String, List<String>> result = CassandraDaemon.parseEntititesToOverrideCompaction();
assertTrue(result.isEmpty());
}
}
@Test
public void testParseEntitiesSingleKeyspace()
{
OVERRIDE_COMPACTION_ENTITIES.setString("ks1");
Map<String, List<String>> result = CassandraDaemon.parseEntititesToOverrideCompaction();
assertEquals(1, result.size());
assertTrue(result.get("ks1").isEmpty());
}
@Test
public void testParseEntitiesMultipleKeyspaces()
{
OVERRIDE_COMPACTION_ENTITIES.setString("ks1,ks2,ks3");
Map<String, List<String>> result = CassandraDaemon.parseEntititesToOverrideCompaction();
assertEquals(3, result.size());
assertTrue(result.get("ks1").isEmpty());
assertTrue(result.get("ks2").isEmpty());
assertTrue(result.get("ks3").isEmpty());
}
@Test
public void testParseEntitiesSpecificTables()
{
OVERRIDE_COMPACTION_ENTITIES.setString("ks1.tbl1,ks1.tbl2");
Map<String, List<String>> result = CassandraDaemon.parseEntititesToOverrideCompaction();
assertEquals(1, result.size());
assertEquals(List.of("tbl1", "tbl2"), result.get("ks1"));
}
@Test
public void testParseEntitiesMixedKeyspacesAndTables()
{
OVERRIDE_COMPACTION_ENTITIES.setString("ks1,ks2.tbl1,ks2.tbl2");
Map<String, List<String>> result = CassandraDaemon.parseEntititesToOverrideCompaction();
assertEquals(2, result.size());
assertTrue(result.get("ks1").isEmpty());
assertEquals(List.of("tbl1", "tbl2"), result.get("ks2"));
}
@Test
public void testParseEntitiesKeyspaceAfterTableOverrides()
{
// keyspace-only entry after table-specific entries overrides to all tables
OVERRIDE_COMPACTION_ENTITIES.setString("ks1.tbl1,ks1.tbl2,ks1");
Map<String, List<String>> result = CassandraDaemon.parseEntititesToOverrideCompaction();
assertEquals(1, result.size());
assertTrue(result.get("ks1").isEmpty());
}
@Test
public void testParseEntitiesTableAfterKeyspaceIsIgnored()
{
// keyspace-only entry selects all tables, subsequent table entries are ignored
OVERRIDE_COMPACTION_ENTITIES.setString("ks1,ks1.tbl1");
Map<String, List<String>> result = CassandraDaemon.parseEntititesToOverrideCompaction();
assertEquals(1, result.size());
assertTrue(result.get("ks1").isEmpty());
}
@Test
public void testParseEntitiesTableAfterKeyspaceOverrideIsIgnored()
{
OVERRIDE_COMPACTION_ENTITIES.setString("ks1.tbl1,ks1,ks1.tbl2");
Map<String, List<String>> result = CassandraDaemon.parseEntititesToOverrideCompaction();
assertEquals(1, result.size());
assertTrue(result.get("ks1").isEmpty());
}
@Test
public void testParseEntitiesWhitespaceTrimmed()
{
OVERRIDE_COMPACTION_ENTITIES.setString(" ks1 , ks2 . tbl1 ");
Map<String, List<String>> result = CassandraDaemon.parseEntititesToOverrideCompaction();
assertEquals(2, result.size());
assertTrue(result.get("ks1").isEmpty());
assertEquals(List.of("tbl1"), result.get("ks2"));
}
}