Don't increase the pressure if close to the compaction throughput limit

datastax · Nov 27, 2024 · f2bfb36 · f2bfb36
1 parent e8cc355
commit f2bfb36
Showing 1 changed file with 9 additions and 3 deletions.
diff --git a/src/java/org/apache/cassandra/io/compress/AdaptiveCompressor.java b/src/java/org/apache/cassandra/io/compress/AdaptiveCompressor.java
@@ -36,6 +36,7 @@
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.compaction.CompactionManager;
 import org.apache.cassandra.db.memtable.AbstractAllocatorMemtable;
+import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.metrics.CassandraMetricsRegistry;
 import org.apache.cassandra.metrics.DefaultNameFactory;
 import org.apache.cassandra.metrics.MetricNameFactory;
@@ -422,13 +423,18 @@ private static double getFlushPressure()
     private static double getCompactionPressure(int maxCompactionQueueLength)
     {
         CompactionManager compactionManager = CompactionManager.instance;
-        double rateLimit = DatabaseDescriptor.getCompactionThroughputMbPerSec() * 1000 * 1000;
-        double actualRate = compactionManager.getMetrics().meanCompactionReadThroughput.getValue();
+        double rateLimit = DatabaseDescriptor.getCompactionThroughputMbPerSec() * FileUtils.ONE_MB;
+        double actualRate = compactionManager.getMetrics().bytesCompactedThroughput.getOneMinuteRate();
+        // We don't want to speed up compression if we can keep up with the configured compression rate limit
+        // 0.0 if actualRate >= rateLimit
+        // 1.0 if actualRate <= 0.8 * rateLimit;
+        double rateLimitFactor = Math.min(1.0, Math.max(0.0, (rateLimit - actualRate) / (0.2 * rateLimit)));
 
         long pendingCompactions = compactionManager.getPendingTasks();
         long activeCompactions = compactionManager.getActiveCompactions();
         long queuedCompactions = pendingCompactions - activeCompactions;
-        return (double) queuedCompactions / (maxCompactionQueueLength * DatabaseDescriptor.getConcurrentCompactors());
+        double compactionQueuePressure = Math.min(1.0, (double) queuedCompactions / (maxCompactionQueueLength * DatabaseDescriptor.getConcurrentCompactors()));
+        return compactionQueuePressure * rateLimitFactor;
     }
 
     private int clampCompressionLevel(long compressionLevel)