CASSANDRA-20937 Implement compression dictionary auto-training

CHANGES.txt

@@ -1,4 +1,8 @@
 6.0-alpha3
+ * Fix Compression dictionary distribution silently leaving a node on a stale dictionary (CASSANDRA-21681)
+ * ZstdCompressionDictionary reference leak when opening dictionary-compressed SSTables concurrently (CASSANDRA-21680)
+ * Avoid ZstdDictionary to allocate a native zstd context per chunk (CASSANDRA-21677)
+ * Avoid ZstdDictionaryCompressor to allocate a native zstd context per chunk (CASSANDRA-21672)
  * Fix non-printable characters in Gossiper log for TOKENS (CASSANDRA-21417)
  * Fix compression dictionary training failing with "insufficient samples" on large-chunk tables (CASSANDRA-21666)
  * AccordExecutor improvements (CASSANDRA-21662)

conf/cassandra.yaml

@@ -3029,3 +3029,18 @@ compression_dictionary_cache_size: 10
 # Expired dictionaries will be removed from memory but can be reloaded if needed.
 # Min unit: s
 compression_dictionary_cache_expire: 24h
+
+# Controls if auto-training of compression dictionaries is enabled on this node or not
+# A node has to be a full CMS member with the lowest id to be selected for auto-training.
+# There is always at most one node who auto-trains in the whole cluster.
+compression_dictionary_auto_training_enabled: false
+
+# Controls the interval auto-training will be conducted at. Each interval, training will be
+# conducted all tables which have auto_training_enabled set to true in their compression parameters
+# when a dictionary compressor is in use.
+compression_dictionary_auto_training_interval: 1h
+
+# Controls initial delay of auto-training. If training interval is, for example, 1 day,
+# we would start auto-training for the first time after one day. If this is set e.g. to 1 hour,
+# Then the initial auto-training will be done after 1 hour and each next one will be done 1 day after it.
+compression_dictionary_auto_training_initial_delay: 1h

conf/cassandra_latest.yaml

@@ -2780,3 +2780,18 @@ compression_dictionary_cache_size: 10
 # Expired dictionaries will be removed from memory but can be reloaded if needed.
 # Min unit: s
 compression_dictionary_cache_expire: 24h
+
+# Controls if auto-training of compression dictionaries is enabled on this node or not
+# A node has to be a full CMS member with the lowest id to be selected for auto-training.
+# There is always at most one node who auto-trains in the whole cluster.
+compression_dictionary_auto_training_enabled: false
+
+# Controls the interval auto-training will be conducted at. Each interval, training will be
+# conducted all tables which have auto_training_enabled set to true in their compression parameters
+# when a dictionary compressor is in use.
+compression_dictionary_auto_training_interval: 1h
+
+# Controls initial delay of auto-training. If training interval is, for example, 1 day,
+# we would start auto-training for the first time after one day. If this is set e.g. to 1 hour,
+# Then the initial auto-training will be done after 1 hour and each next one will be done 1 day after it.
+compression_dictionary_auto_training_initial_delay: 1h

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

@@ -578,6 +578,9 @@ public static class SSTableConfig
     public volatile DurationSpec.IntSecondsBound compression_dictionary_refresh_initial_delay = new DurationSpec.IntSecondsBound("10s"); // 10 seconds default
     public volatile int compression_dictionary_cache_size = 10; // max dictionaries per table
     public volatile DurationSpec.IntSecondsBound compression_dictionary_cache_expire = new DurationSpec.IntSecondsBound("24h");
+    public volatile boolean compression_dictionary_auto_training_enabled = false;
+    public volatile DurationSpec.IntMinutesBound compression_dictionary_auto_training_interval = new DurationSpec.IntMinutesBound("1h");
+    public volatile DurationSpec.IntMinutesBound compression_dictionary_auto_training_initial_delay = new DurationSpec.IntMinutesBound("1h");
 
     public DataStorageSpec.LongMebibytesBound paxos_cache_size = null;
 

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

@@ -1280,6 +1280,15 @@ else if (conf.max_value_size.toMebibytes() >= 2048)
         {
             throw new ConfigurationException(ex.getMessage());
         }
+
+        // The auto-training scheduler uses this as the fixed delay between check cycles, so it must be strictly
+        // positive: ScheduledExecutorService.scheduleWithFixedDelay rejects a non-positive period. A negative value
+        // is already rejected by DurationSpec at parse time; this additionally rejects zero. The initial delay may be
+        // zero (run immediately) and the enabled flag is a boolean, so neither needs validation here.
+        if (conf.compression_dictionary_auto_training_interval == null ||
+            conf.compression_dictionary_auto_training_interval.toSeconds() <= 0)
+            throw new ConfigurationException("compression_dictionary_auto_training_interval must be positive, but was " +
+                                             conf.compression_dictionary_auto_training_interval, false);
     }
 
     @VisibleForTesting
@@ -4645,6 +4654,21 @@ public static int getCompressionDictionaryCacheExpireSeconds()
         return conf.compression_dictionary_cache_expire.toSeconds();
     }
 
+    public static boolean getCompressionDictionaryAutoTrainingEnabled()
+    {
+        return conf.compression_dictionary_auto_training_enabled;
+    }
+
+    public static int getCompressionDictionaryAutoTrainingInterval()
+    {
+        return conf.compression_dictionary_auto_training_interval.toSeconds();
+    }
+
+    public static int getCompressionDictionaryAutoTrainingInitialDelay()
+    {
+        return conf.compression_dictionary_auto_training_initial_delay.toSeconds();
+    }
+
     public static int getStreamingKeepAlivePeriod()
     {
         return conf.streaming_keep_alive_period.toSeconds();

src/java/org/apache/cassandra/cql3/statements/schema/AlterSchemaStatement.java

@@ -51,6 +51,7 @@
 import org.apache.cassandra.transport.Event.SchemaChange;
 import org.apache.cassandra.transport.messages.ResultMessage;
 
+import static org.apache.cassandra.io.compress.IDictionaryCompressor.AUTO_TRAINING_ENABLED;
 import static org.apache.cassandra.io.compress.IDictionaryCompressor.DEFAULT_TRAINING_MIN_FREQUENCY;
 import static org.apache.cassandra.io.compress.IDictionaryCompressor.TRAINING_MIN_FREQUENCY_PARAMETER_NAME;
 import static org.apache.cassandra.schema.KeyspaceMetadata.validateKeyspaceName;
@@ -249,6 +250,26 @@ protected void validateMinimumTrainingFrequencyForDictionaryCompressor(TablePara
         }
     }
 
+    /**
+     * Compression dictionary auto-training is only supported on tables using {@link TimeWindowCompactionStrategy}
+     * for now: the auto-trainer biases its training sample toward the newest time window to detect data drift.
+     * Reject enabling {@code auto_training_enabled} on a table that uses any other compaction strategy.
+     */
+    protected void validateCompactionStrategySupportsAutoTraining(TableParams params)
+    {
+        if (!SchemaConstants.isSystemKeyspace(keyspaceName) &&
+            params.compression.isDictionaryCompressionEnabled() &&
+            Boolean.parseBoolean(params.compression.getOtherOptions().getOrDefault(AUTO_TRAINING_ENABLED, "false")) &&
+            !TimeWindowCompactionStrategy.class.isAssignableFrom(params.compaction.klass()))
+        {
+            throw ire("Compression dictionary auto-training ('%s') is only supported on tables using %s, " +
+                      "but %s is configured for this table",
+                      AUTO_TRAINING_ENABLED,
+                      TimeWindowCompactionStrategy.class.getSimpleName(),
+                      params.compaction.klass().getSimpleName());
+        }
+    }
+
     private void grantPermissionsOnResource(IResource resource, AuthenticatedUser user)
     {
         try

src/java/org/apache/cassandra/cql3/statements/schema/AlterTableStatement.java

@@ -688,6 +688,10 @@ public KeyspaceMetadata apply(Epoch epoch, KeyspaceMetadata keyspace, TableMetad
 
             TableParams params = attrs.asAlteredTableParams(table.params);
 
+            // Validate against the altered (merged) params, not the validate()-time defaults, so we see the
+            // table's actual compaction strategy when only compression (e.g. auto_training) is being changed.
+            validateCompactionStrategySupportsAutoTraining(params);
+
             if (table.isCounter() && params.defaultTimeToLive > 0)
                 throw ire("Cannot set default_time_to_live on a table with counters");
 

src/java/org/apache/cassandra/cql3/statements/schema/CopyTableStatement.java

@@ -222,6 +222,11 @@ public Keyspaces apply(ClusterMetadata metadata)
         TableParams originalParams = targetBuilder.build().params;
         TableParams newTableParams = attrs.asAlteredTableParams(originalParams);
 
+        // Validate against the merged params: compression may be inherited from the source table while the WITH
+        // clause overrides compaction, or vice versa, so neither side alone tells us whether auto-training is
+        // being paired with a non-TWCS strategy.
+        validateCompactionStrategySupportsAutoTraining(newTableParams);
+
         TableMetadata table = targetBuilder.params(newTableParams)
                                            .id(TableId.get(metadata))
                                            .build();

src/java/org/apache/cassandra/cql3/statements/schema/CreateTableStatement.java

@@ -230,6 +230,7 @@ public void validate(ClientState state)
         TableParams paramsForValidation = attrs.asNewTableParams(keyspaceName);
         validateDefaultTimeToLive(paramsForValidation);
         validateMinimumTrainingFrequencyForDictionaryCompressor(paramsForValidation);
+        validateCompactionStrategySupportsAutoTraining(paramsForValidation);
 
         rawColumns.forEach((name, raw) -> raw.validate(state, name));
     }

src/java/org/apache/cassandra/db/ColumnFamilyStore.java

@@ -537,6 +537,12 @@ public ColumnFamilyStore(Keyspace keyspace,
         data.subscribe(StorageService.instance.sstablesTracker);
         data.subscribe(SnapshotManager.instance);
 
+        // Initialize the compression dictionary manager BEFORE openAll() below. Loading dictionary-compressed
+        // sstables resolves each dictionary through owner.compressionDictionaryManager(); if that is still null
+        // here, CompressionMetadata opens an uncached dictionary whose lazily-created selfRef is owned by no one
+        // and never released — leaking one dictionary reference per dict sstable opened at startup (CASSANDRA-21047).
+        compressionDictionaryManager = new CompressionDictionaryManager(this, registerBookeeping);
+
         Collection<SSTableReader> sstables = null;
         // scan for sstables corresponding to this cf and load them
         if (data.loadsstables)
@@ -592,7 +598,6 @@ public ColumnFamilyStore(Keyspace keyspace,
         streamManager = new CassandraStreamManager(this);
         repairManager = new CassandraTableRepairManager(this);
         sstableImporter = new SSTableImporter(this);
-        compressionDictionaryManager = new CompressionDictionaryManager(this, registerBookeeping);
 
         if (DatabaseDescriptor.isClientOrToolInitialized() || SchemaConstants.isSystemKeyspace(getKeyspaceName()))
             topPartitions = null;

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

@@ -254,10 +254,17 @@ static CompressionDictionary deserialize(DataInput input, @Nullable CompressionD
 
         CompressionDictionary dictionary = kind.createDictionary(dictId, dict, checksum);
 
-        // update the dictionary manager if it exists
+        // Register with the manager and adopt the CANONICAL cached instance it returns. Under concurrent
+        // opens (e.g. the multi-threaded SSTableBatchOpen pool) two threads can both miss the cache and
+        // create separate instances for the same dictId; only one wins the cache. Returning our own
+        // (possibly losing) instance here would let the caller tryRef() an uncached dictionary whose
+        // lazily-created selfRef the cache never owns or releases, leaking it once the referrer is
+        // garbage collected (CASSANDRA-21047).
         if (manager != null)
         {
-            manager.add(dictionary);
+            CompressionDictionary canonical = manager.add(dictionary);
+            if (canonical != null)
+                dictionary = canonical;
         }
 
         return dictionary;
@@ -391,7 +398,10 @@ public CompressionDictionary createDictionary(CompressionDictionary.DictId dictI
          * @param createdAt creation date of to-be-constructed dictionary
          * @return a compression dictionary instance
          */
-        public abstract CompressionDictionary createDictionary(CompressionDictionary.DictId dictId, byte[] dict, int checksum, Instant createdAt);
+        public abstract CompressionDictionary createDictionary(CompressionDictionary.DictId dictId,
+                                                               byte[] dict,
+                                                               int checksum,
+                                                               Instant createdAt);
 
         /**
          * Creates a dictionary compressor for this kind

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

@@ -0,0 +1,177 @@
+/*
+ * 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.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Deque;
+import java.util.List;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.utils.FBUtilities;
+
+/**
+ * Node-local record of the auto-training adoption decisions this node has made, exposed by the
+ * {@code system_views.compression_dictionary_auto_training} virtual table.
+ * <p>
+ * Auto-training evaluates a freshly trained candidate dictionary against the table's current one and adopts the
+ * candidate only if it improves the compression ratio by at least the configured threshold. That decision is
+ * otherwise only visible in the log, which makes it awkward to answer "has this table ever produced a better
+ * dictionary, and by how much". Each decision is kept here instead.
+ * <p>
+ * The history is held in memory only: it is bounded, and a restart starts afresh. Only the node that ran the
+ * training records anything, so a cluster-wide picture means querying every node.
+ */
+public class CompressionDictionaryAutoTrainingHistory
+{
+    public static final int DEFAULT_MAX_ENTRIES = 1000;
+
+    public static final CompressionDictionaryAutoTrainingHistory instance =
+        new CompressionDictionaryAutoTrainingHistory(DEFAULT_MAX_ENTRIES);
+
+    private final int maxEntries;
+    // newest first, so the virtual table and the eviction of the oldest entry both read naturally
+    private final Deque<Entry> entries = new ArrayDeque<>();
+
+    @VisibleForTesting
+    public CompressionDictionaryAutoTrainingHistory(int maxEntries)
+    {
+        this.maxEntries = maxEntries;
+    }
+
+    /**
+     * Records one adoption decision.
+     *
+     * @param keyspaceName   keyspace the candidate was trained for
+     * @param tableName      table the candidate was trained for
+     * @param kind           kind of the dictionary that was trained
+     * @param baselineRatio  compressed/uncompressed ratio the current dictionary achieved on the sample
+     * @param candidateRatio compressed/uncompressed ratio the candidate achieved on the same sample
+     * @param improvement    relative improvement of the candidate over the baseline
+     * @param threshold      improvement the candidate had to reach to be adopted
+     * @param promoted       whether the candidate became the table's dictionary
+     */
+    public void record(String keyspaceName,
+                       String tableName,
+                       CompressionDictionary.Kind kind,
+                       double baselineRatio,
+                       double candidateRatio,
+                       double improvement,
+                       double threshold,
+                       boolean promoted)
+    {
+        record(FBUtilities.now().toEpochMilli(), keyspaceName, tableName, kind,
+               baselineRatio, candidateRatio, improvement, threshold, promoted);
+    }
+
+    /**
+     * Records a decision at an explicit time. A table trains one candidate at a time and training is far slower
+     * than the millisecond resolution of the timestamp, so in production two decisions for one table cannot share
+     * a timestamp; tests that need several entries have to space them out themselves.
+     */
+    @VisibleForTesting
+    public void record(long timestampMillis,
+                       String keyspaceName,
+                       String tableName,
+                       CompressionDictionary.Kind kind,
+                       double baselineRatio,
+                       double candidateRatio,
+                       double improvement,
+                       double threshold,
+                       boolean promoted)
+    {
+        Entry entry = new Entry(timestampMillis,
+                                FBUtilities.getBroadcastAddressAndPort(),
+                                keyspaceName,
+                                tableName,
+                                kind,
+                                baselineRatio,
+                                candidateRatio,
+                                improvement,
+                                threshold,
+                                promoted);
+
+        synchronized (entries)
+        {
+            if (entries.size() == maxEntries)
+                entries.removeLast();
+
+            entries.addFirst(entry);
+        }
+    }
+
+    /**
+     * @return the recorded decisions, newest first
+     */
+    public List<Entry> entries()
+    {
+        synchronized (entries)
+        {
+            return new ArrayList<>(entries);
+        }
+    }
+
+    @VisibleForTesting
+    public void clear()
+    {
+        synchronized (entries)
+        {
+            entries.clear();
+        }
+    }
+
+    public static class Entry
+    {
+        public final long timestampMillis;
+        public final InetAddressAndPort node;
+        public final String keyspaceName;
+        public final String tableName;
+        public final CompressionDictionary.Kind kind;
+        public final double baselineRatio;
+        public final double candidateRatio;
+        public final double improvement;
+        public final double threshold;
+        public final boolean promoted;
+
+        Entry(long timestampMillis,
+              InetAddressAndPort node,
+              String keyspaceName,
+              String tableName,
+              CompressionDictionary.Kind kind,
+              double baselineRatio,
+              double candidateRatio,
+              double improvement,
+              double threshold,
+              boolean promoted)
+        {
+            this.timestampMillis = timestampMillis;
+            this.node = node;
+            this.keyspaceName = keyspaceName;
+            this.tableName = tableName;
+            this.kind = kind;
+            this.baselineRatio = baselineRatio;
+            this.candidateRatio = candidateRatio;
+            this.improvement = improvement;
+            this.threshold = threshold;
+            this.promoted = promoted;
+        }
+    }
+}