From 1fe4506757fd83c7c0d2b1a91cd536b224f11794 Mon Sep 17 00:00:00 2001 From: Stefan Miklosovic Date: Thu, 3 Sep 2026 15:06:02 +0200 Subject: [PATCH 1/2] Introduce flush_compression CQL table parameter patch by Stefan Miklosovic; reviewed by TBD for TBD --- NEWS.txt | 6 + conf/cassandra.yaml | 4 + conf/cassandra_latest.yaml | 4 + .../cassandra/pages/developing/cql/ddl.adoc | 26 ++ .../pages/managing/operating/compression.adoc | 8 + pylib/cqlshlib/cql3handling.py | 3 + pylib/cqlshlib/test/test_cqlsh_completion.py | 10 +- pylib/cqlshlib/test/test_cqlsh_output.py | 1 + .../statements/schema/TableAttributes.java | 5 + .../io/sstable/format/DataComponent.java | 53 +++- .../schema/FlushCompressionParams.java | 85 +++++++ .../cassandra/schema/SchemaKeyspace.java | 10 + .../cassandra/schema/TableMetadata.java | 6 + .../apache/cassandra/schema/TableParams.java | 29 ++- .../cassandra/tcm/membership/NodeVersion.java | 2 +- .../cassandra/tcm/serialization/Version.java | 4 + .../test/FlushCompressionTableOptionTest.java | 197 +++++++++++++++ .../CqlConstraintValidationTester.java | 1 + .../cql3/ColumnSpecificationTest.java | 1 + .../statements/DescribeStatementTest.java | 4 + .../FlushCompressionTableOptionTest.java | 208 +++++++++++++++ .../cassandra/db/SchemaCQLHelperTest.java | 1 + .../FlushCompressionTableOptionFlushTest.java | 239 ++++++++++++++++++ .../DataComponentFlushCompressionTest.java | 196 ++++++++++++++ .../schema/FlushCompressionParamsTest.java | 98 +++++++ .../cassandra/schema/SchemaKeyspaceTest.java | 35 +++ .../cassandra/schema/TableParamsTest.java | 91 +++++++ .../cassandra/utils/CassandraGenerators.java | 13 + 28 files changed, 1321 insertions(+), 19 deletions(-) create mode 100644 src/java/org/apache/cassandra/schema/FlushCompressionParams.java create mode 100644 test/distributed/org/apache/cassandra/distributed/test/FlushCompressionTableOptionTest.java create mode 100644 test/unit/org/apache/cassandra/cql3/validation/operations/FlushCompressionTableOptionTest.java create mode 100644 test/unit/org/apache/cassandra/io/compress/FlushCompressionTableOptionFlushTest.java create mode 100644 test/unit/org/apache/cassandra/io/sstable/format/DataComponentFlushCompressionTest.java create mode 100644 test/unit/org/apache/cassandra/schema/FlushCompressionParamsTest.java diff --git a/NEWS.txt b/NEWS.txt index 650ce7c2e372..8ffda4da127a 100644 --- a/NEWS.txt +++ b/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 --------- diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index 92fafa00de2c..ea400f7a8d5a 100644 --- a/conf/cassandra.yaml +++ b/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 diff --git a/conf/cassandra_latest.yaml b/conf/cassandra_latest.yaml index 4093ab962481..38b4095ad31d 100644 --- a/conf/cassandra_latest.yaml +++ b/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 diff --git a/doc/modules/cassandra/pages/developing/cql/ddl.adoc b/doc/modules/cassandra/pages/developing/cql/ddl.adoc index 0f248282a062..20fd44273a68 100644 --- a/doc/modules/cassandra/pages/developing/cql/ddl.adoc +++ b/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 diff --git a/doc/modules/cassandra/pages/managing/operating/compression.adoc b/doc/modules/cassandra/pages/managing/operating/compression.adoc index 3f9a4905b8b5..fb785e8a7b38 100644 --- a/doc/modules/cassandra/pages/managing/operating/compression.adoc +++ b/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 diff --git a/pylib/cqlshlib/cql3handling.py b/pylib/cqlshlib/cql3handling.py index a83c6535b8ea..38192280c0ac 100644 --- a/pylib/cqlshlib/cql3handling.py +++ b/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('')] 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('')] if this_opt == 'incremental_backups': diff --git a/pylib/cqlshlib/test/test_cqlsh_completion.py b/pylib/cqlshlib/test/test_cqlsh_completion.py index aa7904471bc7..94d3e4e5adb4 100644 --- a/pylib/cqlshlib/test/test_cqlsh_completion.py +++ b/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', diff --git a/pylib/cqlshlib/test/test_cqlsh_output.py b/pylib/cqlshlib/test/test_cqlsh_output.py index 302f88bc0082..0e135116800f 100644 --- a/pylib/cqlshlib/test/test_cqlsh_output.py +++ b/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 diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/TableAttributes.java b/src/java/org/apache/cassandra/cql3/statements/schema/TableAttributes.java index 9e84b2d3cc08..07af8e27a835 100644 --- a/src/java/org/apache/cassandra/cql3/statements/schema/TableAttributes.java +++ b/src/java/org/apache/cassandra/cql3/statements/schema/TableAttributes.java @@ -31,6 +31,7 @@ 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; @@ -54,6 +55,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 +145,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))); diff --git a/src/java/org/apache/cassandra/io/sstable/format/DataComponent.java b/src/java/org/apache/cassandra/io/sstable/format/DataComponent.java index bd4a222e5db9..61b700adb23d 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/DataComponent.java +++ b/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) { diff --git a/src/java/org/apache/cassandra/schema/FlushCompressionParams.java b/src/java/org/apache/cassandra/schema/FlushCompressionParams.java new file mode 100644 index 000000000000..70592a55ecd5 --- /dev/null +++ b/src/java/org/apache/cassandra/schema/FlushCompressionParams.java @@ -0,0 +1,85 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.schema; + +import java.util.Arrays; + +import org.apache.cassandra.exceptions.ConfigurationException; + +import static java.lang.String.format; + +public final class FlushCompressionParams +{ + public final Option configurationKey; + + public enum Option + { + auto, + none, + fast, + table; + + private static final String possibleValues = Arrays.toString(Option.values()); + } + + public static final FlushCompressionParams DEFAULT = new FlushCompressionParams(Option.auto); + + public static FlushCompressionParams fromString(String value) + { + try + { + return new FlushCompressionParams(Option.valueOf(value)); + } + catch (Throwable t) + { + throw new ConfigurationException(format("Invalid value used for flush compression parameter: %s, possible values: %s", + value, + Option.possibleValues)); + } + } + + private FlushCompressionParams(Option configurationKey) + { + this.configurationKey = configurationKey; + } + + @Override + public boolean equals(Object o) + { + if (this == o) + return true; + + if (!(o instanceof FlushCompressionParams)) + return false; + + return configurationKey == ((FlushCompressionParams) o).configurationKey; + } + + @Override + public int hashCode() + { + return configurationKey.hashCode(); + } + + @Override + public String toString() + { + return configurationKey.toString(); + } +} diff --git a/src/java/org/apache/cassandra/schema/SchemaKeyspace.java b/src/java/org/apache/cassandra/schema/SchemaKeyspace.java index d466f09ed546..9ed5e0f12122 100644 --- a/src/java/org/apache/cassandra/schema/SchemaKeyspace.java +++ b/src/java/org/apache/cassandra/schema/SchemaKeyspace.java @@ -161,6 +161,7 @@ private SchemaKeyspace() + "comment text," + "compaction frozen>," + "compression frozen>," + + "flush_compression text," + "memtable text," + "crc_check_chance double," + "dclocal_read_repair_chance double," // no longer used, left for drivers' sake @@ -247,6 +248,7 @@ private SchemaKeyspace() + "comment text," + "compaction frozen>," + "compression frozen>," + + "flush_compression text," + "memtable text," + "crc_check_chance double," + "dclocal_read_repair_chance double," // no longer used, left for drivers' sake @@ -653,6 +655,10 @@ public static void addTableParamsToRowBuilder(TableParams params, Row.SimpleBuil if (params.memtable != MemtableParams.DEFAULT) builder.add("memtable", params.memtable.configurationKey()); + // Only add the flush_compression column if the value is not default (auto); absence means auto. + if (!params.flushCompression.equals(FlushCompressionParams.DEFAULT)) + builder.add("flush_compression", params.flushCompression.toString()); + // As above, only add the allow_auto_snapshot column if the value is not default (true) and // auto-snapshotting is enabled, to avoid RTE in pre-4.2 versioned node during upgrades if (!params.allowAutoSnapshot) @@ -1154,6 +1160,10 @@ static TableParams createTableParamsFromRow(UntypedResultSet.Row row) builder.automatedRepair(AutoRepairParams.fromMap(row.getFrozenTextMap("auto_repair"))); } + // flush_compression column was introduced in 7.0 + if (row.has("flush_compression")) + builder.flushCompression(FlushCompressionParams.fromString(row.getString("flush_compression"))); + return builder.build(); } diff --git a/src/java/org/apache/cassandra/schema/TableMetadata.java b/src/java/org/apache/cassandra/schema/TableMetadata.java index 4f9fe1320afb..be57ec59dd1f 100644 --- a/src/java/org/apache/cassandra/schema/TableMetadata.java +++ b/src/java/org/apache/cassandra/schema/TableMetadata.java @@ -1156,6 +1156,12 @@ public Builder compression(CompressionParams val) return this; } + public Builder flushCompression(FlushCompressionParams val) + { + params.flushCompression(val); + return this; + } + public Builder fastPath(FastPathStrategy val) { params.fastPath(val); diff --git a/src/java/org/apache/cassandra/schema/TableParams.java b/src/java/org/apache/cassandra/schema/TableParams.java index c9fb20d3c50c..6b760ce8a578 100644 --- a/src/java/org/apache/cassandra/schema/TableParams.java +++ b/src/java/org/apache/cassandra/schema/TableParams.java @@ -61,6 +61,7 @@ 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.EXTENSIONS; +import static org.apache.cassandra.schema.TableParams.Option.FLUSH_COMPRESSION; import static org.apache.cassandra.schema.TableParams.Option.FAST_PATH; import static org.apache.cassandra.schema.TableParams.Option.GC_GRACE_SECONDS; import static org.apache.cassandra.schema.TableParams.Option.INCREMENTAL_BACKUPS; @@ -76,6 +77,7 @@ public final class TableParams { public static final Serializer serializer = new Serializer(); + public enum Option { ALLOW_AUTO_SNAPSHOT, @@ -101,7 +103,8 @@ public enum Option TRANSACTIONAL_MODE, TRANSACTIONAL_MIGRATION_FROM, PENDING_DROP, - AUTO_REPAIR; + AUTO_REPAIR, + FLUSH_COMPRESSION; @Override public String toString() @@ -126,6 +129,7 @@ public String toString() public final CachingParams caching; public final CompactionParams compaction; public final CompressionParams compression; + public final FlushCompressionParams flushCompression; public final MemtableParams memtable; public final ImmutableMap extensions; public final boolean cdc; @@ -157,6 +161,7 @@ private TableParams(Builder builder) caching = builder.caching; compaction = builder.compaction; compression = builder.compression; + flushCompression = builder.flushCompression; memtable = builder.memtable; extensions = builder.extensions; cdc = builder.cdc; @@ -183,6 +188,7 @@ public static Builder builder(TableParams params) .securityLabel(params.securityLabel) .compaction(params.compaction) .compression(params.compression) + .flushCompression(params.flushCompression) .memtable(params.memtable) .crcCheckChance(params.crcCheckChance) .defaultTimeToLive(params.defaultTimeToLive) @@ -294,6 +300,7 @@ public boolean equals(Object o) && caching.equals(p.caching) && compaction.equals(p.compaction) && compression.equals(p.compression) + && flushCompression.equals(p.flushCompression) && memtable.equals(p.memtable) && extensions.equals(p.extensions) && cdc == p.cdc @@ -324,6 +331,7 @@ public int hashCode() caching, compaction, compression, + flushCompression, memtable, extensions, cdc, @@ -356,6 +364,7 @@ public String toString() .add(CACHING.toString(), caching) .add(COMPACTION.toString(), compaction) .add(COMPRESSION.toString(), compression) + .add(FLUSH_COMPRESSION.toString(), flushCompression) .add(MEMTABLE.toString(), memtable) .add(EXTENSIONS.toString(), extensions) .add(CDC.toString(), cdc) @@ -406,6 +415,8 @@ public void appendCqlTo(CqlBuilder builder, boolean isView) e -> "0x" + ByteBufferUtil.bytesToHex(e.getValue()))), false) .newLine() + .append("AND flush_compression = ").appendWithSingleQuotes(flushCompression.toString()) + .newLine() .append("AND gc_grace_seconds = ").append(gcGraceSeconds) .newLine() .append("AND incremental_backups = ").append(incrementalBackups) @@ -454,6 +465,7 @@ public static final class Builder private CachingParams caching = CachingParams.DEFAULT; private CompactionParams compaction = CompactionParams.DEFAULT; private CompressionParams compression = CompressionParams.DEFAULT; + private FlushCompressionParams flushCompression = FlushCompressionParams.DEFAULT; private MemtableParams memtable = MemtableParams.DEFAULT; private ImmutableMap extensions = ImmutableMap.of(); private boolean cdc; @@ -575,6 +587,12 @@ public Builder compression(CompressionParams val) return this; } + public Builder flushCompression(FlushCompressionParams val) + { + flushCompression = val; + return this; + } + public Builder cdc(boolean val) { cdc = val; @@ -660,6 +678,8 @@ public void serialize(TableParams t, DataOutputPlus out, Version version) throws } if (version.isAtLeast(Version.V8)) out.writeUTF(t.securityLabel); + if (version.isAtLeast(Version.V11)) + out.writeUTF(t.flushCompression.toString()); } public TableParams deserialize(DataInputPlus in, Version version) throws IOException @@ -693,6 +713,9 @@ public TableParams deserialize(DataInputPlus in, Version version) throws IOExcep } if (version.isAtLeast(Version.V8)) builder.securityLabel(in.readUTF()); + if (version.isAtLeast(Version.V11)) + builder.flushCompression(FlushCompressionParams.fromString(in.readUTF())); + return builder.build(); } @@ -725,9 +748,9 @@ public long serializedSize(TableParams t, Version version) sizeof(t.pendingDrop); } if (version.isAtLeast(Version.V8)) - { size += sizeof(t.securityLabel); - } + if (version.isAtLeast(Version.V11)) + size += sizeof(t.flushCompression.toString()); return size; } diff --git a/src/java/org/apache/cassandra/tcm/membership/NodeVersion.java b/src/java/org/apache/cassandra/tcm/membership/NodeVersion.java index ed1e4f561221..fa942a560294 100644 --- a/src/java/org/apache/cassandra/tcm/membership/NodeVersion.java +++ b/src/java/org/apache/cassandra/tcm/membership/NodeVersion.java @@ -36,7 +36,7 @@ public class NodeVersion implements Comparable { public static final Serializer serializer = new Serializer(); - public static final Version CURRENT_METADATA_VERSION = Version.V10; + public static final Version CURRENT_METADATA_VERSION = Version.V11; public static final NodeVersion CURRENT = new NodeVersion(new CassandraVersion(FBUtilities.getReleaseVersionString()), CURRENT_METADATA_VERSION); private static final CassandraVersion SINCE_VERSION = CassandraVersion.CASSANDRA_5_1; diff --git a/src/java/org/apache/cassandra/tcm/serialization/Version.java b/src/java/org/apache/cassandra/tcm/serialization/Version.java index 5134c40ce4d2..beb024852c8a 100644 --- a/src/java/org/apache/cassandra/tcm/serialization/Version.java +++ b/src/java/org/apache/cassandra/tcm/serialization/Version.java @@ -88,6 +88,10 @@ public enum Version * - SchemaTransformation CQL serialized as raw bytes to avoid utf string limit */ V10(10), + /** + * - TableParams has flushCompression + */ + V11(11), UNKNOWN(Integer.MAX_VALUE); diff --git a/test/distributed/org/apache/cassandra/distributed/test/FlushCompressionTableOptionTest.java b/test/distributed/org/apache/cassandra/distributed/test/FlushCompressionTableOptionTest.java new file mode 100644 index 000000000000..4df18d969625 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/FlushCompressionTableOptionTest.java @@ -0,0 +1,197 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.distributed.test; + +import java.util.ArrayList; +import java.util.List; +import java.util.TreeMap; + +import org.junit.Test; + +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.api.IInvokableInstance; +import org.apache.cassandra.io.compress.CompressionMetadata; +import org.apache.cassandra.io.compress.LZ4Compressor; +import org.apache.cassandra.io.compress.NoopCompressor; +import org.apache.cassandra.io.compress.ZstdCompressor; +import org.apache.cassandra.io.sstable.Descriptor; +import org.apache.cassandra.io.sstable.SSTableId; +import org.apache.cassandra.io.sstable.SSTableIdFactory; +import org.apache.cassandra.io.sstable.format.CompressionInfoComponent; +import org.apache.cassandra.io.sstable.format.SSTableFormat.Components; +import org.apache.cassandra.io.util.File; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * {@code flush_compression} table option on a running node: the compressor recorded in the + * CompressionInfo component of flushed SSTables is read from disk, in SSTable id order, and compared + * against the table option and the {@code flush_compression} yaml setting. Automatic compaction is + * disabled on every node start so flushed SSTables are never rewritten. + */ +public class FlushCompressionTableOptionTest extends TestBaseImpl +{ + private static final String ZSTD = "{'class': 'ZstdCompressor'}"; + + @Test + public void tableOptionOverridesYamlTable() throws Throwable + { + try (Cluster cluster = start("table")) + { + createTables(cluster, "auto", "fast", "none", "table"); + flushAll(cluster, "auto", "fast", "none", "table"); + + assertFlushedWith(cluster.get(1), tableFor("auto"), ZstdCompressor.class); + assertFlushedWith(cluster.get(1), tableFor("fast"), LZ4Compressor.class); + assertFlushedWith(cluster.get(1), tableFor("none"), NoopCompressor.class); + assertFlushedWith(cluster.get(1), tableFor("table"), ZstdCompressor.class); + } + } + + @Test + public void tableOptionOverridesYamlNone() throws Throwable + { + try (Cluster cluster = start("none")) + { + createTables(cluster, "auto", "fast", "none", "table"); + flushAll(cluster, "auto", "fast", "none", "table"); + + assertFlushedWith(cluster.get(1), tableFor("auto"), NoopCompressor.class); + assertFlushedWith(cluster.get(1), tableFor("fast"), LZ4Compressor.class); + assertFlushedWith(cluster.get(1), tableFor("none"), NoopCompressor.class); + assertFlushedWith(cluster.get(1), tableFor("table"), ZstdCompressor.class); + } + } + + @Test + public void tableOptionOverridesYamlFast() throws Throwable + { + try (Cluster cluster = start("fast")) + { + createTables(cluster, "auto", "fast", "none", "table"); + flushAll(cluster, "auto", "fast", "none", "table"); + + assertFlushedWith(cluster.get(1), tableFor("auto"), LZ4Compressor.class); + assertFlushedWith(cluster.get(1), tableFor("fast"), LZ4Compressor.class); + assertFlushedWith(cluster.get(1), tableFor("none"), NoopCompressor.class); + assertFlushedWith(cluster.get(1), tableFor("table"), ZstdCompressor.class); + } + } + + @Test + public void alterAppliesToNextFlushAndSurvivesRestart() throws Throwable + { + try (Cluster cluster = start("fast")) + { + cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (k int PRIMARY KEY, v text) WITH compression = " + ZSTD)); + insertAndFlush(cluster, "tbl", 1); + assertFlushedWith(cluster.get(1), "tbl", LZ4Compressor.class); + + cluster.schemaChange(withKeyspace("ALTER TABLE %s.tbl WITH flush_compression = 'table'")); + insertAndFlush(cluster, "tbl", 2); + assertFlushedWith(cluster.get(1), "tbl", LZ4Compressor.class, ZstdCompressor.class); + + cluster.get(1).shutdown().get(); + cluster.get(1).startup(); + disableAutoCompaction(cluster); + + Object[][] rows = cluster.coordinator(1).execute(withKeyspace("SELECT flush_compression FROM system_schema.tables WHERE keyspace_name = '%s' AND table_name = 'tbl'"), ConsistencyLevel.ONE); + assertThat(rows[0][0]).isEqualTo("table"); + + insertAndFlush(cluster, "tbl", 3); + assertFlushedWith(cluster.get(1), "tbl", LZ4Compressor.class, ZstdCompressor.class, ZstdCompressor.class); + } + } + + private static Cluster start(String yamlFlushCompression) throws Throwable + { + Cluster cluster = init(Cluster.build(1).withConfig(c -> c.set("flush_compression", yamlFlushCompression)).start()); + disableAutoCompaction(cluster); + return cluster; + } + + private static void disableAutoCompaction(Cluster cluster) + { + cluster.get(1).nodetoolResult("disableautocompaction").asserts().success(); + } + + /** One table per option, named {@code opt_