natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge 62f8399fad into 2293c58805

This commit is contained in:
Yuqi Yan 2026-07-24 20:30:48 +00:00 committed by GitHub
parent 2293c58805 62f8399fad
commit 2cfac5e143
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 464 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
4.1.12
* Configurable compression algorithm on each level (CASSANDRA-21533)
* Add Paxos v2 option and informatin in cassandra.yaml (CASSANDRA-21316)
* Harden data resurrection startup check with atomic heartbeat file write with fallback (CASSANDRA-21290)
Merged from 4.0:

17
src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java
View File

@ -50,6 +50,19 @@ public class LeveledCompactionStrategy extends AbstractCompactionStrategy
{
private static final Logger logger = LoggerFactory.getLogger(LeveledCompactionStrategy.class);
private static final String SSTABLE_SIZE_OPTION = "sstable_size_in_mb";
/**
* LeveledCompactionStrategy option that lets each level use a different sstable compression than the
* table default. The value is a JSON object mapping level to a compression definition, e.g.
* {@code {"0":{"class":"LZ4Compressor"},"1":{"class":"ZstdCompressor","compression_level":"9"}}}. Each
* level object is a full, independent compression spec, specified exactly like the table
* {@code compression} option (it requires a {@code "class"}; extra keys are that compressor's options; it
* may also set {@code chunk_length_in_kb} / {@code min_compress_ratio}). Nothing is inherited from the
* table compression: unspecified settings use the same defaults as the {@code compression} option (e.g.
* the default chunk length). Levels not listed use the table compression.
*/
public static final String PER_LEVEL_COMPRESSION_OPTION = "per_level_compression";
private static final boolean tolerateSstableSize = Boolean.getBoolean(Config.PROPERTY_PREFIX + "tolerate_sstable_size");
private static final String LEVEL_FANOUT_SIZE_OPTION = "fanout_size";
private static final String SINGLE_SSTABLE_UPLEVEL_OPTION = "single_sstable_uplevel";
@ -621,6 +634,10 @@ public class LeveledCompactionStrategy extends AbstractCompactionStrategy
uncheckedOptions.remove(LEVEL_FANOUT_SIZE_OPTION);
uncheckedOptions.remove(SINGLE_SSTABLE_UPLEVEL_OPTION);
// Validate the per-level compression override (levels not listed use the table compression default)
CompactionParams.parsePerLevelCompression(options.get(PER_LEVEL_COMPRESSION_OPTION));
uncheckedOptions.remove(PER_LEVEL_COMPRESSION_OPTION);
uncheckedOptions.remove(CompactionParams.Option.MIN_THRESHOLD.toString());
uncheckedOptions.remove(CompactionParams.Option.MAX_THRESHOLD.toString());

15
src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java
View File

@ -80,7 +80,7 @@ public class BigTableWriter extends SSTableWriter
TimeUUID pendingRepair,
boolean isTransient,
TableMetadataRef metadata,
MetadataCollector metadataCollector,
MetadataCollector metadataCollector,
SerializationHeader header,
Collection<SSTableFlushObserver> observers,
LifecycleNewTracker lifecycleNewTracker)
@ -122,6 +122,19 @@ public class BigTableWriter extends SSTableWriter
private CompressionParams compressionFor(final OperationType opType)
{
CompressionParams compressionParams = metadata.getLocal().params.compression;
// Per-level compression (LCS-only per-table option): use the compression configured for this
// sstable's level. Each level is a full, independent compression spec (same defaults as the table
// compression option; nothing inherited). Keyed on level. Flush is excluded (it has its own
// flush_compression policy below). No-op for non-LCS tables, which never carry the option.
if (compressionParams.isEnabled() && opType != OperationType.FLUSH)
{
CompressionParams override = metadata.getLocal().params.compaction
.perLevelCompression(metadataCollector.getSSTableLevel());
if (override != null)
compressionParams = override;
}
final ICompressor compressor = compressionParams.getSstableCompressor();
if (null != compressor && opType == OperationType.FLUSH)

5
src/java/org/apache/cassandra/io/sstable/metadata/MetadataCollector.java
View File

@ -250,6 +250,11 @@ public class MetadataCollector implements PartitionStatisticsCollector
return this;
}
public int getSSTableLevel()
{
return sstableLevel;
}
public MetadataCollector updateClusteringValues(ClusteringPrefix<?> clustering)
{
minClustering = minClustering == null || comparator.compare(clustering, minClustering) < 0 ? clustering.minimize() : minClustering;

81
src/java/org/apache/cassandra/schema/CompactionParams.java
View File

@ -17,6 +17,7 @@
*/
package org.apache.cassandra.schema;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.Arrays;
import java.util.HashMap;
@ -30,6 +31,8 @@ import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.cassandra.db.compaction.AbstractCompactionStrategy;
import org.apache.cassandra.db.compaction.LeveledCompactionStrategy;
import org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy;
@ -43,6 +46,8 @@ public final class CompactionParams
{
private static final Logger logger = LoggerFactory.getLogger(CompactionParams.class);
private static final ObjectMapper JSON_MAPPER = new ObjectMapper();
public enum Option
{
CLASS,
@ -90,6 +95,7 @@ public final class CompactionParams
private final ImmutableMap<String, String> options;
private final boolean isEnabled;
private final TombstoneOption tombstoneOption;
private final ImmutableMap<Integer, CompressionParams> perLevelCompression;
private CompactionParams(Class<? extends AbstractCompactionStrategy> klass, Map<String, String> options, boolean isEnabled, TombstoneOption tombstoneOption)
{
@ -97,6 +103,19 @@ public final class CompactionParams
this.options = ImmutableMap.copyOf(options);
this.isEnabled = isEnabled;
this.tombstoneOption = tombstoneOption;
String perLevelSpec = options.get(LeveledCompactionStrategy.PER_LEVEL_COMPRESSION_OPTION);
ImmutableMap<Integer, CompressionParams> perLevel;
try
{
perLevel = ImmutableMap.copyOf(parsePerLevelCompression(perLevelSpec));
}
catch (ConfigurationException e)
{
logger.warn("Ignoring malformed per_level_compression spec '{}': {}", perLevelSpec, e.getMessage());
perLevel = ImmutableMap.of();
}
this.perLevelCompression = perLevel;
}
public static CompactionParams create(Class<? extends AbstractCompactionStrategy> klass, Map<String, String> options)
@ -252,6 +271,68 @@ public final class CompactionParams
return options;
}
/**
* Returns the pre-built compression override for {@code level}, or {@code null} if the level has no
* override and should use the table compression. Each level is built independently from its own
* compression spec, exactly like the table {@code compression} option (same defaults, e.g. chunk length);
* nothing is inherited from the table. See {@link LeveledCompactionStrategy#PER_LEVEL_COMPRESSION_OPTION}.
*/
public CompressionParams perLevelCompression(int level)
{
return perLevelCompression.get(level);
}
/**
* Parses the {@code per_level_compression} spec (see
* {@link LeveledCompactionStrategy#PER_LEVEL_COMPRESSION_OPTION} for the grammar) into a level to
* {@link CompressionParams} map. Each level value is a compression option map that must specify a
* {@code class} and is built and validated exactly like the table {@code compression} option via
* {@link CompressionParams#fromMap}. Returns an empty map for a null or blank spec.
*/
public static Map<Integer, CompressionParams> parsePerLevelCompression(String spec) throws ConfigurationException
{
Map<Integer, CompressionParams> result = new HashMap<>();
if (spec == null || spec.trim().isEmpty())
return result;
Map<String, Map<String, String>> raw;
try
{
raw = JSON_MAPPER.readValue(spec, new TypeReference<>() {});
}
catch (IOException e)
{
throw new ConfigurationException(
format("Invalid per_level_compression JSON '%s': %s", spec, e.getMessage()), e);
}
for (Map.Entry<String, Map<String, String>> entry : raw.entrySet())
{
int level;
try
{
level = Integer.parseInt(entry.getKey().trim());
}
catch (NumberFormatException ex)
{
throw new ConfigurationException(
format("Invalid level '%s' in per_level_compression", entry.getKey()), ex);
}
if (level < 0)
throw new ConfigurationException(format("per_level_compression level must be >= 0, but was %s", level));
Map<String, String> levelOptions = entry.getValue();
if (levelOptions == null || !levelOptions.containsKey(CompressionParams.CLASS))
throw new ConfigurationException(format("per_level_compression entry for level %s must specify a '%s'",
level, CompressionParams.CLASS));
// A level is built exactly like the table 'compression' option (independently, same defaults).
result.put(level, CompressionParams.fromMap(levelOptions));
}
return result;
}
public boolean isEnabled()
{
return isEnabled;

346
test/unit/org/apache/cassandra/io/compress/PerLevelCompressionTest.java Normal file
View File

@ -0,0 +1,346 @@
/*
* 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.io.compress;
import java.util.Map;
import java.util.Set;
import com.google.common.collect.ImmutableMap;
import org.junit.After;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.compaction.LeveledCompactionStrategy;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.schema.CompactionParams;
import org.apache.cassandra.schema.CompressionParams;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
/**
* Tests for the per-table {@code per_level_compression} compaction option on
* {@code LeveledCompactionStrategy}. See {@link LeveledCompactionStrategy#PER_LEVEL_COMPRESSION_OPTION}.
*/
public class PerLevelCompressionTest extends CQLTester
{
private static Config.FlushCompression defaultFlush;
@BeforeClass
public static void setUpClass()
{
CQLTester.setUpClass();
defaultFlush = DatabaseDescriptor.getFlushCompression();
}
@Before
public void setUpCompressionState()
{
// Flush as the table codec so flushed sstables are distinguishable from compaction output.
DatabaseDescriptor.setFlushCompression(Config.FlushCompression.table);
}
@After
public void tearDownCompressionState()
{
DatabaseDescriptor.setFlushCompression(defaultFlush);
}
// ---- Pure parsing / building unit tests ----
@Test
public void parseAndResolveTest()
{
assertTrue(CompactionParams.parsePerLevelCompression(null).isEmpty());
assertTrue(CompactionParams.parsePerLevelCompression(" ").isEmpty());
Map<Integer, CompressionParams> parsed = CompactionParams.parsePerLevelCompression(
"{\"0\":{\"class\":\"LZ4Compressor\"},\"2\":{\"class\":\"ZstdCompressor\",\"compression_level\":\"9\"}}");
assertEquals(2, parsed.size());
assertTrue(parsed.get(0).getSstableCompressor() instanceof LZ4Compressor);
assertTrue(parsed.get(2).getSstableCompressor() instanceof ZstdCompressor);
assertEquals("9", parsed.get(2).getOtherOptions().get("compression_level"));
// Each level is built independently, like the 'compression' option: unspecified chunk length uses the
// compression default; an explicit chunk_length_in_kb is honored (nothing is inherited from the table).
assertEquals(CompressionParams.DEFAULT_CHUNK_LENGTH, parsed.get(0).chunkLength());
assertEquals(1024 * 64, CompactionParams.parsePerLevelCompression(
"{\"1\":{\"class\":\"LZ4Compressor\",\"chunk_length_in_kb\":\"64\"}}").get(1).chunkLength());
// Resolution goes through the cached map on the immutable CompactionParams instance (no re-parsing).
assertTrue(lcsParams("{\"0\":{\"class\":\"LZ4Compressor\"}}").perLevelCompression(0)
.getSstableCompressor() instanceof LZ4Compressor);
assertNull(lcsParams("{\"0\":{\"class\":\"LZ4Compressor\"}}").perLevelCompression(1));
// A malformed spec must degrade gracefully to the table default (null) rather than throw.
assertNull(lcsParams("bogus").perLevelCompression(0));
assertNull(lcsParams("{\"0\":{}}").perLevelCompression(0)); // missing class -> ignored
}
private static CompactionParams lcsParams(String perLevelSpec)
{
return CompactionParams.create(LeveledCompactionStrategy.class,
ImmutableMap.of(LeveledCompactionStrategy.PER_LEVEL_COMPRESSION_OPTION,
perLevelSpec));
}
@Test
public void invalidSpecsRejectedTest()
{
assertBadSpec("not json");
assertBadSpec("[]"); // not a JSON object
assertBadSpec("{\"0\":\"LZ4Compressor\"}"); // level value is not an object
assertBadSpec("{\"x\":{\"class\":\"LZ4Compressor\"}}"); // non-integer level
assertBadSpec("{\"-1\":{\"class\":\"LZ4Compressor\"}}"); // negative level
assertBadSpec("{\"0\":{}}"); // missing class
assertBadSpec("{\"0\":{\"class\":\"NotARealCompressor\"}}"); // class does not resolve
assertBadSpec("{\"0\":{\"class\":\"LZ4Compressor\",\"bogus_option\":\"1\"}}"); // unsupported option
}
private static void assertBadSpec(String spec)
{
try
{
CompactionParams.parsePerLevelCompression(spec);
fail("Expected ConfigurationException for spec: " + spec);
}
catch (ConfigurationException expected)
{
// expected
}
}
// ---- End-to-end tests through a real compaction ----
// NOTE: a maximal compaction (compact()) on LCS uses MajorLeveledCompactionWriter, which writes output
// starting at level 1. For a tiny dataset that yields a single L1 sstable, so the tests below target
// level 1 as the deterministic compaction-output level.
@Test
public void overriddenLevelUsesFastCodecTest() throws Throwable
{
createTable("CREATE TABLE %s (k text PRIMARY KEY, v text) WITH " +
"compaction = {'class':'LeveledCompactionStrategy'," +
"'per_level_compression':'{\"1\":{\"class\":\"LZ4Compressor\"}}'} " +
"AND compression = {'class':'ZstdCompressor'}");
ColumnFamilyStore store = flushTwiceAsZstd();
compact();
assertHasLevel(store, 1);
assertLevelCompressionInvariant(store, 1);
}
@Test
public void unlistedLevelFallsBackToTableCodecTest() throws Throwable
{
// Only level 0 is overridden; the compaction output (L1) must fall back to the table's Zstd.
createTable("CREATE TABLE %s (k text PRIMARY KEY, v text) WITH " +
"compaction = {'class':'LeveledCompactionStrategy'," +
"'per_level_compression':'{\"0\":{\"class\":\"LZ4Compressor\"}}'} " +
"AND compression = {'class':'ZstdCompressor'}");
ColumnFamilyStore store = flushTwiceAsZstd();
compact();
assertHasLevel(store, 1);
assertLevelCompressionInvariant(store, 0);
}
@Test
public void perLevelCompressorOptionsAppliedTest() throws Throwable
{
// L1 override uses Zstd with a non-default compression_level; the table default Zstd carries no level.
createTable("CREATE TABLE %s (k text PRIMARY KEY, v text) WITH " +
"compaction = {'class':'LeveledCompactionStrategy'," +
"'per_level_compression':'{\"1\":{\"class\":\"ZstdCompressor\",\"compression_level\":\"9\"}}'} " +
"AND compression = {'class':'ZstdCompressor'}");
ColumnFamilyStore store = flushTwiceAsZstd();
compact();
assertHasLevel(store, 1);
store.getLiveSSTables().stream()
.filter(s -> s.getSSTableLevel() == 1)
.forEach(s -> {
CompressionParams params = s.getCompressionMetadata().parameters;
assertTrue("Level 1 should be Zstd", params.getSstableCompressor() instanceof ZstdCompressor);
assertEquals("per-level compression_level should be applied and persisted",
"9", params.getOtherOptions().get("compression_level"));
});
}
@Test
public void perLevelChunkLengthOverrideAppliedTest() throws Throwable
{
// Table uses the default 16 KB chunk length; L1 overrides chunk_length_in_kb to 64.
createTable("CREATE TABLE %s (k text PRIMARY KEY, v text) WITH " +
"compaction = {'class':'LeveledCompactionStrategy'," +
"'per_level_compression':'{\"1\":{\"class\":\"LZ4Compressor\",\"chunk_length_in_kb\":\"64\"}}'} " +
"AND compression = {'class':'ZstdCompressor'}");
ColumnFamilyStore store = flushTwiceAsZstd();
compact();
assertHasLevel(store, 1);
store.getLiveSSTables().stream()
.filter(s -> s.getSSTableLevel() == 1)
.forEach(s -> assertEquals("L1 chunk length should be the per-level override (64 KB)",
1024 * 64, s.getCompressionMetadata().chunkLength()));
}
@Test
public void levelChunkLengthIsIndependentOfTableTest() throws Throwable
{
// Table uses a non-default 64 KB chunk length; L1 overrides only the class, so it must fall back to
// the compression default chunk length (16 KB), NOT inherit the table's 64 KB.
createTable("CREATE TABLE %s (k text PRIMARY KEY, v text) WITH " +
"compaction = {'class':'LeveledCompactionStrategy'," +
"'per_level_compression':'{\"1\":{\"class\":\"LZ4Compressor\"}}'} " +
"AND compression = {'class':'ZstdCompressor','chunk_length_in_kb':'64'}");
ColumnFamilyStore store = flushTwiceAsZstd();
compact();
assertHasLevel(store, 1);
store.getLiveSSTables().stream()
.filter(s -> s.getSSTableLevel() == 1)
.forEach(s -> assertEquals("L1 chunk length should be the compression default, not the table's",
CompressionParams.DEFAULT_CHUNK_LENGTH,
s.getCompressionMetadata().chunkLength()));
}
@Test
public void badOptionRejectedAtDdlTest() throws Throwable
{
try
{
createTable("CREATE TABLE %s (k text PRIMARY KEY, v text) WITH " +
"compaction = {'class':'LeveledCompactionStrategy','per_level_compression':'bogus'} " +
"AND compression = {'class':'ZstdCompressor'}");
fail("Expected ConfigurationException for a malformed per_level_compression option");
}
catch (RuntimeException e)
{
assertTrue("Unexpected exception: " + e, hasConfigurationException(e));
}
}
@Test
public void badOptionRejectedAtAlterTest() throws Throwable
{
createTable("CREATE TABLE %s (k text PRIMARY KEY, v text) WITH " +
"compaction = {'class':'LeveledCompactionStrategy'} " +
"AND compression = {'class':'ZstdCompressor'}");
try
{
alterTableMayThrow("ALTER TABLE %s WITH compaction = " +
"{'class':'LeveledCompactionStrategy','per_level_compression':'bogus'}");
fail("Expected ConfigurationException for a malformed per_level_compression option on ALTER");
}
catch (Throwable t)
{
assertTrue("Unexpected exception: " + t, hasConfigurationException(t));
}
}
/**
* Returns true if {@code t} or any exception in its cause chain is a {@link ConfigurationException}.
* The validation failure is wrapped at different depths on CREATE vs ALTER, so walk the whole chain.
*/
private static boolean hasConfigurationException(Throwable t)
{
for (Throwable cause = t; cause != null; cause = cause.getCause())
{
if (cause instanceof ConfigurationException)
return true;
}
return false;
}
@Test
public void validOptionAcceptedAtAlterTest() throws Throwable
{
createTable("CREATE TABLE %s (k text PRIMARY KEY, v text) WITH " +
"compaction = {'class':'LeveledCompactionStrategy'} " +
"AND compression = {'class':'ZstdCompressor'}");
String spec = "{\"1\":{\"class\":\"LZ4Compressor\"}}";
alterTable("ALTER TABLE %s WITH compaction = " +
"{'class':'LeveledCompactionStrategy','per_level_compression':'" + spec + "'}");
Map<String, String> compaction = getCurrentColumnFamilyStore().metadata().params.compaction.options();
assertEquals(spec, compaction.get(LeveledCompactionStrategy.PER_LEVEL_COMPRESSION_OPTION));
}
private ColumnFamilyStore flushTwiceAsZstd() throws Throwable
{
ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
// Keep the two flushed L0 sstables around for the explicit compact() below; otherwise background
// LCS compaction can merge them first and make the sstable count non-deterministic.
cfs.disableAutoCompaction();
execute("INSERT INTO %s (k, v) values (?, ?)", "k1", "v1");
flush();
execute("INSERT INTO %s (k, v) values (?, ?)", "k2", "v2");
flush();
assertEquals(2, cfs.getLiveSSTables().size());
// Flushed sstables use the table codec (flush_compression=table) so they are Zstd.
cfs.getLiveSSTables().forEach(sstable ->
assertTrue(sstable.getCompressionMetadata().parameters.getSstableCompressor() instanceof ZstdCompressor));
return cfs;
}
private static void assertHasLevel(ColumnFamilyStore store, int level)
{
boolean found = store.getLiveSSTables().stream().anyMatch(s -> s.getSSTableLevel() == level);
assertTrue("Expected at least one sstable at level " + level, found);
}
/**
* Asserts every live sstable's compressor matches the override: sstables at {@code fastLevel} use LZ4,
* every other level uses the table's Zstd.
*/
private static void assertLevelCompressionInvariant(ColumnFamilyStore store, int fastLevel)
{
Set<SSTableReader> sstables = store.getLiveSSTables();
assertFalse("Expected at least one sstable after compaction", sstables.isEmpty());
sstables.forEach(sstable -> {
Object compressor = sstable.getCompressionMetadata().parameters.getSstableCompressor();
if (sstable.getSSTableLevel() == fastLevel)
assertTrue("Level " + fastLevel + " should be LZ4 but was " + compressor.getClass().getSimpleName(),
compressor instanceof LZ4Compressor);
else
assertTrue("Level " + sstable.getSSTableLevel() + " should be Zstd but was "
+ compressor.getClass().getSimpleName(),
compressor instanceof ZstdCompressor);
});
}
}