Cassandra 12937 instaclustr

.build/build-rat.xml

@@ -62,6 +62,7 @@
                  <exclude name="**/doc/antora.yml"/>
                  <exclude name="**/test/conf/cassandra.yaml"/>
                  <exclude name="**/test/conf/cassandra-old.yaml"/>
+                 <exclude name="**/test/conf/cassandra-with-sstable-compressor.yaml"/>
                  <exclude name="**/test/conf/cassandra-converters-special-cases-old-names.yaml"/>
                  <exclude name="**/test/conf/cassandra-converters-special-cases.yaml"/>
                  <exclude name="**/test/conf/cassandra_encryption.yaml"/>

conf/cassandra.yaml

@@ -604,6 +604,50 @@ commitlog_segment_size: 32MiB
 #
 # 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 class that implements ICompressor and has a public static 'create' method that accepts
+# a Map<String, String> argument and returns an instance of the 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.
+# chunk_length_in_kb    Same as above but expects an integer.
+# 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.
+#
+# Only one of chunk_length or chunk_length_in_kb may be specified.
+#
+# 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:

pylib/cqlshlib/cql3handling.py

@@ -64,7 +64,7 @@ class Cql3ParsingRuleSet(CqlParsingRuleSet):
             ('class', 'max_threshold', 'tombstone_compaction_interval', 'tombstone_threshold', 'enabled',
              'unchecked_tombstone_compaction', 'only_purge_repaired_tombstones', 'provide_overlapping_tombstones')),
         ('compression', 'compression_parameters',
-            ('sstable_compression', 'chunk_length_kb', 'crc_check_chance')),
+            ('class', 'chunk_length', 'chunk_length_in_kb', 'crc_check_chance', 'enabled', 'min_compress_ratio', 'max_compressed_length')),
         ('caching', None,
             ('rows_per_partition', 'keys')),
     )
@@ -498,7 +498,7 @@ def cf_prop_val_completer(ctxt, cass):
     exist_opts = ctxt.get_binding('propname')
     this_opt = exist_opts[-1]
     if this_opt == 'compression':
-        return ["{'sstable_compression': '"]
+        return ["{'class': '"]
     if this_opt == 'compaction':
         return ["{'class': '"]
     if this_opt == 'caching':
@@ -563,7 +563,7 @@ def cf_prop_val_mapval_completer(ctxt, cass):
             return [Hint('<NONE|ROW|CELL>')]
         return [Hint('<option_value>')]
     elif opt == 'compression':
-        if key == 'sstable_compression':
+        if key == 'class':
             return list(map(escape_value, CqlRuleSet.available_compression_classes))
         return [Hint('<option_value>')]
     elif opt == 'caching':

pylib/cqlshlib/test/test_keyspace_init.cql

@@ -34,7 +34,7 @@ CREATE TABLE has_all_types (
     uuidcol uuid,
     varcharcol varchar,
     varintcol varint
-) WITH compression = {'sstable_compression':'LZ4Compressor'};
+) WITH compression = {'class':'LZ4Compressor'};
 
 INSERT INTO has_all_types (num, intcol, asciicol, bigintcol, blobcol, booleancol,
                            decimalcol, doublecol, durationcol, floatcol, smallintcol, textcol,

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

@@ -1110,6 +1110,9 @@ 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

@@ -226,6 +226,8 @@ public class DatabaseDescriptor
     private static ImmutableMap<String, SSTableFormat<?, ?>> sstableFormats;
     private static volatile SSTableFormat<?, ?> selectedSSTableFormat;
 
+    private static ParameterizedClass sstableCompression;
+
     private static Function<CommitLog, AbstractCommitLogSegmentManager> commitLogSegmentMgrProvider = c -> DatabaseDescriptor.isCDCEnabled()
                                                                                                            ? new CommitLogSegmentManagerCDC(c, DatabaseDescriptor.getCommitLogLocation())
                                                                                                            : new CommitLogSegmentManagerStandard(c, DatabaseDescriptor.getCommitLogLocation());
@@ -867,7 +869,7 @@ else if (conf.repair_session_space.toMebibytes() > (int) (Runtime.getRuntime().m
         if (conf.allow_extra_insecure_udfs)
             logger.warn("Allowing java.lang.System.* access in UDFs is dangerous and not recommended. Set allow_extra_insecure_udfs: false to disable.");
 
-        if(conf.scripted_user_defined_functions_enabled)
+        if (conf.scripted_user_defined_functions_enabled)
             throw new ConfigurationException("JavaScript user-defined functions were removed in CASSANDRA-18252. " +
                                              "Hooks are planned to be introduced as part of CASSANDRA-17280");
 
@@ -966,6 +968,8 @@ else if (conf.max_value_size.toMebibytes() >= 2048)
         if (conf.paxos_state_purging == null)
             conf.paxos_state_purging = PaxosStatePurging.legacy;
 
+        sstableCompression = conf.sstable_compression;
+
         logInitializationOutcome(logger);
 
         if (conf.max_space_usable_for_compactions_in_percentage < 0 || conf.max_space_usable_for_compactions_in_percentage > 1)
@@ -2569,6 +2573,16 @@ public static void setFlushCompression(Config.FlushCompression compression)
         conf.flush_compression = compression;
     }
 
+    public static ParameterizedClass getSSTableCompression()
+    {
+        return sstableCompression;
+    }
+
+    public static void setSSTableCompression(ParameterizedClass compressor)
+    {
+        conf.sstable_compression = compressor;
+    }
+
    /**
     * 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

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

@@ -526,14 +526,14 @@ public void validate(ClientState state)
 
             Guardrails.tableProperties.guard(attrs.updatedProperties(), attrs::removeProperty, state);
 
-            validateDefaultTimeToLive(attrs.asNewTableParams());
+            validateDefaultTimeToLive(attrs.asNewTableParams(keyspaceName, tableName));
         }
 
         public KeyspaceMetadata apply(KeyspaceMetadata keyspace, TableMetadata table)
         {
-            attrs.validate();
+            attrs.validate(keyspaceName, tableName);
 
-            TableParams params = attrs.asAlteredTableParams(table.params);
+            TableParams params = attrs.asAlteredTableParams(table.params, keyspaceName, tableName);
 
             if (table.isCounter() && params.defaultTimeToLive > 0)
                 throw ire("Cannot set default_time_to_live on a table with counters");

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

@@ -68,12 +68,12 @@ public Keyspaces apply(Keyspaces schema)
             throw ire("Materialized view '%s.%s' doesn't exist", keyspaceName, viewName);
         }
 
-        attrs.validate();
+        attrs.validate(keyspaceName, viewName);
 
         // Guardrails on table properties
         Guardrails.tableProperties.guard(attrs.updatedProperties(), attrs::removeProperty, state);
 
-        TableParams params = attrs.asAlteredTableParams(view.metadata.params);
+        TableParams params = attrs.asAlteredTableParams(view.metadata.params, keyspaceName, viewName);
 
         if (params.gcGraceSeconds == 0)
         {

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

@@ -150,7 +150,7 @@ public void validate(ClientState state)
         if (useCompactStorage)
             Guardrails.compactTablesEnabled.ensureEnabled(state);
 
-        validateDefaultTimeToLive(attrs.asNewTableParams());
+        validateDefaultTimeToLive(attrs.asNewTableParams(keyspaceName, tableName));
 
         rawColumns.forEach((name, raw) -> raw.validate(state, name));
     }
@@ -184,8 +184,8 @@ public String toString()
 
     public TableMetadata.Builder builder(Types types)
     {
-        attrs.validate();
-        TableParams params = attrs.asNewTableParams();
+        attrs.validate(keyspaceName, tableName);
+        TableParams params = attrs.asNewTableParams(keyspaceName, tableName);
 
         // use a TreeMap to preserve ordering across JDK versions (see CASSANDRA-9492) - important for stable unit tests
         Map<ColumnIdentifier, ColumnProperties> columns = new TreeMap<>(comparing(o -> o.bytes));

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

@@ -305,7 +305,7 @@ public Keyspaces apply(Keyspaces schema)
          * Validate WITH params
          */
 
-        attrs.validate();
+        attrs.validate(keyspaceName, viewName);
 
         if (attrs.hasOption(TableParams.Option.DEFAULT_TIME_TO_LIVE)
             && attrs.getInt(TableParams.Option.DEFAULT_TIME_TO_LIVE.toString(), 0) != 0)
@@ -324,7 +324,7 @@ public Keyspaces apply(Keyspaces schema)
         if (attrs.hasProperty(TableAttributes.ID))
             builder.id(attrs.getId());
 
-        builder.params(attrs.asNewTableParams())
+        builder.params(attrs.asNewTableParams(keyspaceName, viewName))
                .kind(TableMetadata.Kind.VIEW);
 
         partitionKeyColumns.stream()

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

@@ -54,22 +54,22 @@ public final class TableAttributes extends PropertyDefinitions
         obsoleteKeywords = ImmutableSet.of();
     }
 
-    public void validate()
+    public void validate(String keyspace, String table)
     {
         validate(validKeywords, obsoleteKeywords);
-        build(TableParams.builder()).validate();
+        build(TableParams.builder(), keyspace, table).validate();
     }
 
-    TableParams asNewTableParams()
+    TableParams asNewTableParams(String keyspace, String table)
     {
-        return build(TableParams.builder());
+        return build(TableParams.builder(), keyspace, table);
     }
 
-    TableParams asAlteredTableParams(TableParams previous)
+    TableParams asAlteredTableParams(TableParams previous, String keyspaceName, String tableName)
     {
         if (getId() != null)
             throw new ConfigurationException("Cannot alter table id.");
-        return build(previous.unbuild());
+        return build(previous.unbuild(), keyspaceName, tableName);
     }
 
     public TableId getId() throws ConfigurationException
@@ -95,7 +95,7 @@ public static Set<String> allKeywords()
         return Sets.union(validKeywords, obsoleteKeywords);
     }
 
-    private TableParams build(TableParams.Builder builder)
+    private TableParams build(TableParams.Builder builder, String keyspace, String table)
     {
         if (hasOption(Option.ALLOW_AUTO_SNAPSHOT))
             builder.allowAutoSnapshot(getBoolean(Option.ALLOW_AUTO_SNAPSHOT.toString(), true));
@@ -161,7 +161,7 @@ private TableParams build(TableParams.Builder builder)
         if (hasOption(READ_REPAIR))
             builder.readRepair(ReadRepairStrategy.fromString(getString(READ_REPAIR)));
 
-        return builder.build();
+        return builder.build(keyspace, table);
     }
 
     public boolean hasOption(Option option)

src/java/org/apache/cassandra/db/ColumnFamilyStore.java

@@ -2964,7 +2964,7 @@ public void setCrcCheckChance(double crcCheckChance)
     {
         try
         {
-            TableParams.builder().crcCheckChance(crcCheckChance).build().validate();
+            TableParams.builder().crcCheckChance(crcCheckChance).build(keyspace.getName(), name).validate();
             for (ColumnFamilyStore cfs : concatWithIndexes())
             {
                 cfs.crcCheckChance.set(crcCheckChance);

src/java/org/apache/cassandra/io/compress/LZ4Compressor.java

@@ -66,8 +66,9 @@ public static LZ4Compressor create(Map<String, String> args) throws Configuratio
             if (compressorType.equals(LZ4_FAST_COMPRESSOR) && args.get(LZ4_HIGH_COMPRESSION_LEVEL) != null)
                 logger.warn("'{}' parameter is ignored when '{}' is '{}'", LZ4_HIGH_COMPRESSION_LEVEL, LZ4_COMPRESSOR_TYPE, LZ4_FAST_COMPRESSOR);
             if (compressorType.equals(LZ4_HIGH_COMPRESSOR))
-                logger.info("The ZstdCompressor may be preferable to LZ4 in 'high' mode. Zstd will typically " +
-                            "compress much faster while achieving better ratio, but it may decompress more slowly,");
+                logger.info("The {} may be preferable to LZ4 in 'high' mode. Zstd will typically " +
+                            "compress much faster while achieving better ratio, but it may decompress more slowly.",
+                            ZstdCompressor.class.getSimpleName());
 
             instance = new LZ4Compressor(compressorType, compressionLevel);
             LZ4Compressor instanceFromMap = instances.putIfAbsent(compressorTypeAndLevel, instance);

src/java/org/apache/cassandra/io/sstable/format/DataComponent.java

@@ -89,7 +89,7 @@ private static CompressionParams buildCompressionParams(TableMetadata metadata,
                     if (!compressor.recommendedUses().contains(ICompressor.Uses.FAST_COMPRESSION))
                     {
                         // The default compressor is generally fast (LZ4 with 16KiB block size)
-                        compressionParams = CompressionParams.DEFAULT;
+                        compressionParams = CompressionParams.defaultParams(metadata.keyspace);
                         break;
                     }
                     // else fall through