From 50273d98e4780b57da37400752eab69e65cd41bc Mon Sep 17 00:00:00 2001 From: Yifan Cai Date: Tue, 28 Nov 2023 15:54:39 +0800 Subject: [PATCH] Support max SSTable size in sorted CQLSSTableWriter patch by Yifan Cai; reviewed by Alex Petrov, Francisco Guerrero, Maxwell Guo for CASSANDRA-18941 --- CHANGES.txt | 1 + .../io/sstable/CQLSSTableWriter.java | 53 +++++++++-- .../io/sstable/SSTableMultiWriter.java | 1 + .../sstable/SSTableSimpleUnsortedWriter.java | 13 +-- .../io/sstable/SSTableSimpleWriter.java | 90 +++++++++++++++---- .../io/sstable/SSTableTxnWriter.java | 11 +++ .../io/sstable/SimpleSSTableMultiWriter.java | 6 ++ .../format/RangeAwareSSTableWriter.java | 6 ++ .../format/big/BigTableZeroCopyWriter.java | 8 +- .../io/sstable/CQLSSTableWriterTest.java | 77 +++++++++++++++- .../io/sstable/StressCQLSSTableWriter.java | 2 +- 11 files changed, 234 insertions(+), 34 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index d33268f863..f79af3a59b 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.0.12 + * Support max SSTable size in sorted CQLSSTableWriter (CASSANDRA-18941) * Fix nodetool repair_admin summarize-pending command to not throw exception (CASSANDRA-19014) * Fix cassandra-stress in simplenative mode with prepared statements (CASSANDRA-18744) * Fix filtering system ks sstables for relocation on startup (CASSANDRA-18963) diff --git a/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java b/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java index d2c437b00b..4c299d071e 100644 --- a/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java @@ -54,6 +54,8 @@ import org.apache.cassandra.schema.*; import org.apache.cassandra.service.ClientState; import org.apache.cassandra.transport.ProtocolVersion; import org.apache.cassandra.utils.ByteBufferUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * Utility to write SSTables. @@ -342,17 +344,20 @@ public class CQLSSTableWriter implements Closeable */ public static class Builder { - private File directory; + private static final Logger logger = LoggerFactory.getLogger(Builder.class); + private static final long DEFAULT_BUFFER_SIZE_IN_MIB_FOR_UNSORTED = 128L; + protected SSTableFormat.Type formatType = null; - private CreateTableStatement.Raw schemaStatement; private final List typeStatements; + private File directory; + private CreateTableStatement.Raw schemaStatement; private ModificationStatement.Parsed insertStatement; private IPartitioner partitioner; private boolean sorted = false; - private long bufferSizeInMB = 128; + private long maxSSTableSizeInMiB = -1L; protected Builder() { this.typeStatements = new ArrayList<>(); @@ -458,23 +463,43 @@ public class CQLSSTableWriter implements Closeable return this; } + /** + * Defines the maximum SSTable size in mebibytes when using the sorted writer. + * By default, i.e. not specified, there is no maximum size limit for the produced SSTable + * + * @param size the maximum sizein mebibytes of each individual SSTable allowed + * @return this builder + */ + public Builder withMaxSSTableSizeInMiB(int size) + { + if (size <= 0) + { + logger.warn("A non-positive value for maximum SSTable size is specified, " + + "which disables the size limiting effectively. Please supply a positive value in order " + + "to enforce size limiting for the produced SSTables."); + } + this.maxSSTableSizeInMiB = size; + return this; + } + /** * The size of the buffer to use. *

* This defines how much data will be buffered before being written as - * a new SSTable. This correspond roughly to the data size that will have the created + * a new SSTable. This corresponds roughly to the data size that will have the created * sstable. *

* The default is 128MB, which should be reasonable for a 1GB heap. If you experience * OOM while using the writer, you should lower this value. * + * @deprecated This method is deprecated in favor of the new withMaxSSTableSizeInMiB(int size) * @param size the size to use in MB. * @return this builder. */ + @Deprecated public Builder withBufferSizeInMB(int size) { - this.bufferSizeInMB = size; - return this; + return withMaxSSTableSizeInMiB(size); } /** @@ -510,6 +535,13 @@ public class CQLSSTableWriter implements Closeable if (insertStatement == null) throw new IllegalStateException("No insert statement specified, you should provide an insert statement through using()"); + + // Assign the default max SSTable size if not defined in builder + if (isMaxSSTableSizeUnset()) + { + maxSSTableSizeInMiB = sorted ? -1L : DEFAULT_BUFFER_SIZE_IN_MIB_FOR_UNSORTED; + } + synchronized (Schema.instance) { if (Schema.instance.getKeyspaceMetadata(SchemaConstants.SCHEMA_KEYSPACE_NAME) == null) @@ -543,8 +575,8 @@ public class CQLSSTableWriter implements Closeable TableMetadataRef ref = TableMetadataRef.forOfflineTools(tableMetadata); AbstractSSTableSimpleWriter writer = sorted - ? new SSTableSimpleWriter(directory, ref, preparedInsert.updatedColumns()) - : new SSTableSimpleUnsortedWriter(directory, ref, preparedInsert.updatedColumns(), bufferSizeInMB); + ? new SSTableSimpleWriter(directory, ref, preparedInsert.updatedColumns(), maxSSTableSizeInMiB) + : new SSTableSimpleUnsortedWriter(directory, ref, preparedInsert.updatedColumns(), maxSSTableSizeInMiB); if (formatType != null) writer.setSSTableFormatType(formatType); @@ -553,6 +585,11 @@ public class CQLSSTableWriter implements Closeable } } + private boolean isMaxSSTableSizeUnset() + { + return maxSSTableSizeInMiB <= 0; + } + private Types createTypes(String keyspace) { Types.RawBuilder builder = Types.rawBuilder(keyspace); diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableMultiWriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableMultiWriter.java index 1be79abf8b..bb3ddc6507 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableMultiWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableMultiWriter.java @@ -44,6 +44,7 @@ public interface SSTableMultiWriter extends Transactional String getFilename(); long getFilePointer(); + long getOnDiskBytesWritten(); TableId getTableId(); static void abortOrDie(SSTableMultiWriter writer) diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java index 7ac2ebc14f..87e87e95c2 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java @@ -52,25 +52,26 @@ class SSTableSimpleUnsortedWriter extends AbstractSSTableSimpleWriter private static final Buffer SENTINEL = new Buffer(); private Buffer buffer = new Buffer(); - private final long bufferSize; + private final long maxSSTableSizeInBytes; private long currentSize; // Used to compute the row serialized size private final SerializationHeader header; private final SerializationHelper helper; - private final BlockingQueue writeQueue = new SynchronousQueue(); + private final BlockingQueue writeQueue = new SynchronousQueue<>(); private final DiskWriter diskWriter = new DiskWriter(); - SSTableSimpleUnsortedWriter(File directory, TableMetadataRef metadata, RegularAndStaticColumns columns, long bufferSizeInMB) + SSTableSimpleUnsortedWriter(File directory, TableMetadataRef metadata, RegularAndStaticColumns columns, long maxSSTableSizeInMiB) { super(directory, metadata, columns); - this.bufferSize = bufferSizeInMB * 1024L * 1024L; + this.maxSSTableSizeInBytes = maxSSTableSizeInMiB * 1024L * 1024L; this.header = new SerializationHeader(true, metadata.get(), columns, EncodingStats.NO_STATS); this.helper = new SerializationHelper(this.header); diskWriter.start(); } + @Override PartitionUpdate.Builder getUpdateFor(DecoratedKey key) { assert key != null; @@ -91,7 +92,7 @@ class SSTableSimpleUnsortedWriter extends AbstractSSTableSimpleWriter // Note that the accounting of a row is a bit inaccurate (it doesn't take some of the file format optimization into account) // and the maintaining of the bufferSize is in general not perfect. This has always been the case for this class but we should // improve that. In particular, what we count is closer to the serialized value, but it's debatable that it's the right thing - // to count since it will take a lot more space in memory and the bufferSize if first and foremost used to avoid OOM when + // to count since it will take a lot more space in memory and the bufferSize is first and foremost used to avoid OOM when // using this writer. currentSize += UnfilteredSerializer.serializer.serializedSize(row, helper, 0, formatType.info.getLatestVersion().correspondingMessagingVersion()); } @@ -100,7 +101,7 @@ class SSTableSimpleUnsortedWriter extends AbstractSSTableSimpleWriter { try { - if (currentSize > bufferSize) + if (currentSize > maxSSTableSizeInBytes) sync(); } catch (IOException e) diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableSimpleWriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableSimpleWriter.java index 530a03b776..1cbff412ad 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableSimpleWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableSimpleWriter.java @@ -20,6 +20,7 @@ package org.apache.cassandra.io.sstable; import java.io.File; import java.io.IOException; +import com.google.common.base.Preconditions; import com.google.common.base.Throwables; import org.apache.cassandra.db.*; @@ -35,57 +36,114 @@ import org.apache.cassandra.schema.TableMetadataRef; * means that rows should be added by increasing md5 of the row key. This is * rarely possible and SSTableSimpleUnsortedWriter should most of the time be * prefered. + *

+ * Optionally, the writer can be configured with a max SSTable size for SSTables. + * The output will be a series of SSTables that do not exceed a specified size. + * By default, all sorted data are written into a single SSTable. */ class SSTableSimpleWriter extends AbstractSSTableSimpleWriter { + private final long maxSSTableSizeInBytes; + protected DecoratedKey currentKey; protected PartitionUpdate.Builder update; private SSTableTxnWriter writer; - protected SSTableSimpleWriter(File directory, TableMetadataRef metadata, RegularAndStaticColumns columns) + /** + * Create a SSTable writer for sorted input data. + * When a positive {@param maxSSTableSizeInMiB} is defined, the writer outputs a sequence of SSTables, + * whose sizes do not exceed the specified value. + * + * @param directory directory to store the sstable files + * @param metadata table metadata + * @param columns columns to update + * @param maxSSTableSizeInMiB defines the max SSTable size if the value is positive. + * Any non-positive value indicates the sstable size is unlimited. + */ + protected SSTableSimpleWriter(File directory, TableMetadataRef metadata, RegularAndStaticColumns columns, long maxSSTableSizeInMiB) { super(directory, metadata, columns); + this.maxSSTableSizeInBytes = maxSSTableSizeInMiB * 1024L * 1024L; } - private SSTableTxnWriter getOrCreateWriter() - { - if (writer == null) - writer = createWriter(); - - return writer; - } - + @Override PartitionUpdate.Builder getUpdateFor(DecoratedKey key) throws IOException { - assert key != null; + Preconditions.checkArgument(key != null, "Partition update cannot have null key"); - // If that's not the current key, write the current one if necessary and create a new - // update for the new key. - if (!key.equals(currentKey)) + // update for the first partition or a new partition + if (update == null || !key.equals(currentKey)) { + // write the previous update if not absent if (update != null) - writePartition(update.build()); + writePartition(update.build()); // might switch to a new sstable writer and reset currentSize + currentKey = key; update = new PartitionUpdate.Builder(metadata.get(), currentKey, columns, 4); } - assert update != null; + Preconditions.checkState(update != null, "Partition update to write cannot be null"); return update; } + @Override public void close() + { + writeLastPartitionUpdate(update); + maybeCloseWriter(writer); + } + + /** + * Switch to a new writer when writer is absent or the file size has exceeded the configured max + */ + private boolean shouldSwitchToNewWriter() + { + return writer == null || (maxSSTableSizeInBytes > 0 && writer.getOnDiskBytesWritten() > maxSSTableSizeInBytes); + } + + /** + * Get the current writer, or create a new writer if needed, e.g. writer does not exist + * or writer has reached to the configured max size. + */ + private SSTableTxnWriter getOrCreateWriter() throws IOException + { + if (shouldSwitchToNewWriter()) + { + maybeCloseWriter(writer); + writer = createWriter(); + } + + return writer; + } + + private void writeLastPartitionUpdate(PartitionUpdate.Builder update) { try { if (update != null) writePartition(update.build()); + } + catch (Throwable t) + { + Throwable e = writer == null ? t : writer.abort(t); + Throwables.throwIfUnchecked(e); + throw new RuntimeException(e); + } + } + + private void maybeCloseWriter(SSTableTxnWriter writer) + { + try + { if (writer != null) writer.finish(false); } catch (Throwable t) { - throw Throwables.propagate(writer == null ? t : writer.abort(t)); + Throwable e = writer.abort(t); + Throwables.throwIfUnchecked(e); + throw new RuntimeException(e); } } diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableTxnWriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableTxnWriter.java index cfb1365649..1a4581191c 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableTxnWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableTxnWriter.java @@ -67,6 +67,17 @@ public class SSTableTxnWriter extends Transactional.AbstractTransactional implem return writer.getFilePointer(); } + /** + * Get the amount of data written to disk. Unlike {@link #getFilePointer()}, which returns the position in the + * _uncompressed_ data, this method returns the actual file pointer position of the on disk file. + * + * @return the amount of data already written to disk + */ + public long getOnDiskBytesWritten() + { + return writer.getOnDiskBytesWritten(); + } + protected Throwable doCommit(Throwable accumulate) { return writer.commit(txn.commit(accumulate)); diff --git a/src/java/org/apache/cassandra/io/sstable/SimpleSSTableMultiWriter.java b/src/java/org/apache/cassandra/io/sstable/SimpleSSTableMultiWriter.java index a84f07e949..7dc3bb3a7f 100644 --- a/src/java/org/apache/cassandra/io/sstable/SimpleSSTableMultiWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/SimpleSSTableMultiWriter.java @@ -80,6 +80,12 @@ public class SimpleSSTableMultiWriter implements SSTableMultiWriter return writer.getFilePointer(); } + @Override + public long getOnDiskBytesWritten() + { + return writer.getEstimatedOnDiskBytesWritten(); + } + public TableId getTableId() { return writer.metadata().id; diff --git a/src/java/org/apache/cassandra/io/sstable/format/RangeAwareSSTableWriter.java b/src/java/org/apache/cassandra/io/sstable/format/RangeAwareSSTableWriter.java index 43b1957494..ea3f9594e4 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/RangeAwareSSTableWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/format/RangeAwareSSTableWriter.java @@ -164,6 +164,12 @@ public class RangeAwareSSTableWriter implements SSTableMultiWriter return currentWriter != null ? currentWriter.getFilePointer() : 0L; } + @Override + public long getOnDiskBytesWritten() + { + return currentWriter != null ? currentWriter.getOnDiskBytesWritten() : 0L; + } + @Override public TableId getTableId() { diff --git a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableZeroCopyWriter.java b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableZeroCopyWriter.java index 2564e966d4..b40cd68dbb 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableZeroCopyWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableZeroCopyWriter.java @@ -166,6 +166,12 @@ public class BigTableZeroCopyWriter extends SSTable implements SSTableMultiWrite return 0; } + @Override + public long getOnDiskBytesWritten() + { + return 0; + } + @Override public TableId getTableId() { @@ -231,4 +237,4 @@ public class BigTableZeroCopyWriter extends SSTable implements SSTableMultiWrite throw new FSWriteError(e, writer.getPath()); } } -} \ No newline at end of file +} diff --git a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java index f43294c8ec..13b6e2d927 100644 --- a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java @@ -47,13 +47,11 @@ import org.apache.cassandra.cql3.functions.UDHelper; import org.apache.cassandra.cql3.functions.types.*; import org.apache.cassandra.db.Keyspace; import org.apache.cassandra.db.commitlog.CommitLog; -import org.apache.cassandra.db.rows.Row; import org.apache.cassandra.db.marshal.UTF8Type; import org.apache.cassandra.dht.*; import org.apache.cassandra.exceptions.*; import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.TableMetadataRef; -import org.apache.cassandra.serializers.SimpleDateSerializer; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.transport.ProtocolVersion; import org.apache.cassandra.utils.*; @@ -62,6 +60,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import static org.psjava.util.AssertStatus.assertNotNull; public class CQLSSTableWriterTest { @@ -846,6 +845,80 @@ public class CQLSSTableWriterTest assertEquals(0, filtered.size()); } + @Test + public void testWriteWithSorted() throws Exception + { + String schema = "CREATE TABLE " + qualifiedTable + " (" + + " k int PRIMARY KEY," + + " v blob )"; + CQLSSTableWriter writer = CQLSSTableWriter.builder() + .inDirectory(dataDir) + .forTable(schema) + .using("INSERT INTO " + qualifiedTable + + " (k, v) VALUES (?, textAsBlob(?))" ) + .sorted() + .build(); + int rowCount = 10_000; + for (int i = 0; i < rowCount; i++) + { + writer.addRow(i, UUID.randomUUID().toString()); + } + writer.close(); + loadSSTables(dataDir, keyspace); + + UntypedResultSet resultSet = QueryProcessor.executeInternal("SELECT * FROM " + qualifiedTable); + assertEquals(rowCount, resultSet.size()); + Iterator iter = resultSet.iterator(); + for (int i = 0; i < rowCount; i++) + { + UntypedResultSet.Row row = iter.next(); + assertEquals(i, row.getInt("k")); + } + } + + @Test + public void testWriteWithSortedAndMaxSize() throws Exception + { + String schema = "CREATE TABLE " + qualifiedTable + " (" + + " k int PRIMARY KEY," + + " v blob )"; + CQLSSTableWriter writer = CQLSSTableWriter.builder() + .inDirectory(dataDir) + .forTable(schema) + .using("INSERT INTO " + qualifiedTable + + " (k, v) VALUES (?, textAsBlob(?))") + .sorted() + .withMaxSSTableSizeInMiB(1) + .build(); + int rowCount = 30_000; + // Max SSTable size is 1 MiB + // 30_000 rows should take 30_000 * (4 + 37) = 1.17 MiB > 1 MiB + for (int i = 0; i < rowCount; i++) + { + writer.addRow(i, UUID.randomUUID().toString()); + } + writer.close(); + + File[] dataFiles = dataDir.listFiles(f -> f.getName().endsWith(Component.DATA.type.repr)); + assertNotNull(dataFiles); + assertEquals("The sorted writer should produce 2 sstables when max sstable size is configured", + 2, dataFiles.length); + long closeTo1MiBFileSize = Math.max(dataFiles[0].length(), dataFiles[1].length()); + assertTrue("The file size should be close to 1MiB (with at most 50KiB error rate for the test)", + Math.abs(1024 * 1024 - closeTo1MiBFileSize) < 50 * 1024); + + loadSSTables(dataDir, keyspace); + + UntypedResultSet resultSet = QueryProcessor.executeInternal("SELECT * FROM " + qualifiedTable); + assertEquals(rowCount, resultSet.size()); + Iterator iter = resultSet.iterator(); + for (int i = 0; i < rowCount; i++) + { + UntypedResultSet.Row row = iter.next(); + assertEquals(i, row.getInt("k")); + } + } + private static void loadSSTables(File dataDir, String ks) throws ExecutionException, InterruptedException { SSTableLoader loader = new SSTableLoader(dataDir, new SSTableLoader.Client() diff --git a/tools/stress/src/org/apache/cassandra/io/sstable/StressCQLSSTableWriter.java b/tools/stress/src/org/apache/cassandra/io/sstable/StressCQLSSTableWriter.java index bc6756b498..d7f3377f36 100644 --- a/tools/stress/src/org/apache/cassandra/io/sstable/StressCQLSSTableWriter.java +++ b/tools/stress/src/org/apache/cassandra/io/sstable/StressCQLSSTableWriter.java @@ -554,7 +554,7 @@ public class StressCQLSSTableWriter implements Closeable UpdateStatement preparedInsert = prepareInsert(); AbstractSSTableSimpleWriter writer = sorted - ? new SSTableSimpleWriter(cfs.getDirectories().getDirectoryForNewSSTables(), cfs.metadata, preparedInsert.updatedColumns()) + ? new SSTableSimpleWriter(cfs.getDirectories().getDirectoryForNewSSTables(), cfs.metadata, preparedInsert.updatedColumns(), -1) : new SSTableSimpleUnsortedWriter(cfs.getDirectories().getDirectoryForNewSSTables(), cfs.metadata, preparedInsert.updatedColumns(), bufferSizeInMB); if (formatType != null)