From 80224a97670095e710460f79879462732f1505ae Mon Sep 17 00:00:00 2001 From: Jon Haddad Date: Thu, 1 May 2025 05:58:32 -0700 Subject: [PATCH 1/2] Improved UCS docs with migration advice and example workloads. Patch by Jon Haddad for CASSANDRA-19389 --- .../managing/operating/compaction/ucs.adoc | 81 ++++++++++++++++++- 1 file changed, 80 insertions(+), 1 deletion(-) diff --git a/doc/modules/cassandra/pages/managing/operating/compaction/ucs.adoc b/doc/modules/cassandra/pages/managing/operating/compaction/ucs.adoc index 4798f0b6fd..a41c979ef9 100644 --- a/doc/modules/cassandra/pages/managing/operating/compaction/ucs.adoc +++ b/doc/modules/cassandra/pages/managing/operating/compaction/ucs.adoc @@ -18,7 +18,86 @@ Thus, a compaction is triggered when more than a given number of SSTables are pr * *size* can be replaced by *density*, allowing SSTables to be split at arbitrary points when the output of a compaction is written, while still producing a leveled hierarchy. Density is defined as the size of an SSTable divided by the width of the token range it covers. -Let's look at the first concept in more detail. +== Migration from Other Strategies + +The Unified Compaction Strategy (UCS) can be configured to behave like other compaction strategies, making migration straightforward. It also provides advanced options for optimizing specific workload patterns. + +=== Examples + +Below are examples for migrating from commonly used strategies. UCS can maintain similar behavior while providing additional benefits such as parallel compaction and the ability to change parameters without requiring full recompaction. + +[cols="1,3a", options="header"] +|=== +| Source Strategy | Migration Command +| Migrating From LCS | +[source,plaintext] +---- +ALTER TABLE mykeyspace.foo +WITH COMPACTION = { + 'class': 'UnifiedCompactionStrategy', + 'scaling_parameters': 'L10'}; +---- +| Migration from SizeTieredCompactionStrategy | +[source,plaintext] +---- +ALTER TABLE mykeyspace.foo WITH +COMPACTION = { + 'class': 'UnifiedCompactionStrategy', + 'scaling_parameters': 'T4'}; +---- +|=== + +== Use Case Specific Configurations + +The following configurations are optimized for common workload patterns. The parameters can be adjusted based on your specific requirements. + +These provide a good starting point for common workloads, but you may find you want to tune additional parameters based on your workload characteristics. +Additional details to understand this are in the following section. + +[cols="1,3a,3", options="header"] +|=== +| Use Case | Configuration Example | Explanation +| Read Heavy Key Value | +[source,plaintext] +---- +ALTER TABLE mykeyspace.foo +WITH COMPACTION = { + 'class': 'UnifiedCompactionStrategy', + 'scaling_parameters': 'L10', + 'target_sstable_size': + '256MiB', + 'base_shard_count': '8' +}; +---- +| Optimizes for read-intensive workloads with a leveled approach similar to LCS. The smaller target SSTable size and higher shard count improve read performance by minimizing the number of SSTables that must be consulted for a query. +| Write Heavy | +[source,plaintext] +---- +ALTER TABLE mykeyspace.foo +WITH COMPACTION = { + 'class': 'UnifiedCompactionStrategy', + 'scaling_parameters': 'T4', + 'target_sstable_size': '1GiB', + 'base_shard_count': '4' +}; +---- +| Optimizes for write-intensive workloads using a tiered approach similar to STCS. The larger target SSTable size reduces write amplification by requiring fewer compactions, while the lower shard count reduces the overhead of managing too many SSTables. +| Time Series | +[source,plaintext] +---- +ALTER TABLE mykeyspace.foo WITH COMPACTION = { + 'class': 'UnifiedCompactionStrategy', + 'scaling_parameters': 'T8', + 'target_sstable_size': '512MiB', + 'base_shard_count': '8', + 'expired_sstable_check_frequency_seconds': '300' +}; +---- +| Suitable for time-series data with TTLs. The higher tiered scaling parameter (T8) improves write throughput, while the frequent expired SSTable check helps reclaim space from expired data more quickly. The higher shard count allows for greater parallelism in compaction operations. + +Using `scaling_parameters:T8` will result in more SSTables per read. Consider using T4 for time series use cases where lower read latency is desired, and you can afford to perform additional compaction. +|=== + == Read and write amplification From 4f25e58f267e68df945e0ba47b2a684c1fb33a85 Mon Sep 17 00:00:00 2001 From: David Capwell Date: Fri, 2 May 2025 14:23:17 -0700 Subject: [PATCH 2/2] zero copy streaming allocates direct memory that isn't used, but does help to fragment the memory space patch by David Capwell; reviewed by Yifan Cai for CASSANDRA-20577 --- CHANGES.txt | 1 + .../io/sstable/SSTableZeroCopyWriter.java | 47 ++++++++++++++----- .../cassandra/io/util/SequentialWriter.java | 9 +++- 3 files changed, 44 insertions(+), 13 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 08e0ecfa16..d30a383b7f 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 5.0.5 + * zero copy streaming allocates direct memory that isn't used, but does help to fragment the memory space (CASSANDRA-20577) * CQLSSTableWriter supports setting the format (BTI or Big) (CASSANDRA-20609) * Don't allocate in ThreadLocalReadAheadBuffer#close() (CASSANDRA-20551) * Ensure RowFilter#isMutableIntersection() properly evaluates numeric ranges on a single column (CASSANDRA-20566) diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableZeroCopyWriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableZeroCopyWriter.java index 3bf21f1155..46a490974e 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableZeroCopyWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableZeroCopyWriter.java @@ -38,9 +38,12 @@ import org.apache.cassandra.dht.Token; import org.apache.cassandra.io.FSWriteError; import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.io.util.DataInputPlus; +import org.apache.cassandra.io.util.File; import org.apache.cassandra.io.util.SequentialWriter; +import org.apache.cassandra.io.util.SequentialWriterOption; import org.apache.cassandra.net.AsyncStreamingInputPlus; import org.apache.cassandra.schema.TableId; +import org.apache.cassandra.utils.ByteBufferUtil; import static java.lang.String.format; import static org.apache.cassandra.utils.FBUtilities.prettyPrintMemory; @@ -50,7 +53,7 @@ public class SSTableZeroCopyWriter extends SSTable implements SSTableMultiWriter private static final Logger logger = LoggerFactory.getLogger(SSTableZeroCopyWriter.class); private volatile SSTableReader finalReader; - private final Map componentWriters; // indexed by component name + private final Map componentWriters; // indexed by component name public SSTableZeroCopyWriter(Builder builder, LifecycleNewTracker lifecycleNewTracker, @@ -89,12 +92,12 @@ public class SSTableZeroCopyWriter extends SSTable implements SSTableMultiWriter throw new UnsupportedOperationException(); } - private SequentialWriter makeWriter(Descriptor descriptor, Component component) + private ZeroCopySequentialWriter makeWriter(Descriptor descriptor, Component component) { - return new SequentialWriter(descriptor.fileFor(component), ioOptions.writerOptions, false); + return new ZeroCopySequentialWriter(descriptor.fileFor(component), ioOptions.writerOptions, false); } - private void write(DataInputPlus in, long size, SequentialWriter out) throws FSWriteError + private void write(DataInputPlus in, long size, ZeroCopySequentialWriter out) throws FSWriteError { final int BUFFER_SIZE = 1 << 20; long bytesRead = 0; @@ -128,7 +131,7 @@ public class SSTableZeroCopyWriter extends SSTable implements SSTableMultiWriter { setOpenResult(openResult); - for (SequentialWriter writer : componentWriters.values()) + for (ZeroCopySequentialWriter writer : componentWriters.values()) writer.finish(); return finished(); @@ -170,7 +173,7 @@ public class SSTableZeroCopyWriter extends SSTable implements SSTableMultiWriter @Override public Throwable commit(Throwable accumulate) { - for (SequentialWriter writer : componentWriters.values()) + for (ZeroCopySequentialWriter writer : componentWriters.values()) accumulate = writer.commit(accumulate); return accumulate; } @@ -178,7 +181,7 @@ public class SSTableZeroCopyWriter extends SSTable implements SSTableMultiWriter @Override public Throwable abort(Throwable accumulate) { - for (SequentialWriter writer : componentWriters.values()) + for (ZeroCopySequentialWriter writer : componentWriters.values()) accumulate = writer.abort(accumulate); return accumulate; } @@ -186,29 +189,30 @@ public class SSTableZeroCopyWriter extends SSTable implements SSTableMultiWriter @Override public void prepareToCommit() { - for (SequentialWriter writer : componentWriters.values()) + for (ZeroCopySequentialWriter writer : componentWriters.values()) writer.prepareToCommit(); } @Override public void close() { - for (SequentialWriter writer : componentWriters.values()) + for (ZeroCopySequentialWriter writer : componentWriters.values()) writer.close(); } public void writeComponent(Component component, DataInputPlus in, long size) throws ClosedChannelException { - SequentialWriter writer = componentWriters.get(component.name); + ZeroCopySequentialWriter writer = componentWriters.get(component.name); logger.info("Writing component {} to {} length {}", component, writer.getPath(), prettyPrintMemory(size)); if (in instanceof AsyncStreamingInputPlus) write((AsyncStreamingInputPlus) in, size, writer); else + // this code path is not valid for production and only exists to simplify unit tests write(in, size, writer); } - private void write(AsyncStreamingInputPlus in, long size, SequentialWriter writer) throws ClosedChannelException + private void write(AsyncStreamingInputPlus in, long size, ZeroCopySequentialWriter writer) throws ClosedChannelException { logger.info("Block Writing component to {} length {}", writer.getPath(), prettyPrintMemory(size)); @@ -233,4 +237,25 @@ public class SSTableZeroCopyWriter extends SSTable implements SSTableMultiWriter throw new FSWriteError(e, writer.getPath()); } } + + private static class ZeroCopySequentialWriter extends SequentialWriter + { + private ZeroCopySequentialWriter(File file, SequentialWriterOption option, boolean strictFlushing) + { + super(file, ByteBufferUtil.EMPTY_BYTE_BUFFER, option, strictFlushing); + } + + /** + * In production, we do not expect this method to be called, as only writeDirectlyToChannel should be invoked for zero-copy. + *

+ * This method only exists for tests. + */ + @Override + public void write(byte[] b, int off, int len) throws IOException + { + if (this.buffer == ByteBufferUtil.EMPTY_BYTE_BUFFER) + this.buffer = option.allocateBuffer(); + super.write(b, off, len); + } + } } diff --git a/src/java/org/apache/cassandra/io/util/SequentialWriter.java b/src/java/org/apache/cassandra/io/util/SequentialWriter.java index 69643be987..c3a90732ee 100644 --- a/src/java/org/apache/cassandra/io/util/SequentialWriter.java +++ b/src/java/org/apache/cassandra/io/util/SequentialWriter.java @@ -56,7 +56,7 @@ public class SequentialWriter extends BufferedDataOutputStreamPlus implements Tr // whether to do trickling fsync() to avoid sudden bursts of dirty buffer flushing by kernel causing read // latency spikes - private final SequentialWriterOption option; + protected final SequentialWriterOption option; private int bytesSinceTrickleFsync = 0; protected long lastFlushOffset; @@ -163,7 +163,12 @@ public class SequentialWriter extends BufferedDataOutputStreamPlus implements Tr */ public SequentialWriter(File file, SequentialWriterOption option, boolean strictFlushing) { - super(openChannel(file), option.allocateBuffer()); + this(file, option.allocateBuffer(), option, strictFlushing); + } + + protected SequentialWriter(File file, ByteBuffer buffer, SequentialWriterOption option, boolean strictFlushing) + { + super(openChannel(file), buffer); this.strictFlushing = strictFlushing; this.fchannel = (FileChannel)channel;