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

Merge branch 'cassandra-4.1' into cassandra-5.0

This commit is contained in:
Yifan Cai 2023-11-30 17:45:00 +08:00
parent 806a395b9c 350e968565
commit 9dc0378c2f
7 changed files with 224 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -19,6 +19,7 @@
* Remove deprecated code in Cassandra 1.x and 2.x (CASSANDRA-18959)
* ClientRequestSize metrics should not treat CONTAINS restrictions as being equality-based (CASSANDRA-18896)
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)

77
src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
View File

@ -68,6 +68,8 @@ import org.apache.cassandra.service.ClientState;
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;
@ -374,19 +376,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<CreateTypeStatement.Raw> 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<>();
}
@ -493,21 +498,21 @@ public class CQLSSTableWriter implements Closeable
}
/**
* The size of the buffer to use.
* <p>
* 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.
* <p>
* 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;
}
@ -515,7 +520,27 @@ public class CQLSSTableWriter implements Closeable
* The size of the buffer to use.
* <p>
* 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.
* <p>
* 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.
* <p>
* 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.
* <p>
* The default is 128MiB, which should be reasonable for a 1GiB heap. If you experience
@ -566,6 +591,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)
{
@ -593,8 +625,8 @@ public class CQLSSTableWriter implements Closeable
TableMetadataRef ref = TableMetadataRef.forOfflineTools(tableMetadata);
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);
@ -603,6 +635,11 @@ public class CQLSSTableWriter implements Closeable
}
}
private boolean isMaxSSTableSizeUnset()
{
return maxSSTableSizeInMiB <= 0;
}
private Types createTypes(String keyspace)
{
Types.RawBuilder builder = Types.rawBuilder(keyspace);

11
src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java
View File

@ -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<Buffer> 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)

88
src/java/org/apache/cassandra/io/sstable/SSTableSimpleWriter.java
View File

@ -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.
* <p>
* 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);
}

11
src/java/org/apache/cassandra/io/sstable/SSTableTxnWriter.java
View File

@ -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));

76
test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
View File

@ -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.io.util.File;
import org.junit.Before;
import org.junit.BeforeClass;
@ -58,6 +59,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;
@ -1233,6 +1235,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<UntypedResultSet.Row> 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<UntypedResultSet.Row> 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()

2
tools/stress/src/org/apache/cassandra/io/sstable/StressCQLSSTableWriter.java
View File

@ -586,7 +586,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)