diff --git a/CHANGES.txt b/CHANGES.txt index ecc330bf8b..d1488c7c58 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -52,6 +52,7 @@ Merged from 4.1: * Internode legacy SSL storage port certificate is not hot reloaded on update (CASSANDRA-18681) * Nodetool paxos-only repair is no longer incremental (CASSANDRA-18466) Merged from 4.0: + * 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 0d804bf4f2..60e2179c0f 100644 --- a/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java @@ -69,6 +69,8 @@ import org.apache.cassandra.tcm.ClusterMetadataService; import org.apache.cassandra.transport.ProtocolVersion; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.JavaDriverUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis; @@ -376,19 +378,22 @@ 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 format = null; - private CreateTableStatement.Raw schemaStatement; private final List typeStatements; + + private File directory; + private CreateTableStatement.Raw schemaStatement; private ModificationStatement.Parsed modificationStatement; private IPartitioner partitioner; - private boolean sorted = false; - private long bufferSizeInMiB = 128; + private long maxSSTableSizeInMiB = -1L; - protected Builder() { + protected Builder() + { this.typeStatements = new ArrayList<>(); } @@ -495,21 +500,21 @@ public class CQLSSTableWriter implements Closeable } /** - * 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 - * sstable. - *

- * The default is 128MiB, which should be reasonable for a 1GiB heap. If you experience - * OOM while using the writer, you should lower this value. + * 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 size to use in MiB. - * @return this builder. + * @param size the maximum sizein mebibytes of each individual SSTable allowed + * @return this builder */ - public Builder withBufferSizeInMiB(int size) + public Builder withMaxSSTableSizeInMiB(int size) { - this.bufferSizeInMiB = 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; } @@ -517,7 +522,27 @@ public class CQLSSTableWriter implements Closeable * 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 128MiB, which should be reasonable for a 1GiB 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 MiB. + * @return this builder. + */ + @Deprecated(since = "5.0") + public Builder withBufferSizeInMiB(int size) + { + return withMaxSSTableSizeInMiB(size); + } + + /** + * The size of the buffer to use. + *

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

* The default is 128MiB, which should be reasonable for a 1GiB heap. If you experience @@ -568,6 +593,13 @@ public class CQLSSTableWriter implements Closeable Preconditions.checkState(Sets.difference(SchemaConstants.LOCAL_SYSTEM_KEYSPACE_NAMES, Schema.instance.getKeyspaces()).isEmpty(), "Local keyspaces were not loaded. If this is running as a client, please make sure to add %s=true system property.", CassandraRelevantProperties.FORCE_LOAD_LOCAL_KEYSPACES.getKey()); + + // Assign the default max SSTable size if not defined in builder + if (isMaxSSTableSizeUnset()) + { + maxSSTableSizeInMiB = sorted ? -1L : DEFAULT_BUFFER_SIZE_IN_MIB_FOR_UNSORTED; + } + synchronized (CQLSSTableWriter.class) { String keyspaceName = schemaStatement.keyspace(); @@ -599,8 +631,8 @@ public class CQLSSTableWriter implements Closeable TableMetadataRef ref = tableMetadata.ref; AbstractSSTableSimpleWriter writer = sorted - ? new SSTableSimpleWriter(directory, ref, preparedModificationStatement.updatedColumns()) - : new SSTableSimpleUnsortedWriter(directory, ref, preparedModificationStatement.updatedColumns(), bufferSizeInMiB); + ? new SSTableSimpleWriter(directory, ref, preparedModificationStatement.updatedColumns(), maxSSTableSizeInMiB) + : new SSTableSimpleUnsortedWriter(directory, ref, preparedModificationStatement.updatedColumns(), maxSSTableSizeInMiB); if (format != null) writer.setSSTableFormatType(format); @@ -609,6 +641,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/SSTableSimpleUnsortedWriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java index 08515747d0..ce927d9947 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java @@ -56,7 +56,7 @@ 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 @@ -66,15 +66,16 @@ class SSTableSimpleUnsortedWriter extends AbstractSSTableSimpleWriter private final BlockingQueue writeQueue = newBlockingQueue(0); 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; @@ -95,7 +96,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, format.getLatestVersion().correspondingMessagingVersion()); } @@ -104,7 +105,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 744f35a41e..ce423f2ec5 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableSimpleWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableSimpleWriter.java @@ -19,6 +19,7 @@ package org.apache.cassandra.io.sstable; import java.io.IOException; +import com.google.common.base.Preconditions; import com.google.common.base.Throwables; import org.apache.cassandra.db.DecoratedKey; @@ -36,57 +37,112 @@ 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() throws IOException - { - if (writer == null) - writer = createWriter(null); - - 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(null); + } + + 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) { - Throwable e = 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 76fb83e913..3b43dcfdf4 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableTxnWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableTxnWriter.java @@ -65,6 +65,17 @@ public class SSTableTxnWriter extends Transactional.AbstractTransactional implem return writer.getBytesWritten(); } + /** + * 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/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java index 6039f0aba7..ee78862f83 100644 --- a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java @@ -31,6 +31,7 @@ import java.util.stream.StreamSupport; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; +import org.apache.cassandra.io.sstable.format.big.BigFormat; import org.apache.cassandra.ServerTestUtils; import org.apache.cassandra.io.util.File; import org.junit.Before; @@ -59,6 +60,7 @@ import org.apache.cassandra.utils.*; import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -1235,6 +1237,80 @@ public class CQLSSTableWriterTest assertFalse(iter.hasNext()); } + @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 (?, text_as_blob(?))" ) + .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 (?, text_as_blob(?))" ) + .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.list(f -> f.name().endsWith(BigFormat.Components.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 c8cc0e3daf..90d71abaa7 100644 --- a/tools/stress/src/org/apache/cassandra/io/sstable/StressCQLSSTableWriter.java +++ b/tools/stress/src/org/apache/cassandra/io/sstable/StressCQLSSTableWriter.java @@ -587,7 +587,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(), bufferSizeInMiB); if (format != null)