diff --git a/CHANGES.txt b/CHANGES.txt index 0f61587cd1..9d0aafacdb 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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) diff --git a/conf/jvm-server.options b/conf/jvm-server.options index b977e710bd..4a397fbfb9 100644 --- a/conf/jvm-server.options +++ b/conf/jvm-server.options @@ -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 # ######################## diff --git a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java index cf514b7dfc..9184b78395 100644 --- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java +++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java @@ -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))), diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java index 0144fcae07..4a5c2f1d9d 100644 --- a/src/java/org/apache/cassandra/service/CassandraDaemon.java +++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java @@ -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 keyspaces) + { + if (StringUtils.isBlank(OVERRIDE_COMPACTION_ENTITIES.getString()) || StringUtils.isBlank(OVERRIDE_COMPACTION_PARAMS.getString())) + { + return; + } + + Map> 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 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> 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> 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 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); diff --git a/test/distributed/org/apache/cassandra/distributed/test/CompactionStrategyOverrideTest.java b/test/distributed/org/apache/cassandra/distributed/test/CompactionStrategyOverrideTest.java new file mode 100644 index 0000000000..ccf4bcd5e3 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/CompactionStrategyOverrideTest.java @@ -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 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 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 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 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()); + } + } +} diff --git a/test/unit/org/apache/cassandra/service/CassandraDaemonTest.java b/test/unit/org/apache/cassandra/service/CassandraDaemonTest.java new file mode 100644 index 0000000000..965e00b047 --- /dev/null +++ b/test/unit/org/apache/cassandra/service/CassandraDaemonTest.java @@ -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> result = CassandraDaemon.parseEntititesToOverrideCompaction(); + assertTrue(result.isEmpty()); + } + } + + @Test + public void testParseEntitiesSingleKeyspace() + { + OVERRIDE_COMPACTION_ENTITIES.setString("ks1"); + Map> result = CassandraDaemon.parseEntititesToOverrideCompaction(); + assertEquals(1, result.size()); + assertTrue(result.get("ks1").isEmpty()); + } + + @Test + public void testParseEntitiesMultipleKeyspaces() + { + OVERRIDE_COMPACTION_ENTITIES.setString("ks1,ks2,ks3"); + Map> 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> 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> 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> 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> 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> result = CassandraDaemon.parseEntititesToOverrideCompaction(); + assertEquals(1, result.size()); + assertTrue(result.get("ks1").isEmpty()); + } + + @Test + public void testParseEntitiesWhitespaceTrimmed() + { + OVERRIDE_COMPACTION_ENTITIES.setString(" ks1 , ks2 . tbl1 "); + Map> result = CassandraDaemon.parseEntititesToOverrideCompaction(); + assertEquals(2, result.size()); + assertTrue(result.get("ks1").isEmpty()); + assertEquals(List.of("tbl1"), result.get("ks2")); + } +}