Leveled Compaction doesn't validate maxBytesForLevel when the table is altered/created

patch by Nikhil Kumawat, Nikhil; reviewed by David Capwell, guo Maxwell for CASSANDRA-20570

Author: nikhilkumawat03

CHANGES.txt

@@ -1,4 +1,5 @@
 4.0.20
+ * Leveled Compaction doesn't validate maxBytesForLevel when the table is altered/created (CASSANDRA-20570)
  * Updated dtest-api to 0.0.18 and removed JMX-related classes that now live in the dtest-api (CASSANDRA-20884)
 
 4.0.19

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

@@ -18,6 +18,7 @@
 package org.apache.cassandra.db.compaction;
 
 import java.util.*;
+import java.math.BigInteger;
 
 
 import com.google.common.annotations.VisibleForTesting;
@@ -43,6 +44,8 @@
 import org.apache.cassandra.io.sstable.ISSTableScanner;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 
+import static org.apache.cassandra.db.compaction.LeveledGenerations.MAX_LEVEL_COUNT;
+
 public class LeveledCompactionStrategy extends AbstractCompactionStrategy
 {
     private static final Logger logger = LoggerFactory.getLogger(LeveledCompactionStrategy.class);
@@ -560,10 +563,14 @@ public static Map<String, String> validateOptions(Map<String, String> options) t
     {
         Map<String, String> uncheckedOptions = AbstractCompactionStrategy.validateOptions(options);
 
+        int ssSize;
+        int fanoutSize;
+
+        // Validate the sstable_size option
         String size = options.containsKey(SSTABLE_SIZE_OPTION) ? options.get(SSTABLE_SIZE_OPTION) : "1";
         try
         {
-            int ssSize = Integer.parseInt(size);
+            ssSize = Integer.parseInt(size);
             if (ssSize < 1)
             {
                 throw new ConfigurationException(String.format("%s must be larger than 0, but was %s", SSTABLE_SIZE_OPTION, ssSize));
@@ -580,15 +587,31 @@ public static Map<String, String> validateOptions(Map<String, String> options) t
         String levelFanoutSize = options.containsKey(LEVEL_FANOUT_SIZE_OPTION) ? options.get(LEVEL_FANOUT_SIZE_OPTION) : String.valueOf(DEFAULT_LEVEL_FANOUT_SIZE);
         try
         {
-            int fanoutSize = Integer.parseInt(levelFanoutSize);
+            fanoutSize = Integer.parseInt(levelFanoutSize);
             if (fanoutSize < 1)
             {
                 throw new ConfigurationException(String.format("%s must be larger than 0, but was %s", LEVEL_FANOUT_SIZE_OPTION, fanoutSize));
             }
         }
         catch (NumberFormatException ex)
         {
-            throw new ConfigurationException(String.format("%s is not a parsable int (base10) for %s", size, LEVEL_FANOUT_SIZE_OPTION), ex);
+            throw new ConfigurationException(String.format("%s is not a parsable int (base10) for %s", levelFanoutSize, LEVEL_FANOUT_SIZE_OPTION), ex);
+        }
+
+        // Validate max Bytes for a level
+        try
+        {
+            long maxSSTableSizeInBytes = Math.multiplyExact(ssSize, 1024L * 1024L); // Convert MB to Bytes
+            BigInteger fanoutPower = BigInteger.valueOf(fanoutSize).pow(MAX_LEVEL_COUNT - 1);
+            BigInteger maxBytes = fanoutPower.multiply(BigInteger.valueOf(maxSSTableSizeInBytes));
+            BigInteger longMaxValue = BigInteger.valueOf(Long.MAX_VALUE);
+            if (maxBytes.compareTo(longMaxValue) > 0)
+                throw new ConfigurationException(String.format("At most %s bytes may be in a compaction level; " +
+                        "your maxSSTableSize must be absurdly high to compute %s", Long.MAX_VALUE, maxBytes));
+        }
+        catch (ArithmeticException ex)
+        {
+            throw new ConfigurationException(String.format("sstable_size_in_mb=%d is too large; resulting bytes exceed Long.MAX_VALUE (%d)", ssSize, Long.MAX_VALUE), ex);
         }
 
         uncheckedOptions.remove(LEVEL_FANOUT_SIZE_OPTION);

test/unit/org/apache/cassandra/db/compaction/LeveledCompactionStrategyTest.java

@@ -906,6 +906,24 @@ public void testReduceScopeL0L1() throws IOException
         }
     }
 
+    @Test()
+    public void testInvalidFanoutAndSSTableSize()
+    {
+        try
+        {
+            Map<String, String> options = new HashMap<>();
+            options.put("class", "LeveledCompactionStrategy");
+            options.put("fanout_size", "90");
+            options.put("sstable_size_in_mb", "1089");
+            LeveledCompactionStrategy.validateOptions(options);
+            Assert.fail("fanout_sizeed and sstable_size_in_mb are invalid, but did not throw ConfigurationException");
+        }
+        catch (ConfigurationException e)
+        {
+            assertTrue(e.getMessage().contains("your maxSSTableSize must be absurdly high to compute"));
+        }
+    }
+
     @Test
     public void testReduceScopeL0()
     {

test/unit/org/apache/cassandra/schema/CreateTableValidationTest.java

@@ -28,6 +28,7 @@
 import org.apache.cassandra.cql3.QueryOptions;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.exceptions.RequestValidationException;
 import org.apache.cassandra.transport.Message;
 import org.apache.cassandra.transport.ProtocolVersion;
 import org.apache.cassandra.transport.SimpleClient;
@@ -202,9 +203,22 @@ public void testNonAlphanummericTableName() throws Throwable
                              String.format("CREATE TABLE %s.\"    \" (key int PRIMARY KEY, val int)", KEYSPACE));
     }
 
-    private void expectedFailure(String statement, String errorMsg)
+    @Test
+    public void testInvalidCompactionOptions()
+    {
+        expectedFailure(ConfigurationException.class, "CREATE TABLE %s (k int PRIMARY KEY, v int) WITH compaction = {'class': 'LeveledCompactionStrategy', 'fanout_size': '90', 'sstable_size_in_mb': '1089'}",
+                        "your maxSSTableSize must be absurdly high to compute");
+    }
+
+    private void expectedFailure(final Class<? extends RequestValidationException> exceptionType, String statement, String errorMsg)
     {
 
+        assertThatExceptionOfType(exceptionType)
+        .isThrownBy(() -> createTableMayThrow(statement)) .withMessageContaining(errorMsg);
+    }
+
+    private void expectedFailure(String statement, String errorMsg)
+    {
         assertThatExceptionOfType(InvalidRequestException.class)
         .isThrownBy(() -> createTableMayThrow(statement)) .withMessageContaining(errorMsg);
     }