moved definition in yaml file, reverted MessagingService version checks

Author: Claudenw

conf/cassandra.yaml

@@ -646,46 +646,6 @@ commitlog_disk_access_mode: legacy
 #
 # flush_compression: fast
 
-# Defines the default compression used on tables when none is specified
-# in the CQL command.
-#
-# The class_name is the compressor class name. It may be one of the aliases,
-# the class name of a system ICompressor implementation, or fully qualified
-# name of a ICompressor class.
-#
-# class aliases are:
-#  Alias    System compressor impl.
-#  deflate  DeflateCompressor
-#  lz4      LZ4Compressor
-#  none     (null) -- compresson disabled
-#  noop     NoopCompressor
-#  snappy   SnappyCompressor
-#  zstd     ZstdCompressor
-#
-# The standard parameters are any required or optional parameter for the instantiation of the
-# specified class, or one of the following standard parameters:
-# Parameter             Usage
-# enabled               Disables compression if set to false. Defaults to true.
-# chunk_length          The length of the compresson chunks, must include KiB, MiB or GiB suffix, defaults to 16KiB.
-# min_compress_ratio    The minimal acceptable compression, must greater than or equal to 1.0.
-# max_compressed_length The maximum size for a compressed block. Must be less than or equal to chunk_length.
-#                       Must include KiB, MiB or GiB suffix. Defaults to Integer.MAX_VALUE
-#
-# Only one of the min_compress_ratio and max_compressed_length options can be specified.
-# They are mathematically related in that 'chunk_length / max_compressed_length = min_compress_ratio'.
-# If neither option is specified a min_compress_ratio of 0.0 and a max_compressed_length of
-# Integer.MAX_VALUE KB is the default.
-#
-# Additional class specific parameters may be added to the parameters section. The value of the class specific
-# parameter must be a string.
-#
-#sstable_compression:
-#   - class_name: lz4
-#     parameters:
-#       - enabled: "true"
-#         chunk_length: 16KiB
-#         max_compressed_length: 16KiB
-
 # any class that implements the SeedProvider interface and has a
 # constructor that takes a Map<String, String> of parameters will do.
 seed_provider:
@@ -1153,6 +1113,46 @@ snapshot_links_per_second: 0
 #sstable:
 #  selected_format: big
 
+# Defines the default compression used on tables when none is specified
+# in the CQL command.
+#
+# The class_name is the compressor class name. It may be one of the aliases,
+# the class name of a system ICompressor implementation, or fully qualified
+# name of a ICompressor class.
+#
+# class aliases are:
+#  Alias    System compressor impl.
+#  deflate  DeflateCompressor
+#  lz4      LZ4Compressor
+#  none     (null) -- compresson disabled
+#  noop     NoopCompressor
+#  snappy   SnappyCompressor
+#  zstd     ZstdCompressor
+#
+# The standard parameters are any required or optional parameter for the instantiation of the
+# specified class, or one of the following standard parameters:
+# Parameter             Usage
+# enabled               Disables compression if set to false. Defaults to true.
+# chunk_length          The length of the compresson chunks, must include KiB, MiB or GiB suffix, defaults to 16KiB.
+# min_compress_ratio    The minimal acceptable compression, must greater than or equal to 1.0.
+# max_compressed_length The maximum size for a compressed block. Must be less than or equal to chunk_length.
+#                       Must include KiB, MiB or GiB suffix. Defaults to Integer.MAX_VALUE
+#
+# Only one of the min_compress_ratio and max_compressed_length options can be specified.
+# They are mathematically related in that 'chunk_length / max_compressed_length = min_compress_ratio'.
+# If neither option is specified a min_compress_ratio of 0.0 and a max_compressed_length of
+# Integer.MAX_VALUE KB is the default.
+#
+# Additional class specific parameters may be added to the parameters section. The value of the class specific
+# parameter must be a string.
+#
+#  sstable_compression:
+#     - class_name: lz4
+#       parameters:
+#         - enabled: "true"
+#           chunk_length: 16KiB
+#           max_compressed_length: 16KiB
+
 # Granularity of the collation index of rows within a partition.
 # Applies to both BIG and BTI SSTable formats. In both formats,
 # a smaller granularity results in faster lookup of rows within

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

@@ -371,6 +371,9 @@ public static class SSTableConfig
     {
         public String selected_format = BigFormat.NAME;
         public Map<String, Map<String, String>> format = new HashMap<>();
+
+        @Nullable
+        public ParameterizedClass default_compression;
     }
 
     public final SSTableConfig sstable = new SSTableConfig();
@@ -1126,9 +1129,6 @@ public enum PaxosOnLinearizabilityViolation
      */
     public ParameterizedClass default_compaction = null;
 
-    @Nullable
-    public ParameterizedClass sstable_compression;
-
     public static Supplier<Config> getOverrideLoadConfig()
     {
         return overrideLoadConfig;

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

@@ -977,7 +977,7 @@ else if (conf.max_value_size.toMebibytes() >= 2048)
         if (conf.paxos_state_purging == null)
             conf.paxos_state_purging = PaxosStatePurging.legacy;
 
-        sstableCompression = conf.sstable_compression;
+        sstableCompression = conf.sstable.default_compression;
 
         logInitializationOutcome(logger);
 
@@ -2677,7 +2677,7 @@ public static ParameterizedClass getSSTableCompression()
 
     public static void setSSTableCompression(ParameterizedClass compressor)
     {
-        conf.sstable_compression = compressor;
+        conf.sstable.default_compression = compressor;
     }
 
    /**

src/java/org/apache/cassandra/schema/CompressionParams.java

@@ -651,6 +651,7 @@ static class Serializer implements IVersionedSerializer<CompressionParams>
     {
         public void serialize(CompressionParams parameters, DataOutputPlus out, int version) throws IOException
         {
+            assert version >= MessagingService.VERSION_40;
             out.writeUTF(parameters.sstableCompressor.getClass().getSimpleName());
             out.writeInt(parameters.otherOptions.size());
             for (Map.Entry<String, String> entry : parameters.otherOptions.entrySet())
@@ -659,15 +660,12 @@ public void serialize(CompressionParams parameters, DataOutputPlus out, int vers
                 out.writeUTF(entry.getValue());
             }
             out.writeInt(parameters.chunkLength());
-            if (version >= MessagingService.VERSION_40)
-                out.writeInt(parameters.maxCompressedLength);
-            else
-                if (parameters.maxCompressedLength != Integer.MAX_VALUE)
-                    throw new UnsupportedOperationException("Cannot stream SSTables with uncompressed chunks to pre-4.0 nodes.");
+            out.writeInt(parameters.maxCompressedLength);
         }
 
         public CompressionParams deserialize(DataInputPlus in, int version) throws IOException
         {
+            assert version >= MessagingService.VERSION_40;
             String compressorName = in.readUTF();
             int optionCount = in.readInt();
             Map<String, String> options = new HashMap<>();
@@ -678,9 +676,7 @@ public CompressionParams deserialize(DataInputPlus in, int version) throws IOExc
                 options.put(key, value);
             }
             int chunkLength = in.readInt();
-            int minCompressRatio = Integer.MAX_VALUE;   // Earlier Cassandra cannot use uncompressed chunks.
-            if (version >= MessagingService.VERSION_40)
-                minCompressRatio = in.readInt();
+            int minCompressRatio = in.readInt();
 
             CompressionParams parameters;
             try
@@ -696,6 +692,7 @@ public CompressionParams deserialize(DataInputPlus in, int version) throws IOExc
 
         public long serializedSize(CompressionParams parameters, int version)
         {
+            assert version >= MessagingService.VERSION_40;
             long size = TypeSizes.sizeof(parameters.sstableCompressor.getClass().getSimpleName());
             size += TypeSizes.sizeof(parameters.otherOptions.size());
             for (Map.Entry<String, String> entry : parameters.otherOptions.entrySet())
@@ -704,8 +701,7 @@ public long serializedSize(CompressionParams parameters, int version)
                 size += TypeSizes.sizeof(entry.getValue());
             }
             size += TypeSizes.sizeof(parameters.chunkLength());
-            if (version >= MessagingService.VERSION_40)
-                size += TypeSizes.sizeof(parameters.maxCompressedLength());
+            size += TypeSizes.sizeof(parameters.maxCompressedLength());
             return size;
         }
     }

test/conf/cassandra-with-sstable-compressor.yaml

@@ -101,10 +101,11 @@ memtable:
         test_invalid_factory_field:
             class_name: org.apache.cassandra.cql3.validation.operations.CreateTest$InvalidMemtableFactoryField
 
-sstable_compression:
-    class_name: lz4
-    parameters:
-        lz4_compressor_type: high
-        lz4_high_compressor_level: "8"
-        chunk_length: 32MiB
-        min_compress_ratio: "1.5"
+sstable:
+    default_compression:
+        class_name: lz4
+        parameters:
+            lz4_compressor_type: high
+            lz4_high_compressor_level: "8"
+            chunk_length: 32MiB
+            min_compress_ratio: "1.5"

test/unit/org/apache/cassandra/config/YamlConfigurationLoaderTest.java

@@ -223,16 +223,16 @@ public void readSSTableCompressionFromConfig()
     {
         Config c = load("test/conf/cassandra.yaml");
 
-        assertNull(c.sstable_compression);
+        assertNull(c.sstable.default_compression);
 
         c = load("test/conf/cassandra-with-sstable-compressor.yaml");
 
-        assertNotNull(c.sstable_compression);
-        assertThat(c.sstable_compression.class_name).isEqualTo("lz4");
-        assertThat(c.sstable_compression.parameters.remove(CompressionParams.CHUNK_LENGTH)).isEqualTo("32MiB");
-        assertThat(c.sstable_compression.parameters.remove(CompressionParams.MIN_COMPRESS_RATIO)).isEqualTo("1.5");
-        assertThat(c.sstable_compression.parameters.remove(CompressionParams.ENABLED)).isNull();
-        assertThat(c.sstable_compression.parameters.size()).isEqualTo(2);
+        assertNotNull(c.sstable.default_compression);
+        assertThat(c.sstable.default_compression.class_name).isEqualTo("lz4");
+        assertThat(c.sstable.default_compression.parameters.remove(CompressionParams.CHUNK_LENGTH)).isEqualTo("32MiB");
+        assertThat(c.sstable.default_compression.parameters.remove(CompressionParams.MIN_COMPRESS_RATIO)).isEqualTo("1.5");
+        assertThat(c.sstable.default_compression.parameters.remove(CompressionParams.ENABLED)).isNull();
+        assertThat(c.sstable.default_compression.parameters.size()).isEqualTo(2);
     }
 
     @Test