CASSANDRA-21642 Consider the introduction of flush_compression CQL table parameter

NEWS.txt

@@ -90,6 +90,12 @@ New features
       deployments that authenticate over mutual TLS. PasswordDefaultRoleInitializer
       remains the default and produces the historical behaviour. See CASSANDRA-21546
       for more information.
+    - New table option flush_compression selects the compressor used when a memtable is
+      flushed to an SSTable, per table. Values are 'auto' (default, follow the flush_compression
+      setting in cassandra.yaml), 'none', 'fast' and 'table', with the same meaning as the
+      yaml values. Compaction is unaffected and always uses the table's compression. The
+      option is stored in a new flush_compression column of system_schema.tables and
+      system_schema.views, which is only populated for non-default values.
 
 Upgrading
 ---------

conf/cassandra.yaml

@@ -748,6 +748,10 @@ cursor_compaction_enabled: false
 # table: Always flush with the same compressor that the table uses. This
 #        was the pre 4.0 behavior.
 #
+# This is the node-wide default. A table can override it with the
+# flush_compression table option (values: none, fast, table); the default
+# table value 'auto' uses this setting.
+#
 # flush_compression: fast
 
 # any class that implements the SeedProvider interface and has a

conf/cassandra_latest.yaml

@@ -748,6 +748,10 @@ background_write_disk_access_mode: direct
 # table: Always flush with the same compressor that the table uses. This
 #        was the pre 4.0 behavior.
 #
+# This is the node-wide default. A table can override it with the
+# flush_compression table option (values: none, fast, table); the default
+# table value 'auto' uses this setting.
+#
 # flush_compression: fast
 
 # any class that implements the SeedProvider interface and has a

doc/modules/cassandra/pages/developing/cql/ddl.adoc

@@ -509,6 +509,7 @@ impacts the size of bloom filters in-memory and on-disk.
 | `default_time_to_live` |_simple_ |0 |Default expiration time (“TTL”) in seconds for a table
 | `compaction` |_map_ |_see below_ | xref:cassandra:managing/operating/compaction/index.adoc#cql-compaction-options[Compaction options]
 | `compression` |_map_ |_see below_ | xref:cassandra:managing/operating/compression/index.adoc#cql-compression-options[Compression options]
+| `flush_compression` |_simple_ |auto | xref:cassandra:developing/cql/ddl.adoc#cql-flush-compression-option[Flush compression option]
 | `caching` |_map_ |_see below_ |Caching options
 | `memtable_flush_period_in_ms` |_simple_ |0 |Time (in ms) before Cassandra flushes memtables to disk
 | `read_repair` |_simple_ |BLOCKING |Sets read repair behavior (see below)
@@ -657,6 +658,31 @@ For instance, to create a table with LZ4Compressor and a `chunk_length_in_kb` of
 include::cassandra:example$CQL/chunk_length.cql[]
 ----
 
+[[cql-flush-compression-option]]
+===== Flush compression option
+
+The `flush_compression` option selects the compressor used when a memtable is flushed to an SSTable.
+Flushes are on the write path, so a slow compressor (`ZstdCompressor`, `DeflateCompressor`, LZ4 in high compression mode)
+can back up flushes and block writes. Compaction always rewrites SSTables with the compressor configured
+in `compression`, so the choice only affects freshly flushed SSTables.
+
+[cols=",",options="header",]
+|===
+|Value |Description
+| `auto` | Default. Use the node-wide `flush_compression` setting from `cassandra.yaml`.
+| `none` | Flush without compressing blocks. Checksums are still written.
+| `fast` | Flush with a fast compressor. If the table compressor is already fast it is used as is, otherwise the default LZ4 compressor is used.
+| `table` | Flush with the compressor configured in `compression`.
+|===
+
+The option has no effect on tables with compression disabled. For example, to flush a Zstd compressed
+table with its own compressor regardless of the node setting:
+
+[source,cql]
+----
+ALTER TABLE cycling.cyclist_name WITH flush_compression = 'table';
+----
+
 [[cql-caching-options]]
 ===== Caching options
 

doc/modules/cassandra/pages/managing/operating/compression.adoc

@@ -330,6 +330,14 @@ this is a performance problem it is highly encouraged not to turn this
 off as it is Cassandra's only protection against bitrot. In earlier versions of Cassandra a
 duplicate of this option existed in the compression configuration.
 The latter was deprecated in Cassandra 3.0 and removed in Cassandra 5.0.
+* `flush_compression` (default: `auto`): selects the compressor used when a
+memtable is flushed. Flushes sit on the write path, so a "good" ratio compressor
+can back them up and block writes; by default Cassandra flushes with a fast
+compressor and lets compaction rewrite the data with the table compressor.
+`auto` follows the `flush_compression` setting in `cassandra.yaml` (default `fast`).
+`none`, `fast` and `table` override the node setting for this table. See the
+xref:cassandra:developing/cql/ddl.adoc#cql-flush-compression-option[CQL reference]
+for the semantics of each value.
 
 == Benefits and Uses
 

pylib/cqlshlib/cql3handling.py

@@ -56,6 +56,7 @@ class Cql3ParsingRuleSet(CqlParsingRuleSet):
         ('memtable_flush_period_in_ms', None),
         ('cdc', None),
         ('read_repair', None),
+        ('flush_compression', None),
     )
 
     columnfamily_layout_map_options = (
@@ -568,6 +569,8 @@ def cf_prop_val_completer(ctxt, cass):
         return [Hint('<true|false>')]
     if this_opt in ('read_repair'):
         return [Hint('<\'none\'|\'blocking\'>')]
+    if this_opt == 'flush_compression':
+        return ["'auto'", "'none'", "'fast'", "'table'"]
     if this_opt == 'allow_auto_snapshot':
         return [Hint('<boolean>')]
     if this_opt == 'incremental_backups':

pylib/cqlshlib/test/test_cqlsh_completion.py

@@ -716,7 +716,7 @@ def create_columnfamily_table_template(self, name):
                             choices=['allow_auto_snapshot',
                                      'bloom_filter_fp_chance', 'compaction',
                                      'compression',
-                                     'default_time_to_live', 'gc_grace_seconds',
+                                     'default_time_to_live', 'flush_compression', 'gc_grace_seconds',
                                      'incremental_backups',
                                      'max_index_interval',
                                      'memtable',
@@ -728,7 +728,7 @@ def create_columnfamily_table_template(self, name):
                             choices=['allow_auto_snapshot',
                                      'bloom_filter_fp_chance', 'compaction',
                                      'compression',
-                                     'default_time_to_live', 'gc_grace_seconds',
+                                     'default_time_to_live', 'flush_compression', 'gc_grace_seconds',
                                      'incremental_backups',
                                      'max_index_interval',
                                      'memtable',
@@ -778,7 +778,7 @@ def create_columnfamily_table_template(self, name):
                             + "{'class': 'SizeTieredCompactionStrategy'} AND ",
                             choices=['allow_auto_snapshot', 'bloom_filter_fp_chance', 'compaction',
                                      'compression',
-                                     'default_time_to_live', 'gc_grace_seconds',
+                                     'default_time_to_live', 'flush_compression', 'gc_grace_seconds',
                                      'incremental_backups',
                                      'max_index_interval',
                                      'memtable',
@@ -847,7 +847,7 @@ def test_complete_in_create_table_like(self):
                             choices=['allow_auto_snapshot',
                                      'bloom_filter_fp_chance', 'compaction',
                                      'compression',
-                                     'default_time_to_live', 'gc_grace_seconds',
+                                     'default_time_to_live', 'flush_compression', 'gc_grace_seconds',
                                      'incremental_backups',
                                      'max_index_interval',
                                      'memtable',
@@ -900,7 +900,7 @@ def test_complete_in_create_table_like(self):
                             + "{'class': 'SizeTieredCompactionStrategy'} AND ",
                             choices=['allow_auto_snapshot', 'bloom_filter_fp_chance', 'compaction',
                                      'compression',
-                                     'default_time_to_live', 'gc_grace_seconds',
+                                     'default_time_to_live', 'flush_compression', 'gc_grace_seconds',
                                      'incremental_backups',
                                      'max_index_interval',
                                      'memtable',

pylib/cqlshlib/test/test_cqlsh_output.py

@@ -759,6 +759,7 @@ def test_describe_columnfamily_output(self):
                 AND fast_path = 'keyspace'
                 AND default_time_to_live = 0
                 AND extensions = {}
+                AND flush_compression = 'auto'
                 AND gc_grace_seconds = 864000
                 AND incremental_backups = true
                 AND max_index_interval = 2048

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

@@ -679,7 +679,7 @@ private TableParams validateAndUpdateTransactionalMigration(boolean isCounter, T
         @Override
         public boolean compatibleWith(ClusterMetadata metadata)
         {
-            return metadata.directory.commonSerializationVersion.isAtLeast(Version.V0);
+            return metadata.directory.commonSerializationVersion.isAtLeast(attrs.minimumSerializationVersion());
         }
 
         public KeyspaceMetadata apply(Epoch epoch, KeyspaceMetadata keyspace, TableMetadata table, ClusterMetadata metadata)

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

@@ -31,7 +31,6 @@
 import org.apache.cassandra.schema.ViewMetadata;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.tcm.ClusterMetadata;
-import org.apache.cassandra.tcm.serialization.Version;
 import org.apache.cassandra.transport.Event.SchemaChange;
 import org.apache.cassandra.transport.Event.SchemaChange.Change;
 import org.apache.cassandra.transport.Event.SchemaChange.Target;
@@ -63,7 +62,7 @@ public void validate(ClientState state)
     @Override
     public boolean compatibleWith(ClusterMetadata metadata)
     {
-        return metadata.directory.commonSerializationVersion.isAtLeast(Version.V0);
+        return metadata.directory.commonSerializationVersion.isAtLeast(attrs.minimumSerializationVersion());
     }
 
     @Override

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

@@ -115,7 +115,8 @@ public AuditLogContext getAuditLogContext()
     @Override
     public boolean compatibleWith(ClusterMetadata metadata)
     {
-        return metadata.directory.commonSerializationVersion.isAtLeast(Version.V5);
+        Version version = metadata.directory.commonSerializationVersion;
+        return version.isAtLeast(Version.V5) && version.isAtLeast(attrs.minimumSerializationVersion());
     }
 
     @Override

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

@@ -75,7 +75,6 @@
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.reads.repair.ReadRepairStrategy;
 import org.apache.cassandra.tcm.ClusterMetadata;
-import org.apache.cassandra.tcm.serialization.Version;
 import org.apache.cassandra.transport.Event.SchemaChange;
 import org.apache.cassandra.transport.Event.SchemaChange.Change;
 import org.apache.cassandra.transport.Event.SchemaChange.Target;
@@ -141,7 +140,7 @@ public String cql()
     @Override
     public boolean compatibleWith(ClusterMetadata metadata)
     {
-        return metadata.directory.commonSerializationVersion.isAtLeast(Version.V0);
+        return metadata.directory.commonSerializationVersion.isAtLeast(attrs.minimumSerializationVersion());
     }
 
     public Keyspaces apply(ClusterMetadata metadata)

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

@@ -58,7 +58,6 @@
 import org.apache.cassandra.schema.ViewMetadata;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.tcm.ClusterMetadata;
-import org.apache.cassandra.tcm.serialization.Version;
 import org.apache.cassandra.transport.Event.SchemaChange;
 import org.apache.cassandra.transport.Event.SchemaChange.Change;
 import org.apache.cassandra.transport.Event.SchemaChange.Target;
@@ -133,7 +132,7 @@ public void validate(ClientState state)
     @Override
     public boolean compatibleWith(ClusterMetadata metadata)
     {
-        return metadata.directory.commonSerializationVersion.isAtLeast(Version.V0);
+        return metadata.directory.commonSerializationVersion.isAtLeast(attrs.minimumSerializationVersion());
     }
 
     @Override

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

@@ -31,12 +31,14 @@
 import org.apache.cassandra.schema.CachingParams;
 import org.apache.cassandra.schema.CompactionParams;
 import org.apache.cassandra.schema.CompressionParams;
+import org.apache.cassandra.schema.FlushCompressionParams;
 import org.apache.cassandra.schema.MemtableParams;
 import org.apache.cassandra.schema.Schema;
 import org.apache.cassandra.schema.SchemaConstants;
 import org.apache.cassandra.schema.TableId;
 import org.apache.cassandra.schema.TableParams;
 import org.apache.cassandra.schema.TableParams.Option;
+import org.apache.cassandra.tcm.serialization.Version;
 import org.apache.cassandra.service.accord.topology.FastPathStrategy;
 import org.apache.cassandra.service.consensus.TransactionalMode;
 import org.apache.cassandra.service.consensus.migration.TransactionalMigrationFromMode;
@@ -54,6 +56,7 @@
 import static org.apache.cassandra.schema.TableParams.Option.COMPRESSION;
 import static org.apache.cassandra.schema.TableParams.Option.CRC_CHECK_CHANCE;
 import static org.apache.cassandra.schema.TableParams.Option.DEFAULT_TIME_TO_LIVE;
+import static org.apache.cassandra.schema.TableParams.Option.FLUSH_COMPRESSION;
 import static org.apache.cassandra.schema.TableParams.Option.GC_GRACE_SECONDS;
 import static org.apache.cassandra.schema.TableParams.Option.INCREMENTAL_BACKUPS;
 import static org.apache.cassandra.schema.TableParams.Option.MAX_INDEX_INTERVAL;
@@ -143,6 +146,9 @@ private TableParams build(TableParams.Builder builder)
         if (hasOption(COMPRESSION))
             builder.compression(CompressionParams.fromMap(getMap(COMPRESSION)));
 
+        if (hasOption(FLUSH_COMPRESSION))
+            builder.flushCompression(FlushCompressionParams.fromString(getString(FLUSH_COMPRESSION)));
+
         if (hasOption(Option.MEMTABLE))
             builder.memtable(MemtableParams.getWithFallback(getString(Option.MEMTABLE)));
 
@@ -208,6 +214,15 @@ public boolean hasOption(Option option)
         return hasProperty(option.toString());
     }
 
+    /**
+     * Minimum cluster-wide metadata serialization version required to enact a schema change carrying these options.
+     * Options unknown to older nodes must not be committed while such nodes can still replay the statement.
+     */
+    public Version minimumSerializationVersion()
+    {
+        return hasOption(FLUSH_COMPRESSION) ? Version.V11 : Version.V0;
+    }
+
     private String getString(Option option)
     {
         String value = getString(option.toString());

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

@@ -38,6 +38,7 @@
 import org.apache.cassandra.io.util.SequentialWriter;
 import org.apache.cassandra.io.util.SequentialWriterOption;
 import org.apache.cassandra.schema.CompressionParams;
+import org.apache.cassandra.schema.FlushCompressionParams;
 import org.apache.cassandra.schema.TableMetadata;
 
 import static org.apache.cassandra.io.DirectIoSupport.NOT_A_WRITER;
@@ -143,23 +144,20 @@ && isDirectWriteSupported(operationType))
      *
      * @return {@link CompressionParams}
      */
-    private static CompressionParams buildCompressionParams(TableMetadata metadata, OperationType operationType, FlushCompression flushCompression)
+    @VisibleForTesting
+    static CompressionParams buildCompressionParams(TableMetadata metadata, OperationType operationType, FlushCompression flushCompression)
     {
         CompressionParams compressionParams = metadata.params.compression;
+        FlushCompressionParams flushCompressionParams = metadata.params.flushCompression;
         final ICompressor compressor = compressionParams.getSstableCompressor();
 
         if (null != compressor && operationType == OperationType.FLUSH)
         {
-            // When we are flushing out of the memtable throughput of the compressor is critical as flushes,
-            // especially of large tables, can queue up and potentially block writes.
-            // This optimization allows us to fall back to a faster compressor if a particular
-            // compression algorithm indicates we should. See CASSANDRA-15379 for more details.
-            switch (flushCompression)
+            switch (flushCompressionParams.configurationKey)
             {
-                // It is relatively easier to insert a Noop compressor than to disable compressed writing
-                // entirely as the "compression" member field is provided outside the scope of this class.
-                // It may make sense in the future to refactor the ownership of the compression flag so that
-                // We can bypass the CompressedSequentialWriter in this case entirely.
+                case auto:
+                    compressionParams = resolveCompressionParams(flushCompression, compressor, compressionParams);
+                    break;
                 case none:
                     compressionParams = CompressionParams.NOOP;
                     break;
@@ -179,6 +177,41 @@ private static CompressionParams buildCompressionParams(TableMetadata metadata,
         return compressionParams;
     }
 
+    private static CompressionParams resolveCompressionParams(FlushCompression flushCompression,
+                                                              ICompressor compressor,
+                                                              CompressionParams tableCompressionParams)
+    {
+        CompressionParams compressionParams = tableCompressionParams;
+
+        // When we are flushing out of the memtable throughput of the compressor is critical as flushes,
+        // especially of large tables, can queue up and potentially block writes.
+        // This optimization allows us to fall back to a faster compressor if a particular
+        // compression algorithm indicates we should. See CASSANDRA-15379 for more details.
+        switch (flushCompression)
+        {
+            // It is relatively easier to insert a Noop compressor than to disable compressed writing
+            // entirely as the "compression" member field is provided outside the scope of this class.
+            // It may make sense in the future to refactor the ownership of the compression flag so that
+            // We can bypass the CompressedSequentialWriter in this case entirely.
+            case none:
+                compressionParams = CompressionParams.NOOP;
+                break;
+            case fast:
+                if (!compressor.recommendedUses().contains(ICompressor.Uses.FAST_COMPRESSION))
+                {
+                    // The default compressor is generally fast (LZ4 with 16KiB block size)
+                    compressionParams = CompressionParams.DEFAULT;
+                    break;
+                }
+                // else fall through
+            case table:
+            default:
+                break;
+        }
+
+        return compressionParams;
+    }
+
     @VisibleForTesting
     static boolean isDirectWriteSupported(OperationType operationType)
     {