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

Optimize memtable flush logic 

add more flushing stats: partitions/rows, bytes rate, CPU and heap allocation for the flushing thread
avoid columns filtering overheads for unfilteredIterator
do not re-map colums in serializeRowBody if they haven't changed
reduce allocations during serialization of NativeClustering
add fast return for BTreeRow.hasComplexDeletion, avoid column.name.bytes.hashCode if not needed, avoid capturing lambda allocation in UnfilteredSerializer.serializeRowBody
check if Guardrails enabled at the beginning of writing, avoid hidden auto-boxing for logging of primitive parameters
split call sites for in Cell serialize logic, make isCounterCell cheaper (avoid megamorphic call + cache isCounterColumn)
invoke metadataCollector.updateClusteringValues only for first and last clustering key in a partition
enforce inlining for MinMaxIntTracker/MinMaxLongTracker

Patch by Dmitry Konstantinov; reviewed by Branimir Lambov for CASSANDRA-21083
This commit is contained in:
Dmitry Konstantinov 2025-12-18 12:25:56 +00:00
parent aa4da838ec
commit 39d1149c74
28 changed files with 548 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
5.1
* Optimize memtable flush logic (CASSANDRA-21083)
* No need to evict already prepared statements, as it creates a race condition between multiple threads (CASSANDRA-17401)
* Include Level information for UnifiedCompactionStrategy in nodetool tablestats output (CASSANDRA-20820)
* Support low-overhead async profiling (CASSANDRA-20854)

39
src/java/org/apache/cassandra/db/ClusteringPrefix.java
View File

@ -260,6 +260,33 @@ public interface ClusteringPrefix<V> extends IMeasurableMemory, Clusterable<V>
*/
public V get(int i);
/**
* A dedicated method to write ith value of this prefix,
* it is introduced for optimization reasons to avoid retrieval (and potential allocation) of ith value object.
* For the same reason null and empty check are performed inside the method.
*/
default void writeValueSkippingNullAndEmpty(AbstractType<?> type, int i, DataOutputPlus out) throws IOException
{
V v = get(i);
if (v != null && !isEmpty(i))
type.writeValue(v, accessor(), out);
}
/**
* A dedicated method to get a written length of ith value of this prefix,
* it is introduced for optimization reasons to avoid retrieval (and potential allocation) of ith value object.
* For the same reason null and empty check are performed inside the method.
*/
default long writtenLengthSkippingNullAndEmpty(AbstractType<?> type, int i)
{
V v = get(i);
if (v == null || isEmpty(i))
return 0;
return type.writtenLength(v, accessor());
}
/**
* The method is introduced to allow to avoid a value object retrieval/allocation for simple checks
*/
@ -478,7 +505,6 @@ public interface ClusteringPrefix<V> extends IMeasurableMemory, Clusterable<V>
{
int offset = 0;
int clusteringSize = clustering.size();
ValueAccessor<V> accessor = clustering.accessor();
// serialize in batches of 32, to avoid garbage when deserializing headers
while (offset < clusteringSize)
{
@ -490,9 +516,7 @@ public interface ClusteringPrefix<V> extends IMeasurableMemory, Clusterable<V>
out.writeUnsignedVInt(makeHeader(clustering, offset, limit));
while (offset < limit)
{
V v = clustering.get(offset);
if (v != null && !accessor.isEmpty(v))
types.get(offset).writeValue(v, accessor, out);
clustering.writeValueSkippingNullAndEmpty(types.get(offset), offset, out);
offset++;
}
}
@ -509,14 +533,9 @@ public interface ClusteringPrefix<V> extends IMeasurableMemory, Clusterable<V>
result += TypeSizes.sizeofUnsignedVInt(makeHeader(clustering, offset, limit));
offset = limit;
}
ValueAccessor<V> accessor = clustering.accessor();
for (int i = 0; i < clusteringSize; i++)
{
V v = clustering.get(i);
if (v == null || accessor.isEmpty(v))
continue; // handled in the header
result += types.get(i).writtenLength(v, accessor);
result += clustering.writtenLengthSkippingNullAndEmpty(types.get(i), i);
}
return result;
}

54
src/java/org/apache/cassandra/db/NativeClustering.java
View File

@ -18,14 +18,18 @@
*/
package org.apache.cassandra.db;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.AddressBasedNativeData;
import org.apache.cassandra.db.marshal.ByteBufferAccessor;
import org.apache.cassandra.db.marshal.IndexedValueHolder;
import org.apache.cassandra.db.marshal.NativeAccessor;
import org.apache.cassandra.db.marshal.NativeData;
import org.apache.cassandra.db.marshal.ValueAccessor;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.concurrent.OpOrder;
@ -34,7 +38,7 @@ import org.apache.cassandra.utils.memory.HeapCloner;
import org.apache.cassandra.utils.memory.MemoryUtil;
import org.apache.cassandra.utils.memory.NativeEndianMemoryUtil;
public class NativeClustering implements Clustering<NativeData>
public class NativeClustering implements Clustering<NativeData>, IndexedValueHolder<NativeData>
{
private static final long EMPTY_SIZE = ObjectSizes.measure(new NativeClustering());
@ -118,11 +122,59 @@ public class NativeClustering implements Clustering<NativeData>
return buildDataObject(i, AddressBasedNativeData::new);
}
public void writeValueSkippingNullAndEmpty(AbstractType<?> type, int i, DataOutputPlus out) throws IOException
{
if (!isEmpty(i)) // is null is checked as a part of isEmpty
type.writeValue(this, i, NativeAccessor.instance, out);
}
public long writtenLengthSkippingNullAndEmpty(AbstractType<?> type, int i)
{
if (isEmpty(i)) // is null is checked as a part of isEmpty
return 0;
return type.writtenLength(this, i, NativeAccessor.instance);
}
@Override
public int size(int i)
{
int size = size();
if (isNull(peer, size, i))
return 0;
int startOffset = NativeEndianMemoryUtil.getUnsignedShort(peer + 2 + i * 2);
int endOffset = NativeEndianMemoryUtil.getUnsignedShort(peer + 4 + i * 2);
return (endOffset - startOffset);
}
public boolean isNull(int i)
{
return isNull(peer, size(), i);
}
@Override
public void write(int i, DataOutputPlus out) throws IOException
{
int size = size();
if (i >= size)
throw new IndexOutOfBoundsException();
int metadataSize = (size * 2) + 4;
int bitmapSize = ((size + 7) >>> 3);
long bitmapStart = peer + metadataSize;
int b = NativeEndianMemoryUtil.getByte(bitmapStart + (i >>> 3));
if ((b & (1 << (i & 7))) != 0)
return;
int startOffset = NativeEndianMemoryUtil.getUnsignedShort(peer + 2 + i * 2);
int endOffset = NativeEndianMemoryUtil.getUnsignedShort(peer + 4 + i * 2);
long address = bitmapStart + bitmapSize + startOffset;
int length = endOffset - startOffset;
out.writeMemory(address, length);
}
private static boolean isNull(long peer, int size, int i)
{
if (i >= size)

37
src/java/org/apache/cassandra/db/SerializationHeader.java
View File

@ -63,12 +63,25 @@ public class SerializationHeader
private final Map<ByteBuffer, AbstractType<?>> typeMap;
private final boolean columnsMayChanged;
private SerializationHeader(boolean isForSSTable,
AbstractType<?> keyType,
List<AbstractType<?>> clusteringTypes,
RegularAndStaticColumns columns,
EncodingStats stats,
Map<ByteBuffer, AbstractType<?>> typeMap)
{
this(isForSSTable, keyType, clusteringTypes, columns, stats, typeMap, true);
}
private SerializationHeader(boolean isForSSTable,
AbstractType<?> keyType,
List<AbstractType<?>> clusteringTypes,
RegularAndStaticColumns columns,
EncodingStats stats,
Map<ByteBuffer, AbstractType<?>> typeMap,
boolean columnsMayChanged)
{
this.isForSSTable = isForSSTable;
this.keyType = keyType;
@ -76,6 +89,7 @@ public class SerializationHeader
this.columns = columns;
this.stats = stats;
this.typeMap = typeMap;
this.columnsMayChanged = columnsMayChanged;
}
public static SerializationHeader makeWithoutStats(TableMetadata metadata)
@ -118,6 +132,21 @@ public class SerializationHeader
return readers;
}
public SerializationHeader(boolean isForSSTable,
TableMetadata metadata,
RegularAndStaticColumns columns,
EncodingStats stats,
boolean columnsMayChanged)
{
this(isForSSTable,
metadata.partitionKeyType,
metadata.comparator.subtypes(),
columns,
stats,
null,
columnsMayChanged);
}
public SerializationHeader(boolean isForSSTable,
TableMetadata metadata,
RegularAndStaticColumns columns,
@ -128,7 +157,8 @@ public class SerializationHeader
metadata.comparator.subtypes(),
columns,
stats,
null);
null,
true);
}
public RegularAndStaticColumns columns()
@ -146,6 +176,11 @@ public class SerializationHeader
return isForSSTable;
}
public boolean columnsMayChanged()
{
return columnsMayChanged;
}
public EncodingStats stats()
{
return stats;

8
src/java/org/apache/cassandra/db/compaction/unified/ShardedMultiWriter.java
View File

@ -205,6 +205,14 @@ public class ShardedMultiWriter implements SSTableMultiWriter
return bytesWritten;
}
public long getTotalRows()
{
long totalRows = 0;
for (int i = 0; i <= currentWriter; ++i)
totalRows += writers[i].getTotalRows();
return totalRows;
}
@Override
public TableId getTableId()
{

27
src/java/org/apache/cassandra/db/marshal/AbstractType.java
View File

@ -586,6 +586,25 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>, Assignm
}
}
public <V> void writeValue(IndexedValueHolder<V> valueHolder, int i, ValueAccessor<V> accessor, DataOutputPlus out) throws IOException
{
assert !valueHolder.isNull(i) : "bytes should not be null for type " + this;
int expectedValueLength = valueLengthIfFixed();
if (expectedValueLength >= 0)
{
int actualValueLength = valueHolder.size(i);
if (actualValueLength == expectedValueLength)
accessor.write(valueHolder, i, out);
else
throw new IOException(String.format("Expected exactly %d bytes, but was %d",
expectedValueLength, actualValueLength));
}
else
{
accessor.writeWithVIntLength(valueHolder, i, out);
}
}
public long writtenLength(ByteBuffer value)
{
return writtenLength(value, ByteBufferAccessor.instance);
@ -599,6 +618,14 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>, Assignm
: accessor.sizeWithVIntLength(value);
}
public <V> long writtenLength(IndexedValueHolder<V> valueHolder, int i, ValueAccessor<V> accessor)
{
assert !valueHolder.isNull(i) : "bytes should not be null for type " + this;
return valueLengthIfFixed() >= 0
? valueHolder.size(i) // if the size is wrong, this will be detected in writeValue
: accessor.sizeWithVIntLength(valueHolder, i);
}
public ByteBuffer readBuffer(DataInputPlus in) throws IOException
{
return readBuffer(in, Integer.MAX_VALUE);

32
src/java/org/apache/cassandra/db/marshal/IndexedValueHolder.java Normal file
View File

@ -0,0 +1,32 @@
/*
* 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.db.marshal;
import java.io.IOException;
import org.apache.cassandra.io.util.DataOutputPlus;
public interface IndexedValueHolder<V>
{
V get(int i);
int size(int i);
boolean isEmpty(int i);
boolean isNull(int i);
void write(int i, DataOutputPlus out) throws IOException;
}

7
src/java/org/apache/cassandra/db/marshal/NativeAccessor.java
View File

@ -70,6 +70,13 @@ public class NativeAccessor implements ValueAccessor<NativeData>
out.writeMemory(sourceValue.getAddress(), sourceValue.nativeDataSize());
}
public void writeWithVIntLength(IndexedValueHolder<NativeData> valueHolder, int i, DataOutputPlus out) throws IOException
{
int size = valueHolder.size(i);
out.writeUnsignedVInt32(size);
valueHolder.write(i, out);
}
@Override
public ByteBuffer toBuffer(NativeData value)
{

18
src/java/org/apache/cassandra/db/marshal/ValueAccessor.java
View File

@ -125,6 +125,12 @@ public interface ValueAccessor<V>
return TypeSizes.sizeofUnsignedVInt(size) + size;
}
default int sizeWithVIntLength(IndexedValueHolder<V> valueHolder, int i)
{
int size = valueHolder.size(i);
return TypeSizes.sizeofUnsignedVInt(size) + size;
}
/** serialized size including a short length prefix */
default int sizeWithShortLength(V value)
{
@ -173,12 +179,24 @@ public interface ValueAccessor<V>
*/
void write(V value, DataOutputPlus out) throws IOException;
default void write(IndexedValueHolder<V> valueHolder, int i, DataOutputPlus out) throws IOException
{
write(valueHolder.get(i), out);
}
default void writeWithVIntLength(V value, DataOutputPlus out) throws IOException
{
out.writeUnsignedVInt32(size(value));
write(value, out);
}
default void writeWithVIntLength(IndexedValueHolder<V> valueHolder, int i, DataOutputPlus out) throws IOException
{
out.writeUnsignedVInt32(valueHolder.size(i));
write(valueHolder.get(i), out);
}
/**
* Write the contents of the given value into the ByteBuffer
*/

39
src/java/org/apache/cassandra/db/memtable/Flushing.java
View File

@ -22,6 +22,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
@ -45,7 +46,9 @@ import org.apache.cassandra.io.sstable.SSTableMultiWriter;
import org.apache.cassandra.io.sstable.format.SSTableFormat;
import org.apache.cassandra.metrics.TableMetrics;
import org.apache.cassandra.service.ActiveRepairService;
import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.ThreadStats;
public class Flushing
{
@ -152,6 +155,9 @@ public class Flushing
private void writeSortedContents()
{
logger.info("Writing {}, flushed range = [{}, {})", toFlush.memtable(), toFlush.from(), toFlush.to());
long startTimeNs = Clock.Global.nanoTime();
long startCpuTime = ThreadStats.getCurrentThreadCpuTimeNano();
long startAllocatedBytes = ThreadStats.getCurrentThreadAllocatedBytes();
// (we can't clear out the map as-we-go to free up memory,
// since the memtable is being used for queries in the "pending flush" category)
@ -176,11 +182,34 @@ public class Flushing
if (logCompletion)
{
long endTimeNs = Clock.Global.nanoTime();
long endCpuTime = ThreadStats.getCurrentThreadCpuTimeNano();
long endAllocatedBytes = ThreadStats.getCurrentThreadAllocatedBytes();
long durationMs = TimeUnit.NANOSECONDS.toMillis(endTimeNs - startTimeNs);
long durationSec = TimeUnit.MILLISECONDS.toSeconds(durationMs);
durationSec = durationSec == 0 ? 1 : durationSec;
long bytesFlushed = writer.getBytesWritten();
logger.info("Completed flushing {} ({}) for commitlog position {}",
long byteFlushedPerSec = bytesFlushed / durationSec;
long partitionsPerSec = toFlush.partitionCount() / durationSec;
long rowsPerSec = writer.getTotalRows() / durationSec;
logger.info("Completed flushing {} ({}) for commitlog position {}, " +
"time spent: {} ms, " +
"bytes flushed: {} / (rate: {}), " +
"partitions flushed: {} / (rate: {}/s), " +
"rows: {} / (rate: {}/s), " +
"cpu time: {} ms, heap allocated: {}",
writer.getFilename(),
FBUtilities.prettyPrintMemory(bytesFlushed),
toFlush.memtable().getFinalCommitLogUpperBound());
toFlush.memtable().getFinalCommitLogUpperBound(),
durationMs,
bytesFlushed, FBUtilities.prettyPrintMemoryPerSecond(byteFlushedPerSec),
toFlush.partitionCount(), partitionsPerSec,
writer.getTotalRows(), rowsPerSec,
startCpuTime < 0 ? "n/a": TimeUnit.NANOSECONDS.toMillis(endCpuTime - startCpuTime),
endAllocatedBytes < 0 ? "n/a" : FBUtilities.prettyPrintMemory(endAllocatedBytes - startAllocatedBytes)
);
// Update the metrics
metrics.bytesFlushed.inc(bytesFlushed);
}
@ -207,6 +236,9 @@ public class Flushing
Descriptor descriptor,
long partitionCount)
{
// column types altering is not allowed after CASSANDRA-12443
// removal and re-adding of a column with a different type is limited to serialization compatible types only, CASSANDRA-16905
boolean columnsMayChanged = false;
return cfs.createSSTableMultiWriter(descriptor,
partitionCount,
ActiveRepairService.UNREPAIRED_SSTABLE,
@ -217,7 +249,8 @@ public class Flushing
new SerializationHeader(true,
flushSet.metadata(),
flushSet.columns(),
flushSet.encodingStats()),
flushSet.encodingStats(),
columnsMayChanged),
txn);
}
}

2
src/java/org/apache/cassandra/db/memtable/TrieMemtable.java
View File

@ -737,7 +737,7 @@ public class TrieMemtable extends AbstractShardedMemtable
@Override
public UnfilteredRowIterator unfilteredIterator()
{
return unfilteredIterator(ColumnFilter.selection(super.columns()), Slices.ALL, false);
return unfilteredIterator(ColumnFilter.all(super.columns()), Slices.ALL, false);
}
@Override

2
src/java/org/apache/cassandra/db/partitions/AbstractBTreePartition.java
View File

@ -168,7 +168,7 @@ public abstract class AbstractBTreePartition implements Partition, Iterable<Row>
public UnfilteredRowIterator unfilteredIterator()
{
return unfilteredIterator(ColumnFilter.selection(columns()), Slices.ALL, false);
return unfilteredIterator(ColumnFilter.all(columns()), Slices.ALL, false);
}
public UnfilteredRowIterator unfilteredIterator(ColumnFilter selection, Slices slices, boolean reversed)

4
src/java/org/apache/cassandra/db/rows/AbstractCell.java
View File

@ -51,12 +51,12 @@ public abstract class AbstractCell<V> extends Cell<V>
public boolean isCounterCell()
{
return !isTombstone() && column.isCounterColumn();
return column.isCounterColumn() && !isTombstone();
}
public boolean isLive(long nowInSec)
{
return localDeletionTime() == NO_DELETION_TIME || (ttl() != NO_TTL && nowInSec < localDeletionTime());
return isLive(nowInSec, localDeletionTime(), ttl());
}
public boolean isTombstone()

5
src/java/org/apache/cassandra/db/rows/BTreeRow.java
View File

@ -354,7 +354,8 @@ public class BTreeRow extends AbstractRow
if (!inclusionTester.test(column))
return null;
DroppedColumn dropped = droppedColumns.get(column.name.bytes);
// we check isEmpty here to avoid bytes.hashCode calculation if it is not needed
DroppedColumn dropped = droppedColumns.isEmpty() ? null : droppedColumns.get(column.name.bytes);
if (column.isComplex())
return ((ComplexColumnData) cd).filter(filter, mayHaveShadowed ? activeDeletion : DeletionTime.LIVE, dropped, rowLiveness);
@ -402,6 +403,8 @@ public class BTreeRow extends AbstractRow
public boolean hasComplexDeletion()
{
if (minLocalDeletionTime == Cell.MAX_DELETION_TIME || !hasComplex())
return false;
long result = accumulate((cd, v) -> ((ComplexColumnData) cd).complexDeletion().isLive() ? 0 : STOP_SENTINEL_VALUE,
COLUMN_COMPARATOR, isStatic() ? FIRST_COMPLEX_STATIC : FIRST_COMPLEX_REGULAR, 0L);
return result == STOP_SENTINEL_VALUE;

76
src/java/org/apache/cassandra/db/rows/Cell.java
View File

@ -177,6 +177,11 @@ public abstract class Cell<V> extends ColumnData
*/
public abstract boolean isLive(long nowInSec);
public final boolean isLive(long nowInSec, long localDeletionTime, int ttl)
{
return localDeletionTime == NO_DELETION_TIME || ttl != NO_TTL && nowInSec < localDeletionTime;
}
/**
* For cells belonging to complex types (non-frozen collection and UDT), the
* path to the cell.
@ -280,11 +285,64 @@ public abstract class Cell<V> extends ColumnData
public <T> void serialize(Cell<T> cell, ColumnMetadata column, DataOutputPlus out, LivenessInfo rowLiveness, SerializationHeader header) throws IOException
{
assert cell != null;
boolean hasValue = cell.valueSize() > 0;
boolean isDeleted = cell.isTombstone();
boolean isExpiring = cell.isExpiring();
boolean useRowTimestamp = !rowLiveness.isEmpty() && cell.timestamp() == rowLiveness.timestamp();
boolean useRowTTL = isExpiring && rowLiveness.isExpiring() && cell.ttl() == rowLiveness.ttl() && cell.localDeletionTime() == rowLiveness.localExpirationTime();
int valueSize;
boolean hasValue;
boolean isDeleted;
boolean isExpiring;
long cellTimestamp;
long localDeletionTime;
int ttl;
T value;
ValueAccessor<T> accessor;
// To avoid megamorphic calls we split call sites.
// We have ArrayCell, BufferCell and NativeCell and all of them can be used here in different scenarios.
// The type check is executed once per 8 Cell method calls, so it amortized.
// While this is a micro-optimization, given that it is invoked per cell,
// the invocation frequency is extremely large and the overhead is noticeable.
Class<?> cellClass = cell.getClass();
if (cellClass == NativeCell.class)
{
valueSize = cell.valueSize();
hasValue = valueSize > 0;
isDeleted = cell.isTombstone();
isExpiring = cell.isExpiring();
cellTimestamp = cell.timestamp();
localDeletionTime = cell.localDeletionTime();
ttl = cell.ttl();
value = cell.value();
accessor = cell.accessor();
}
else if (cellClass == ArrayCell.class)
{
valueSize = cell.valueSize();
hasValue = valueSize > 0;
isDeleted = cell.isTombstone();
isExpiring = cell.isExpiring();
cellTimestamp = cell.timestamp();
localDeletionTime = cell.localDeletionTime();
ttl = cell.ttl();
value = cell.value();
accessor = cell.accessor();
}
else
{
valueSize = cell.valueSize();
hasValue = valueSize > 0;
isDeleted = cell.isTombstone();
isExpiring = cell.isExpiring();
cellTimestamp = cell.timestamp();
localDeletionTime = cell.localDeletionTime();
ttl = cell.ttl();
value = cell.value();
accessor = cell.accessor();
}
boolean useRowTimestamp = !rowLiveness.isEmpty() && cellTimestamp == rowLiveness.timestamp();
boolean useRowTTL = isExpiring
&& rowLiveness.isExpiring()
&& ttl == rowLiveness.ttl()
&& localDeletionTime == rowLiveness.localExpirationTime();
int flags = 0;
if (!hasValue)
flags |= HAS_EMPTY_VALUE_MASK;
@ -302,18 +360,18 @@ public abstract class Cell<V> extends ColumnData
out.writeByte((byte)flags);
if (!useRowTimestamp)
header.writeTimestamp(cell.timestamp(), out);
header.writeTimestamp(cellTimestamp, out);
if ((isDeleted || isExpiring) && !useRowTTL)
header.writeLocalDeletionTime(cell.localDeletionTime(), out);
header.writeLocalDeletionTime(localDeletionTime, out);
if (isExpiring && !useRowTTL)
header.writeTTL(cell.ttl(), out);
header.writeTTL(ttl, out);
if (column.isComplex())
column.cellPathSerializer().serialize(cell.path(), out);
if (hasValue)
header.getType(column).writeValue(cell.value(), cell.accessor(), out);
header.getType(column).writeValue(value, accessor, out);
}
public <V> Cell<V> deserialize(DataInputPlus in, LivenessInfo rowLiveness, ColumnMetadata column, SerializationHeader header, DeserializationHelper helper, ValueAccessor<V> accessor) throws IOException

17
src/java/org/apache/cassandra/db/rows/SerializationHelper.java
View File

@ -18,17 +18,34 @@
package org.apache.cassandra.db.rows;
import javax.annotation.concurrent.NotThreadSafe;
import org.apache.cassandra.db.LivenessInfo;
import org.apache.cassandra.db.SerializationHeader;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.utils.SearchIterator;
import org.apache.cassandra.utils.btree.BTreeSearchIterator;
/**
* The class has a mutable state such as iterators and reusable fields to avoid extra allocation during cells processing,
* so it is not safe to share between mutiple threads.
*/
@NotThreadSafe
public class SerializationHelper
{
public final SerializationHeader header;
private BTreeSearchIterator<ColumnMetadata, ColumnMetadata> statics = null;
private BTreeSearchIterator<ColumnMetadata, ColumnMetadata> regulars = null;
// reusable fields to avoid extra allocation during cells processing
// within org.apache.cassandra.db.rows.UnfilteredSerializer.serializeRowBody
int flags;
LivenessInfo pkLiveness;
DataOutputPlus out;
SearchIterator<ColumnMetadata, ColumnMetadata> si;
public SerializationHelper(SerializationHeader header)
{
this.header = header;

48
src/java/org/apache/cassandra/db/rows/UnfilteredSerializer.java
View File

@ -242,31 +242,15 @@ public class UnfilteredSerializer
if ((flags & HAS_ALL_COLUMNS) == 0)
Columns.serializer.serializeSubset(row.columns(), headerColumns, out);
SearchIterator<ColumnMetadata, ColumnMetadata> si = helper.iterator(isStatic);
SearchIterator<ColumnMetadata, ColumnMetadata> si = helper.header.columnsMayChanged() ? helper.iterator(isStatic) : null;
helper.flags = flags;
helper.pkLiveness = pkLiveness;
helper.out = out;
helper.si = si;
try
{
row.apply(cd -> {
// We can obtain the column for data directly from data.column(). However, if the cell/complex data
// originates from a sstable, the column we'll get will have the type used when the sstable was serialized,
// and if that type have been recently altered, that may not be the type we want to serialize the column
// with. So we use the ColumnMetadata from the "header" which is "current". Also see #11810 for what
// happens if we don't do that.
ColumnMetadata column = si.next(cd.column());
assert column != null : cd.column.toString();
try
{
if (cd.column.isSimple())
Cell.serializer.serialize((Cell<?>) cd, column, out, pkLiveness, header);
else
writeComplexColumn((ComplexColumnData) cd, column, hasComplexDeletion(flags), pkLiveness, header, out);
}
catch (IOException e)
{
throw new WrappedException(e);
}
});
row.apply(UnfilteredSerializer::serializeColumnData, helper);
}
catch (WrappedException e)
{
@ -277,7 +261,25 @@ public class UnfilteredSerializer
}
}
private void writeComplexColumn(ComplexColumnData data, ColumnMetadata column, boolean hasComplexDeletion, LivenessInfo rowLiveness, SerializationHeader header, DataOutputPlus out)
private static void serializeColumnData(SerializationHelper helper, ColumnData cd)
{
ColumnMetadata column = helper.header.columnsMayChanged() ? helper.si.next(cd.column()) : cd.column();
assert column != null : cd.column.toString();
try
{
if (cd.column.isSimple())
Cell.serializer.serialize((Cell<?>) cd, column, helper.out, helper.pkLiveness, helper.header);
else
writeComplexColumn((ComplexColumnData) cd, column, hasComplexDeletion(helper.flags), helper.pkLiveness, helper.header, helper.out);
}
catch (IOException e)
{
throw new WrappedException(e);
}
}
private static void writeComplexColumn(ComplexColumnData data, ColumnMetadata column, boolean hasComplexDeletion, LivenessInfo rowLiveness, SerializationHeader header, DataOutputPlus out)
throws IOException
{
if (hasComplexDeletion)

6
src/java/org/apache/cassandra/io/sstable/RangeAwareSSTableWriter.java
View File

@ -154,6 +154,12 @@ public class RangeAwareSSTableWriter implements SSTableMultiWriter
return currentWriter != null ? currentWriter.getBytesWritten() : 0L;
}
@Override
public long getTotalRows()
{
return currentWriter != null ? currentWriter.getTotalRows() : 0L;
}
@Override
public long getOnDiskBytesWritten()
{

1
src/java/org/apache/cassandra/io/sstable/SSTableMultiWriter.java
View File

@ -44,6 +44,7 @@ public interface SSTableMultiWriter extends Transactional
String getFilename();
long getBytesWritten();
long getOnDiskBytesWritten();
long getTotalRows();
TableId getTableId();
static void abortOrDie(SSTableMultiWriter writer)

6
src/java/org/apache/cassandra/io/sstable/SSTableTxnSingleStreamWriter.java
View File

@ -106,6 +106,12 @@ public class SSTableTxnSingleStreamWriter implements SSTableMultiWriter
return writer.getOnDiskBytesWritten();
}
@Override
public long getTotalRows()
{
return writer.getTotalRows();
}
@Override
public TableId getTableId()
{

6
src/java/org/apache/cassandra/io/sstable/SSTableZeroCopyWriter.java
View File

@ -164,6 +164,12 @@ public class SSTableZeroCopyWriter extends SSTable implements SSTableMultiWriter
return 0;
}
@Override
public long getTotalRows()
{
return 0;
}
@Override
public TableId getTableId()
{

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

@ -81,6 +81,11 @@ public class SimpleSSTableMultiWriter implements SSTableMultiWriter
return writer.getEstimatedOnDiskBytesWritten();
}
public long getTotalRows()
{
return writer.getTotalRows();
}
public TableId getTableId()
{
return writer.metadata().id;

5
src/java/org/apache/cassandra/io/sstable/format/SSTableWriter.java
View File

@ -215,6 +215,11 @@ public abstract class SSTableWriter extends SSTable implements Transactional
return getOnDiskFilePointer();
}
public long getTotalRows()
{
return metadataCollector.getTotalRows();
}
/**
* Reset the data file to the marked position (see {@link #mark()}) and truncate the rest of the file.
*/

54
src/java/org/apache/cassandra/io/sstable/format/SortedTableWriter.java
View File

@ -89,6 +89,11 @@ public abstract class SortedTableWriter<P extends SortedTablePartitionWriter, I
private long lastEarlyOpenLength;
private final Supplier<Double> crcCheckChanceSupplier;
private final boolean isPartitionSizeGuardEnabled;
private final boolean isPartitionTombstonesGuardEnabled;
private final boolean areCollectionGuardsDisabled;
public SortedTableWriter(Builder<P, I, ?, ?> builder, ILifecycleTransaction txn, SSTable.Owner owner)
{
super(builder, txn, owner);
@ -119,6 +124,9 @@ public abstract class SortedTableWriter<P extends SortedTablePartitionWriter, I
handleConstructionFailure(ex);
throw ex;
}
isPartitionSizeGuardEnabled = Guardrails.partitionSize.enabled();
isPartitionTombstonesGuardEnabled = Guardrails.partitionTombstones.enabled();
areCollectionGuardsDisabled = !Guardrails.collectionSize.enabled() && !Guardrails.itemsPerCollection.enabled();
}
/**
@ -146,8 +154,20 @@ public abstract class SortedTableWriter<P extends SortedTablePartitionWriter, I
if (header.hasStatic())
addStaticRow(partition.partitionKey(), partition.staticRow());
while (partition.hasNext())
addUnfiltered(partition.partitionKey(), partition.next());
int i = 0;
boolean hasNext = false;
while (hasNext || partition.hasNext())
{
Unfiltered current = partition.next();
hasNext = partition.hasNext();
boolean isRowFirstOrLast;
if (i == 0)
isRowFirstOrLast = true;
else
isRowFirstOrLast = !hasNext;
addUnfiltered(partition.partitionKey(), current, isRowFirstOrLast);
i++;
}
indexEntry = endPartition(partition.partitionKey(), partition.partitionLevelDeletion());
@ -198,30 +218,32 @@ public abstract class SortedTableWriter<P extends SortedTablePartitionWriter, I
onStaticRow(row);
}
private void addUnfiltered(DecoratedKey key, Unfiltered unfiltered) throws IOException
private void addUnfiltered(DecoratedKey key, Unfiltered unfiltered, boolean isRowFirstOrLast) throws IOException
{
if (unfiltered.isRow())
addRow(key, (Row) unfiltered);
addRow(key, (Row) unfiltered, isRowFirstOrLast);
else
addRangeTomstoneMarker((RangeTombstoneMarker) unfiltered);
addRangeTomstoneMarker((RangeTombstoneMarker) unfiltered, isRowFirstOrLast);
}
private void addRow(DecoratedKey key, Row row) throws IOException
private void addRow(DecoratedKey key, Row row, boolean isRowFirstOrLast) throws IOException
{
guardCollectionSize(key, row);
partitionWriter.addUnfiltered(row);
metadataCollector.updateClusteringValues(row.clustering());
if (isRowFirstOrLast)
metadataCollector.updateClusteringValues(row.clustering());
Rows.collectStats(row, metadataCollector);
onRow(row);
}
private void addRangeTomstoneMarker(RangeTombstoneMarker marker) throws IOException
private void addRangeTomstoneMarker(RangeTombstoneMarker marker, boolean isRowFirstOrLast) throws IOException
{
partitionWriter.addUnfiltered(marker);
metadataCollector.updateClusteringValuesByBoundOrBoundary(marker.clustering());
if (isRowFirstOrLast)
metadataCollector.updateClusteringValuesByBoundOrBoundary(marker.clustering());
if (marker.isBoundary())
{
RangeTombstoneBoundaryMarker bm = (RangeTombstoneBoundaryMarker) marker;
@ -242,10 +264,12 @@ public abstract class SortedTableWriter<P extends SortedTablePartitionWriter, I
long finishResult = partitionWriter.finish();
long endPosition = dataWriter.position();
// inclusive of last byte
long partitionSize = endPosition - partitionWriter.getPartitionStartPosition();
guardPartitionThreshold(Guardrails.partitionSize, key, partitionSize);
guardPartitionThreshold(Guardrails.partitionTombstones, key, metadataCollector.totalTombstones);
// inclusive of last byte
if (isPartitionSizeGuardEnabled)
guardPartitionThreshold(Guardrails.partitionSize, key, partitionSize);
if (isPartitionTombstonesGuardEnabled)
guardPartitionThreshold(Guardrails.partitionTombstones, key, metadataCollector.totalTombstones);
metadataCollector.addPartitionSizeInBytes(partitionSize);
metadataCollector.addKey(key.getKey());
metadataCollector.addCellPerPartitionCount();
@ -255,7 +279,8 @@ public abstract class SortedTableWriter<P extends SortedTablePartitionWriter, I
if (first == null)
first = lastWrittenKey;
logger.trace("wrote {} at {}", key, endPosition);
if (logger.isTraceEnabled())
logger.trace("wrote {} at {}", key, endPosition);
return createRowIndexEntry(key, partitionLevelDeletion, finishResult);
}
@ -402,6 +427,9 @@ public abstract class SortedTableWriter<P extends SortedTablePartitionWriter, I
private void guardCollectionSize(DecoratedKey partitionKey, Row row)
{
if (areCollectionGuardsDisabled)
return;
if (!Guardrails.collectionSize.enabled() && !Guardrails.itemsPerCollection.enabled())
return;

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

@ -295,7 +295,8 @@ public class BigTableWriter extends SortedTableWriter<BigFormatPartitionWriter,
}
long indexEnd = writer.position();
logger.trace("wrote index entry: {} at {}", indexEntry, indexStart);
if (logger.isTraceEnabled())
logger.trace("wrote index entry: {} at {}", indexEntry, indexStart);
summary.maybeAddEntry(key, indexStart, indexEnd, dataEnd);
}

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

@ -26,6 +26,8 @@ import java.util.UUID;
import com.clearspring.analytics.stream.cardinality.HyperLogLogPlus;
import com.clearspring.analytics.stream.cardinality.ICardinality;
import net.nicoulaj.compilecommand.annotations.Inline;
import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.ClusteringBound;
import org.apache.cassandra.db.ClusteringBoundOrBoundary;
@ -39,7 +41,9 @@ import org.apache.cassandra.db.commitlog.CommitLogPosition;
import org.apache.cassandra.db.commitlog.IntervalSet;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.partitions.PartitionStatisticsCollector;
import org.apache.cassandra.db.rows.ArrayCell;
import org.apache.cassandra.db.rows.Cell;
import org.apache.cassandra.db.rows.NativeCell;
import org.apache.cassandra.db.rows.Unfiltered;
import org.apache.cassandra.io.sstable.ClusteringDescriptor;
import org.apache.cassandra.io.sstable.SSTable;
@ -246,10 +250,39 @@ public class MetadataCollector implements PartitionStatisticsCollector
public void update(Cell<?> cell)
{
++currentPartitionCells;
updateTimestamp(cell.timestamp());
updateTTL(cell.ttl());
updateLocalDeletionTime(cell.localDeletionTime());
if (!cell.isLive(nowInSec))
long timestamp;
int ttl;
long localDeletionTime;
// This method may process several implementations of Cell.
// To improve inlining of Cell method calls, we split the call sites.
// This is a very hot path, invoked for every cell (potentially millions of times per second),
// so this micro-optimization is justified.
// The type check is executed once per 3 Cell method calls, so it amortized.
Class<?> cellClass = cell.getClass();
if (cellClass == NativeCell.class)
{
timestamp = cell.timestamp();
ttl = cell.ttl();
localDeletionTime = cell.localDeletionTime();
}
else if (cellClass == ArrayCell.class)
{
timestamp = cell.timestamp();
ttl = cell.ttl();
localDeletionTime = cell.localDeletionTime();
}
else
{
timestamp = cell.timestamp();
ttl = cell.ttl();
localDeletionTime = cell.localDeletionTime();
}
updateTimestamp(timestamp);
updateTTL(ttl);
updateLocalDeletionTime(localDeletionTime);
// isLive(nowInSec) is not used to avoid additional non-monomorphic calls of Cell methods
if (!cell.isLive(nowInSec, localDeletionTime, ttl))
updateTombstoneCount();
}
@ -406,6 +439,11 @@ public class MetadataCollector implements PartitionStatisticsCollector
this.hasLegacyCounterShards = this.hasLegacyCounterShards || hasLegacyCounterShards;
}
public long getTotalRows()
{
return totalRows;
}
public Map<MetadataType, MetadataComponent> finalizeMetadata(String partitioner, double bloomFilterFPChance, long repairedAt, TimeUUID pendingRepair, boolean isTransient, SerializationHeader header, ByteBuffer firstKey, ByteBuffer lastKey)
{
assert minClustering.kind() == ClusteringPrefix.Kind.CLUSTERING || minClustering.kind().isStart();
@ -478,6 +516,7 @@ public class MetadataCollector implements PartitionStatisticsCollector
this.defaultMax = defaultMax;
}
@Inline
public void update(long value)
{
if (!isSet)
@ -525,6 +564,7 @@ public class MetadataCollector implements PartitionStatisticsCollector
this.defaultMax = defaultMax;
}
@Inline
public void update(int value)
{
if (!isSet)

8
src/java/org/apache/cassandra/schema/ColumnMetadata.java
View File

@ -136,6 +136,8 @@ public final class ColumnMetadata extends ColumnSpecification implements Selecta
*/
private final long comparisonOrder;
private final boolean isCounterColumn;
/**
* Masking function used to dynamically mask the contents of this column.
*/
@ -289,6 +291,7 @@ public final class ColumnMetadata extends ColumnSpecification implements Selecta
this.cellComparator = cellPathComparator == null ? ColumnData.comparator : (a, b) -> cellPathComparator.compare(a.path(), b.path());
this.asymmetricCellPathComparator = cellPathComparator == null ? null : (a, b) -> cellPathComparator.compare(((Cell<?>)a).path(), (CellPath) b);
this.comparisonOrder = comparisonOrder(kind, isComplex(), Math.max(0, position), name);
this.isCounterColumn = isCounterColumn(type);
this.mask = mask;
this.columnConstraints = columnConstraints;
this.columnConstraints.setColumnName(name);
@ -730,6 +733,11 @@ public final class ColumnMetadata extends ColumnSpecification implements Selecta
* Check if column is counter type.
*/
public boolean isCounterColumn()
{
return isCounterColumn;
}
private static boolean isCounterColumn(AbstractType<?> type)
{
if (type instanceof CollectionType) // Possible with, for example, supercolumns
return ((CollectionType) type).valueComparator().isCounter();

60
src/java/org/apache/cassandra/utils/ThreadStats.java Normal file
View File

@ -0,0 +1,60 @@
/*
* 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.utils;
import java.lang.management.ManagementFactory;
import com.sun.management.ThreadMXBean;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ThreadStats
{
private final static Logger logger = LoggerFactory.getLogger(ThreadStats.class);
private static final ThreadMXBean threadMxBean;
static
{
ThreadMXBean threadMxBeanToInit;
try
{
threadMxBeanToInit = (com.sun.management.ThreadMXBean) ManagementFactory.getThreadMXBean();
threadMxBeanToInit.setThreadAllocatedMemoryEnabled(true);
}
catch (Throwable e)
{
threadMxBeanToInit = null;
logger.debug("Per thread stats are not available: {}", e.getMessage());
}
threadMxBean = threadMxBeanToInit;
}
public static long getCurrentThreadCpuTimeNano()
{
return threadMxBean == null || !threadMxBean.isCurrentThreadCpuTimeSupported()
? -1 : threadMxBean.getCurrentThreadCpuTime();
}
public static long getCurrentThreadAllocatedBytes()
{
return threadMxBean == null || !threadMxBean.isThreadAllocatedMemorySupported()
? -1 : threadMxBean.getCurrentThreadAllocatedBytes();
}
}