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.0 into cassandra-4.1

This commit is contained in:
Benjamin Lerer 2022-07-07 13:57:19 +02:00
parent a250126f0f c378874a9f
commit 50e7a3f5df
64 changed files with 1908 additions and 893 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -11,6 +11,7 @@
* Revert removal of withBufferSizeInMB(int size) in CQLSSTableWriter.Builder class and deprecate it in favor of withBufferSizeInMiB(int size) (CASSANDRA-17675)
* Remove expired snapshots of dropped tables after restart (CASSANDRA-17619)
Merged from 4.0:
* Utilise BTree improvements to reduce garbage and improve throughput (CASSANDRA-15511)
* SSL storage port in sstableloader is deprecated (CASSANDRA-17602)
* Fix counter write timeouts at ONE (CASSANDRA-17411)
* Fix NPE in getLocalPrimaryRangeForEndpoint (CASSANDRA-17680)

6
src/java/org/apache/cassandra/cql3/ColumnIdentifier.java
View File

@ -30,7 +30,7 @@ import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.memory.AbstractAllocator;
import org.apache.cassandra.utils.memory.ByteBufferCloner;
/**
* Represents an identifer for a CQL column definition.
@ -209,9 +209,9 @@ public class ColumnIdentifier implements IMeasurableMemory, Comparable<ColumnIde
+ ObjectSizes.sizeOf(text);
}
public ColumnIdentifier clone(AbstractAllocator allocator)
public ColumnIdentifier clone(ByteBufferCloner cloner)
{
return interned ? this : new ColumnIdentifier(allocator.clone(bytes), text, false);
return interned ? this : new ColumnIdentifier(cloner.clone(bytes), text, false);
}
public int compareTo(ColumnIdentifier that)

7
src/java/org/apache/cassandra/db/ArrayClusteringBound.java
View File

@ -23,7 +23,7 @@ import java.nio.ByteBuffer;
import com.google.common.base.Preconditions;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.memory.AbstractAllocator;
import org.apache.cassandra.utils.memory.ByteBufferCloner;
public class ArrayClusteringBound extends ArrayClusteringBoundOrBoundary implements ClusteringBound<byte[]>
{
@ -45,9 +45,10 @@ public class ArrayClusteringBound extends ArrayClusteringBoundOrBoundary impleme
return create(kind().invert(), values);
}
public ClusteringBound<ByteBuffer> copy(AbstractAllocator allocator)
@Override
public ClusteringBound<ByteBuffer> clone(ByteBufferCloner cloner)
{
return (ClusteringBound<ByteBuffer>) super.copy(allocator);
return (ClusteringBound<ByteBuffer>) super.clone(cloner);
}
public static ArrayClusteringBound create(ClusteringPrefix.Kind kind, byte[][] values)

6
src/java/org/apache/cassandra/db/BufferClusteringBound.java
View File

@ -23,7 +23,7 @@ import java.nio.ByteBuffer;
import com.google.common.base.Preconditions;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.memory.AbstractAllocator;
import org.apache.cassandra.utils.memory.ByteBufferCloner;
public class BufferClusteringBound extends BufferClusteringBoundOrBoundary implements ClusteringBound<ByteBuffer>
{
@ -45,9 +45,9 @@ public class BufferClusteringBound extends BufferClusteringBoundOrBoundary imple
return create(kind().invert(), values);
}
public ClusteringBound<ByteBuffer> copy(AbstractAllocator allocator)
public ClusteringBound<ByteBuffer> clone(ByteBufferCloner cloner)
{
return (ClusteringBound<ByteBuffer>) super.copy(allocator);
return (ClusteringBound<ByteBuffer>) super.clone(cloner);
}
public static BufferClusteringBound create(ClusteringPrefix.Kind kind, ByteBuffer[] values)

6
src/java/org/apache/cassandra/db/BufferClusteringBoundary.java
View File

@ -23,7 +23,7 @@ import java.nio.ByteBuffer;
import com.google.common.base.Preconditions;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.memory.AbstractAllocator;
import org.apache.cassandra.utils.memory.ByteBufferCloner;
public class BufferClusteringBoundary extends BufferClusteringBoundOrBoundary implements ClusteringBoundary<ByteBuffer>
{
@ -52,9 +52,9 @@ public class BufferClusteringBoundary extends BufferClusteringBoundOrBoundary im
}
@Override
public ClusteringBoundary<ByteBuffer> copy(AbstractAllocator allocator)
public ClusteringBoundary<ByteBuffer> clone(ByteBufferCloner cloner)
{
return (ClusteringBoundary<ByteBuffer>) super.copy(allocator);
return (ClusteringBoundary<ByteBuffer>) super.clone(cloner);
}
public ClusteringBound<ByteBuffer> openBound(boolean reversed)

6
src/java/org/apache/cassandra/db/Clustering.java
View File

@ -31,7 +31,7 @@ import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.utils.memory.AbstractAllocator;
import org.apache.cassandra.utils.memory.ByteBufferCloner;
import static org.apache.cassandra.db.AbstractBufferClusteringPrefix.EMPTY_VALUES_ARRAY;
@ -41,7 +41,7 @@ public interface Clustering<V> extends ClusteringPrefix<V>, IMeasurableMemory
public long unsharedHeapSizeExcludingData();
public default Clustering<?> copy(AbstractAllocator allocator)
public default Clustering<?> clone(ByteBufferCloner cloner)
{
// Important for STATIC_CLUSTERING (but must copy empty native clustering types).
if (size() == 0)
@ -51,7 +51,7 @@ public interface Clustering<V> extends ClusteringPrefix<V>, IMeasurableMemory
for (int i = 0; i < size(); i++)
{
ByteBuffer val = accessor().toBuffer(get(i));
newValues[i] = val == null ? null : allocator.clone(val);
newValues[i] = val == null ? null : cloner.clone(val);
}
return new BufferClustering(newValues);
}

4
src/java/org/apache/cassandra/db/ClusteringBound.java
View File

@ -24,7 +24,7 @@ import java.nio.ByteBuffer;
import java.util.List;
import org.apache.cassandra.db.marshal.ByteBufferAccessor;
import org.apache.cassandra.utils.memory.AbstractAllocator;
import org.apache.cassandra.utils.memory.ByteBufferCloner;
/**
* The start or end of a range of clusterings, either inclusive or exclusive.
@ -47,7 +47,7 @@ public interface ClusteringBound<V> extends ClusteringBoundOrBoundary<V>
ClusteringBound<V> invert();
@Override
ClusteringBound<ByteBuffer> copy(AbstractAllocator allocator);
ClusteringBound<ByteBuffer> clone(ByteBufferCloner cloner);
default boolean isStart()
{

6
src/java/org/apache/cassandra/db/ClusteringBoundOrBoundary.java
View File

@ -30,7 +30,7 @@ import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.utils.memory.AbstractAllocator;
import org.apache.cassandra.utils.memory.ByteBufferCloner;
/**
* This class defines a threshold between ranges of clusterings. It can either be a start or end bound of a range, or
@ -62,11 +62,11 @@ public interface ClusteringBoundOrBoundary<V> extends ClusteringPrefix<V>
return kind().isClose(reversed);
}
default ClusteringBoundOrBoundary<ByteBuffer> copy(AbstractAllocator allocator)
default ClusteringBoundOrBoundary<ByteBuffer> clone(ByteBufferCloner cloner)
{
ByteBuffer[] newValues = new ByteBuffer[size()];
for (int i = 0; i < size(); i++)
newValues[i] = allocator.clone(get(i), accessor());
newValues[i] = cloner.clone(get(i), accessor());
return ByteBufferAccessor.instance.factory().boundOrBoundary(kind(), newValues);
}

133
src/java/org/apache/cassandra/db/Columns.java
View File

@ -32,6 +32,8 @@ import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.cql3.ColumnIdentifier;
import org.apache.cassandra.db.marshal.SetType;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.db.rows.ColumnData;
import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.utils.ByteBufferUtil;
@ -96,8 +98,32 @@ public class Columns extends AbstractCollection<ColumnMetadata> implements Colle
return new Columns(BTree.singleton(c), c.isComplex() ? 0 : 1);
}
/**
* Returns a new {@code Columns} object holing the same columns as the provided Row.
*
* @param row the row from which to create the new {@code Columns}.
* @return the newly created {@code Columns} containing the columns from {@code row}.
*/
public static Columns from(Row row)
{
try (BTree.FastBuilder<ColumnMetadata> builder = BTree.fastBuilder())
{
for (ColumnData cd : row)
builder.add(cd.column());
Object[] tree = builder.build();
return new Columns(tree, findFirstComplexIdx(tree));
}
}
public static Columns from(BTree.Builder<ColumnMetadata> builder)
{
Object[] tree = builder.build();
return new Columns(tree, findFirstComplexIdx(tree));
}
/**
* Returns a new {@code Columns} object holing the same columns than the provided set.
* Returns a new {@code Columns} object holding the same columns than the provided set.
* This method assumes nothing about the order of {@code s}.
*
* @param s the set from which to create the new {@code Columns}.
* @return the newly created {@code Columns} containing the columns from {@code s}.
@ -446,25 +472,26 @@ public class Columns extends AbstractCollection<ColumnMetadata> implements Colle
public Columns deserialize(DataInputPlus in, TableMetadata metadata) throws IOException
{
int length = (int)in.readUnsignedVInt();
BTree.Builder<ColumnMetadata> builder = BTree.builder(Comparator.naturalOrder());
builder.auto(false);
for (int i = 0; i < length; i++)
try (BTree.FastBuilder<ColumnMetadata> builder = BTree.fastBuilder())
{
ByteBuffer name = ByteBufferUtil.readWithVIntLength(in);
ColumnMetadata column = metadata.getColumn(name);
if (column == null)
for (int i = 0; i < length; i++)
{
// If we don't find the definition, it could be we have data for a dropped column, and we shouldn't
// fail deserialization because of that. So we grab a "fake" ColumnMetadata that ensure proper
// deserialization. The column will be ignore later on anyway.
column = metadata.getDroppedColumn(name);
ByteBuffer name = ByteBufferUtil.readWithVIntLength(in);
ColumnMetadata column = metadata.getColumn(name);
if (column == null)
throw new RuntimeException("Unknown column " + UTF8Type.instance.getString(name) + " during deserialization");
{
// If we don't find the definition, it could be we have data for a dropped column, and we shouldn't
// fail deserialization because of that. So we grab a "fake" ColumnMetadata that ensure proper
// deserialization. The column will be ignore later on anyway.
column = metadata.getDroppedColumn(name);
if (column == null)
throw new RuntimeException("Unknown column " + UTF8Type.instance.getString(name) + " during deserialization");
}
builder.add(column);
}
builder.add(column);
return new Columns(builder.build());
}
return new Columns(builder.build());
}
/**
@ -532,21 +559,23 @@ public class Columns extends AbstractCollection<ColumnMetadata> implements Colle
}
else
{
BTree.Builder<ColumnMetadata> builder = BTree.builder(Comparator.naturalOrder());
int firstComplexIdx = 0;
for (ColumnMetadata column : superset)
try (BTree.FastBuilder<ColumnMetadata> builder = BTree.fastBuilder())
{
if ((encoded & 1) == 0)
int firstComplexIdx = 0;
for (ColumnMetadata column : superset)
{
builder.add(column);
if (column.isSimple())
++firstComplexIdx;
if ((encoded & 1) == 0)
{
builder.add(column);
if (column.isSimple())
++firstComplexIdx;
}
encoded >>>= 1;
}
encoded >>>= 1;
if (encoded != 0)
throw new IOException("Invalid Columns subset bytes; too many bits set:" + Long.toBinaryString(encoded));
return new Columns(builder.build(), firstComplexIdx);
}
if (encoded != 0)
throw new IOException("Invalid Columns subset bytes; too many bits set:" + Long.toBinaryString(encoded));
return new Columns(builder.build(), firstComplexIdx);
}
}
@ -615,37 +644,39 @@ public class Columns extends AbstractCollection<ColumnMetadata> implements Colle
int supersetCount = superset.size();
int columnCount = supersetCount - delta;
BTree.Builder<ColumnMetadata> builder = BTree.builder(Comparator.naturalOrder());
if (columnCount < supersetCount / 2)
try (BTree.FastBuilder<ColumnMetadata> builder = BTree.fastBuilder())
{
for (int i = 0 ; i < columnCount ; i++)
if (columnCount < supersetCount / 2)
{
int idx = (int) in.readUnsignedVInt();
builder.add(BTree.findByIndex(superset.columns, idx));
}
}
else
{
Iterator<ColumnMetadata> iter = superset.iterator();
int idx = 0;
int skipped = 0;
while (true)
{
int nextMissingIndex = skipped < delta ? (int)in.readUnsignedVInt() : supersetCount;
while (idx < nextMissingIndex)
for (int i = 0 ; i < columnCount ; i++)
{
ColumnMetadata def = iter.next();
builder.add(def);
idx++;
int idx = (int) in.readUnsignedVInt();
builder.add(BTree.findByIndex(superset.columns, idx));
}
if (idx == supersetCount)
break;
iter.next();
idx++;
skipped++;
}
else
{
Iterator<ColumnMetadata> iter = superset.iterator();
int idx = 0;
int skipped = 0;
while (true)
{
int nextMissingIndex = skipped < delta ? (int)in.readUnsignedVInt() : supersetCount;
while (idx < nextMissingIndex)
{
ColumnMetadata def = iter.next();
builder.add(def);
idx++;
}
if (idx == supersetCount)
break;
iter.next();
idx++;
skipped++;
}
}
return new Columns(builder.build());
}
return new Columns(builder.build());
}
@DontInline

5
src/java/org/apache/cassandra/db/DeletionInfo.java
View File

@ -21,7 +21,7 @@ import java.util.Iterator;
import org.apache.cassandra.cache.IMeasurableMemory;
import org.apache.cassandra.db.rows.EncodingStats;
import org.apache.cassandra.utils.memory.AbstractAllocator;
import org.apache.cassandra.utils.memory.ByteBufferCloner;
/**
* A combination of a top-level (partition) tombstone and range tombstones describing the deletions
@ -70,5 +70,6 @@ public interface DeletionInfo extends IMeasurableMemory
public boolean mayModify(DeletionInfo delInfo);
public MutableDeletionInfo mutableCopy();
public DeletionInfo copy(AbstractAllocator allocator);
public DeletionInfo clone(ByteBufferCloner cloner);
}

2
src/java/org/apache/cassandra/db/EmptyIterators.java
View File

@ -185,7 +185,7 @@ public class EmptyIterators
{
RegularAndStaticColumns columns = RegularAndStaticColumns.NONE;
if (!staticRow.isEmpty())
columns = new RegularAndStaticColumns(Columns.from(staticRow.columns()), Columns.NONE);
columns = new RegularAndStaticColumns(Columns.from(staticRow), Columns.NONE);
else
staticRow = Rows.EMPTY_STATIC_ROW;

7
src/java/org/apache/cassandra/db/MutableDeletionInfo.java
View File

@ -25,7 +25,7 @@ import com.google.common.base.Objects;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.memory.AbstractAllocator;
import org.apache.cassandra.utils.memory.ByteBufferCloner;
/**
* A mutable implementation of {@code DeletionInfo}.
@ -83,11 +83,12 @@ public class MutableDeletionInfo implements DeletionInfo
return new MutableDeletionInfo(partitionDeletion, ranges == null ? null : ranges.copy());
}
public MutableDeletionInfo copy(AbstractAllocator allocator)
@Override
public MutableDeletionInfo clone(ByteBufferCloner cloner)
{
RangeTombstoneList rangesCopy = null;
if (ranges != null)
rangesCopy = ranges.copy(allocator);
rangesCopy = ranges.clone(cloner);
return new MutableDeletionInfo(partitionDeletion, rangesCopy);
}

12
src/java/org/apache/cassandra/db/RangeTombstoneList.java
View File

@ -29,7 +29,7 @@ import com.google.common.collect.Iterators;
import org.apache.cassandra.cache.IMeasurableMemory;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.memory.AbstractAllocator;
import org.apache.cassandra.utils.memory.ByteBufferCloner;
/**
* Data structure holding the range tombstones of a ColumnFamily.
@ -105,7 +105,7 @@ public class RangeTombstoneList implements Iterable<RangeTombstone>, IMeasurable
boundaryHeapSize, size);
}
public RangeTombstoneList copy(AbstractAllocator allocator)
public RangeTombstoneList clone(ByteBufferCloner cloner)
{
RangeTombstoneList copy = new RangeTombstoneList(comparator,
new ClusteringBound<?>[size],
@ -117,18 +117,18 @@ public class RangeTombstoneList implements Iterable<RangeTombstone>, IMeasurable
for (int i = 0; i < size; i++)
{
copy.starts[i] = clone(starts[i], allocator);
copy.ends[i] = clone(ends[i], allocator);
copy.starts[i] = clone(starts[i], cloner);
copy.ends[i] = clone(ends[i], cloner);
}
return copy;
}
private static <T> ClusteringBound<ByteBuffer> clone(ClusteringBound<T> bound, AbstractAllocator allocator)
private static <T> ClusteringBound<ByteBuffer> clone(ClusteringBound<T> bound, ByteBufferCloner cloner)
{
ByteBuffer[] values = new ByteBuffer[bound.size()];
for (int i = 0; i < values.length; i++)
values[i] = allocator.clone(bound.get(i), bound.accessor());
values[i] = cloner.clone(bound.get(i), bound.accessor());
return new BufferClusteringBound(bound.kind(), values);
}

18
src/java/org/apache/cassandra/db/RegularAndStaticColumns.java
View File

@ -23,7 +23,7 @@ import com.google.common.collect.Iterators;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.btree.BTreeSet;
import org.apache.cassandra.utils.btree.BTree;
import static java.util.Comparator.naturalOrder;
@ -150,22 +150,22 @@ public class RegularAndStaticColumns implements Iterable<ColumnMetadata>
// Note that we do want to use sorted sets because we want the column definitions to be compared
// through compareTo, not equals. The former basically check it's the same column name, while the latter
// check it's the same object, including the same type.
private BTreeSet.Builder<ColumnMetadata> regularColumns;
private BTreeSet.Builder<ColumnMetadata> staticColumns;
private BTree.Builder<ColumnMetadata> regularColumns;
private BTree.Builder<ColumnMetadata> staticColumns;
public Builder add(ColumnMetadata c)
{
if (c.isStatic())
{
if (staticColumns == null)
staticColumns = BTreeSet.builder(naturalOrder());
staticColumns = BTree.builder(naturalOrder());
staticColumns.add(c);
}
else
{
assert c.isRegular();
if (regularColumns == null)
regularColumns = BTreeSet.builder(naturalOrder());
regularColumns = BTree.builder(naturalOrder());
regularColumns.add(c);
}
return this;
@ -181,13 +181,13 @@ public class RegularAndStaticColumns implements Iterable<ColumnMetadata>
public Builder addAll(RegularAndStaticColumns columns)
{
if (regularColumns == null && !columns.regulars.isEmpty())
regularColumns = BTreeSet.builder(naturalOrder());
regularColumns = BTree.builder(naturalOrder());
for (ColumnMetadata c : columns.regulars)
regularColumns.add(c);
if (staticColumns == null && !columns.statics.isEmpty())
staticColumns = BTreeSet.builder(naturalOrder());
staticColumns = BTree.builder(naturalOrder());
for (ColumnMetadata c : columns.statics)
staticColumns.add(c);
@ -197,8 +197,8 @@ public class RegularAndStaticColumns implements Iterable<ColumnMetadata>
public RegularAndStaticColumns build()
{
return new RegularAndStaticColumns(staticColumns == null ? Columns.NONE : Columns.from(staticColumns.build()),
regularColumns == null ? Columns.NONE : Columns.from(regularColumns.build()));
return new RegularAndStaticColumns(staticColumns == null ? Columns.NONE : Columns.from(staticColumns),
regularColumns == null ? Columns.NONE : Columns.from(regularColumns));
}
}
}

32
src/java/org/apache/cassandra/db/compaction/Scrubber.java
View File

@ -45,7 +45,7 @@ import org.apache.cassandra.io.util.RandomAccessReader;
import org.apache.cassandra.service.ActiveRepairService;
import org.apache.cassandra.utils.*;
import org.apache.cassandra.utils.concurrent.Refs;
import org.apache.cassandra.utils.memory.HeapAllocator;
import org.apache.cassandra.utils.memory.HeapCloner;
import static org.apache.cassandra.utils.TimeUUID.Generator.nextTimeUUID;
@ -824,30 +824,26 @@ public class Scrubber implements Closeable
private Unfiltered fixNegativeLocalExpirationTime(Row row)
{
Row.Builder builder = HeapAllocator.instance.cloningBTreeRowBuilder();
builder.newRow(row.clustering());
builder.addPrimaryKeyLivenessInfo(row.primaryKeyLivenessInfo().isExpiring() && row.primaryKeyLivenessInfo().localExpirationTime() < 0 ?
row.primaryKeyLivenessInfo().withUpdatedTimestampAndLocalDeletionTime(row.primaryKeyLivenessInfo().timestamp() + 1, AbstractCell.MAX_DELETION_TIME)
:row.primaryKeyLivenessInfo());
builder.addRowDeletion(row.deletion());
for (ColumnData cd : row)
{
LivenessInfo livenessInfo = row.primaryKeyLivenessInfo();
if (livenessInfo.isExpiring() && livenessInfo.localExpirationTime() < 0)
livenessInfo = livenessInfo.withUpdatedTimestampAndLocalDeletionTime(livenessInfo.timestamp() + 1, AbstractCell.MAX_DELETION_TIME);
return row.transformAndFilter(livenessInfo, row.deletion(), cd -> {
if (cd.column().isSimple())
{
Cell<?> cell = (Cell<?>)cd;
builder.addCell(cell.isExpiring() && cell.localDeletionTime() < 0 ? cell.withUpdatedTimestampAndLocalDeletionTime(cell.timestamp() + 1, AbstractCell.MAX_DELETION_TIME) : cell);
Cell cell = (Cell)cd;
return cell.isExpiring() && cell.localDeletionTime() < 0
? cell.withUpdatedTimestampAndLocalDeletionTime(cell.timestamp() + 1, AbstractCell.MAX_DELETION_TIME)
: cell;
}
else
{
ComplexColumnData complexData = (ComplexColumnData)cd;
builder.addComplexDeletion(complexData.column(), complexData.complexDeletion());
for (Cell<?> cell : complexData)
{
builder.addCell(cell.isExpiring() && cell.localDeletionTime() < 0 ? cell.withUpdatedTimestampAndLocalDeletionTime(cell.timestamp() + 1, AbstractCell.MAX_DELETION_TIME) : cell);
}
return complexData.transformAndFilter(cell -> cell.isExpiring() && cell.localDeletionTime() < 0
? cell.withUpdatedTimestampAndLocalDeletionTime(cell.timestamp() + 1, AbstractCell.MAX_DELETION_TIME)
: cell);
}
}
return builder.build();
}).clone(HeapCloner.instance);
}
}
}

13
src/java/org/apache/cassandra/db/filter/ClusteringIndexNamesFilter.java
View File

@ -31,6 +31,7 @@ import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.utils.btree.BTree;
import org.apache.cassandra.utils.btree.BTreeSet;
/**
@ -244,12 +245,14 @@ public class ClusteringIndexNamesFilter extends AbstractClusteringIndexFilter
public ClusteringIndexFilter deserialize(DataInputPlus in, int version, TableMetadata metadata, boolean reversed) throws IOException
{
ClusteringComparator comparator = metadata.comparator;
BTreeSet.Builder<Clustering<?>> clusterings = BTreeSet.builder(comparator);
int size = (int)in.readUnsignedVInt();
for (int i = 0; i < size; i++)
clusterings.add(Clustering.serializer.deserialize(in, version, comparator.subtypes()));
return new ClusteringIndexNamesFilter(clusterings.build(), reversed);
try (BTree.FastBuilder<Clustering<?>> builder = BTree.fastBuilder())
{
for (int i = 0; i < size; i++)
builder.add(Clustering.serializer.deserialize(in, version, comparator.subtypes()));
BTreeSet<Clustering<?>> clusterings = BTreeSet.wrap(builder.build(), comparator);
return new ClusteringIndexNamesFilter(clusterings, reversed);
}
}
}
}

6
src/java/org/apache/cassandra/db/memtable/ShardedSkipListMemtable.java
View File

@ -56,6 +56,7 @@ import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.schema.TableMetadataRef;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.concurrent.OpOrder;
import org.apache.cassandra.utils.memory.Cloner;
import org.apache.cassandra.utils.memory.MemtableAllocator;
import org.github.jamm.Unmetered;
@ -367,12 +368,13 @@ public class ShardedSkipListMemtable extends AbstractAllocatorMemtable
public long put(DecoratedKey key, PartitionUpdate update, UpdateTransaction indexer, OpOrder.Group opGroup)
{
Cloner cloner = allocator.cloner(opGroup);
AtomicBTreePartition previous = partitions.get(key);
long initialSize = 0;
if (previous == null)
{
final DecoratedKey cloneKey = allocator.clone(key, opGroup);
final DecoratedKey cloneKey = cloner.clone(key);
AtomicBTreePartition empty = new AtomicBTreePartition(metadata, cloneKey, allocator);
// We'll add the columns later. This avoids wasting works if we get beaten in the putIfAbsent
previous = partitions.putIfAbsent(cloneKey, empty);
@ -387,7 +389,7 @@ public class ShardedSkipListMemtable extends AbstractAllocatorMemtable
}
}
long[] pair = previous.addAllWithSizeDelta(update, opGroup, indexer);
long[] pair = previous.addAllWithSizeDelta(update, cloner, opGroup, indexer);
updateMin(minTimestamp, update.stats().minTimestamp);
updateMin(minLocalDeletionTime, update.stats().minLocalDeletionTime);
liveDataSize.addAndGet(initialSize + pair[0]);

9
src/java/org/apache/cassandra/db/memtable/SkipListMemtable.java
View File

@ -55,6 +55,7 @@ import org.apache.cassandra.schema.TableMetadataRef;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.concurrent.OpOrder;
import org.apache.cassandra.utils.memory.Cloner;
import org.apache.cassandra.utils.memory.MemtableAllocator;
import static org.apache.cassandra.config.CassandraRelevantProperties.MEMTABLE_OVERHEAD_COMPUTE_STEPS;
@ -109,12 +110,13 @@ public class SkipListMemtable extends AbstractAllocatorMemtable
@Override
public long put(PartitionUpdate update, UpdateTransaction indexer, OpOrder.Group opGroup)
{
Cloner cloner = allocator.cloner(opGroup);
AtomicBTreePartition previous = partitions.get(update.partitionKey());
long initialSize = 0;
if (previous == null)
{
final DecoratedKey cloneKey = allocator.clone(update.partitionKey(), opGroup);
final DecoratedKey cloneKey = cloner.clone(update.partitionKey());
AtomicBTreePartition empty = new AtomicBTreePartition(metadata, cloneKey, allocator);
// We'll add the columns later. This avoids wasting works if we get beaten in the putIfAbsent
previous = partitions.putIfAbsent(cloneKey, empty);
@ -129,7 +131,7 @@ public class SkipListMemtable extends AbstractAllocatorMemtable
}
}
long[] pair = previous.addAllWithSizeDelta(update, opGroup, indexer);
long[] pair = previous.addAllWithSizeDelta(update, cloner, opGroup, indexer);
updateMin(minTimestamp, update.stats().minTimestamp);
updateMin(minLocalDeletionTime, update.stats().minLocalDeletionTime);
liveDataSize.addAndGet(initialSize + pair[0]);
@ -222,10 +224,11 @@ public class SkipListMemtable extends AbstractAllocatorMemtable
{
int rowOverhead;
MemtableAllocator allocator = MEMORY_POOL.newAllocator("");
Cloner cloner = allocator.cloner(group);
ConcurrentNavigableMap<PartitionPosition, Object> partitions = new ConcurrentSkipListMap<>();
final Object val = new Object();
for (int i = 0 ; i < count ; i++)
partitions.put(allocator.clone(new BufferDecoratedKey(new LongToken(i), ByteBufferUtil.EMPTY_BYTE_BUFFER), group), val);
partitions.put(cloner.clone(new BufferDecoratedKey(new LongToken(i), ByteBufferUtil.EMPTY_BYTE_BUFFER)), val);
double avgSize = ObjectSizes.measureDeep(partitions) / (double) count;
rowOverhead = (int) ((avgSize - Math.floor(avgSize)) < 0.05 ? Math.floor(avgSize) : Math.ceil(avgSize));
rowOverhead -= ObjectSizes.measureDeep(new LongToken(0));

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

@ -22,6 +22,7 @@ import java.util.Collections;
import java.util.Iterator;
import java.util.NavigableSet;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Iterators;
import org.apache.cassandra.schema.TableMetadata;
@ -49,14 +50,15 @@ public abstract class AbstractBTreePartition implements Partition, Iterable<Row>
this.partitionKey = partitionKey;
}
protected static final class Holder
@VisibleForTesting
public static final class Holder
{
final RegularAndStaticColumns columns;
final DeletionInfo deletionInfo;
public final RegularAndStaticColumns columns;
public final DeletionInfo deletionInfo;
// the btree of rows
final Object[] tree;
final Row staticRow;
final EncodingStats stats;
public final Object[] tree;
public final Row staticRow;
public final EncodingStats stats;
Holder(RegularAndStaticColumns columns, Object[] tree, DeletionInfo deletionInfo, Row staticRow, EncodingStats stats)
{
@ -356,25 +358,25 @@ public abstract class AbstractBTreePartition implements Partition, Iterable<Row>
// Note that when building with a RowIterator, deletion will generally be LIVE, but we allow to pass it nonetheless because PartitionUpdate
// passes a MutableDeletionInfo that it mutates later.
protected static Holder build(RowIterator rows, DeletionInfo deletion, boolean buildEncodingStats, int initialRowCapacity)
protected static Holder build(RowIterator rows, DeletionInfo deletion, boolean buildEncodingStats)
{
TableMetadata metadata = rows.metadata();
RegularAndStaticColumns columns = rows.columns();
boolean reversed = rows.isReverseOrder();
BTree.Builder<Row> builder = BTree.builder(metadata.comparator, initialRowCapacity);
builder.auto(false);
while (rows.hasNext())
builder.add(rows.next());
try (BTree.FastBuilder<Row> builder = BTree.fastBuilder())
{
while (rows.hasNext())
builder.add(rows.next());
if (reversed)
builder.reverse();
Row staticRow = rows.staticRow();
Object[] tree = builder.build();
EncodingStats stats = buildEncodingStats ? EncodingStats.Collector.collect(staticRow, BTree.iterator(tree), deletion)
: EncodingStats.NO_STATS;
return new Holder(columns, tree, deletion, staticRow, stats);
Object[] tree = reversed ? builder.buildReverse()
: builder.build();
Row staticRow = rows.staticRow();
EncodingStats stats = buildEncodingStats ? EncodingStats.Collector.collect(staticRow, BTree.iterator(tree), deletion)
: EncodingStats.NO_STATS;
return new Holder(columns, tree, deletion, staticRow, stats);
}
}
@Override
@ -444,4 +446,16 @@ public abstract class AbstractBTreePartition implements Partition, Iterable<Row>
return BTree.findByIndex(tree, BTree.size(tree) - 1);
}
@VisibleForTesting
public static Holder unsafeGetEmptyHolder()
{
return EMPTY;
}
@VisibleForTesting
public static Holder unsafeConstructHolder(RegularAndStaticColumns columns, Object[] tree, DeletionInfo deletionInfo, Row staticRow, EncodingStats stats)
{
return new Holder(columns, tree, deletionInfo, staticRow, stats);
}
}

103
src/java/org/apache/cassandra/db/partitions/AtomicBTreePartition.java
View File

@ -36,8 +36,10 @@ import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.btree.BTree;
import org.apache.cassandra.utils.btree.UpdateFunction;
import org.apache.cassandra.utils.concurrent.OpOrder;
import org.apache.cassandra.utils.memory.HeapAllocator;
import org.apache.cassandra.utils.memory.Cloner;
import org.apache.cassandra.utils.memory.HeapCloner;
import org.apache.cassandra.utils.memory.MemtableAllocator;
import com.google.common.annotations.VisibleForTesting;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
@ -113,7 +115,6 @@ public final class AtomicBTreePartition extends AbstractBTreePartition
private long[] addAllWithSizeDeltaInternal(RowUpdater updater, PartitionUpdate update, UpdateTransaction indexer)
{
Holder current = ref;
updater.ref = current;
updater.reset();
if (!update.deletionInfo().getPartitionDeletion().isLive())
@ -126,7 +127,7 @@ public final class AtomicBTreePartition extends AbstractBTreePartition
if (update.deletionInfo().mayModify(current.deletionInfo))
{
if (updater.inputDeletionInfoCopy == null)
updater.inputDeletionInfoCopy = update.deletionInfo().copy(HeapAllocator.instance);
updater.inputDeletionInfoCopy = update.deletionInfo().clone(HeapCloner.instance);
deletionInfo = current.deletionInfo.mutableCopy().add(updater.inputDeletionInfoCopy);
updater.onAllocatedOnHeap(deletionInfo.unsharedHeapSize() - current.deletionInfo.unsharedHeapSize());
@ -141,7 +142,7 @@ public final class AtomicBTreePartition extends AbstractBTreePartition
Row newStatic = update.staticRow();
Row staticRow = newStatic.isEmpty()
? current.staticRow
: (current.staticRow.isEmpty() ? updater.apply(newStatic) : updater.apply(current.staticRow, newStatic));
: (current.staticRow.isEmpty() ? updater.insert(newStatic) : updater.merge(current.staticRow, newStatic));
Object[] tree = BTree.update(current.tree, update.holder().tree, update.metadata().comparator, updater);
EncodingStats newStats = current.stats.mergeWith(update.stats());
updater.onAllocatedOnHeap(newStats.unsharedHeapSize() - current.stats.unsharedHeapSize());
@ -162,9 +163,12 @@ public final class AtomicBTreePartition extends AbstractBTreePartition
* @return an array containing first the difference in size seen after merging the updates, and second the minimum
* time detla between updates.
*/
public long[] addAllWithSizeDelta(final PartitionUpdate update, OpOrder.Group writeOp, UpdateTransaction indexer)
public long[] addAllWithSizeDelta(final PartitionUpdate update,
Cloner cloner,
OpOrder.Group writeOp,
UpdateTransaction indexer)
{
RowUpdater updater = new RowUpdater(this, allocator, writeOp, indexer);
RowUpdater updater = new RowUpdater(allocator, cloner, writeOp, indexer);
try
{
boolean shouldLock = shouldLock(writeOp);
@ -331,15 +335,25 @@ public final class AtomicBTreePartition extends AbstractBTreePartition
return wasteTracker;
}
// the function we provide to the btree utilities to perform any column replacements
private static final class RowUpdater implements UpdateFunction<Row, Row>
@VisibleForTesting
public void unsafeSetHolder(Holder holder)
{
ref = holder;
}
@VisibleForTesting
public Holder unsafeGetHolder()
{
return ref;
}
// the function we provide to the btree utilities to perform any column replacements
private static final class RowUpdater implements UpdateFunction<Row, Row>, ColumnData.PostReconciliationFunction
{
final AtomicBTreePartition updating;
final MemtableAllocator allocator;
final OpOrder.Group writeOp;
final UpdateTransaction indexer;
Holder ref;
Row.Builder regularBuilder;
final Cloner cloner;
long dataSize;
long heapSize;
long colUpdateTimeDelta = Long.MAX_VALUE;
@ -347,29 +361,18 @@ public final class AtomicBTreePartition extends AbstractBTreePartition
DeletionInfo inputDeletionInfoCopy = null;
private RowUpdater(AtomicBTreePartition updating, MemtableAllocator allocator, OpOrder.Group writeOp, UpdateTransaction indexer)
private RowUpdater(MemtableAllocator allocator, Cloner cloner, OpOrder.Group writeOp, UpdateTransaction indexer)
{
this.updating = updating;
this.allocator = allocator;
this.writeOp = writeOp;
this.indexer = indexer;
this.cloner = cloner;
}
private Row.Builder builder(Clustering<?> clustering)
@Override
public Row insert(Row insert)
{
boolean isStatic = clustering == Clustering.STATIC_CLUSTERING;
// We know we only insert/update one static per PartitionUpdate, so no point in saving the builder
if (isStatic)
return allocator.rowBuilder(writeOp);
if (regularBuilder == null)
regularBuilder = allocator.rowBuilder(writeOp);
return regularBuilder;
}
public Row apply(Row insert)
{
Row data = Rows.copy(insert, builder(insert.clustering())).build();
Row data = insert.clone(cloner);
indexer.onInserted(insert);
this.dataSize += data.dataSize();
@ -380,17 +383,11 @@ public final class AtomicBTreePartition extends AbstractBTreePartition
return data;
}
public Row apply(Row existing, Row update)
public Row merge(Row existing, Row update)
{
Row.Builder builder = builder(existing.clustering());
colUpdateTimeDelta = Math.min(colUpdateTimeDelta, Rows.merge(existing, update, builder));
Row reconciled = builder.build();
Row reconciled = Rows.merge(existing, update, this);
indexer.onUpdated(existing, reconciled);
dataSize += reconciled.dataSize() - existing.dataSize();
onAllocatedOnHeap(reconciled.unsharedHeapSizeExcludingData() - existing.unsharedHeapSizeExcludingData());
if (inserted == null)
inserted = new ArrayList<>();
inserted.add(reconciled);
@ -398,6 +395,11 @@ public final class AtomicBTreePartition extends AbstractBTreePartition
return reconciled;
}
public Row retain(Row existing)
{
return existing;
}
protected void reset()
{
this.dataSize = 0;
@ -405,9 +407,36 @@ public final class AtomicBTreePartition extends AbstractBTreePartition
if (inserted != null)
inserted.clear();
}
public boolean abortEarly()
public Cell<?> merge(Cell<?> previous, Cell<?> insert)
{
return updating.ref != ref;
if (insert != previous)
{
long timeDelta = Math.abs(insert.timestamp() - previous.timestamp());
if (timeDelta < colUpdateTimeDelta)
colUpdateTimeDelta = timeDelta;
}
if (cloner != null)
insert = cloner.clone(insert);
dataSize += insert.dataSize() - previous.dataSize();
heapSize += insert.unsharedHeapSizeExcludingData() - previous.unsharedHeapSizeExcludingData();
return insert;
}
public ColumnData insert(ColumnData insert)
{
if (cloner != null)
insert = insert.clone(cloner);
dataSize += insert.dataSize();
heapSize += insert.unsharedHeapSizeExcludingData();
return insert;
}
@Override
public void delete(ColumnData existing)
{
dataSize -= existing.dataSize();
heapSize -= existing.unsharedHeapSizeExcludingData();
}
public void onAllocatedOnHeap(long heapSize)

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

@ -29,7 +29,7 @@ public class FilteredPartition extends ImmutableBTreePartition
{
public FilteredPartition(RowIterator rows)
{
super(rows.metadata(), rows.partitionKey(), build(rows, DeletionInfo.LIVE, false, 16));
super(rows.metadata(), rows.partitionKey(), build(rows, DeletionInfo.LIVE, false));
}
/**

29
src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java
View File

@ -25,6 +25,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Set;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.primitives.Ints;
@ -132,8 +133,8 @@ public class PartitionUpdate extends AbstractBTreePartition
MutableDeletionInfo deletionInfo = MutableDeletionInfo.live();
Holder holder = new Holder(
new RegularAndStaticColumns(
staticRow == null ? Columns.NONE : Columns.from(staticRow.columns()),
row == null ? Columns.NONE : Columns.from(row.columns())
staticRow == null ? Columns.NONE : Columns.from(staticRow),
row == null ? Columns.NONE : Columns.from(row)
),
row == null ? BTree.empty() : BTree.singleton(row),
deletionInfo,
@ -207,7 +208,7 @@ public class PartitionUpdate extends AbstractBTreePartition
{
iterator = RowIterators.withOnlyQueriedData(iterator, filter);
MutableDeletionInfo deletionInfo = MutableDeletionInfo.live();
Holder holder = build(iterator, deletionInfo, true, 16);
Holder holder = build(iterator, deletionInfo, true);
return new PartitionUpdate(iterator.metadata(), iterator.partitionKey(), holder, deletionInfo, false);
}
@ -490,6 +491,16 @@ public class PartitionUpdate extends AbstractBTreePartition
IndexRegistry.obtain(metadata()).validate(this);
}
@VisibleForTesting
public static PartitionUpdate unsafeConstruct(TableMetadata metadata,
DecoratedKey key,
Holder holder,
MutableDeletionInfo deletionInfo,
boolean canHaveShadowedData)
{
return new PartitionUpdate(metadata, key, holder, deletionInfo, canHaveShadowedData);
}
/**
* Interface for building partition updates geared towards human.
* <p>
@ -661,25 +672,25 @@ public class PartitionUpdate extends AbstractBTreePartition
assert header.rowEstimate >= 0;
MutableDeletionInfo.Builder deletionBuilder = MutableDeletionInfo.builder(header.partitionDeletion, metadata.comparator, false);
BTree.Builder<Row> rows = BTree.builder(metadata.comparator, header.rowEstimate);
rows.auto(false);
try (UnfilteredRowIterator partition = UnfilteredRowIteratorSerializer.serializer.deserialize(in, version, metadata, flag, header))
Object[] rows;
try (BTree.FastBuilder<Row> builder = BTree.fastBuilder();
UnfilteredRowIterator partition = UnfilteredRowIteratorSerializer.serializer.deserialize(in, version, metadata, flag, header))
{
while (partition.hasNext())
{
Unfiltered unfiltered = partition.next();
if (unfiltered.kind() == Unfiltered.Kind.ROW)
rows.add((Row)unfiltered);
builder.add((Row)unfiltered);
else
deletionBuilder.add((RangeTombstoneMarker)unfiltered);
}
rows = builder.build();
}
MutableDeletionInfo deletionInfo = deletionBuilder.build();
return new PartitionUpdate(metadata,
header.key,
new Holder(header.sHeader.columns(), rows.build(), deletionInfo, header.staticRow, header.sHeader.stats()),
new Holder(header.sHeader.columns(), rows, deletionInfo, header.staticRow, header.sHeader.stats()),
deletionInfo,
false);
}

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

@ -30,7 +30,7 @@ import org.apache.cassandra.db.marshal.ValueAccessor;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.serializers.MarshalException;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.memory.AbstractAllocator;
import org.apache.cassandra.utils.memory.ByteBufferCloner;
/**
* Base abstract class for {@code Cell} implementations.
@ -98,15 +98,17 @@ public abstract class AbstractCell<V> extends Cell<V>
return this;
}
public Cell<?> purgeDataOlderThan(long timestamp)
{
return this.timestamp() < timestamp ? null : this;
}
public Cell<?> copy(AbstractAllocator allocator)
@Override
public Cell<?> clone(ByteBufferCloner cloner)
{
CellPath path = path();
return new BufferCell(column, timestamp(), ttl(), localDeletionTime(), allocator.clone(buffer()), path == null ? null : path.copy(allocator));
return new BufferCell(column, timestamp(), ttl(), localDeletionTime(), cloner.clone(buffer()), path == null ? null : path.clone(cloner));
}
// note: while the cell returned may be different, the value is the same, so if the value is offheap it must be referenced inside a guarded context (or copied)

7
src/java/org/apache/cassandra/db/rows/ArrayCell.java
View File

@ -26,7 +26,7 @@ import org.apache.cassandra.db.marshal.ValueAccessor;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.memory.AbstractAllocator;
import org.apache.cassandra.utils.memory.ByteBufferCloner;
import static org.apache.cassandra.utils.ByteArrayUtil.EMPTY_BYTE_ARRAY;
@ -102,12 +102,13 @@ public class ArrayCell extends AbstractCell<byte[]>
return new ArrayCell(column, timestamp, ttl, localDeletionTime, EMPTY_BYTE_ARRAY, path);
}
public Cell<?> copy(AbstractAllocator allocator)
@Override
public Cell<?> clone(ByteBufferCloner cloner)
{
if (value.length == 0)
return this;
return new BufferCell(column, timestamp, ttl, localDeletionTime, allocator.clone(value), path == null ? null : path.copy(allocator));
return super.clone(cloner);
}
@Override

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

@ -20,19 +20,17 @@ package org.apache.cassandra.db.rows;
import java.nio.ByteBuffer;
import java.util.AbstractCollection;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;
import com.google.common.base.Function;
import com.google.common.collect.Collections2;
import com.google.common.collect.Iterators;
import com.google.common.primitives.Ints;
@ -53,12 +51,14 @@ import org.apache.cassandra.schema.DroppedColumn;
import org.apache.cassandra.utils.AbstractIterator;
import org.apache.cassandra.utils.BiLongAccumulator;
import org.apache.cassandra.utils.BulkIterator;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.LongAccumulator;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.btree.BTree;
import org.apache.cassandra.utils.btree.BTreeSearchIterator;
import org.apache.cassandra.utils.btree.UpdateFunction;
import org.apache.cassandra.utils.memory.Cloner;
/**
* Immutable implementation of a Row object.
@ -227,16 +227,7 @@ public class BTreeRow extends AbstractRow
private static int minDeletionTime(Object[] btree, LivenessInfo info, DeletionTime rowDeletion)
{
long min = Math.min(minDeletionTime(info), minDeletionTime(rowDeletion));
min = BTree.<ColumnData>accumulate(btree, (cd, l) -> {
int m = Math.min((int) l, minDeletionTime(cd));
return m != Integer.MIN_VALUE ? m : Long.MAX_VALUE;
}, min);
if (min == Long.MAX_VALUE)
return Integer.MIN_VALUE;
return Ints.checkedCast(min);
return (int) BTree.<ColumnData>accumulate(btree, (cd, l) -> Math.min(l, minDeletionTime(cd)), min);
}
public Clustering<?> clustering()
@ -410,9 +401,8 @@ public class BTreeRow extends AbstractRow
public Row markCounterLocalToBeCleared()
{
return transformAndFilter(primaryKeyLivenessInfo, deletion, (cd) -> cd.column().isCounterColumn()
? cd.markCounterLocalToBeCleared()
: cd);
return transform((cd) -> cd.column().isCounterColumn() ? cd.markCounterLocalToBeCleared()
: cd);
}
public boolean hasDeletion(int nowInSec)
@ -486,18 +476,40 @@ public class BTreeRow extends AbstractRow
return transformAndFilter(newInfo, newDeletion, cd -> cd.purgeDataOlderThan(timestamp));
}
private Row transformAndFilter(LivenessInfo info, Deletion deletion, Function<ColumnData, ColumnData> function)
@Override
public Row transformAndFilter(LivenessInfo info, Deletion deletion, Function<ColumnData, ColumnData> function)
{
Object[] transformed = BTree.transformAndFilter(btree, function);
return update(info, deletion, BTree.transformAndFilter(btree, function));
}
if (btree == transformed && info == this.primaryKeyLivenessInfo && deletion == this.deletion)
private Row update(LivenessInfo info, Deletion deletion, Object[] newTree)
{
if (btree == newTree && info == this.primaryKeyLivenessInfo && deletion == this.deletion)
return this;
if (info.isEmpty() && deletion.isLive() && BTree.isEmpty(transformed))
if (info.isEmpty() && deletion.isLive() && BTree.isEmpty(newTree))
return null;
int minDeletionTime = minDeletionTime(transformed, info, deletion.time());
return BTreeRow.create(clustering, info, deletion, transformed, minDeletionTime);
int minDeletionTime = minDeletionTime(newTree, info, deletion.time());
return BTreeRow.create(clustering, info, deletion, newTree, minDeletionTime);
}
@Override
public Row transformAndFilter(Function<ColumnData, ColumnData> function)
{
return transformAndFilter(primaryKeyLivenessInfo, deletion, function);
}
public Row transform(Function<ColumnData, ColumnData> function)
{
return update(primaryKeyLivenessInfo, deletion, BTree.transform(btree, function));
}
@Override
public Row clone(Cloner cloner)
{
Object[] tree = BTree.<ColumnData, ColumnData>transform(btree, c -> c.clone(cloner));
return BTreeRow.create(cloner.clone(clustering), primaryKeyLivenessInfo, deletion, tree);
}
public int dataSize()
@ -561,6 +573,43 @@ public class BTreeRow extends AbstractRow
return () -> new CellInLegacyOrderIterator(metadata, reversed);
}
public static Row merge(BTreeRow existing,
BTreeRow update,
ColumnData.PostReconciliationFunction reconcileF)
{
Object[] existingBtree = existing.btree;
Object[] updateBtree = update.btree;
LivenessInfo existingInfo = existing.primaryKeyLivenessInfo();
LivenessInfo updateInfo = update.primaryKeyLivenessInfo();
LivenessInfo livenessInfo = existingInfo.supersedes(updateInfo) ? existingInfo : updateInfo;
Row.Deletion rowDeletion = existing.deletion().supersedes(update.deletion()) ? existing.deletion() : update.deletion();
if (rowDeletion.deletes(livenessInfo))
livenessInfo = LivenessInfo.EMPTY;
else if (rowDeletion.isShadowedBy(livenessInfo))
rowDeletion = Row.Deletion.LIVE;
DeletionTime deletion = rowDeletion.time();
try (ColumnData.Reconciler reconciler = ColumnData.reconciler(reconcileF, deletion))
{
if (!rowDeletion.isLive())
{
if (rowDeletion == existing.deletion())
{
updateBtree = BTree.transformAndFilter(updateBtree, reconciler::retain);
}
else
{
existingBtree = BTree.transformAndFilter(existingBtree, reconciler::retain);
}
}
Object[] tree = BTree.update(existingBtree, updateBtree, ColumnData.comparator, reconciler);
return new BTreeRow(existing.clustering, livenessInfo, rowDeletion, tree, minDeletionTime(tree, livenessInfo, deletion));
}
}
private class CellIterator extends AbstractIterator<Cell<?>>
{
private Iterator<ColumnData> columnData = iterator();
@ -724,7 +773,8 @@ public class BTreeRow extends AbstractRow
lb++;
}
List<Object> buildFrom = new ArrayList<>(ub - lb);
Object[] buildFrom = new Object[ub - lb];
int buildFromCount = 0;
Cell<?> previous = null;
for (int i = lb; i < ub; i++)
{
@ -735,18 +785,21 @@ public class BTreeRow extends AbstractRow
if (previous != null && column.cellComparator().compare(previous, c) == 0)
{
c = Cells.reconcile(previous, c);
buildFrom.set(buildFrom.size() - 1, c);
buildFrom[buildFromCount - 1] = c;
}
else
{
buildFrom.add(c);
buildFrom[buildFromCount++] = c;
}
previous = c;
}
}
Object[] btree = BTree.build(buildFrom);
return new ComplexColumnData(column, btree, deletion);
try (BulkIterator<Cell> iterator = BulkIterator.of(buildFrom))
{
Object[] btree = BTree.build(iterator, buildFromCount, UpdateFunction.noOp());
return new ComplexColumnData(column, btree, deletion);
}
}
}

10
src/java/org/apache/cassandra/db/rows/BufferCell.java
View File

@ -26,6 +26,7 @@ import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.db.marshal.ByteType;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.memory.ByteBufferCloner;
import static java.lang.String.format;
@ -139,6 +140,15 @@ public class BufferCell extends AbstractCell<ByteBuffer>
return EMPTY_SIZE + ObjectSizes.sizeOnHeapOf(value) + (path == null ? 0 : path.unsharedHeapSize());
}
@Override
public Cell<?> clone(ByteBufferCloner cloner)
{
if (!value.hasRemaining())
return this;
return super.clone(cloner);
}
@Override
public long unsharedHeapSizeExcludingData()
{

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

@ -27,7 +27,8 @@ import org.apache.cassandra.db.marshal.ValueAccessor;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.utils.memory.AbstractAllocator;
import org.apache.cassandra.utils.memory.ByteBufferCloner;
import org.apache.cassandra.utils.memory.Cloner;
/**
* A cell is our atomic unit for a single value of a single column.
@ -157,7 +158,13 @@ public abstract class Cell<V> extends ColumnData
*/
public abstract Cell<?> withSkippedValue();
public abstract Cell<?> copy(AbstractAllocator allocator);
@Override
public final Cell<?> clone(Cloner cloner)
{
return cloner.clone(this);
}
public abstract Cell<?> clone(ByteBufferCloner cloner);
@Override
// Overrides super type to provide a more precise return type.

12
src/java/org/apache/cassandra/db/rows/CellPath.java
View File

@ -27,7 +27,7 @@ import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.memory.AbstractAllocator;
import org.apache.cassandra.utils.memory.ByteBufferCloner;
/**
* A path for a cell belonging to a complex column type (non-frozen collection or UDT).
@ -61,7 +61,7 @@ public abstract class CellPath implements IMeasurableMemory
digest.update(get(i));
}
public abstract CellPath copy(AbstractAllocator allocator);
public abstract CellPath clone(ByteBufferCloner cloner);
public abstract long unsharedHeapSizeExcludingData();
@ -123,9 +123,10 @@ public abstract class CellPath implements IMeasurableMemory
return value;
}
public CellPath copy(AbstractAllocator allocator)
@Override
public CellPath clone(ByteBufferCloner cloner)
{
return new SingleItemCellPath(allocator.clone(value));
return new SingleItemCellPath(cloner.clone(value));
}
@Override
@ -153,7 +154,8 @@ public abstract class CellPath implements IMeasurableMemory
throw new UnsupportedOperationException();
}
public CellPath copy(AbstractAllocator allocator)
@Override
public CellPath clone(ByteBufferCloner cloner)
{
return this;
}

112
src/java/org/apache/cassandra/db/rows/Cells.java
View File

@ -49,55 +49,6 @@ public abstract class Cells
collector.updateHasLegacyCounterShards(CounterCells.hasLegacyShards(cell));
}
/**
* Reconciles/merges two cells, one being an update to an existing cell,
* yielding index updates if appropriate.
* <p>
* Note that this method assumes that the provided cells can meaningfully
* be reconciled together, that is that those cells are for the same row and same
* column (and same cell path if the column is complex).
* <p>
* Also note that which cell is provided as {@code existing} and which is
* provided as {@code update} matters for index updates.
*
* @param existing the pre-existing cell, the one that is updated. This can be
* {@code null} if this reconciliation correspond to an insertion.
* @param update the newly added cell, the update. This can be {@code null} out
* of convenience, in which case this function simply copy {@code existing} to
* {@code writer}.
* @param deletion the deletion time that applies to the cells being considered.
* This deletion time may delete both {@code existing} or {@code update}.
* @param builder the row builder to which the result of the reconciliation is written.
*
* @return the timestamp delta between existing and update, or {@code Long.MAX_VALUE} if one
* of them is {@code null} or deleted by {@code deletion}).
*/
public static long reconcile(Cell<?> existing,
Cell<?> update,
DeletionTime deletion,
Row.Builder builder)
{
existing = existing == null || deletion.deletes(existing) ? null : existing;
update = update == null || deletion.deletes(update) ? null : update;
if (existing == null || update == null)
{
if (update != null)
{
builder.addCell(update);
}
else if (existing != null)
{
builder.addCell(existing);
}
return Long.MAX_VALUE;
}
Cell<?> reconciled = reconcile(existing, update);
builder.addCell(reconciled);
return Math.abs(existing.timestamp() - update.timestamp());
}
/**
* Reconciles/merge two cells.
* <p>
@ -210,69 +161,6 @@ public abstract class Cells
return new BufferCell(left.column(), timestamp, Cell.NO_TTL, Cell.NO_DELETION_TIME, merged, left.path());
}
/**
* Computes the reconciliation of a complex column given its pre-existing
* cells and the ones it is updated with, and generating index update if
* appropriate.
* <p>
* Note that this method assumes that the provided cells can meaningfully
* be reconciled together, that is that the cells are for the same row and same
* complex column.
* <p>
* Also note that which cells is provided as {@code existing} and which are
* provided as {@code update} matters for index updates.
*
* @param column the complex column the cells are for.
* @param existing the pre-existing cells, the ones that are updated. This can be
* {@code null} if this reconciliation correspond to an insertion.
* @param update the newly added cells, the update. This can be {@code null} out
* of convenience, in which case this function simply copy the cells from
* {@code existing} to {@code writer}.
* @param deletion the deletion time that applies to the cells being considered.
* This deletion time may delete cells in both {@code existing} and {@code update}.
* @param builder the row build to which the result of the reconciliation is written.
*
* @return the smallest timestamp delta between corresponding cells from existing and update. A
* timestamp delta being computed as the difference between a cell from {@code update} and the
* cell in {@code existing} having the same cell path (if such cell exists). If the intersection
* of cells from {@code existing} and {@code update} having the same cell path is empty, this
* returns {@code Long.MAX_VALUE}.
*/
public static long reconcileComplex(ColumnMetadata column,
Iterator<Cell<?>> existing,
Iterator<Cell<?>> update,
DeletionTime deletion,
Row.Builder builder)
{
Comparator<CellPath> comparator = column.cellPathComparator();
Cell<?> nextExisting = getNext(existing);
Cell<?> nextUpdate = getNext(update);
long timeDelta = Long.MAX_VALUE;
while (nextExisting != null || nextUpdate != null)
{
int cmp = nextExisting == null ? 1
: (nextUpdate == null ? -1
: comparator.compare(nextExisting.path(), nextUpdate.path()));
if (cmp < 0)
{
reconcile(nextExisting, null, deletion, builder);
nextExisting = getNext(existing);
}
else if (cmp > 0)
{
reconcile(null, nextUpdate, deletion, builder);
nextUpdate = getNext(update);
}
else
{
timeDelta = Math.min(timeDelta, reconcile(nextExisting, nextUpdate, deletion, builder));
nextExisting = getNext(existing);
nextUpdate = getNext(update);
}
}
return timeDelta;
}
/**
* Adds to the builder a representation of the given existing cell that, when merged/reconciled with the given
* update cell, produces the same result as merging the original with the update.

172
src/java/org/apache/cassandra/db/rows/ColumnData.java
View File

@ -23,8 +23,13 @@ import org.apache.cassandra.cache.IMeasurableMemory;
import org.apache.cassandra.db.Digest;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.db.DeletionPurger;
import org.apache.cassandra.db.DeletionTime;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.serializers.MarshalException;
import org.apache.cassandra.utils.btree.BTree;
import org.apache.cassandra.utils.btree.UpdateFunction;
import org.apache.cassandra.utils.caching.TinyThreadLocalPool;
import org.apache.cassandra.utils.memory.Cloner;
/**
* Generic interface for the data of a given column (inside a row).
@ -36,6 +41,171 @@ public abstract class ColumnData implements IMeasurableMemory
{
public static final Comparator<ColumnData> comparator = (cd1, cd2) -> cd1.column().compareTo(cd2.column());
/**
* Construct an UpdateFunction for reconciling normal ColumnData
* (i.e. not suitable for ComplexColumnDeletion sentinels, but suitable ComplexColumnData or Cell)
*
* @param updateF a consumer receiving all pairs of reconciled cells
* @param activeDeletion the row or partition deletion time to use for purging
*/
public static Reconciler reconciler(PostReconciliationFunction updateF, DeletionTime activeDeletion)
{
TinyThreadLocalPool.TinyPool<Reconciler> pool = Reconciler.POOL.get();
Reconciler reconciler = pool.poll();
if (reconciler == null)
reconciler = new Reconciler();
reconciler.init(updateF, activeDeletion);
reconciler.pool = pool;
return reconciler;
}
public static PostReconciliationFunction noOp = new PostReconciliationFunction()
{
@Override
public Cell<?> merge(Cell<?> previous, Cell<?> insert)
{
return insert;
}
@Override
public ColumnData insert(ColumnData insert)
{
return insert;
}
@Override
public void delete(ColumnData existing)
{
}
public void onAllocatedOnHeap(long delta)
{
}
};
public interface PostReconciliationFunction
{
ColumnData insert(ColumnData insert);
Cell<?> merge(Cell<?> previous, Cell<?> insert);
void delete(ColumnData existing);
void onAllocatedOnHeap(long delta);
}
public static class Reconciler implements UpdateFunction<ColumnData, ColumnData>, AutoCloseable
{
private static final TinyThreadLocalPool<Reconciler> POOL = new TinyThreadLocalPool<>();
private PostReconciliationFunction modifier;
private DeletionTime activeDeletion;
private TinyThreadLocalPool.TinyPool<Reconciler> pool;
private void init(PostReconciliationFunction modifier, DeletionTime activeDeletion)
{
this.modifier = modifier;
this.activeDeletion = activeDeletion;
}
public ColumnData merge(ColumnData existing, ColumnData update)
{
if (!(existing instanceof ComplexColumnData))
{
Cell<?> existingCell = (Cell) existing, updateCell = (Cell) update;
Cell<?> result = Cells.reconcile(existingCell, updateCell);
return modifier.merge(existingCell, result);
}
else
{
ComplexColumnData existingComplex = (ComplexColumnData) existing;
ComplexColumnData updateComplex = (ComplexColumnData) update;
DeletionTime existingDeletion = existingComplex.complexDeletion();
DeletionTime updateDeletion = updateComplex.complexDeletion();
DeletionTime maxComplexDeletion = existingDeletion.supersedes(updateDeletion) ? existingDeletion : updateDeletion;
Object[] existingTree = existingComplex.tree();
Object[] updateTree = updateComplex.tree();
Object[] cells;
try (Reconciler reconciler = reconciler(modifier, maxComplexDeletion))
{
if (!maxComplexDeletion.isLive())
{
if (maxComplexDeletion == existingDeletion)
{
updateTree = BTree.transformAndFilter(updateTree, reconciler::retain);
}
else
{
existingTree = BTree.transformAndFilter(existingTree, reconciler::retain);
}
}
cells = BTree.update(existingTree, updateTree, existingComplex.column.cellComparator(), (UpdateFunction) reconciler);
}
return new ComplexColumnData(existingComplex.column, cells, maxComplexDeletion);
}
}
@Override
public void onAllocatedOnHeap(long heapSize)
{
modifier.onAllocatedOnHeap(heapSize);
}
@Override
public ColumnData insert(ColumnData insert)
{
return modifier.insert(insert);
}
/**
* Checks if the specified value should be deleted or not.
*
* @param existing the existing value to check
* @return {@code null} if the value should be removed from the BTree or the existing value if it should not.
*/
public ColumnData retain(ColumnData existing)
{
if (!(existing instanceof ComplexColumnData))
{
if (activeDeletion.deletes((Cell) existing))
{
modifier.delete(existing);
return null;
}
}
else
{
ComplexColumnData existingComplex = (ComplexColumnData) existing;
if (activeDeletion.supersedes(existingComplex.complexDeletion()))
{
Object[] cells = BTree.transformAndFilter(existingComplex.tree(), this::retain);
return BTree.isEmpty(cells) ? null : new ComplexColumnData(existingComplex.column, cells, DeletionTime.LIVE);
}
}
return existing;
}
public void close()
{
activeDeletion = null;
modifier = null;
TinyThreadLocalPool.TinyPool<Reconciler> tmp = pool;
pool = null;
tmp.offer(this);
}
}
protected final ColumnMetadata column;
protected ColumnData(ColumnMetadata column)
{
@ -86,6 +256,8 @@ public abstract class ColumnData implements IMeasurableMemory
cd.digest(digest);
}
public abstract ColumnData clone(Cloner cloner);
/**
* Returns a copy of the data where all timestamps for live data have replaced by {@code newTimestamp} and
* all deletion timestamp by {@code newTimestamp - 1}.

42
src/java/org/apache/cassandra/db/rows/ComplexColumnData.java
View File

@ -20,16 +20,19 @@ package org.apache.cassandra.db.rows;
import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.Objects;
import java.util.function.BiFunction;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Function;
import org.apache.cassandra.db.DeletionPurger;
import org.apache.cassandra.db.DeletionTime;
import org.apache.cassandra.db.Digest;
import org.apache.cassandra.db.LivenessInfo;
import org.apache.cassandra.db.context.CounterContext;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.ByteType;
import org.apache.cassandra.db.marshal.CollectionType;
import org.apache.cassandra.db.marshal.SetType;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.DroppedColumn;
@ -38,6 +41,7 @@ import org.apache.cassandra.utils.LongAccumulator;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.SearchIterator;
import org.apache.cassandra.utils.btree.BTree;
import org.apache.cassandra.utils.memory.Cloner;
/**
* The data for a complex column, that is it's cells and potential complex
@ -54,7 +58,6 @@ public class ComplexColumnData extends ColumnData implements Iterable<Cell<?>>
private final DeletionTime complexDeletion;
// Only ArrayBackedRow should call this.
ComplexColumnData(ColumnMetadata column, Object[] cells, DeletionTime complexDeletion)
{
super(column);
@ -95,6 +98,11 @@ public class ComplexColumnData extends ColumnData implements Iterable<Cell<?>>
return complexDeletion;
}
Object[] tree()
{
return cells;
}
public Iterator<Cell<?>> iterator()
{
return BTree.iterator(cells);
@ -226,14 +234,25 @@ public class ComplexColumnData extends ColumnData implements Iterable<Cell<?>>
return new ComplexColumnData(column, newCells, newDeletion);
}
public ComplexColumnData transformAndFilter(Function<? super Cell<?>, ? extends Cell<?>> function)
{
return update(complexDeletion, BTree.transformAndFilter(cells, function));
}
public ComplexColumnData transformAndFilter(DeletionTime newDeletion, Function<? super Cell, ? extends Cell> function)
{
return update(newDeletion, BTree.transformAndFilter(cells, function));
}
public <V> ComplexColumnData transformAndFilter(BiFunction<? super Cell, ? super V, ? extends Cell> function, V param)
public <V> ComplexColumnData transform(Function<? super Cell<?>, ? extends Cell<?>> function)
{
return update(complexDeletion, BTree.transformAndFilter(cells, function, param));
return update(complexDeletion, BTree.transform(cells, function));
}
@Override
public ColumnData clone(Cloner cloner)
{
return transform(c -> cloner.clone(c));
}
public ComplexColumnData updateAllTimestamp(long newTimestamp)
@ -279,6 +298,21 @@ public class ComplexColumnData extends ColumnData implements Iterable<Cell<?>>
return Objects.hash(column(), complexDeletion(), BTree.hashCode(cells));
}
@Override
public String toString()
{
return String.format("[%s=%s %s]",
column().name,
complexDeletion.toString(),
BTree.toString(cells));
}
@VisibleForTesting
public static ComplexColumnData unsafeConstruct(ColumnMetadata column, Object[] cells, DeletionTime complexDeletion)
{
return new ComplexColumnData(column, cells, complexDeletion);
}
public static Builder builder()
{
return new Builder();

7
src/java/org/apache/cassandra/db/rows/RangeTombstoneBoundMarker.java
View File

@ -23,7 +23,7 @@ import org.apache.cassandra.db.marshal.ValueAccessor;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.db.*;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.memory.AbstractAllocator;
import org.apache.cassandra.utils.memory.ByteBufferCloner;
/**
* A range tombstone marker that indicates the bound of a range tombstone (start or end).
@ -140,9 +140,10 @@ public class RangeTombstoneBoundMarker extends AbstractRangeTombstoneMarker<Clus
return isClose(reversed) ? clustering() : null;
}
public RangeTombstoneBoundMarker copy(AbstractAllocator allocator)
@Override
public RangeTombstoneBoundMarker clone(ByteBufferCloner cloner)
{
return new RangeTombstoneBoundMarker(clustering().copy(allocator), deletion);
return new RangeTombstoneBoundMarker(clustering().clone(cloner), deletion);
}
public RangeTombstoneBoundMarker withNewOpeningDeletionTime(boolean reversed, DeletionTime newDeletionTime)

9
src/java/org/apache/cassandra/db/rows/RangeTombstoneBoundaryMarker.java
View File

@ -23,8 +23,10 @@ import java.util.Objects;
import org.apache.cassandra.db.marshal.ValueAccessor;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.db.*;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.memory.AbstractAllocator;
import org.apache.cassandra.utils.memory.ByteBufferCloner;
/**
* A range tombstone marker that represents a boundary between 2 range tombstones (i.e. it closes one range and open another).
@ -148,9 +150,10 @@ public class RangeTombstoneBoundaryMarker extends AbstractRangeTombstoneMarker<C
return !startDeletion.validate() || !endDeletion.validate();
}
public RangeTombstoneBoundaryMarker copy(AbstractAllocator allocator)
@Override
public RangeTombstoneBoundaryMarker clone(ByteBufferCloner cloner)
{
return new RangeTombstoneBoundaryMarker((ClusteringBoundary<ByteBuffer>) clustering().copy(allocator), endDeletion, startDeletion);
return new RangeTombstoneBoundaryMarker((ClusteringBoundary<ByteBuffer>) clustering().clone(cloner), endDeletion, startDeletion);
}
public RangeTombstoneBoundaryMarker withNewOpeningDeletionTime(boolean reversed, DeletionTime newDeletionTime)

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

@ -21,7 +21,7 @@ import java.util.*;
import org.apache.cassandra.cache.IMeasurableMemory;
import org.apache.cassandra.db.*;
import org.apache.cassandra.utils.memory.AbstractAllocator;
import org.apache.cassandra.utils.memory.ByteBufferCloner;
/**
* A marker for a range tombstone bound.
@ -46,7 +46,7 @@ public interface RangeTombstoneMarker extends Unfiltered, IMeasurableMemory
public ClusteringBound<?> openBound(boolean reversed);
public ClusteringBound<?> closeBound(boolean reversed);
public RangeTombstoneMarker copy(AbstractAllocator allocator);
public RangeTombstoneMarker clone(ByteBufferCloner cloner);
default public boolean isEmpty()
{

24
src/java/org/apache/cassandra/db/rows/Row.java
View File

@ -20,6 +20,7 @@ package org.apache.cassandra.db.rows;
import java.util.*;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Function;
import org.apache.cassandra.cache.IMeasurableMemory;
import org.apache.cassandra.db.*;
@ -33,7 +34,7 @@ import org.apache.cassandra.utils.MergeIterator;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.SearchIterator;
import org.apache.cassandra.utils.btree.BTree;
import org.apache.cassandra.utils.btree.UpdateFunction;
import org.apache.cassandra.utils.memory.Cloner;
/**
* Storage engine representation of a row.
@ -219,6 +220,27 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
*/
public Row filter(ColumnFilter filter, DeletionTime activeDeletion, boolean setActiveDeletionToRow, TableMetadata metadata);
/**
* Requires that {@code function} returns either {@code null} or {@code ColumnData} for the same column.
*
* Returns a copy of this row that:
* 1) {@code function} has been applied to the members of
* 2) doesn't include any {@code null} results of {@code function}
* 3) has precisely the provided {@code LivenessInfo} and {@code Deletion}
*/
public Row transformAndFilter(LivenessInfo info, Deletion deletion, Function<ColumnData, ColumnData> function);
/**
* Requires that {@code function} returns either {@code null} or {@code ColumnData} for the same column.
*
* Returns a copy of this row that:
* 1) {@code function} has been applied to the members of
* 2) doesn't include any {@code null} results of {@code function}
*/
public Row transformAndFilter(Function<ColumnData, ColumnData> function);
public Row clone(Cloner cloner);
/**
* Returns a copy of this row without any deletion info that should be purged according to {@code purger}.
*

120
src/java/org/apache/cassandra/db/rows/Rows.java
View File

@ -37,28 +37,6 @@ public abstract class Rows
public static final Row EMPTY_STATIC_ROW = BTreeRow.emptyRow(Clustering.STATIC_CLUSTERING);
public static Row.Builder copy(Row row, Row.Builder builder)
{
builder.newRow(row.clustering());
builder.addPrimaryKeyLivenessInfo(row.primaryKeyLivenessInfo());
builder.addRowDeletion(row.deletion());
for (ColumnData cd : row)
{
if (cd.column().isSimple())
{
builder.addCell((Cell<?>)cd);
}
else
{
ComplexColumnData complexData = (ComplexColumnData)cd;
builder.addComplexDeletion(complexData.column(), complexData.complexDeletion());
for (Cell<?> cell : complexData)
builder.addCell(cell);
}
}
return builder;
}
/**
* Creates a new simple row builder.
*
@ -253,90 +231,28 @@ public abstract class Rows
iter.next();
}
public static Row merge(Row row1, Row row2)
public static Row merge(Row existing, Row update)
{
Row.Builder builder = BTreeRow.sortedBuilder();
merge(row1, row2, builder);
return builder.build();
return merge(existing, update, ColumnData.noOp);
}
/**
* Merges two rows into the given builder, mainly for merging memtable rows. In addition to reconciling the cells
* in each row, the liveness info, and deletion times for the row and complex columns are also merged.
* Merges two rows. In addition to reconciling the cells in each row, the liveness info, and deletion times for
* the row and complex columns are also merged.
* <p>
* Note that this method assumes that the provided rows can meaningfully be reconciled together. That is,
* that the rows share the same clustering value, and belong to the same partition.
*
* @param existing
* @param update
* @param builder the row build to which the result of the reconciliation is written.
*
* @return the smallest timestamp delta between corresponding rows from existing and update. A
* timestamp delta being computed as the difference between the cells and DeletionTimes from {@code existing}
* and those in {@code update}.
* @return the row resulting from the merge.
*/
public static long merge(Row existing,
Row update,
Row.Builder builder)
public static Row merge(Row existing, Row update, ColumnData.PostReconciliationFunction onReconcile)
{
Clustering<?> clustering = existing.clustering();
builder.newRow(clustering);
LivenessInfo existingInfo = existing.primaryKeyLivenessInfo();
LivenessInfo updateInfo = update.primaryKeyLivenessInfo();
LivenessInfo mergedInfo = existingInfo.supersedes(updateInfo) ? existingInfo : updateInfo;
long timeDelta = Math.abs(existingInfo.timestamp() - mergedInfo.timestamp());
Row.Deletion rowDeletion = existing.deletion().supersedes(update.deletion()) ? existing.deletion() : update.deletion();
if (rowDeletion.deletes(mergedInfo))
mergedInfo = LivenessInfo.EMPTY;
else if (rowDeletion.isShadowedBy(mergedInfo))
rowDeletion = Row.Deletion.LIVE;
builder.addPrimaryKeyLivenessInfo(mergedInfo);
builder.addRowDeletion(rowDeletion);
DeletionTime deletion = rowDeletion.time();
Iterator<ColumnData> a = existing.iterator();
Iterator<ColumnData> b = update.iterator();
ColumnData nexta = a.hasNext() ? a.next() : null, nextb = b.hasNext() ? b.next() : null;
while (nexta != null | nextb != null)
{
int comparison = nexta == null ? 1 : nextb == null ? -1 : nexta.column.compareTo(nextb.column);
ColumnData cura = comparison <= 0 ? nexta : null;
ColumnData curb = comparison >= 0 ? nextb : null;
ColumnMetadata column = getColumnMetadata(cura, curb);
if (column.isSimple())
{
timeDelta = Math.min(timeDelta, Cells.reconcile((Cell<?>) cura, (Cell<?>) curb, deletion, builder));
}
else
{
ComplexColumnData existingData = (ComplexColumnData) cura;
ComplexColumnData updateData = (ComplexColumnData) curb;
DeletionTime existingDt = existingData == null ? DeletionTime.LIVE : existingData.complexDeletion();
DeletionTime updateDt = updateData == null ? DeletionTime.LIVE : updateData.complexDeletion();
DeletionTime maxDt = existingDt.supersedes(updateDt) ? existingDt : updateDt;
if (maxDt.supersedes(deletion))
builder.addComplexDeletion(column, maxDt);
else
maxDt = deletion;
Iterator<Cell<?>> existingCells = existingData == null ? null : existingData.iterator();
Iterator<Cell<?>> updateCells = updateData == null ? null : updateData.iterator();
timeDelta = Math.min(timeDelta, Cells.reconcileComplex(column, existingCells, updateCells, maxDt, builder));
}
if (cura != null)
nexta = a.hasNext() ? a.next() : null;
if (curb != null)
nextb = b.hasNext() ? b.next() : null;
}
return timeDelta;
assert existing instanceof BTreeRow;
assert update instanceof BTreeRow;
return BTreeRow.merge((BTreeRow) existing, (BTreeRow) update, onReconcile);
}
/**
@ -411,22 +327,4 @@ public abstract class Rows
Row row = builder.build();
return row != null && !row.isEmpty() ? row : null;
}
/**
* Returns the {@code ColumnMetadata} to use for merging the columns.
* If the 2 column metadata are different the latest one will be returned.
*/
private static ColumnMetadata getColumnMetadata(ColumnData cura, ColumnData curb)
{
if (cura == null)
return curb.column;
if (curb == null)
return cura.column;
if (ColumnMetadataVersionComparator.INSTANCE.compare(cura.column, curb.column) >= 0)
return cura.column;
return curb.column;
}
}

2
src/java/org/apache/cassandra/db/rows/WrappingUnfilteredRowIterator.java
View File

@ -29,7 +29,7 @@ import org.apache.cassandra.db.*;
* some of the methods.
* <p>
* Note that if most of what you want to do is modifying/filtering the returned
* {@code Unfiltered}, {@link org.apache.cassandra.db.transform.Transformation#apply(UnfilteredRowIterator,Transformation)} can be a simpler option.
* {@code Unfiltered}, {@link org.apache.cassandra.db.transform.Transformation#merge(UnfilteredRowIterator,Transformation)} can be a simpler option.
*/
public abstract class WrappingUnfilteredRowIterator extends UnmodifiableIterator<Unfiltered> implements UnfilteredRowIterator
{

24
src/java/org/apache/cassandra/db/view/TableViews.java
View File

@ -39,6 +39,7 @@ import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.schema.TableMetadataRef;
import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.btree.BTree;
import org.apache.cassandra.utils.btree.BTreeSet;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
@ -423,20 +424,23 @@ public class TableViews extends AbstractCollection<View>
// If we had some slices from the deletions above, we'll continue using that. Otherwise, it's more efficient to build
// a names query.
BTreeSet.Builder<Clustering<?>> namesBuilder = sliceBuilder == null ? BTreeSet.builder(metadata.comparator) : null;
for (Row row : updates)
NavigableSet<Clustering<?>> names;
try (BTree.FastBuilder<Clustering<?>> namesBuilder = sliceBuilder == null ? BTree.fastBuilder() : null)
{
// Don't read the existing state if we can prove the update won't affect any views
if (!affectsAnyViews(key, row, views))
continue;
for (Row row : updates)
{
// Don't read the existing state if we can prove the update won't affect any views
if (!affectsAnyViews(key, row, views))
continue;
if (namesBuilder == null)
sliceBuilder.add(Slice.make(row.clustering()));
else
namesBuilder.add(row.clustering());
if (namesBuilder == null)
sliceBuilder.add(Slice.make(row.clustering()));
else
namesBuilder.add(row.clustering());
}
names = namesBuilder == null ? null : BTreeSet.wrap(namesBuilder.build(), metadata.comparator);
}
NavigableSet<Clustering<?>> names = namesBuilder == null ? null : namesBuilder.build();
// If we have a slice builder, it means we had some deletions and we have to read. But if we had
// only row updates, it's possible none of them affected the views, in which case we have nothing
// to do.

4
src/java/org/apache/cassandra/dht/LocalPartitioner.java
View File

@ -28,7 +28,7 @@ import org.apache.cassandra.db.CachedHashDecoratedKey;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.memory.HeapAllocator;
import org.apache.cassandra.utils.memory.HeapCloner;
public class LocalPartitioner implements IPartitioner
{
@ -140,7 +140,7 @@ public class LocalPartitioner implements IPartitioner
public LocalToken(ByteBuffer token)
{
super(HeapAllocator.instance.clone(token));
super(HeapCloner.instance.clone(token));
}
@Override

6
src/java/org/apache/cassandra/index/internal/CassandraIndexSearcher.java
View File

@ -95,10 +95,8 @@ public abstract class CassandraIndexSearcher implements Index.Searcher
if (filter instanceof ClusteringIndexNamesFilter)
{
NavigableSet<Clustering<?>> requested = ((ClusteringIndexNamesFilter)filter).requestedRows();
BTreeSet.Builder<Clustering<?>> clusterings = BTreeSet.builder(index.getIndexComparator());
for (Clustering<?> c : requested)
clusterings.add(makeIndexClustering(pk, c));
return new ClusteringIndexNamesFilter(clusterings.build(), filter.isReversed());
BTreeSet<Clustering<?>> clusterings = BTreeSet.copy(requested, index.getIndexComparator());
return new ClusteringIndexNamesFilter(clusterings, filter.isReversed());
}
else
{

4
src/java/org/apache/cassandra/io/sstable/SSTable.java
View File

@ -54,7 +54,7 @@ import org.apache.cassandra.schema.TableMetadataRef;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.TimeUUID;
import org.apache.cassandra.utils.memory.HeapAllocator;
import org.apache.cassandra.utils.memory.HeapCloner;
import static org.apache.cassandra.io.util.File.WriteMode.APPEND;
import static org.apache.cassandra.service.ActiveRepairService.NO_PENDING_REPAIR;
@ -164,7 +164,7 @@ public abstract class SSTable
public static DecoratedKey getMinimalKey(DecoratedKey key)
{
return key.getKey().position() > 0 || key.getKey().hasRemaining() || !key.getKey().hasArray()
? new BufferDecoratedKey(key.getToken(), HeapAllocator.instance.clone(key.getKey()))
? new BufferDecoratedKey(key.getToken(), HeapCloner.instance.clone(key.getKey()))
: key;
}

85
src/java/org/apache/cassandra/utils/btree/BTree.java
View File

@ -149,7 +149,7 @@ public class BTree
{
updateF.onAllocatedOnHeap(ObjectSizes.sizeOfReferenceArray(values.length));
for (int i = 0; i < size; i++)
values[i] = updateF.apply((I) values[i]);
values[i] = updateF.insert((I) values[i]);
}
return values;
}
@ -166,7 +166,7 @@ public class BTree
if (!isSimple(updateF))
{
for (int i = 0; i < size; i++)
values[i] = updateF.apply((I) values[i]);
values[i] = updateF.insert((I) values[i]);
}
return values;
}
@ -220,7 +220,7 @@ public class BTree
while (remaining >= threshold)
{
branch[keyCount + i] = buildLeaf(source, MAX_KEYS, updateF);
branch[i] = isSimple(updateF) ? source.next() : updateF.apply(source.next());
branch[i] = isSimple(updateF) ? source.next() : updateF.insert(source.next());
remaining -= MAX_KEYS + 1;
sizeMap[i++] = size - remaining - 1;
}
@ -228,7 +228,7 @@ public class BTree
{
int childSize = remaining / 2;
branch[keyCount + i] = buildLeaf(source, childSize, updateF);
branch[i] = isSimple(updateF) ? source.next() : updateF.apply(source.next());
branch[i] = isSimple(updateF) ? source.next() : updateF.insert(source.next());
remaining -= childSize + 1;
sizeMap[i++] = size - remaining - 1;
}
@ -250,7 +250,7 @@ public class BTree
while (remaining >= threshold)
{
branch[keyCount + i] = buildPerfectDense(source, height, updateF);
branch[i] = isSimple(updateF) ? source.next() : updateF.apply(source.next());
branch[i] = isSimple(updateF) ? source.next() : updateF.insert(source.next());
remaining -= denseChildSize + 1;
sizeMap[i++] = size - remaining - 1;
}
@ -263,7 +263,7 @@ public class BTree
assert grandChildCount >= MIN_KEYS + 1;
int childSize = grandChildCount * (denseGrandChildSize + 1) - 1;
branch[keyCount + i] = buildMaximallyDense(source, grandChildCount, childSize, height, updateF);
branch[i] = isSimple(updateF) ? source.next() : updateF.apply(source.next());
branch[i] = isSimple(updateF) ? source.next() : updateF.insert(source.next());
remaining -= childSize + 1;
sizeMap[i++] = size - remaining - 1;
}
@ -304,7 +304,7 @@ public class BTree
for (int i = 0; i < keyCount; i++)
{
node[keyCount + i] = buildLeafWithoutSizeTracking(source, childSize, updateF);
node[i] = isSimple(updateF) ? source.next() : updateF.apply(source.next());
node[i] = isSimple(updateF) ? source.next() : updateF.insert(source.next());
}
node[2 * keyCount] = buildLeafWithoutSizeTracking(source, childSize, updateF);
}
@ -314,7 +314,7 @@ public class BTree
{
Object[] child = buildPerfectDenseWithoutSizeTracking(source, height - 1, updateF);
node[keyCount + i] = child;
node[i] = isSimple(updateF) ? source.next() : updateF.apply(source.next());
node[i] = isSimple(updateF) ? source.next() : updateF.insert(source.next());
}
node[2 * keyCount] = buildPerfectDenseWithoutSizeTracking(source, height - 1, updateF);
}
@ -334,40 +334,44 @@ public class BTree
* <p>
* Note that {@code UpdateFunction.noOp} is assumed to indicate a lack of interest in which value survives.
*/
public static <Compare, Existing extends Compare, Insert extends Compare> Object[] update(Object[] update, Object[] insert, Comparator<? super Compare> comparator, UpdateFunction<Insert, Existing> updateF)
public static <Compare, Existing extends Compare, Insert extends Compare> Object[] update(Object[] toUpdate,
Object[] insert,
Comparator<? super Compare> comparator,
UpdateFunction<Insert, Existing> updateF)
{
// perform some initial obvious optimisations
if (isEmpty(insert))
return update; // do nothing if update is empty
return toUpdate; // do nothing if update is empty
if (isEmpty(update))
if (isEmpty(toUpdate))
{
if (isSimple(updateF))
return insert; // if update is empty and updateF is trivial, return our new input
// if update is empty and updateF is non-trivial, perform a simple fast transformation of the input tree
insert = BTree.transform(insert, updateF);
insert = BTree.transform(insert, updateF::insert);
updateF.onAllocatedOnHeap(sizeOnHeapOf(insert));
return insert;
}
if (isLeaf(update) && isLeaf(insert))
if (isLeaf(toUpdate) && isLeaf(insert))
{
// if both are leaves, perform a tight-loop leaf variant of update
// possibly flipping the input order if sizes suggest and updateF permits
if (updateF == (UpdateFunction) UpdateFunction.noOp && update.length < insert.length)
if (updateF == (UpdateFunction) UpdateFunction.noOp && toUpdate.length < insert.length)
{
Object[] tmp = update;
update = insert;
Object[] tmp = toUpdate;
toUpdate = insert;
insert = tmp;
}
return updateLeaves(update, insert, comparator, updateF);
return updateLeaves(toUpdate, insert, comparator, updateF);
}
if (!isLeaf(insert) && isSimple(updateF))
{
// consider flipping the order of application, if update is much larger than insert and applying unary no-op
int updateSize = size(update);
int updateSize = size(toUpdate);
int insertSize = size(insert);
int scale = Integer.numberOfLeadingZeros(updateSize) - Integer.numberOfLeadingZeros(insertSize);
if (scale >= 4)
@ -375,8 +379,8 @@ public class BTree
// i.e. at roughly 16x the size, or one tier deeper - very arbitrary, should pick more carefully
// experimentally, at least at 64x the size the difference in performance is ~10x
Object[] tmp = insert;
insert = update;
update = tmp;
insert = toUpdate;
toUpdate = tmp;
if (updateF != (UpdateFunction) UpdateFunction.noOp)
updateF = ((UpdateFunction.Simple) updateF).flip();
}
@ -384,14 +388,17 @@ public class BTree
try (Updater<Compare, Existing, Insert> updater = Updater.get())
{
return updater.update(update, insert, comparator, updateF);
return updater.update(toUpdate, insert, comparator, updateF);
}
}
/**
* A fast tight-loop variant of updating one btree with another, when both are leaves.
*/
public static <Compare, Existing extends Compare, Insert extends Compare> Object[] updateLeaves(Object[] unode, Object[] inode, Comparator<? super Compare> comparator, UpdateFunction<Insert, Existing> updateF)
public static <Compare, Existing extends Compare, Insert extends Compare> Object[] updateLeaves(Object[] unode,
Object[] inode,
Comparator<? super Compare> comparator,
UpdateFunction<Insert, Existing> updateF)
{
int upos = -1, usz = sizeOfLeaf(unode);
Existing uk = (Existing) unode[0];
@ -414,7 +421,7 @@ public class BTree
}
else // c == 0
{
merged = updateF.apply(uk, ik);
merged = updateF.merge(uk, ik);
if (merged != uk)
break;
if (++ipos == isz)
@ -434,8 +441,7 @@ public class BTree
if (upos > 0)
{
// copy any initial section that is unmodified
System.arraycopy(unode, 0, builder.leaf().buffer, 0, upos);
builder.leaf().count = upos;
builder.leaf().copy(unode, 0, upos);
}
// handle prior loop's exit condition
@ -450,7 +456,7 @@ public class BTree
}
else // c > 0
{
builder.add(updateF.apply(ik));
builder.add(updateF.insert(ik));
}
if (++ipos < isz)
ik = (Insert) inode[ipos];
@ -463,7 +469,7 @@ public class BTree
{
if (c == 0)
{
builder.leaf().addKey(updateF.apply(uk, ik));
builder.leaf().addKey(updateF.merge(uk, ik));
++upos;
++ipos;
if (upos == usz || ipos == isz)
@ -2569,12 +2575,11 @@ public class BTree
}
/**
* Copy the contents of {@code source[from..to)} to {@code buffer}, overflowing as necessary.
* Applies {@code updateF} to the contents before insertion.
* Copy the contents of the data to {@code buffer}, overflowing as necessary.
*/
<Insert, Existing> void copy(Object[] source, int offset, int length, UpdateFunction<Insert, Existing> updateF)
<Insert, Existing> void copy(Object[] source, int offset, int length, UpdateFunction<Insert, Existing> apply)
{
if (isSimple(updateF))
if (isSimple(apply))
{
copy(source, offset, length);
return;
@ -2584,15 +2589,17 @@ public class BTree
{
int copy = MAX_KEYS - count;
for (int i = 0; i < copy; ++i)
buffer[count + i] = updateF.apply((Insert) source[offset + i]);
buffer[count + i] = apply.insert((Insert) source[offset + i]);
offset += copy;
// implicitly: leaf().count = MAX_KEYS;
overflow(updateF.apply((Insert) source[offset++]));
overflow(apply.insert((Insert) source[offset++]));
length -= 1 + copy;
}
for (int i = 0; i < length; ++i)
buffer[count + i] = updateF.apply((Insert) source[offset + i]);
buffer[count + i] = apply.insert((Insert) source[offset + i]);
count += length;
}
/**
@ -3487,7 +3494,7 @@ public class BTree
if (c == 0)
{
// ik matches next key
builder.addKey(updateF.apply(nextUKey, ik));
builder.addKey(updateF.merge(nextUKey, ik));
ik = insert.next();
}
else
@ -3516,7 +3523,7 @@ public class BTree
{
if (c == 0)
{
leaf().addKey(updateF.apply(uk, ik));
leaf().addKey(updateF.merge(uk, ik));
if (++upos < usz)
uk = (Existing) unode[upos];
ik = insert.next();
@ -3542,7 +3549,7 @@ public class BTree
}
else
{
builder.addKey(isSimple(updateF) ? ik : updateF.apply(ik));
builder.addKey(isSimple(updateF) ? ik : updateF.insert(ik));
c = insert.copyKeysSmallerThan(uk, comparator, builder, updateF); // 0 on match, -1 otherwise
ik = insert.next();
if (ik == null)
@ -3554,7 +3561,7 @@ public class BTree
}
if (uub == null || comparator.compare(ik, uub) < 0)
{
builder.addKey(isSimple(updateF) ? ik : updateF.apply(ik));
builder.addKey(isSimple(updateF) ? ik : updateF.insert(ik));
insert.copyKeysSmallerThan(uub, comparator, builder, updateF); // 0 on match, -1 otherwise
ik = insert.next();
}
@ -4162,7 +4169,7 @@ public class BTree
int cmp = compareWithMaybeInfinity(comparator, branchKey, bound);
if (cmp >= 0)
return -cmp;
builder.addKey(isSimple(transformer) ? branchKey : transformer.apply(branchKey));
builder.addKey(isSimple(transformer) ? branchKey : transformer.insert(branchKey));
advanceBranch(node, position + 1);
}
}

23
src/java/org/apache/cassandra/utils/btree/UpdateFunction.java
View File

@ -20,19 +20,28 @@ package org.apache.cassandra.utils.btree;
import java.util.function.BiFunction;
import com.google.common.base.Function;
/**
* An interface defining a function to be applied to both the object we are replacing in a BTree and
* the object that is intended to replace it, returning the object to actually replace it.
* An interface defining the method to be applied to the existing and replacing object in a BTree. The objects returned
* by the methods will be the object that need to be stored in the BTree.
*/
public interface UpdateFunction<K, V> extends Function<K, V>
public interface UpdateFunction<K, V>
{
/**
* Computes the value that should be inserted in the BTree.
*
* @param insert the update value
* @return the value that should be inserted in the BTree
*/
V insert(K insert);
/**
* Computes the result of merging the existing value with the one from the update.
*
* @param replacing the value in the original tree we have matched
* @param update the value in the updating collection that matched
* @return the value to insert into the new tree
*/
V apply(V replacing, K update);
V merge(V replacing, K update);
/**
* @param heapSize extra heap space allocated (over previous tree)
@ -48,13 +57,13 @@ public interface UpdateFunction<K, V> extends Function<K, V>
}
@Override
public V apply(V v)
public V insert(V v)
{
return v;
}
@Override
public V apply(V replacing, V update)
public V merge(V replacing, V update)
{
return wrapped.apply(replacing, update);
}

84
src/java/org/apache/cassandra/utils/memory/AbstractAllocator.java → src/java/org/apache/cassandra/utils/memory/ByteBufferCloner.java
View File

@ -7,46 +7,68 @@
* "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
* 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.
* 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.memory;
import java.nio.ByteBuffer;
import org.apache.cassandra.db.BufferDecoratedKey;
import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.marshal.ByteArrayAccessor;
import org.apache.cassandra.db.marshal.ByteBufferAccessor;
import org.apache.cassandra.db.marshal.ValueAccessor;
import org.apache.cassandra.db.rows.BTreeRow;
import org.apache.cassandra.db.rows.Cell;
import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.utils.ByteBufferUtil;
public abstract class AbstractAllocator
/**
* Cloner class that can be use to clone partition elements using on-heap or off-heap buffers.
*
*/
public abstract class ByteBufferCloner implements Cloner
{
/**
* Allocate a slice of the given length.
*/
public ByteBuffer clone(ByteBuffer buffer)
public abstract ByteBuffer allocate(int size);
@Override
public DecoratedKey clone(DecoratedKey key)
{
return new BufferDecoratedKey(key.getToken(), clone(key.getKey()));
}
@Override
public Clustering<?> clone(Clustering<?> clustering)
{
return clustering.clone(this);
}
@Override
public Cell<?> clone(Cell<?> cell)
{
return cell.clone(this);
}
public final ByteBuffer clone(ByteBuffer buffer)
{
return clone(buffer, ByteBufferAccessor.instance);
}
/**
* Allocate a slice of the given length.
*/
public ByteBuffer clone(byte[] bytes)
public final ByteBuffer clone(byte[] bytes)
{
return clone(bytes, ByteArrayAccessor.instance);
}
public <V> ByteBuffer clone(V value, ValueAccessor<V> accessor)
public final <V> ByteBuffer clone(V value, ValueAccessor<V> accessor)
{
assert value != null;
int size = accessor.size(value);
@ -59,34 +81,4 @@ public abstract class AbstractAllocator
cloned.reset();
return cloned;
}
public abstract ByteBuffer allocate(int size);
public Row.Builder cloningBTreeRowBuilder()
{
return new CloningBTreeRowBuilder(this);
}
private static class CloningBTreeRowBuilder extends BTreeRow.Builder
{
private final AbstractAllocator allocator;
private CloningBTreeRowBuilder(AbstractAllocator allocator)
{
super(true);
this.allocator = allocator;
}
@Override
public void newRow(Clustering<?> clustering)
{
super.newRow(clustering.copy(allocator));
}
@Override
public void addCell(Cell<?> cell)
{
super.addCell(cell.copy(allocator));
}
}
}
}

54
src/java/org/apache/cassandra/utils/memory/Cloner.java Normal file
View File

@ -0,0 +1,54 @@
/*
* 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.memory;
import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.rows.Cell;
/**
* Allow cloning of partition elements
*
*/
public interface Cloner
{
/**
* Clones the specified key.
*
* @param key the key to clone
* @return the cloned key
*/
DecoratedKey clone(DecoratedKey key);
/**
* Clones the specified clustering.
*
* @param clustering the clustering to clone
* @return the cloned clustering
*/
Clustering<?> clone(Clustering<?> clustering);
/**
* Clones the specified cell.
*
* @param cell the cell to clone
* @return the cloned cell
*/
Cell<?> clone(Cell<?> cell);
}

59
src/java/org/apache/cassandra/utils/memory/ContextAllocator.java
View File

@ -1,59 +0,0 @@
/*
* 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.memory;
import java.nio.ByteBuffer;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.concurrent.OpOrder;
/**
* Wraps calls to a PoolAllocator with the provided writeOp. Also doubles as a Function that clones Cells
* using itself
*/
public final class ContextAllocator extends AbstractAllocator
{
private final OpOrder.Group opGroup;
private final MemtableBufferAllocator allocator;
public ContextAllocator(OpOrder.Group opGroup, MemtableBufferAllocator allocator)
{
this.opGroup = opGroup;
this.allocator = allocator;
}
@Override
public ByteBuffer clone(ByteBuffer buffer)
{
assert buffer != null;
if (buffer.remaining() == 0)
return ByteBufferUtil.EMPTY_BYTE_BUFFER;
ByteBuffer cloned = allocate(buffer.remaining());
cloned.mark();
cloned.put(buffer.duplicate());
cloned.reset();
return cloned;
}
public ByteBuffer allocate(int size)
{
return allocator.allocate(size, opGroup);
}
}

8
src/java/org/apache/cassandra/utils/memory/EnsureOnHeap.java
View File

@ -52,14 +52,14 @@ public abstract class EnsureOnHeap extends Transformation
public DecoratedKey applyToPartitionKey(DecoratedKey key)
{
return new BufferDecoratedKey(key.getToken(), HeapAllocator.instance.clone(key.getKey()));
return new BufferDecoratedKey(key.getToken(), HeapCloner.instance.clone(key.getKey()));
}
public Row applyToRow(Row row)
{
if (row == null)
return null;
return Rows.copy(row, HeapAllocator.instance.cloningBTreeRowBuilder()).build();
return row.clone(HeapCloner.instance);
}
public Row applyToStatic(Row row)
@ -71,7 +71,7 @@ public abstract class EnsureOnHeap extends Transformation
public RangeTombstoneMarker applyToMarker(RangeTombstoneMarker marker)
{
return marker.copy(HeapAllocator.instance);
return marker.clone(HeapCloner.instance);
}
public UnfilteredRowIterator applyToPartition(UnfilteredRowIterator partition)
@ -111,7 +111,7 @@ public abstract class EnsureOnHeap extends Transformation
public DeletionInfo applyToDeletionInfo(DeletionInfo deletionInfo)
{
return deletionInfo.copy(HeapAllocator.instance);
return deletionInfo.clone(HeapCloner.instance);
}
}

41
src/java/org/apache/cassandra/utils/memory/HeapAllocator.java
View File

@ -1,41 +0,0 @@
/*
* 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.memory;
import java.nio.ByteBuffer;
public final class HeapAllocator extends AbstractAllocator
{
public static final HeapAllocator instance = new HeapAllocator();
/**
* Normally you should use HeapAllocator.instance, since there is no per-Allocator state.
* This is exposed so that the reflection done by Memtable works when SlabAllocator is disabled.
*/
private HeapAllocator() {}
public ByteBuffer allocate(int size)
{
return ByteBuffer.allocate(size);
}
public boolean allocatingOnHeap()
{
return true;
}
}

37
src/java/org/apache/cassandra/utils/memory/HeapCloner.java Normal file
View File

@ -0,0 +1,37 @@
/*
* 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.memory;
import java.nio.ByteBuffer;
/**
* Cloner class that can be use to clone partition elements on heap.
*
*/
public final class HeapCloner extends ByteBufferCloner
{
public static final HeapCloner instance = new HeapCloner();
private HeapCloner() {}
public ByteBuffer allocate(int size)
{
return ByteBuffer.allocate(size);
}
}

19
src/java/org/apache/cassandra/utils/memory/HeapPool.java
View File

@ -20,11 +20,14 @@ package org.apache.cassandra.utils.memory;
import java.nio.ByteBuffer;
import com.google.common.annotations.VisibleForTesting;
import org.apache.cassandra.utils.Shared;
import org.apache.cassandra.utils.concurrent.OpOrder;
import static org.apache.cassandra.utils.Shared.Scope.SIMULATION;
public class HeapPool extends MemtablePool
{
private static final EnsureOnHeap ENSURE_NOOP = new EnsureOnHeap.NoOp();
@ -39,9 +42,11 @@ public class HeapPool extends MemtablePool
return new Allocator(this);
}
private static class Allocator extends MemtableBufferAllocator
@VisibleForTesting
public static class Allocator extends MemtableBufferAllocator
{
Allocator(HeapPool pool)
@VisibleForTesting
public Allocator(HeapPool pool)
{
super(pool.onHeap.newAllocator(), pool.offHeap.newAllocator());
}
@ -56,6 +61,11 @@ public class HeapPool extends MemtablePool
{
return ENSURE_NOOP;
}
public Cloner cloner(OpOrder.Group opGroup)
{
return allocator(opGroup);
}
}
public static class Logged extends MemtablePool
@ -116,6 +126,11 @@ public class HeapPool extends MemtablePool
{
return ENSURE_NOOP;
}
public Cloner cloner(OpOrder.Group opGroup)
{
return allocator(opGroup);
}
}
SubPool getSubPool(long limit, float cleanThreshold)

7
src/java/org/apache/cassandra/utils/memory/MemtableAllocator.java
View File

@ -24,8 +24,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.codahale.metrics.Timer;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.utils.concurrent.OpOrder;
import org.apache.cassandra.utils.concurrent.WaitQueue;
@ -63,10 +62,10 @@ public abstract class MemtableAllocator
this.offHeap = offHeap;
}
public abstract Row.Builder rowBuilder(OpOrder.Group opGroup);
public abstract DecoratedKey clone(DecoratedKey key, OpOrder.Group opGroup);
public abstract EnsureOnHeap ensureOnHeap();
public abstract Cloner cloner(OpOrder.Group opGroup);
public SubAllocator onHeap()
{
return onHeap;

23
src/java/org/apache/cassandra/utils/memory/MemtableBufferAllocator.java
View File

@ -19,8 +19,6 @@ package org.apache.cassandra.utils.memory;
import java.nio.ByteBuffer;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.utils.concurrent.OpOrder;
public abstract class MemtableBufferAllocator extends MemtableAllocator
@ -30,20 +28,17 @@ public abstract class MemtableBufferAllocator extends MemtableAllocator
super(onHeap, offHeap);
}
public Row.Builder rowBuilder(OpOrder.Group writeOp)
{
return allocator(writeOp).cloningBTreeRowBuilder();
}
public DecoratedKey clone(DecoratedKey key, OpOrder.Group writeOp)
{
return new BufferDecoratedKey(key.getToken(), allocator(writeOp).clone(key.getKey()));
}
public abstract ByteBuffer allocate(int size, OpOrder.Group opGroup);
protected AbstractAllocator allocator(OpOrder.Group writeOp)
protected Cloner allocator(OpOrder.Group opGroup)
{
return new ContextAllocator(writeOp, this);
return new ByteBufferCloner()
{
@Override
public ByteBuffer allocate(int size)
{
return MemtableBufferAllocator.this.allocate(size, opGroup);
}
};
}
}

31
src/java/org/apache/cassandra/utils/memory/NativeAllocator.java
View File

@ -27,6 +27,7 @@ import org.apache.cassandra.db.*;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.utils.concurrent.OpOrder;
import org.apache.cassandra.utils.concurrent.Semaphore;
import org.apache.cassandra.utils.concurrent.OpOrder.Group;
import static org.apache.cassandra.utils.concurrent.Semaphore.newSemaphore;
@ -100,6 +101,35 @@ public class NativeAllocator extends MemtableAllocator
return new NativeDecoratedKey(key.getToken(), this, writeOp, key.getKey());
}
@Override
public Cloner cloner(Group opGroup)
{
return new Cloner()
{
@Override
public DecoratedKey clone(DecoratedKey key)
{
return NativeAllocator.this.clone(key, opGroup);
}
@Override
public Clustering<?> clone(Clustering<?> clustering)
{
if (clustering != Clustering.STATIC_CLUSTERING)
return new NativeClustering(NativeAllocator.this, opGroup, clustering);
return Clustering.STATIC_CLUSTERING;
}
@Override
public Cell<?> clone(Cell<?> cell)
{
return new NativeCell(NativeAllocator.this, opGroup, cell);
}
};
}
public EnsureOnHeap ensureOnHeap()
{
return cloneToHeap;
@ -254,5 +284,4 @@ public class NativeAllocator extends MemtableAllocator
"waste=" + Math.max(0, capacity - nextFreeOffset.get());
}
}
}

4
src/java/org/apache/cassandra/utils/memory/SlabAllocator.java
View File

@ -152,9 +152,9 @@ public class SlabAllocator extends MemtableBufferAllocator
}
}
protected AbstractAllocator allocator(OpOrder.Group writeOp)
public Cloner cloner(OpOrder.Group writeOp)
{
return new ContextAllocator(writeOp, this);
return allocator(writeOp);
}
/**

8
test/burn/org/apache/cassandra/utils/LongBTreeTest.java
View File

@ -1109,14 +1109,18 @@ public class LongBTreeTest
public static final class InverseNoOp<V> implements UpdateFunction<V, V>
{
public static final InverseNoOp instance = new InverseNoOp();
public V apply(V replacing, V update)
public V merge(V replacing, V update)
{
return update;
}
public void onAllocatedOnHeap(long heapSize)
{
}
public V apply(V v)
public V insert(V v)
{
return v;
}
public V retain(V v)
{
return v;
}

615
test/microbench/org/apache/cassandra/test/microbench/btree/AtomicBTreePartitionUpdateBench.java Normal file
View File

@ -0,0 +1,615 @@
/*
* 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.test.microbench.btree;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.IntToLongFunction;
import java.util.stream.IntStream;
import com.google.common.collect.ImmutableList;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.ColumnIdentifier;
import org.apache.cassandra.db.BufferDecoratedKey;
import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.DeletionInfo;
import org.apache.cassandra.db.DeletionTime;
import org.apache.cassandra.db.LivenessInfo;
import org.apache.cassandra.db.MutableDeletionInfo;
import org.apache.cassandra.db.RegularAndStaticColumns;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.db.marshal.CompositeType;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.db.marshal.MapType;
import org.apache.cassandra.db.partitions.AbstractBTreePartition;
import org.apache.cassandra.db.partitions.AtomicBTreePartition;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.rows.BTreeRow;
import org.apache.cassandra.db.rows.BufferCell;
import org.apache.cassandra.db.rows.Cell;
import org.apache.cassandra.db.rows.CellPath;
import org.apache.cassandra.db.rows.ColumnData;
import org.apache.cassandra.db.rows.ComplexColumnData;
import org.apache.cassandra.db.rows.EncodingStats;
import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.db.rows.Rows;
import org.apache.cassandra.dht.ByteOrderedPartitioner;
import org.apache.cassandra.index.transactions.UpdateTransaction;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.schema.TableMetadataRef;
import org.apache.cassandra.utils.btree.BTree;
import org.apache.cassandra.utils.btree.UpdateFunction;
import org.apache.cassandra.utils.concurrent.ImmediateFuture;
import org.apache.cassandra.utils.concurrent.OpOrder;
import org.apache.cassandra.utils.BulkIterator;
import org.apache.cassandra.utils.memory.ByteBufferCloner;
import org.apache.cassandra.utils.memory.Cloner;
import org.apache.cassandra.utils.memory.HeapPool;
import org.apache.cassandra.utils.memory.MemtableAllocator;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
import org.openjdk.jmh.annotations.Fork;
import org.openjdk.jmh.annotations.Level;
import org.openjdk.jmh.annotations.Measurement;
import org.openjdk.jmh.annotations.Mode;
import org.openjdk.jmh.annotations.OutputTimeUnit;
import org.openjdk.jmh.annotations.Param;
import org.openjdk.jmh.annotations.Scope;
import org.openjdk.jmh.annotations.Setup;
import org.openjdk.jmh.annotations.State;
import org.openjdk.jmh.annotations.Threads;
import org.openjdk.jmh.annotations.Warmup;
import static java.lang.Long.min;
import static java.lang.Math.max;
@BenchmarkMode(Mode.Throughput)
@OutputTimeUnit(TimeUnit.MILLISECONDS)
@Warmup(iterations = 10, time = 1, timeUnit = TimeUnit.SECONDS)
@Measurement(iterations = 5, time = 2, timeUnit = TimeUnit.SECONDS)
@Fork(value = 2)
@Threads(4)
@State(Scope.Benchmark)
public class AtomicBTreePartitionUpdateBench
{
private static final OpOrder NO_ORDER = new OpOrder();
private static final MutableDeletionInfo NO_DELETION_INFO = new MutableDeletionInfo(DeletionTime.LIVE);
private static final HeapPool POOL = new HeapPool(Long.MAX_VALUE, 1.0f, () -> ImmediateFuture.success(Boolean.TRUE));
private static final ByteBuffer zero = Int32Type.instance.decompose(0);
private static final DecoratedKey decoratedKey = new BufferDecoratedKey(new ByteOrderedPartitioner().getToken(zero), zero);
static
{
DatabaseDescriptor.setPartitionerUnsafe(ByteOrderedPartitioner.instance);
}
public enum Distribution { RANDOM, SEQUENTIAL }
final AtomicInteger uniqueThreadInitialisation = new AtomicInteger();
@Param({"4"})
int clusteringCount;
// a value of -1 indicates to send all inserts to a single row,
// using the clusterings to build CellPath and write our rows into a Map
@Param({"-1", "1", "8"})
int columnCount;
@Param({"0", "0.5"})
float insertRowOverlap;
@Param({"1", "32", "256"})
int insertRowCount;
@Param({"8", "256"})
int valueSize;
@Param({"256"})
int rolloverAfterInserts;
@Param({"RANDOM", "SEQUENTIAL"})
Distribution distribution;
@Param({"RANDOM", "SEQUENTIAL"})
Distribution timestamps;
@Param({"false"})
boolean uniquePerTrial;
// hacky way to pass in the number of threads we're executing concurrently in JMH
@Param({"4"})
int threadCount;
@State(Scope.Benchmark)
public static class GlobalState
{
RegularAndStaticColumns partitionColumns;
TableMetadata metadata;
ColumnMetadata[] columns;
Clustering[] clusterings;
CellPath[] complexPaths;
ByteBuffer value;
@Setup(Level.Trial)
public void setup(AtomicBTreePartitionUpdateBench bench)
{
ColumnMetadata[] partitionKeyColumns = bench.partitionKeyColumns();
ColumnMetadata[] clusteringColumns = bench.clusteringColumns();
ColumnMetadata[] regularColumns = bench.regularColumns();
partitionColumns = RegularAndStaticColumns.builder().addAll(Arrays.asList(regularColumns)).build();
columns = regularColumns;
metadata = bench.metadata(partitionKeyColumns, clusteringColumns, columns);
clusterings = bench.clusterings();
complexPaths = bench.complexPaths(regularColumns);
value = ByteBuffer.allocate(bench.valueSize);
}
}
// stateful; cannot be shared between threads
private static class UpdateGenerator
{
final Random random;
final TableMetadata metadata;
final RegularAndStaticColumns partitionColumns;
final boolean isComplex;
final ColumnMetadata[] columns;
final Clustering[] clusterings;
final CellPath[] complexPaths;
final float insertRowOverlap;
final Distribution distribution;
final IntToLongFunction timestamps;
final ByteBuffer value;
final IntVisitor insertRowCount;
final Row[] insertBuffer;
final ColumnData[] columnBuffer;
final Cell[] complexBuffer;
int offset;
UpdateGenerator(GlobalState global, AtomicBTreePartitionUpdateBench bench, long seed)
{
this.random = new Random(seed);
this.metadata = global.metadata;
this.partitionColumns = global.partitionColumns;
this.columns = global.columns.clone();
this.isComplex = this.columns[0].isComplex();
this.clusterings = global.clusterings.clone();
this.complexPaths = global.complexPaths.clone();
this.value = global.value;
this.insertRowCount = new IntVisitor(bench.insertRowCount);
this.insertRowOverlap = bench.insertRowOverlap;
this.distribution = bench.distribution;
this.timestamps = bench.timestamps == Distribution.RANDOM ? i -> random.nextLong() : i -> i;
this.insertBuffer = new Row[bench.insertRowCount * 2];
this.columnBuffer = new ColumnData[columns.length];
this.complexBuffer = new Cell[complexPaths.length];
}
void reset()
{
insertRowCount.randomise(random);
offset = 0;
}
final Function<Clustering, Row> simpleRow = this::simpleRow;
final Function<CellPath, Cell> complexCell = this::complexCell;
PartitionUpdate next()
{
int rowCount;
if (!isComplex)
{
rowCount = selectSortAndTransform(insertBuffer, clusterings, metadata.comparator, simpleRow);
}
else
{
rowCount = 1;
insertBuffer[0] = complexRow();
}
try (BulkIterator<Row> iter = BulkIterator.of(insertBuffer))
{
Object[] tree = BTree.build(iter, rowCount, UpdateFunction.noOp());
return PartitionUpdate.unsafeConstruct(metadata, decoratedKey, AbstractBTreePartition.unsafeConstructHolder(partitionColumns, tree, DeletionInfo.LIVE, Rows.EMPTY_STATIC_ROW, EncodingStats.NO_STATS), NO_DELETION_INFO, false);
}
}
private <I, O> int selectSortAndTransform(O[] out, I[] in, Comparator<? super I> comparator, Function<I, O> transform)
{
int prevRowCount = offset == 0 ? 0 : insertRowCount.cur();
int rowCount = insertRowCount.next();
switch (distribution)
{
case SEQUENTIAL:
{
for (int i = 0 ; i < rowCount ; ++i)
out[i] = transform.apply(in[i + offset]);
offset += rowCount * (1f - insertRowOverlap);
break;
}
case RANDOM:
{
int rowOverlap = (int) (insertRowOverlap * min(rowCount, prevRowCount));
shuffle(random, in, 0, rowOverlap, 0, prevRowCount);
shuffle(random, in, rowOverlap, rowCount - rowOverlap, prevRowCount, in.length - prevRowCount);
Arrays.sort(in, 0, rowCount, comparator);
for (int i = 0 ; i < rowCount ; ++i)
out[i] = transform.apply(in[i]);
++offset;
break;
}
}
return rowCount;
}
Row simpleRow(Clustering clustering)
{
for (int i = 0 ; i < columns.length ; ++i)
columnBuffer[i] = cell(columns[i], null);
return bufferToRow(clustering);
}
Row complexRow()
{
int mapCount = selectSortAndTransform(complexBuffer, complexPaths, columns[0].cellPathComparator(), complexCell);
try (BulkIterator<ColumnData> iter = BulkIterator.of(complexBuffer))
{
columnBuffer[0] = ComplexColumnData.unsafeConstruct(columns[0], BTree.build(iter, mapCount, UpdateFunction.noOp()), DeletionTime.LIVE);
}
return bufferToRow(clusterings[0]);
}
Row bufferToRow(Clustering clustering)
{
try (BulkIterator<ColumnData> iter = BulkIterator.of(columnBuffer))
{
return BTreeRow.create(clustering, LivenessInfo.EMPTY, Row.Deletion.LIVE, BTree.build(iter, columns.length, UpdateFunction.noOp()));
}
}
Cell simpleCell(ColumnMetadata column)
{
return cell(column, null);
}
Cell complexCell(CellPath path)
{
return cell(columns[0], path);
}
Cell cell(ColumnMetadata column, CellPath path)
{
return new BufferCell(column, timestamps.applyAsLong(offset), Cell.NO_TTL, Cell.NO_DELETION_TIME, value, path);
}
}
private static class Batch
{
final AtomicBTreePartition update;
final PartitionUpdate[] insert;
// low 20 bits contain the next insert we're performing this generation
// next 20 bits are inserts we've performed this generation
// next 24 bits are generation (i.e. number of times we've run this update)
final AtomicLong state = new AtomicLong();
final AtomicLong activeThreads = new AtomicLong();
final MemtableAllocator allocator;
final Cloner cloner;
final int waitForActiveThreads;
/** Signals to replace the reference in {@code update} after this many invocations of {@code allocator.allocate} */
private int invalidateOn;
public Batch(int threads, TableMetadata metadata, UpdateGenerator generator, int rolloverAfterInserts)
{
waitForActiveThreads = threads;
allocator = new HeapPool.Allocator(POOL)
{
public Cloner cloner(OpOrder.Group opGroup)
{
return new ByteBufferCloner()
{
@Override
public ByteBuffer allocate(int size)
{
if (invalidateOn > 0 && --invalidateOn == 0)
{
AbstractBTreePartition.Holder holder = update.unsafeGetHolder();
if (!BTree.isEmpty(holder.tree))
update.unsafeSetHolder(AbstractBTreePartition.unsafeConstructHolder(
holder.columns, Arrays.copyOf(holder.tree, holder.tree.length), holder.deletionInfo, holder.staticRow, holder.stats));
}
return ByteBuffer.allocate(size);
}
};
}
};
update = new AtomicBTreePartition(TableMetadataRef.forOfflineTools(metadata), decoratedKey, allocator);
cloner = allocator.cloner(NO_ORDER.getCurrent());
generator.reset();
insert = IntStream.range(0, rolloverAfterInserts).mapToObj(i -> generator.next()).toArray(PartitionUpdate[]::new);
}
boolean performOne(int ifGeneration, Consumer<Batch> invokeBefore)
{
int index;
ifGeneration &= 0xffffff;
while (true)
{
long cur = state.get();
int curGeneration = ((int) (cur >>> 40)) & 0xffffff;
if (curGeneration == ifGeneration)
{
index = ((int) cur) & 0xfffff;
if (index == this.insert.length)
return false;
if (state.compareAndSet(cur, cur + 1)) break;
else continue;
}
if (ifGeneration < curGeneration)
return false;
// should never really happen
Thread.yield();
}
try
{
if (activeThreads.get() < waitForActiveThreads)
{
// try to prevent threads scattering to the four winds and updating different partitions
// we don't wait forever, as we can't synchronise with JMH to ensure all threads stop together
activeThreads.incrementAndGet();
long start = System.nanoTime();
while (activeThreads.get() < waitForActiveThreads && TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start) < 50)
ThreadLocalRandom.current().nextLong();
}
invokeBefore.accept(this);
update.addAllWithSizeDelta(insert[index], cloner, NO_ORDER.getCurrent(), UpdateTransaction.NO_OP);
return true;
}
finally
{
if (state.addAndGet(0x100000L) == ((((long)ifGeneration) << 40) | (((long)insert.length) << 20) | insert.length))
{
activeThreads.set(0);
update.unsafeSetHolder(AbstractBTreePartition.unsafeGetEmptyHolder());
// reset the state and rollover the generation
state.set((ifGeneration + 1L) << 40);
}
}
}
}
public static class Batches
{
Batch[] batches;
void setup(int threadsForBatch, int threadsForTest, AtomicBTreePartitionUpdateBench bench, GlobalState global)
{
UpdateGenerator generator = new UpdateGenerator(global, bench, bench.uniqueThreadInitialisation.incrementAndGet());
int batchCount = Math.min(256, max(1, (int) (Runtime.getRuntime().maxMemory() / (2 * ((threadsForTest + threadsForBatch - 1) / threadsForBatch) * bench.sizeInBytesOfOneBatch()))));
batches = IntStream.range(0, batchCount).mapToObj(i -> new Batch(threadsForBatch, global.metadata, generator, bench.rolloverAfterInserts))
.toArray(Batch[]::new);
}
Batch get(int i)
{
return batches[i % batches.length];
}
}
@State(Scope.Thread)
public static class SingleThreadUpdates extends Batches
{
int i;
int gen;
@Setup(Level.Trial)
public void setup(AtomicBTreePartitionUpdateBench bench, GlobalState global)
{
super.setup(1, bench.threadCount, bench, global);
}
void performOne(Consumer<Batch> invokeFirst)
{
while (true)
{
if (get(i).performOne(gen, invokeFirst))
break;
if (++i == batches.length) { i = 0; ++gen; }
}
}
}
@State(Scope.Benchmark)
public static class ConcurrentUpdates extends Batches
{
final AtomicLong state = new AtomicLong();
@Setup(Level.Trial)
public void setup(AtomicBTreePartitionUpdateBench bench, GlobalState global)
{
super.setup(bench.threadCount, bench.threadCount, bench, global);
}
void performOne(Consumer<Batch> invokeFirst)
{
while (true)
{
long cur = state.get();
int generation = (int) (cur >>> 32);
int index = (int) cur;
if (get(index).performOne(generation, invokeFirst))
break;
state.compareAndSet(cur, (cur & 0xffffffffL) == batches.length ? ((cur & 0xffffffff00000000L) + 0x100000000L) : cur + 1);
}
}
}
@Benchmark
public void success(SingleThreadUpdates updates)
{
updates.performOne(batch -> {});
}
@Benchmark
public void oneFailure(SingleThreadUpdates updates)
{
updates.performOne(batch -> batch.invalidateOn = ThreadLocalRandom.current().nextInt(insertRowCount * max(1, columnCount)));
}
@Benchmark
public void concurrent(ConcurrentUpdates updates)
{
updates.performOne(batch -> {});
}
private int sizeInBytesOfOneBatch()
{
// 50 bytes ~= size of a Cell
return 50 * max(columnCount, 1) * (insertRowCount + insertRowCount/2) * rolloverAfterInserts;
}
private ColumnMetadata[] regularColumns()
{
if (columnCount >= 0)
return regularColumns(BytesType.instance, columnCount);
AbstractType[] types = new AbstractType[clusteringCount];
Arrays.fill(types, BytesType.instance);
return regularColumns(MapType.getInstance(CompositeType.getInstance(types), BytesType.instance, true), 1);
}
private ColumnMetadata[] partitionKeyColumns()
{
return columns(Int32Type.instance, ColumnMetadata.Kind.PARTITION_KEY, 1, "pk");
}
private ColumnMetadata[] clusteringColumns()
{
return columns(Int32Type.instance, ColumnMetadata.Kind.CLUSTERING, clusteringCount, "c");
}
private TableMetadata metadata(ColumnMetadata[] partitionKeyColumns, ColumnMetadata[] clusteringColumns, ColumnMetadata[] regularColumns)
{
List<ColumnMetadata> columns = ImmutableList.<ColumnMetadata>builder()
.add(partitionKeyColumns)
.add(clusteringColumns)
.add(regularColumns)
.build();
return TableMetadata.builder("", "")
.id(TableId.fromUUID(UUID.randomUUID()))
.addColumns(columns)
.partitioner(ByteOrderedPartitioner.instance)
.build();
}
private int uniqueRowCount()
{
return (int) (insertRowCount * (1 + (rolloverAfterInserts * (1f - insertRowOverlap))));
}
private Clustering[] clusterings()
{
Clustering<ByteBuffer> prefix = Clustering.make(IntStream.range(0, clusteringCount - 1).mapToObj(i -> zero).toArray(ByteBuffer[]::new));
int rowCount = columnCount >= 0 ? uniqueRowCount() : 1;
return clusterings(rowCount, prefix);
}
private CellPath[] complexPaths(ColumnMetadata[] columns)
{
if (columnCount >= 0)
return new CellPath[0];
Clustering<ByteBuffer> prefix = Clustering.make(IntStream.range(0, clusteringCount - 1).mapToObj(i -> zero).toArray(ByteBuffer[]::new));
return complexPaths((CompositeType) ((MapType)columns[0].type).getKeysType(), uniqueRowCount(), prefix);
}
private static ColumnMetadata[] regularColumns(AbstractType<?> type, int count)
{
return columns(type, ColumnMetadata.Kind.REGULAR, count, "v");
}
private static ColumnMetadata[] columns(AbstractType<?> type, ColumnMetadata.Kind kind, int count, String prefix)
{
return IntStream.range(0, count)
.mapToObj(i -> new ColumnMetadata("", "", new ColumnIdentifier(prefix + i, true), type, kind != ColumnMetadata.Kind.REGULAR ? i : ColumnMetadata.NO_POSITION, kind))
.toArray(ColumnMetadata[]::new);
}
private static Clustering[] clusterings(int rowCount, Clustering<ByteBuffer> prefix)
{
return IntStream.range(0, rowCount)
.mapToObj(i -> {
ByteBuffer[] values = Arrays.copyOf(prefix.getBufferArray(), prefix.size() + 1);
values[prefix.size()] = Int32Type.instance.decompose(i);
return Clustering.make(values);
}).toArray(Clustering[]::new);
}
private static CellPath[] complexPaths(CompositeType type, int pathCount, Clustering<ByteBuffer> prefix)
{
return IntStream.range(0, pathCount)
.mapToObj(i -> {
Object[] values = Arrays.copyOf(prefix.getRawValues(), prefix.size() + 1);
values[prefix.size()] = Int32Type.instance.decompose(i);
return CellPath.create(type.decompose(values));
}).toArray(CellPath[]::new);
}
private static <T> void shuffleAndSort(Random random, T[] data, int size, Comparator<T> comparator)
{
shuffle(random, data, size);
Arrays.sort(data, 0, size, comparator);
}
private static void shuffle(Random random, Object[] data, int size)
{
shuffle(random, data, 0, size, 0, data.length);
}
private static void shuffle(Random random, Object[] data, int trgOffset, int trgSize, int srcOffset, int srcSize)
{
for (int i = 0 ; i < trgSize ; ++i)
{
int swap = srcOffset + srcSize == 0 ? 0 : random.nextInt(srcSize);
Object tmp = data[swap];
data[swap] = data[i + trgOffset];
data[i + trgOffset] = tmp;
}
}
}

14
test/microbench/org/apache/cassandra/test/microbench/btree/Megamorphism.java
View File

@ -39,16 +39,22 @@ public class Megamorphism
private static final UpdateFunction UNSIMPLE_KEEP_OLD = new UpdateFunction()
{
public Object apply(Object replacing, Object update) { return replacing; }
@Override
public Object merge(Object replacing, Object update) { return replacing; }
@Override
public void onAllocatedOnHeap(long heapSize) { }
public Object apply(Object v) { return v; }
@Override
public Object insert(Object v) { return v; }
};
private static final UpdateFunction UNSIMPLE_KEEP_NEW = new UpdateFunction()
{
public Object apply(Object replacing, Object update) { return update; }
@Override
public Object merge(Object replacing, Object update) { return update; }
@Override
public void onAllocatedOnHeap(long heapSize) { }
public Object apply(Object v) { return v; }
@Override
public Object insert(Object v) { return v; }
};
static <V> IntFunction<UpdateFunction<V, V>> updateFGetter(boolean keepOld, BTreeBench.UpdateF updateF)

1
test/unit/org/apache/cassandra/cql3/validation/operations/DeleteTest.java
View File

@ -28,6 +28,7 @@ import org.junit.Test;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.utils.btree.BTree;
import static org.apache.cassandra.utils.ByteBufferUtil.EMPTY_BYTE_BUFFER;
import static org.apache.cassandra.utils.ByteBufferUtil.bytes;

46
test/unit/org/apache/cassandra/db/CellTest.java
View File

@ -371,29 +371,29 @@ public class CellTest
return FieldIdentifier.forQuoted(field);
}
@Test
public void testComplexCellReconcile()
{
ColumnMetadata m = cfm2.getColumn(new ColumnIdentifier("m", false));
int now1 = FBUtilities.nowInSeconds();
long ts1 = now1*1000000L;
Cell<?> r1m1 = BufferCell.live(m, ts1, bb(1), CellPath.create(bb(1)));
Cell<?> r1m2 = BufferCell.live(m, ts1, bb(2), CellPath.create(bb(2)));
List<Cell<?>> cells1 = Lists.newArrayList(r1m1, r1m2);
int now2 = now1 + 1;
long ts2 = now2*1000000L;
Cell<?> r2m2 = BufferCell.live(m, ts2, bb(1), CellPath.create(bb(2)));
Cell<?> r2m3 = BufferCell.live(m, ts2, bb(2), CellPath.create(bb(3)));
Cell<?> r2m4 = BufferCell.live(m, ts2, bb(3), CellPath.create(bb(4)));
List<Cell<?>> cells2 = Lists.newArrayList(r2m2, r2m3, r2m4);
RowBuilder builder = new RowBuilder();
Cells.reconcileComplex(m, cells1.iterator(), cells2.iterator(), DeletionTime.LIVE, builder);
Assert.assertEquals(Lists.newArrayList(r1m1, r2m2, r2m3, r2m4), builder.cells);
}
// @Test
// public void testComplexCellReconcile()
// {
// ColumnMetadata m = cfm2.getColumn(new ColumnIdentifier("m", false));
// int now1 = FBUtilities.nowInSeconds();
// long ts1 = now1*1000000L;
//
//
// Cell<?> r1m1 = BufferCell.live(m, ts1, bb(1), CellPath.create(bb(1)));
// Cell<?> r1m2 = BufferCell.live(m, ts1, bb(2), CellPath.create(bb(2)));
// List<Cell<?>> cells1 = Lists.newArrayList(r1m1, r1m2);
//
// int now2 = now1 + 1;
// long ts2 = now2*1000000L;
// Cell<?> r2m2 = BufferCell.live(m, ts2, bb(1), CellPath.create(bb(2)));
// Cell<?> r2m3 = BufferCell.live(m, ts2, bb(2), CellPath.create(bb(3)));
// Cell<?> r2m4 = BufferCell.live(m, ts2, bb(3), CellPath.create(bb(4)));
// List<Cell<?>> cells2 = Lists.newArrayList(r2m2, r2m3, r2m4);
//
// RowBuilder builder = new RowBuilder();
// Cells.reconcileComplex(m, cells1.iterator(), cells2.iterator(), DeletionTime.LIVE, builder);
// Assert.assertEquals(Lists.newArrayList(r1m1, r2m2, r2m3, r2m4), builder.cells);
// }
private int testExpiring(String n1, String v1, long t1, int et1, String n2, String v2, Long t2, Integer et2)
{

12
test/unit/org/apache/cassandra/db/NativeCellTest.java
View File

@ -36,7 +36,7 @@ import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.utils.concurrent.ImmediateFuture;
import org.apache.cassandra.utils.concurrent.OpOrder;
import org.apache.cassandra.utils.memory.HeapAllocator;
import org.apache.cassandra.utils.memory.HeapCloner;
import org.apache.cassandra.utils.memory.NativeAllocator;
import org.apache.cassandra.utils.memory.NativePool;
@ -151,8 +151,8 @@ public class NativeCellTest
private static void test(Row row)
{
Row nrow = clone(row, nativeAllocator.rowBuilder(group));
Row brow = clone(row, HeapAllocator.instance.cloningBTreeRowBuilder());
Row nrow = row.clone(nativeAllocator.cloner(group));
Row brow = row.clone(HeapCloner.instance);
Assert.assertEquals(row, nrow);
Assert.assertEquals(row, brow);
Assert.assertEquals(nrow, brow);
@ -166,10 +166,4 @@ public class NativeCellTest
Assert.assertEquals(0, comparator.compare(row.clustering(), brow.clustering()));
Assert.assertEquals(0, comparator.compare(nrow.clustering(), brow.clustering()));
}
private static Row clone(Row row, Row.Builder builder)
{
return Rows.copy(row, builder).build();
}
}

91
test/unit/org/apache/cassandra/db/rows/RowBuilder.java
View File

@ -1,91 +0,0 @@
/*
* 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.rows;
import java.util.LinkedList;
import java.util.List;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.DeletionTime;
import org.apache.cassandra.db.LivenessInfo;
import org.apache.cassandra.db.rows.Row.Builder;
import org.apache.cassandra.utils.Pair;
/**
* Instrumented Builder implementation for testing the
* behavior of Cells and Rows static methods
*/
public class RowBuilder implements Row.Builder
{
public List<Cell<?>> cells = new LinkedList<>();
public Clustering<?> clustering = null;
public LivenessInfo livenessInfo = null;
public Row.Deletion deletionTime = null;
public List<Pair<ColumnMetadata, DeletionTime>> complexDeletions = new LinkedList<>();
@Override
public Builder copy()
{
throw new UnsupportedOperationException();
}
public void addCell(Cell<?> cell)
{
cells.add(cell);
}
public boolean isSorted()
{
throw new UnsupportedOperationException();
}
public void newRow(Clustering<?> clustering)
{
assert this.clustering == null;
this.clustering = clustering;
}
public Clustering<?> clustering()
{
return clustering;
}
public void addPrimaryKeyLivenessInfo(LivenessInfo info)
{
assert livenessInfo == null;
livenessInfo = info;
}
public void addRowDeletion(Row.Deletion deletion)
{
assert deletionTime == null;
deletionTime = deletion;
}
public void addComplexDeletion(ColumnMetadata column, DeletionTime complexDeletion)
{
complexDeletions.add(Pair.create(column, complexDeletion));
}
public Row build()
{
throw new UnsupportedOperationException();
}
}

286
test/unit/org/apache/cassandra/db/rows/RowsMergingTest.java Normal file
View File

@ -0,0 +1,286 @@
/*
* 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.rows;
import java.util.Arrays;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.commons.lang3.StringUtils;
import org.junit.BeforeClass;
import org.junit.Test;
import com.google.common.base.Joiner;
public class RowsMergingTest extends CQLTester
{
@BeforeClass
public static void setUpClass()
{
System.setProperty("cassandra.btree.branchshift", "2");
CQLTester.setUpClass();
}
@Test
public void testInsertion() throws Throwable
{
checker().schema("CREATE TABLE %s (pk int, ck int, a int, b int, c int, d int, e int, f int, PRIMARY KEY (pk, ck))")
.queries("INSERT INTO %s (pk, ck, a, b, c, d, e, f) VALUES (?, 1, 1, 1, 1, 1, 1, 1) USING TIMESTAMP 1001",
"INSERT INTO %s (pk, ck, a, b, c, d) VALUES (?, 1, 2, 2, 2, 2) USING TIMESTAMP 1002")
.expectedRow(1, 2, 2, 2, 2, 1, 1)
.check();
}
@Test
public void testRowDeletions() throws Throwable
{
checker().schema("CREATE TABLE %s (pk int, ck int, a int, b int, c int, d int, e int, f int, PRIMARY KEY (pk, ck))")
.queries("INSERT INTO %s (pk, ck, a, b, c, d, e, f) VALUES (?, 1, 1, 1, 1, 1, 1, 1) USING TIMESTAMP 1001",
"DELETE FROM %s USING TIMESTAMP 1002 WHERE pk = ? AND ck = 1",
"INSERT INTO %s (pk, ck, a) VALUES (?, 1, 2) USING TIMESTAMP 1003")
.expectedRow(1, 2, null, null, null, null, null)
.check();
}
@Test
public void testRowDeletionWithUpdate() throws Throwable
{
checker().schema("CREATE TABLE %s (pk int, ck int, a int, b int, c int, d int, e int, f int, PRIMARY KEY (pk, ck))")
.queries("INSERT INTO %s (pk, ck, a, b, c, d, e, f) VALUES (?, 1, 1, 1, 1, 1, 1, 1) USING TIMESTAMP 1001",
"DELETE FROM %s USING TIMESTAMP 1002 WHERE pk = ? AND ck = 1",
"UPDATE %s USING TIMESTAMP 1003 SET a = 2 WHERE pk = ? AND ck = 1")
.expectedRow(1, 2, null, null, null, null, null)
.check();
}
@Test
public void testRowDeletionWithOnlyUpdates() throws Throwable
{
checker().schema("CREATE TABLE %s (pk int, ck int, a int, b int, c int, d int, e int, f int, PRIMARY KEY (pk, ck))")
.queries("UPDATE %s USING TIMESTAMP 1001 SET a = 1, b = 1, c = 1, d = 1, e = 1, f = 1 WHERE pk = ? AND ck = 1",
"DELETE FROM %s USING TIMESTAMP 1002 WHERE pk = ? AND ck = 1",
"UPDATE %s USING TIMESTAMP 1003 SET a = 2 WHERE pk = ? AND ck = 1")
.expectedRow(1, 2, null, null, null, null, null)
.check();
}
@Test
public void testRowWithMultipleDeletions() throws Throwable
{
checker().schema("CREATE TABLE %s (pk int, ck int, a int, b int, c int, d int, e int, f int, PRIMARY KEY (pk, ck))")
.queries("INSERT INTO %s (pk, ck, a, b, c, d, e, f) VALUES (?, 1, 1, 1, 1, 1, 1, 1) USING TIMESTAMP 1001",
"DELETE FROM %s USING TIMESTAMP 1002 WHERE pk = ? AND ck = 1",
"UPDATE %s USING TIMESTAMP 1003 SET a = 2 WHERE pk = ? AND ck = 1",
"DELETE FROM %s USING TIMESTAMP 1004 WHERE pk = ? AND ck = 1")
.check();
}
@Test
public void testRowWithComplexCollectionOverride() throws Throwable
{
checker().schema("CREATE TABLE %s (pk int, ck int, s set<text>, PRIMARY KEY (pk, ck))")
.queries("INSERT INTO %s (pk, ck, s) VALUES (?, 1, {'a', 'b', 'c'}) USING TIMESTAMP 1001",
"UPDATE %s USING TIMESTAMP 1002 SET s = {'m', 'n'} WHERE pk = ? AND ck = 1")
.expectedRow(1, set("m", "n"))
.check();
}
@Test
public void testRowWithDeletionAndComplexCollection() throws Throwable
{
checker().schema("CREATE TABLE %s (pk int, ck int, s set<text>, PRIMARY KEY (pk, ck))")
.queries("INSERT INTO %s (pk, ck, s) VALUES (?, 1, {'a', 'b', 'c'}) USING TIMESTAMP 1001",
"DELETE FROM %s USING TIMESTAMP 1002 WHERE pk = ? AND ck = 1")
.check();
}
@Test
public void testRowWithDeletionAndComplexCollectionOverride() throws Throwable
{
checker().schema("CREATE TABLE %s (pk int, ck int, s set<text>, PRIMARY KEY (pk, ck))")
.queries("INSERT INTO %s (pk, ck, s) VALUES (?, 1, {'a', 'b', 'c'}) USING TIMESTAMP 1001",
"DELETE FROM %s USING TIMESTAMP 1002 WHERE pk = ? AND ck = 1",
"UPDATE %s USING TIMESTAMP 1003 SET s = {'m', 'n'} WHERE pk = ? AND ck = 1")
.expectedRow(1, set("m", "n"))
.check();
}
@Test
public void testRowWithComplexDeletionAfterRowDeletion() throws Throwable
{
checker().schema("CREATE TABLE %s (pk int, ck int, s set<text>, PRIMARY KEY (pk, ck))")
.queries("DELETE FROM %s USING TIMESTAMP 1001 WHERE pk = ? AND ck = 1",
"INSERT INTO %s (pk, ck, s) VALUES (?, 1, {'a', 'b', 'c'}) USING TIMESTAMP 1002",
"DELETE s FROM %s USING TIMESTAMP 1003 WHERE pk = ? AND ck = 1")
.expectedRow(1, null)
.check();
}
@Test
public void testRowDeletionsWithBatchWithBatchTimestamp() throws Throwable
{
checker().schema("CREATE TABLE %s (pk int, ck int, a int, b int, c int, d int, e int, f int, PRIMARY KEY (pk, ck))")
.queries("BEGIN BATCH USING TIMESTAMP 1001 \n" +
"INSERT INTO %s (pk, ck, a, b, c, d, e, f) VALUES (?, 1, 1, 1, 1, 1, 1, 1); \n" +
"DELETE FROM %s WHERE pk = ? AND ck = 1; \n" +
"APPLY BATCH;",
"INSERT INTO %s (pk, ck, a) VALUES (?, 1, 2) USING TIMESTAMP 1002")
.expectedRow(1, 2, null, null, null, null, null)
.check();
}
@Test
public void testRowDeletionsWithBatchWithTimestampPerOperation() throws Throwable
{
checker().schema("CREATE TABLE %s (pk int, ck int, a int, b int, c int, d int, e int, f int, PRIMARY KEY (pk, ck))")
.queries("BEGIN BATCH \n" +
"INSERT INTO %s (pk, ck, a, b, c, d, e, f) VALUES (?, 1, 1, 1, 1, 1, 1, 1) USING TIMESTAMP 1001 ; \n" +
"DELETE FROM %s USING TIMESTAMP 1002 WHERE pk = ? AND ck = 1; \n" +
"APPLY BATCH;",
"INSERT INTO %s (pk, ck, a) VALUES (?, 1, 2) USING TIMESTAMP 1003")
.expectedRow(1, 2, null, null, null, null, null)
.check();
}
public MergeChecker checker()
{
return new MergeChecker();
}
/**
* Utility class to check that a merge result in the expected row not no matter in which order the operation are applied.
*/
private class MergeChecker
{
private int pk;
private String schema;
private String[] queries;
private Object[] expectedRow;
public MergeChecker schema(String schema)
{
this.schema = schema;
return this;
}
public MergeChecker queries(String... queries)
{
this.queries = queries;
return this;
}
public MergeChecker expectedRow(Object... columnValues)
{
this.expectedRow = new Object[columnValues.length + 1];
System.arraycopy(columnValues, 0, this.expectedRow, 1, columnValues.length);
return this;
}
public void check() throws Throwable
{
createTable(schema);
checkAllPermutations(queries.length, queries);
}
private void check(String[] queries) throws Throwable
{
for (String query : queries)
{
try
{
int count = StringUtils.countMatches(query, "%s");
Object[] parameters1 = new Object[count];
Arrays.fill(parameters1, pk);
Object[] parameters = parameters1;
executeFormattedQuery(formatQueries(query, count), parameters);
}
catch (Throwable e)
{
throw new AssertionError("Executing the following queries did not lead to the expected result: \n"
+ Joiner.on("; \n").join(queries)
+ "\n when executing: \n" + query, e);
}
}
try
{
if (expectedRow != null)
{
expectedRow[0] = pk;
assertRows(execute("SELECT * FROM %s WHERE pk = ?" , pk),
expectedRow);
}
else
{
assertEmpty(execute("SELECT * FROM %s WHERE pk = ?" , pk));
}
}
catch (Throwable e)
{
throw new AssertionError("Executing the following queries did not lead to the expected result: \n" + Joiner.on("; \n").join(queries), e);
}
pk++;
}
private String formatQueries(String query, int numberOfqueries)
{
String table = keyspace() + '.' + currentTable();
return String.format(query, createFilledArray(numberOfqueries, table));
}
private Object[] createFilledArray(int length, Object fillingValue)
{
Object[] tables = new Object[length];
Arrays.fill(tables, fillingValue);
return tables;
}
private void checkAllPermutations(int n, String[] queries) throws Throwable
{
if (n == 1)
{
check(queries);
}
else
{
for (int i = 0, m = n - 1; i < m; i++)
{
checkAllPermutations(n - 1, queries);
if ((i & 1) == 0)
{
swap(queries, i, n - 1);
}
else
{
swap(queries, 0, n - 1);
}
}
checkAllPermutations(n - 1, queries);
}
}
private void swap(String[] queries, int i, int j)
{
String tmp = queries[i];
queries[i] = queries[j];
queries[j] = tmp;
}
}
}

90
test/unit/org/apache/cassandra/db/rows/RowsTest.java
View File

@ -20,8 +20,8 @@ package org.apache.cassandra.db.rows;
import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
@ -44,7 +44,6 @@ import org.apache.cassandra.db.marshal.*;
import org.apache.cassandra.db.partitions.PartitionStatisticsCollector;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
public class RowsTest
{
@ -208,11 +207,17 @@ public class RowsTest
return now * 1000000L;
}
private static Row.Builder createBuilder(Clustering<?> c)
{
Row.Builder builder = BTreeRow.unsortedBuilder();
builder.newRow(c);
return builder;
}
private static Row.Builder createBuilder(Clustering<?> c, int now, ByteBuffer vVal, ByteBuffer mKey, ByteBuffer mVal)
{
long ts = secondToTs(now);
Row.Builder builder = BTreeRow.unsortedBuilder();
builder.newRow(c);
Row.Builder builder = createBuilder(c);
builder.addPrimaryKeyLivenessInfo(LivenessInfo.create(ts, now));
if (vVal != null)
{
@ -227,35 +232,6 @@ public class RowsTest
return builder;
}
@Test
public void copy()
{
int now = FBUtilities.nowInSeconds();
long ts = secondToTs(now);
Row.Builder originalBuilder = BTreeRow.unsortedBuilder();
originalBuilder.newRow(c1);
LivenessInfo liveness = LivenessInfo.create(ts, now);
originalBuilder.addPrimaryKeyLivenessInfo(liveness);
DeletionTime complexDeletion = new DeletionTime(ts-1, now);
originalBuilder.addComplexDeletion(m, complexDeletion);
List<Cell<?>> expectedCells = Lists.newArrayList(BufferCell.live(v, secondToTs(now), BB1),
BufferCell.live(m, secondToTs(now), BB1, CellPath.create(BB1)),
BufferCell.live(m, secondToTs(now), BB2, CellPath.create(BB2)));
expectedCells.forEach(originalBuilder::addCell);
// We need to use ts-1 so the deletion doesn't shadow what we've created
Row.Deletion rowDeletion = new Row.Deletion(new DeletionTime(ts-1, now), false);
originalBuilder.addRowDeletion(rowDeletion);
RowBuilder builder = new RowBuilder();
Rows.copy(originalBuilder.build(), builder);
Assert.assertEquals(c1, builder.clustering);
Assert.assertEquals(liveness, builder.livenessInfo);
Assert.assertEquals(rowDeletion, builder.deletionTime);
Assert.assertEquals(Lists.newArrayList(Pair.create(m, complexDeletion)), builder.complexDeletions);
Assert.assertEquals(Sets.newHashSet(expectedCells), Sets.newHashSet(builder.cells));
}
@Test
public void collectStats()
{
@ -493,61 +469,55 @@ public class RowsTest
updateBuilder.addComplexDeletion(m, expectedComplexDeletionTime);
updateBuilder.addCell(expectedMCell);
RowBuilder builder = new RowBuilder();
long td = Rows.merge(existingBuilder.build(), updateBuilder.build(), builder);
Row merged = Rows.merge(existingBuilder.build(), updateBuilder.build());
Assert.assertEquals(c1, builder.clustering);
Assert.assertEquals(LivenessInfo.create(ts2, now2), builder.livenessInfo);
Assert.assertEquals(Lists.newArrayList(Pair.create(m, new DeletionTime(ts2-1, now2))), builder.complexDeletions);
Assert.assertEquals(c1, merged.clustering());
Assert.assertEquals(LivenessInfo.create(ts2, now2), merged.primaryKeyLivenessInfo());
Assert.assertEquals(2, builder.cells.size());
Assert.assertEquals(Lists.newArrayList(expectedVCell, expectedMCell), Lists.newArrayList(builder.cells));
Assert.assertEquals(ts2 - secondToTs(now1), td);
Iterator<Cell<?>> iter = merged.cells().iterator();
Assert.assertTrue(iter.hasNext());
Assert.assertEquals(expectedVCell, iter.next());
Assert.assertTrue(iter.hasNext());
Assert.assertEquals(expectedMCell, iter.next());
Assert.assertFalse(iter.hasNext());
}
@Test
public void mergeComplexDeletionSupersededByRowDeletion()
{
int now1 = FBUtilities.nowInSeconds();
Row.Builder existingBuilder = createBuilder(c1, now1, null, null, null);
Row.Builder existingBuilder = createBuilder(c1, now1, null, BB2, BB2);
int now2 = now1 + 1;
Row.Builder updateBuilder = createBuilder(c1, now2, null, BB1, BB1);
Row.Builder updateBuilder = createBuilder(c1);
int now3 = now2 + 1;
Row.Deletion expectedDeletion = new Row.Deletion(new DeletionTime(secondToTs(now3), now3), false);
updateBuilder.addRowDeletion(expectedDeletion);
RowBuilder builder = new RowBuilder();
Rows.merge(existingBuilder.build(), updateBuilder.build(), builder);
Row merged = Rows.merge(existingBuilder.build(), updateBuilder.build());
Assert.assertEquals(expectedDeletion, builder.deletionTime);
Assert.assertEquals(Collections.emptyList(), builder.complexDeletions);
Assert.assertEquals(Collections.emptyList(), builder.cells);
Assert.assertEquals(expectedDeletion, merged.deletion());
Assert.assertFalse(merged.hasComplexDeletion());
Assert.assertFalse(merged.cells().iterator().hasNext());
}
/**
* If a row's deletion time deletes a row's liveness info, the new row should have it's
* liveness info set to empty
*/
@Test
public void mergeRowDeletionSupercedesLiveness()
{
int now1 = FBUtilities.nowInSeconds();
Row.Builder existingBuilder = createBuilder(c1, now1, null, null, null);
Row.Builder existingBuilder = createBuilder(c1, now1, BB1, BB1, BB1);
int now2 = now1 + 1;
Row.Builder updateBuilder = createBuilder(c1, now2, BB1, BB1, BB1);
Row.Builder updateBuilder = createBuilder(c1);
int now3 = now2 + 1;
Row.Deletion expectedDeletion = new Row.Deletion(new DeletionTime(secondToTs(now3), now3), false);
updateBuilder.addRowDeletion(expectedDeletion);
RowBuilder builder = new RowBuilder();
Rows.merge(existingBuilder.build(), updateBuilder.build(), builder);
Row merged = Rows.merge(existingBuilder.build(), updateBuilder.build());
Assert.assertEquals(expectedDeletion, builder.deletionTime);
Assert.assertEquals(LivenessInfo.EMPTY, builder.livenessInfo);
Assert.assertEquals(Collections.emptyList(), builder.complexDeletions);
Assert.assertEquals(Collections.emptyList(), builder.cells);
Assert.assertEquals(expectedDeletion, merged.deletion());
Assert.assertEquals(LivenessInfo.EMPTY, merged.primaryKeyLivenessInfo());
Assert.assertEquals(0, merged.columns().size());
}
// Creates a dummy cell for a (regular) column for the provided name and without a cellPath.

13
test/unit/org/apache/cassandra/utils/btree/BTreeTest.java
View File

@ -39,7 +39,7 @@ public class BTreeTest
static final UpdateFunction<Integer, Integer> updateF = new UpdateFunction<Integer, Integer>()
{
public Integer apply(Integer replacing, Integer update)
public Integer merge(Integer replacing, Integer update)
{
return ints[update];
}
@ -48,7 +48,12 @@ public class BTreeTest
{
}
public Integer apply(Integer integer)
public Integer insert(Integer integer)
{
return ints[integer];
}
public Integer retain(Integer integer)
{
return ints[integer];
}
@ -423,7 +428,7 @@ public class BTreeTest
private int[] numberOfCalls = new int[20];
@Override
public Integer apply(Integer replacing, Integer update)
public Integer merge(Integer replacing, Integer update)
{
numberOfCalls[update] = numberOfCalls[update] + 1;
return update;
@ -436,7 +441,7 @@ public class BTreeTest
}
@Override
public Integer apply(Integer integer)
public Integer insert(Integer integer)
{
numberOfCalls[integer] = numberOfCalls[integer] + 1;
return integer;