Add configs to cassandra.yaml and refactoring

Author: yifan-c

conf/cassandra.yaml

@@ -617,6 +617,54 @@ counter_cache_save_period: 7200s
 # Disabled by default, meaning all keys are going to be saved
 # counter_cache_keys_to_save: 100
 
+# Dictionary compression settings for ZSTD dictionary-based compression
+# These settings control the automatic training and caching of compression dictionaries
+# for tables that use ZSTD dictionary compression.
+
+# How often to refresh compression dictionaries across the cluster.
+# During refresh, nodes will check for newer dictionary versions and update their caches.
+# Min unit: s
+compression_dictionary_refresh_interval: 3600s
+
+# Initial delay before starting the first dictionary refresh cycle after node startup.
+# This prevents all nodes from refreshing simultaneously when the cluster starts.
+# Min unit: s
+compression_dictionary_refresh_initial_delay: 10s
+
+# Maximum number of compression dictionaries to cache per table.
+# Each table using dictionary compression can have multiple dictionaries cached
+# (current version plus recently used versions for reading older SSTables).
+compression_dictionary_cache_size: 10
+
+# How long to keep compression dictionaries in the cache before they expire.
+# Expired dictionaries will be removed from memory but can be reloaded if needed.
+# Min unit: s
+compression_dictionary_cache_expire: 3600s
+
+# Dictionary training configuration (advanced settings)
+# These settings control how compression dictionaries are trained from sample data.
+
+# Maximum size of a trained compression dictionary in bytes.
+# Larger dictionaries may provide better compression but use more memory.
+# Min unit: B
+compression_dictionary_training_max_dictionary_size: 65536
+
+# Maximum total size of sample data to collect for dictionary training.
+# More sample data generally produces better dictionaries but takes longer to train.
+# The recommended sample size is 100x the dictionary size.
+# Min unit: B
+compression_dictionary_training_max_total_sample_size: 10485760
+
+# Enable automatic dictionary training based on sampling of write operations.
+# When enabled, the system will automatically collect samples and train new dictionaries.
+# Manual training via nodetool is always available regardless of this setting.
+compression_dictionary_training_auto_train_enabled: false
+
+# Sampling rate for automatic dictionary training (1-10000).
+# Value of 100 means 1% of writes are sampled. Lower values reduce overhead but may
+# result in less representative sample data for dictionary training.
+compression_dictionary_training_sampling_rate: 100
+
 # saved caches
 # If not set, the default directory is $CASSANDRA_HOME/data/saved_caches.
 # saved_caches_directory: /var/lib/cassandra/saved_caches

src/java/org/apache/cassandra/config/Config.java

@@ -519,12 +519,10 @@ public static class SSTableConfig
     public volatile int compression_dictionary_cache_size = 10; // max dictionaries per table
     public volatile DurationSpec.IntSecondsBound compression_dictionary_cache_expire = new DurationSpec.IntSecondsBound("3600s"); // 1 hour default
 
-    // TODO: add to cassandra.yaml
     // Dictionary training settings
     public volatile int compression_dictionary_training_max_dictionary_size = 65536; // 64KB
     public volatile int compression_dictionary_training_max_total_sample_size = 10485760; // 10MB total
     public volatile boolean compression_dictionary_training_auto_train_enabled = false;
-    public volatile DurationSpec.IntSecondsBound compression_dictionary_training_manual_sampling_duration = new DurationSpec.IntSecondsBound("600s"); // 10 minutes default
     public volatile int compression_dictionary_training_sampling_rate = 100; // samples 1%; using int since random.nextInt is generally faster than random.nextDouble
 
     public DataStorageSpec.LongMebibytesBound paxos_cache_size = null;

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -4396,10 +4396,6 @@ public static boolean getCompressionDictionaryTrainingAutoTrainEnabled()
         return conf.compression_dictionary_training_auto_train_enabled;
     }
 
-    public static int getCompressionDictionaryTrainingManualSamplingDurationSeconds()
-    {
-        return conf.compression_dictionary_training_manual_sampling_duration.toSeconds();
-    }
 
     public static int getCompressionDictionaryTrainingSamplingRate()
     {

src/java/org/apache/cassandra/db/compression/CompressionDictionaryManager.java

@@ -212,8 +212,11 @@ public synchronized void train(Map<String, String> options)
             throw new IllegalStateException("Dictionary trainer is not available for table " + keyspaceName + '.' + tableName);
         }
 
+        // Parse and validate training options
+        ManualTrainingOptions trainingOptions = ManualTrainingOptions.fromStringMap(options);
+
         trainer.start(true);
-        scheduler.scheduleManualTraining(options, trainer);
+        scheduler.scheduleManualTraining(trainingOptions, trainer);
     }
 
     @Override

src/java/org/apache/cassandra/db/compression/CompressionDictionaryScheduler.java

@@ -18,7 +18,6 @@
 
 package org.apache.cassandra.db.compression;
 
-import java.util.Map;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 
@@ -79,28 +78,14 @@ public void scheduleRefreshTask()
     }
 
     @Override
-    public void scheduleManualTraining(Map<String, String> options, ICompressionDictionaryTrainer trainer)
+    public void scheduleManualTraining(ManualTrainingOptions options, ICompressionDictionaryTrainer trainer)
     {
         if (scheduledManualTrainingTask != null)
         {
             throw new IllegalStateException("Training already in progress for table " + keyspaceName + '.' + tableName);
         }
 
-        // Parse max sampling duration from options (default from configuration)
-        int maxSamplingDurationSeconds = DatabaseDescriptor.getCompressionDictionaryTrainingManualSamplingDurationSeconds();
-        if (options.containsKey("maxSamplingDurationSeconds"))
-        {
-            String durationStr = options.get("maxSamplingDurationSeconds");
-            try
-            {
-                maxSamplingDurationSeconds = Integer.parseInt(durationStr);
-            }
-            catch (NumberFormatException e)
-            {
-                logger.warn("Invalid maxSamplingDurationSeconds value: {}, using default: {}",
-                            durationStr, maxSamplingDurationSeconds);
-            }
-        }
+        int maxSamplingDurationSeconds = options.getMaxSamplingDurationSeconds();
 
         logger.info("Starting manual dictionary training for {}.{} with max sampling duration: {} seconds",
                     keyspaceName, tableName, maxSamplingDurationSeconds);

src/java/org/apache/cassandra/db/compression/ICompressionDictionaryScheduler.java

@@ -18,8 +18,6 @@
 
 package org.apache.cassandra.db.compression;
 
-import java.util.Map;
-
 /**
  * Interface for managing scheduled tasks for compression dictionary operations.
  * <p>
@@ -38,11 +36,11 @@ public interface ICompressionDictionaryScheduler extends AutoCloseable
     /**
      * Schedules manual training with the specified options.
      *
-     * @param options training options including maxSamplingDurationSeconds
+     * @param options parsed and validated training options
      * @param trainer the trainer to use
      * @throws IllegalStateException if training is already in progress
      */
-    void scheduleManualTraining(Map<String, String> options, ICompressionDictionaryTrainer trainer);
+    void scheduleManualTraining(ManualTrainingOptions options, ICompressionDictionaryTrainer trainer);
 
     /**
      * Cancel the in-progress manual training

src/java/org/apache/cassandra/db/compression/ManualTrainingOptions.java

@@ -0,0 +1,80 @@
+/*
+ * 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.db.compression;
+
+import java.util.Map;
+
+/**
+ * Configuration options for manual compression dictionary training.
+ * This class encapsulates the parsed and validated parameters needed for training.
+ */
+public class ManualTrainingOptions
+{
+    private final int maxSamplingDurationSeconds;
+
+    public ManualTrainingOptions(int maxSamplingDurationSeconds)
+    {
+        if (maxSamplingDurationSeconds <= 0)
+        {
+            throw new IllegalArgumentException("maxSamplingDurationSeconds must be positive, got: " + maxSamplingDurationSeconds);
+        }
+        this.maxSamplingDurationSeconds = maxSamplingDurationSeconds;
+    }
+
+    /**
+     * Parse options from a string map, typically from JMX/MBean calls.
+     *
+     * @param options the string map containing training options
+     * @return parsed and validated ManualTrainingOptions
+     * @throws IllegalArgumentException if required parameters are missing or invalid
+     */
+    public static ManualTrainingOptions fromStringMap(Map<String, String> options)
+    {
+        if (options == null || !options.containsKey("maxSamplingDurationSeconds"))
+        {
+            throw new IllegalArgumentException("maxSamplingDurationSeconds parameter is required for manual dictionary training");
+        }
+
+        String durationStr = options.get("maxSamplingDurationSeconds");
+        int maxSamplingDurationSeconds;
+        try
+        {
+            maxSamplingDurationSeconds = Integer.parseInt(durationStr);
+        }
+        catch (NumberFormatException e)
+        {
+            throw new IllegalArgumentException("Invalid maxSamplingDurationSeconds value: " + durationStr, e);
+        }
+
+        return new ManualTrainingOptions(maxSamplingDurationSeconds);
+    }
+
+    public int getMaxSamplingDurationSeconds()
+    {
+        return maxSamplingDurationSeconds;
+    }
+
+    @Override
+    public String toString()
+    {
+        return "ManualTrainingOptions{" +
+               "maxSamplingDurationSeconds=" + maxSamplingDurationSeconds +
+               '}';
+    }
+}

test/unit/org/apache/cassandra/db/compression/CompressionDictionaryIntegrationTest.java

@@ -112,13 +112,13 @@ public void testEnableDisableDictionaryCompression()
 
         assertThatNoException()
         .as("Should allow manual training")
-        .isThrownBy(() -> manager.train(Map.of()));
+        .isThrownBy(() -> manager.train(Map.of("maxSamplingDurationSeconds", "600")));
 
         // Disable dictionary compression
         CompressionParams nonDictParams = CompressionParams.lz4();
         manager.maybeReloadFromSchema(nonDictParams);
 
-        assertThatThrownBy(() -> manager.train(Map.of()))
+        assertThatThrownBy(() -> manager.train(Map.of("maxSamplingDurationSeconds", "600")))
         .as("Should disallow manual training when using lz4")
         .isInstanceOf(IllegalArgumentException.class)
         .hasMessageContaining("does not support dictionary compression");
@@ -130,7 +130,7 @@ public void testEnableDisableDictionaryCompression()
 
         assertThatNoException()
         .as("Should allow manual training after switching back to dictionary compression")
-        .isThrownBy(() -> manager.train(Map.of()));
+        .isThrownBy(() -> manager.train(Map.of("maxSamplingDurationSeconds", "600")));
     }
 
     @Test

test/unit/org/apache/cassandra/db/compression/CompressionDictionaryManagerTest.java

@@ -153,7 +153,7 @@ public void testMaybeReloadFromSchemaEnableDictionaryCompression()
 
         managerWithoutDict.maybeReloadFromSchema(dictParams);
 
-        managerWithoutDict.train(Map.of());
+        managerWithoutDict.train(Map.of("maxSamplingDurationSeconds", "600"));
         // Should now have training capability
         String newStatus = managerWithoutDict.getTrainingStatus();
         assertThat(newStatus)
@@ -164,7 +164,7 @@ public void testMaybeReloadFromSchemaEnableDictionaryCompression()
     @Test
     public void testMaybeReloadFromSchemaDisableDictionaryCompression()
     {
-        managerWithDict.train(Map.of());
+        managerWithDict.train(Map.of("maxSamplingDurationSeconds", "600"));
         String status = managerWithDict.getTrainingStatus();
         assertThat(status)
         .as("Should be sampling")
@@ -184,7 +184,7 @@ public void testMaybeReloadFromSchemaDisableDictionaryCompression()
     @Test
     public void testTrainerCompatibilityCheck()
     {
-        managerWithDict.train(Map.of());
+        managerWithDict.train(Map.of("maxSamplingDurationSeconds", "600"));
         String initialStatus = managerWithDict.getTrainingStatus();
         assertThat(initialStatus)
         .as("Should be sampling")
@@ -221,11 +221,36 @@ public void testAddSample()
     @Test
     public void testTrainManualWithNonDictionaryTable()
     {
-        assertThatThrownBy(() -> managerWithoutDict.train(Map.of()))
+        assertThatThrownBy(() -> managerWithoutDict.train(Map.of("maxSamplingDurationSeconds", "600")))
         .isInstanceOf(IllegalArgumentException.class)
         .hasMessageContaining("does not support dictionary compression");
     }
 
+    @Test
+    public void testTrainManualWithMissingParameters()
+    {
+        assertThatThrownBy(() -> managerWithDict.train(Map.of()))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("maxSamplingDurationSeconds parameter is required");
+
+        assertThatThrownBy(() -> managerWithDict.train(null))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("maxSamplingDurationSeconds parameter is required");
+    }
+
+    @Test
+    public void testTrainManualWithInvalidParameters()
+    {
+        assertThatThrownBy(() -> managerWithDict.train(Map.of("maxSamplingDurationSeconds", "invalid")))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("Invalid maxSamplingDurationSeconds value: invalid")
+        .hasCauseInstanceOf(NumberFormatException.class);
+
+        assertThatThrownBy(() -> managerWithDict.train(Map.of("maxSamplingDurationSeconds", "-1")))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("maxSamplingDurationSeconds must be positive, got: -1");
+    }
+
     @Test
     public void testTrainManualWithOptions()
     {
@@ -249,7 +274,7 @@ public void testSchemaChangeWorkflow()
         CompressionParams dictParams = CompressionParams.zstd(CompressionParams.DEFAULT_CHUNK_LENGTH, true,
                                                               Map.of("compression_level", "3"));
         managerWithoutDict.maybeReloadFromSchema(dictParams);
-        managerWithoutDict.train(Map.of());
+        managerWithoutDict.train(Map.of("maxSamplingDurationSeconds", "600"));
         // Should now support training
         String enabledStatus = managerWithoutDict.getTrainingStatus();
         assertThat(enabledStatus).isEqualTo(TrainingStatus.SAMPLING.toString());
@@ -262,7 +287,7 @@ public void testSchemaChangeWorkflow()
         // Should still support training with new parameters
         String updatedStatus = managerWithoutDict.getTrainingStatus();
         assertThat(updatedStatus).isEqualTo(TrainingStatus.NOT_STARTED.toString());
-        managerWithoutDict.train(Map.of());
+        managerWithoutDict.train(Map.of("maxSamplingDurationSeconds", "600"));
         assertThat(enabledStatus).isEqualTo(TrainingStatus.SAMPLING.toString());
 
         // Disable dictionary compression
@@ -278,7 +303,7 @@ public void testSchemaChangeWorkflow()
     public void testUpdateSamplingRate()
     {
         // Test with enabled dictionary manager
-        managerWithDict.train(Map.of());
+        managerWithDict.train(Map.of("maxSamplingDurationSeconds", "600"));
 
         // Should be able to update sampling rate
         assertThatNoException().isThrownBy(() -> managerWithDict.updateSamplingRate(5));
@@ -299,7 +324,7 @@ public void testUpdateSamplingRateWithoutTrainer()
     public void testUpdateSamplingRateValidation()
     {
         // Test with enabled dictionary manager
-        managerWithDict.train(Map.of());
+        managerWithDict.train(Map.of("maxSamplingDurationSeconds", "600"));
 
         // Test invalid sampling rates are rejected by the trainer
         assertThatThrownBy(() -> managerWithDict.updateSamplingRate(0))