CNDB-11613 SAI compressed indexes

CHANGES.txt

@@ -1,5 +1,7 @@
 Future version (tbd)
  * Require only MODIFY permission on base when updating table with MV (STAR-564)
+Merged from 5.1:
+ * Expose current compaction throughput in nodetool (CASSANDRA-13890)
 Merged from 5.0:
  * Disable chronicle analytics (CASSANDRA-19656)
  * Remove mocking in InternalNodeProbe spying on StorageServiceMBean (CASSANDRA-18152)

conf/cassandra.yaml

@@ -479,6 +479,8 @@ commitlog_segment_size_in_mb: 32
 # none : Flush without compressing blocks but while still doing checksums.
 # fast : Flush with a fast compressor. If the table is already using a
 #        fast compressor that compressor is used.
+# adaptive : Flush with a fast adaptive compressor. If the table is already using a
+#        fast compressor that compressor is used.
 # table: Always flush with the same compressor that the table uses. This
 #        was the pre 4.0 behavior.
 #

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

@@ -347,6 +347,9 @@ public enum CassandraRelevantProperties
     /** Watcher used when opening sstables to discover extra components, eg. archive component */
     CUSTOM_SSTABLE_WATCHER("cassandra.custom_sstable_watcher"),
 
+    /** When enabled, a user can set compression options in the index schema */
+    INDEX_COMPRESSION("cassandra.index.compression_enabled", "false"),
+
     /** Controls the maximum top-k limit for vector search */
     SAI_VECTOR_SEARCH_MAX_TOP_K("cassandra.sai.vector_search.max_top_k", "1000"),
 

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

@@ -20,6 +20,7 @@
 import java.lang.reflect.Field;
 import java.lang.reflect.Modifier;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -37,6 +38,7 @@
 import org.apache.cassandra.fql.FullQueryLoggerOptions;
 import org.apache.cassandra.db.ConsistencyLevel;
 import org.apache.cassandra.guardrails.GuardrailsConfig;
+import org.apache.cassandra.io.compress.AdaptiveCompressor;
 import org.apache.cassandra.utils.FBUtilities;
 
 /**
@@ -294,8 +296,9 @@ public class Config
     public double commitlog_sync_group_window_in_ms = Double.NaN;
     public int commitlog_sync_period_in_ms;
     public int commitlog_segment_size_in_mb = 32;
+
     public ParameterizedClass commitlog_compression;
-    public FlushCompression flush_compression = FlushCompression.fast;
+    public FlushCompression flush_compression;
     public int commitlog_max_compression_buffers_in_pool = 3;
     public Integer periodic_commitlog_sync_lag_block_in_ms;
     public TransparentDataEncryptionOptions transparent_data_encryption_options = new TransparentDataEncryptionOptions();
@@ -666,6 +669,7 @@ public enum FlushCompression
     {
         none,
         fast,
+        adaptive,
         table
     }
 

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

@@ -2118,15 +2118,22 @@ public static void setCommitLogCompression(ParameterizedClass compressor)
 
     public static Config.FlushCompression getFlushCompression()
     {
-        return conf.flush_compression;
+        return Objects.requireNonNullElseGet(conf.flush_compression, () -> shouldUseAdaptiveCompressionByDefault()
+                                                                           ? Config.FlushCompression.adaptive
+                                                                           : Config.FlushCompression.fast);
     }
 
     public static void setFlushCompression(Config.FlushCompression compression)
     {
         conf.flush_compression = compression;
     }
 
-   /**
+    public static boolean shouldUseAdaptiveCompressionByDefault()
+    {
+        return System.getProperty("cassandra.default_sstable_compression", "fast").equals("adaptive");
+    }
+
+    /**
     * Maximum number of buffers in the compression pool. The default value is 3, it should not be set lower than that
     * (one segment in compression, one written to, one in reserve); delays in compression may cause the log to use
     * more, depending on how soon the sync policy stops all writing threads.

src/java/org/apache/cassandra/cql3/CqlBuilder.java

@@ -21,6 +21,7 @@
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.TreeMap;
+import java.util.function.Consumer;
 
 import com.google.common.annotations.VisibleForTesting;
 
@@ -238,4 +239,36 @@ public String toString()
     {
         return builder.toString();
     }
+
+    /**
+     * Builds a `WITH option1 = ... AND option2 = ... AND option3 = ... clause
+     * @param builder a receiver to receive a builder allowing to add each option
+     */
+    public CqlBuilder appendOptions(Consumer<OptionsBuilder> builder)
+    {
+        builder.accept(new OptionsBuilder(this));
+        return this;
+    }
+
+    public static class OptionsBuilder
+    {
+        private CqlBuilder builder;
+        boolean empty = true;
+
+        OptionsBuilder(CqlBuilder builder)
+        {
+            this.builder = builder;
+        }
+
+        public OptionsBuilder append(String name, Map<String, String> options)
+        {
+            if (options.isEmpty())
+                return this;
+
+            builder.append((empty ? " WITH " : " AND ") + name + " = ");
+            empty = false;
+            builder.append(options);
+            return this;
+        }
+    }
 }

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

@@ -28,7 +28,9 @@
 import org.apache.cassandra.audit.AuditLogContext;
 import org.apache.cassandra.audit.AuditLogEntryType;
 import org.apache.cassandra.auth.Permission;
+import org.apache.cassandra.config.CassandraRelevantProperties;
 import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.ParameterizedClass;
 import org.apache.cassandra.cql3.ColumnIdentifier;
 import org.apache.cassandra.cql3.QualifiedName;
 import org.apache.cassandra.cql3.statements.RawKeyspaceAwareStatement;
@@ -161,7 +163,21 @@ public Keyspaces apply(Keyspaces schema)
 
         Map<String, String> options = attrs.isCustom ? attrs.getOptions() : Collections.emptyMap();
 
-        IndexMetadata index = IndexMetadata.fromIndexTargets(indexTargets, name, kind, options);
+        Map<String, String> keyCompressionOptions = attrs.getMap("key_compression");
+        CompressionParams keyCompression = keyCompressionOptions != null
+                                        ? CompressionParams.fromMap(keyCompressionOptions)
+                                        : CompressionParams.noCompression();
+
+        Map<String, String> valueCompressionOptions = attrs.getMap("value_compression");
+        CompressionParams valueCompression = valueCompressionOptions != null
+                                           ? CompressionParams.fromMap(valueCompressionOptions)
+                                           : CompressionParams.noCompression();
+
+        if ((keyCompression.isEnabled() || valueCompression.isEnabled()) && !CassandraRelevantProperties.INDEX_COMPRESSION.getBoolean())
+            throw ire("Cannot create a compressed index, because index compression is disabled. " +
+                      "Please set " + CassandraRelevantProperties.INDEX_COMPRESSION.getKey() + " property to enable it.");
+
+        IndexMetadata index = IndexMetadata.fromIndexTargets(indexTargets, name, kind, options, keyCompression, valueCompression);
 
         String className = index.getIndexClassName();
         IndexGuardrails guardRails = IndexGuardrails.forClassName(className);
@@ -194,7 +210,10 @@ public Keyspaces apply(Keyspaces schema)
             throw ire("Index %s is a duplicate of existing index %s", index.name, equalIndex.name);
         }
 
-        TableMetadata newTable = table.withSwapped(table.indexes.with(index));
+        // All indexes on one table must use the same key_compression.
+        // The newly created index forces key_compression on the previous indexes.
+        Indexes newIndexes = table.indexes.withKeyCompression(index.keyCompression).with(index);
+        TableMetadata newTable = table.withSwapped(newIndexes);
         newTable.validate();
 
         return schema.withAddedOrUpdated(keyspace.withSwapped(keyspace.tables.withSwapped(newTable)));

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

@@ -29,6 +29,8 @@ public class IndexAttributes extends PropertyDefinitions
     private static final String DEFAULT_INDEX_CLASS_PROPERTY = "cassandra.default_index_implementation_class";
 
     private static final String KW_OPTIONS = "options";
+    private static final String KW_KEY_COMPRESSION = "key_compression";
+    private static final String KW_VALUE_COMPRESSION = "value_compression";
 
     private static final Set<String> keywords = new HashSet<>();
     private static final Set<String> obsoleteKeywords = new HashSet<>();
@@ -39,6 +41,8 @@ public class IndexAttributes extends PropertyDefinitions
     static
     {
         keywords.add(KW_OPTIONS);
+        keywords.add(KW_KEY_COMPRESSION);
+        keywords.add(KW_VALUE_COMPRESSION);
     }
 
     public void maybeApplyDefaultIndex()

src/java/org/apache/cassandra/db/compaction/CompactionManager.java

@@ -65,7 +65,7 @@
 import com.google.common.util.concurrent.Uninterruptibles;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-
+import com.codahale.metrics.Meter;
 import io.netty.util.concurrent.FastThreadLocal;
 import org.apache.cassandra.cache.AutoSavingCache;
 import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor;
@@ -290,6 +290,11 @@ public void setRate(final double throughPutMbPerSec)
             compactionRateLimiter.setRate(throughput);
     }
 
+    public Meter getCompactionThroughput()
+    {
+        return metrics.bytesCompactedThroughput;
+    }
+
     /**
      * Call this whenever a compaction might be needed on the given column family store.
      * It's okay to over-call (within reason) if a call is unnecessary, it will
@@ -314,7 +319,7 @@ public CompletableFuture<?>[] startCompactionTasks(ColumnFamilyStore cfs, Collec
     {
         return backgroundCompactionRunner.startCompactionTasks(cfs, tasks);
     }
-    
+
     public int getOngoingBackgroundUpgradesCount()
     {
         return backgroundCompactionRunner.getOngoingUpgradesCount();
@@ -1373,7 +1378,7 @@ private void doCleanupOne(final ColumnFamilyStore cfs,
 
     }
 
-    static boolean compactionRateLimiterAcquire(RateLimiter limiter, long bytesScanned, long lastBytesScanned, double compressionRatio)
+    protected boolean compactionRateLimiterAcquire(RateLimiter limiter, long bytesScanned, long lastBytesScanned, double compressionRatio)
     {
         if (DatabaseDescriptor.getCompactionThroughputMbPerSec() == 0)
             return false;
@@ -1386,8 +1391,9 @@ static boolean compactionRateLimiterAcquire(RateLimiter limiter, long bytesScann
         return actuallyAcquire(limiter, lengthRead);
     }
 
-    private static boolean actuallyAcquire(RateLimiter limiter, long lengthRead)
+    private boolean actuallyAcquire(RateLimiter limiter, long lengthRead)
     {
+        metrics.bytesCompactedThroughput.mark(lengthRead);
         while (lengthRead >= Integer.MAX_VALUE)
         {
             limiter.acquire(Integer.MAX_VALUE);

src/java/org/apache/cassandra/db/compaction/CompactionTask.java

@@ -64,7 +64,6 @@
 
 import static org.apache.cassandra.config.CassandraRelevantProperties.COMPACTION_HISTORY_ENABLED;
 import static org.apache.cassandra.config.CassandraRelevantProperties.CURSORS_ENABLED;
-import static org.apache.cassandra.db.compaction.CompactionManager.compactionRateLimiterAcquire;
 import static org.apache.cassandra.utils.FBUtilities.prettyPrintMemory;
 import static org.apache.cassandra.utils.FBUtilities.prettyPrintMemoryPerSecond;
 
@@ -715,7 +714,7 @@ void execute0()
                 long bytesScanned = compactionIterator.bytesRead();
 
                 // Rate limit the scanners, and account for compression
-                if (compactionRateLimiterAcquire(limiter, bytesScanned, lastBytesScanned, compressionRatio))
+                if (CompactionManager.instance.compactionRateLimiterAcquire(limiter, bytesScanned, lastBytesScanned, compressionRatio))
                     lastBytesScanned = bytesScanned;
 
                 maybeStopOrUpdateState();

src/java/org/apache/cassandra/index/sai/IndexContext.java

@@ -87,6 +87,7 @@
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.schema.ColumnMetadata;
+import org.apache.cassandra.schema.CompressionParams;
 import org.apache.cassandra.schema.IndexMetadata;
 import org.apache.cassandra.schema.TableId;
 import org.apache.cassandra.utils.CloseableIterator;
@@ -609,6 +610,16 @@ public String getIndexName()
         return this.config == null ? null : config.name;
     }
 
+    public CompressionParams getKeyCompression()
+    {
+        return this.config.keyCompression;
+    }
+
+    public CompressionParams getValueCompression()
+    {
+        return this.config.valueCompression;
+    }
+
     public int getIntOption(String name, int defaultValue)
     {
         String value = this.config.options.get(name);

src/java/org/apache/cassandra/index/sai/disk/StorageAttachedIndexWriter.java

@@ -44,6 +44,7 @@
 import org.apache.cassandra.io.sstable.SSTable;
 import org.apache.cassandra.io.sstable.format.SSTableFlushObserver;
 import org.apache.cassandra.metrics.TableMetrics;
+import org.apache.cassandra.schema.CompressionParams;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.utils.ApproximateTime;
 import org.apache.cassandra.utils.Throwables;
@@ -107,11 +108,13 @@ public StorageAttachedIndexWriter(IndexDescriptor indexDescriptor,
                                       .filter(Objects::nonNull) // a null here means the column had no data to flush
                                       .collect(Collectors.toList());
 
+        CompressionParams keyCompression = indices.iterator().next().getIndexMetadata().keyCompression;
+
         // If the SSTable components are already being built by another index build then we don't want
         // to build them again so use a NO-OP writer
         this.perSSTableWriter = perIndexComponentsOnly
                                 ? PerSSTableWriter.NONE
-                                : onDiskFormat.newPerSSTableWriter(indexDescriptor);
+                                : onDiskFormat.newPerSSTableWriter(indexDescriptor, keyCompression);
         this.tableMetrics = tableMetrics;
     }
 

src/java/org/apache/cassandra/index/sai/disk/format/IndexComponent.java

@@ -23,9 +23,11 @@
 
 import org.apache.cassandra.index.sai.disk.io.IndexInput;
 import org.apache.cassandra.index.sai.disk.io.IndexOutputWriter;
+import org.apache.cassandra.io.compress.CompressionMetadata;
 import org.apache.cassandra.io.sstable.Component;
 import org.apache.cassandra.io.util.File;
 import org.apache.cassandra.io.util.FileHandle;
+import org.apache.cassandra.schema.CompressionParams;
 import org.apache.lucene.store.ChecksumIndexInput;
 
 public interface IndexComponent
@@ -38,8 +40,11 @@ public interface IndexComponent
 
     String fileNamePart();
     Component asCustomComponent();
+
     File file();
 
+    File compressionMetaFile();
+
     default boolean isCompletionMarker()
     {
         return componentType() == parent().completionMarkerComponent();
@@ -77,7 +82,18 @@ default IndexOutputWriter openOutput() throws IOException
             return openOutput(false);
         }
 
-        IndexOutputWriter openOutput(boolean append) throws IOException;
+        default IndexOutputWriter openOutput(boolean append) throws IOException
+        {
+            return openOutput(append, CompressionParams.noCompression());
+        }
+
+        default IndexOutputWriter openOutput(CompressionParams compression) throws IOException
+        {
+            return openOutput(false, compression);
+        }
+
+        IndexOutputWriter openOutput(boolean append, CompressionParams compression) throws IOException;
+
 
         void createEmpty() throws IOException;
     }