CNDB-11532: Adaptive compression

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/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;
 
 /**
@@ -292,8 +294,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();
@@ -664,6 +667,7 @@ public enum FlushCompression
     {
         none,
         fast,
+        adaptive,
         table
     }
 

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

@@ -2119,15 +2119,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/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

@@ -63,7 +63,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;
 
@@ -658,7 +657,7 @@ void execute0()
                 long bytesScanned = compactionIterator.getTotalBytesScanned();
 
                 // 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();