mirror of https://github.com/apache/cassandra
Optimize TrieMemtable#getFlushSet
A flushing range is a combination of partial and full shards, for full shards we use write time captured values for count and total size of partition keys, for partial shards we iterate over keys When we iterate over partial shards we count key bytes only instead of reading them and also try to skip tokens full parsing Patch by Dmitry Konstantinov; reviewed by Branimir Lambov for CASSANDRA-20760
This commit is contained in:
Dmitry Konstantinov
parent
429b2f8be4
commit
19118a6cff
|
|
@ -1,4 +1,5 @@
|
|||
5.1
|
||||
* Optimize TrieMemtable#getFlushSet (CASSANDRA-20760)
|
||||
* Support manual secondary index selection at the CQL level (CASSANDRA-18112)
|
||||
* When regulars CQL mutations run on Accord use the txn timestamp rather than server timestamp (CASSANDRA-20744)
|
||||
* When using BEGIN TRANSACTION if a complex mutation exists in the same statement as one that uses a reference, then the complex delete is dropped (CASSANDRA-20788)
|
||||
|
|
|
|||
|
|
@ -153,6 +153,13 @@ public abstract class DecoratedKey implements PartitionPosition, FilterKey
|
|||
return false;
|
||||
}
|
||||
|
||||
public boolean isMaximum()
|
||||
{
|
||||
// A DecoratedKey can never be the maximum position on the ring
|
||||
return false;
|
||||
}
|
||||
|
||||
|
||||
public PartitionPosition.Kind kind()
|
||||
{
|
||||
return PartitionPosition.Kind.ROW_KEY;
|
||||
|
|
@ -244,11 +251,28 @@ public abstract class DecoratedKey implements PartitionPosition, FilterKey
|
|||
assert version != Version.LEGACY; // reverse translation is not supported for LEGACY version.
|
||||
// Decode the token from the first component of the multi-component sequence representing the whole decorated key.
|
||||
// We won't use it, but the decoding also positions the byte source after it.
|
||||
partitioner.getTokenFactory().fromComparableBytes(ByteSourceInverse.nextComponentSource(peekableByteSource), version);
|
||||
partitioner.getTokenFactory().skipComparableBytes(ByteSourceInverse.nextComponentSource(peekableByteSource), version, partitioner);
|
||||
// Decode the key bytes from the second component.
|
||||
byte[] keyBytes = ByteSourceInverse.getUnescapedBytes(ByteSourceInverse.nextComponentSource(peekableByteSource));
|
||||
int terminator = peekableByteSource.next();
|
||||
assert terminator == ByteSource.TERMINATOR : "Decorated key encoding must end in terminator.";
|
||||
return keyBytes;
|
||||
}
|
||||
|
||||
public static int keySizeFromByteSource(ByteSource.Peekable peekableByteSource,
|
||||
Version version,
|
||||
IPartitioner partitioner)
|
||||
{
|
||||
assert version != Version.LEGACY; // reverse translation is not supported for LEGACY version.
|
||||
// Decode the token from the first component of the multi-component sequence representing the whole decorated key.
|
||||
// We won't use it, but the decoding also positions the byte source after it.
|
||||
partitioner.getTokenFactory().skipComparableBytes(ByteSourceInverse.nextComponentSource(peekableByteSource), version, partitioner);
|
||||
// Decode the key bytes from the second component.
|
||||
int keyBytesCount = ByteSourceInverse.getUnescapedBytesCount(ByteSourceInverse.nextComponentSource(peekableByteSource));
|
||||
int terminator = peekableByteSource.next();
|
||||
assert terminator == ByteSource.TERMINATOR : "Decorated key encoding must end in terminator.";
|
||||
return keyBytesCount;
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
|
@ -56,6 +56,8 @@ public interface PartitionPosition extends RingPosition<PartitionPosition>, Byte
|
|||
public Kind kind();
|
||||
public boolean isMinimum();
|
||||
|
||||
public boolean isMaximum();
|
||||
|
||||
/**
|
||||
* Produce a prefix-free byte-comparable representation of the key, i.e. such a sequence of bytes that any pair x, y
|
||||
* of valid positions (with the same key column types and partitioner),
|
||||
|
|
|
|||
|
|
@ -68,7 +68,7 @@ public class ShardBoundaries
|
|||
{
|
||||
for (int i = 0; i < boundaries.length; i++)
|
||||
{
|
||||
if (tk.compareTo(boundaries[i]) < 0)
|
||||
if (tk.compareTo(boundaries[i]) <= 0) // boundaries are end-inclusive
|
||||
return i;
|
||||
}
|
||||
return boundaries.length;
|
||||
|
|
@ -87,6 +87,20 @@ public class ShardBoundaries
|
|||
return getShardForToken(key.getToken());
|
||||
}
|
||||
|
||||
public Token getShardStartBoundary(int shardId)
|
||||
{
|
||||
if (shardId <= 0 || shardId >= boundaries.length)
|
||||
return null;
|
||||
return boundaries[shardId - 1];
|
||||
}
|
||||
|
||||
public Token getShardEndBoundary(int shardId)
|
||||
{
|
||||
if (shardId < 0 || shardId >= boundaries.length)
|
||||
return null;
|
||||
return boundaries[shardId];
|
||||
}
|
||||
|
||||
/**
|
||||
* The number of shards that this boundaries support, that is how many different shard ids {@link #getShardForToken} might
|
||||
* possibly return.
|
||||
|
|
|
|||
|
|
@ -59,8 +59,10 @@ import org.apache.cassandra.db.tries.InMemoryTrie;
|
|||
import org.apache.cassandra.db.tries.Trie;
|
||||
import org.apache.cassandra.dht.AbstractBounds;
|
||||
import org.apache.cassandra.dht.Bounds;
|
||||
import org.apache.cassandra.dht.IPartitioner;
|
||||
import org.apache.cassandra.dht.IncludingExcludingBounds;
|
||||
import org.apache.cassandra.dht.Range;
|
||||
import org.apache.cassandra.dht.Token;
|
||||
import org.apache.cassandra.index.transactions.UpdateTransaction;
|
||||
import org.apache.cassandra.io.compress.BufferType;
|
||||
import org.apache.cassandra.io.sstable.SSTableReadsListener;
|
||||
|
|
@ -235,6 +237,14 @@ public class TrieMemtable extends AbstractShardedMemtable
|
|||
return total;
|
||||
}
|
||||
|
||||
public long partitionKeysTotalSize()
|
||||
{
|
||||
long total = 0;
|
||||
for (MemtableShard shard : shards)
|
||||
total += shard.partitionKeysSize();
|
||||
return total;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the minTS if one available, otherwise NO_MIN_TIMESTAMP.
|
||||
*
|
||||
|
|
@ -348,21 +358,93 @@ public class TrieMemtable extends AbstractShardedMemtable
|
|||
@Override
|
||||
public FlushablePartitionSet<MemtablePartition> getFlushSet(PartitionPosition from, PartitionPosition to)
|
||||
{
|
||||
Trie<BTreePartitionData> toFlush = mergedTrie.subtrie(from, true, to, false);
|
||||
boolean allPositionsToFlush = from == null && to == null
|
||||
|| from != null && to != null
|
||||
&& from.isMinimum()
|
||||
&& to.isMaximum();
|
||||
|
||||
// to avoid an unnessesary wrapping of mergedTrie into SlicedTrie
|
||||
Trie<BTreePartitionData> toFlush = allPositionsToFlush ? mergedTrie.subtrie(null, true, null, false) :
|
||||
mergedTrie.subtrie(from, true, to, false);
|
||||
|
||||
long partitionKeySize;
|
||||
long partitionCount;
|
||||
|
||||
/*
|
||||
In total, we have boundaries.shardCount() shards, with indexes from 0 to boundaries.shardCount() - 1
|
||||
We start with 0 or 1 partial shards, then we may have several full shards, and we finish with 0 or 1 partial shards
|
||||
(token_m .......... token_m+1](token_m+1 ... ... ... ... token_n-1](token_n ........... token_n+1]
|
||||
↑ ↑ ↑ ↑
|
||||
[---partial shard--|---full shard--- ... ---full shard---|---partial shard---)
|
||||
|
||||
*/
|
||||
int fromShardFull;
|
||||
int fromShardPartial;
|
||||
int toShardFull;
|
||||
int toShardPartial;
|
||||
|
||||
if (from == null || from.isMinimum())
|
||||
fromShardFull = fromShardPartial = 0;
|
||||
else
|
||||
{
|
||||
fromShardPartial = boundaries.getShardForToken(from.getToken());
|
||||
// not symmetric to "to" logic because the start part of shard ranges is exclusive
|
||||
Token fromEndBoundaryToken = boundaries.getShardEndBoundary(fromShardPartial);
|
||||
if (fromEndBoundaryToken != null && from.equals(fromEndBoundaryToken.maxKeyBound()))
|
||||
{
|
||||
fromShardPartial++;
|
||||
fromShardFull = fromShardPartial;
|
||||
}
|
||||
else
|
||||
fromShardFull = fromShardPartial + 1;
|
||||
}
|
||||
|
||||
if (to == null || to.isMaximum())
|
||||
toShardFull = toShardPartial = boundaries.shardCount() - 1;
|
||||
else
|
||||
{
|
||||
toShardPartial = boundaries.getShardForToken(to.getToken());
|
||||
Token toEndBoundaryToken = boundaries.getShardEndBoundary(toShardPartial);
|
||||
if (toEndBoundaryToken != null && to.equals(toEndBoundaryToken.maxKeyBound()))
|
||||
toShardFull = toShardPartial;
|
||||
else
|
||||
toShardFull = toShardPartial - 1;
|
||||
}
|
||||
|
||||
List<Trie<BTreePartitionData>> partialShardTries = new ArrayList<>(2);
|
||||
boolean fromShardAdded = false;
|
||||
if (fromShardPartial != fromShardFull)
|
||||
{
|
||||
partialShardTries.add(shards[fromShardPartial].data);
|
||||
fromShardAdded = true;
|
||||
}
|
||||
if (toShardPartial != toShardFull && (toShardPartial != fromShardPartial || !fromShardAdded) )
|
||||
{
|
||||
partialShardTries.add(shards[toShardPartial].data);
|
||||
}
|
||||
|
||||
long keySize = 0;
|
||||
int keyCount = 0;
|
||||
|
||||
for (Iterator<Map.Entry<ByteComparable, BTreePartitionData>> it = toFlush.entryIterator(); it.hasNext(); )
|
||||
Trie<BTreePartitionData> partialShardsTrie = Trie.mergeDistinct(partialShardTries).subtrie(from, true, to, false);
|
||||
IPartitioner partitioner = metadata().partitioner;
|
||||
for (Iterator<Map.Entry<ByteComparable, BTreePartitionData>> it = partialShardsTrie.entryIterator(); it.hasNext(); )
|
||||
{
|
||||
Map.Entry<ByteComparable, BTreePartitionData> en = it.next();
|
||||
byte[] keyBytes = DecoratedKey.keyFromByteSource(ByteSource.peekable(en.getKey().asComparableBytes(BYTE_COMPARABLE_VERSION)),
|
||||
BYTE_COMPARABLE_VERSION,
|
||||
metadata().partitioner);
|
||||
keySize += keyBytes.length;
|
||||
int keyByteSize = DecoratedKey.keySizeFromByteSource(ByteSource.peekable(en.getKey().asComparableBytes(BYTE_COMPARABLE_VERSION)),
|
||||
BYTE_COMPARABLE_VERSION,
|
||||
partitioner);
|
||||
keySize += keyByteSize;
|
||||
keyCount++;
|
||||
}
|
||||
long partitionKeySize = keySize;
|
||||
int partitionCount = keyCount;
|
||||
|
||||
for (int i = fromShardFull; i <= toShardFull; i++)
|
||||
{
|
||||
keySize += shards[i].partitionKeysSize();
|
||||
keyCount += shards[i].size();
|
||||
}
|
||||
|
||||
partitionKeySize = keySize;
|
||||
partitionCount = keyCount;
|
||||
|
||||
return new AbstractFlushablePartitionSet<MemtablePartition>()
|
||||
{
|
||||
|
|
@ -416,6 +498,8 @@ public class TrieMemtable extends AbstractShardedMemtable
|
|||
|
||||
private volatile long currentOperations = 0;
|
||||
|
||||
private volatile long partitionKeysSize = 0;
|
||||
|
||||
@Unmetered
|
||||
private final ReentrantLock writeLock = new ReentrantLock();
|
||||
|
||||
|
|
@ -489,6 +573,7 @@ public class TrieMemtable extends AbstractShardedMemtable
|
|||
minLocalDeletionTime = Math.min(minLocalDeletionTime, update.stats().minLocalDeletionTime);
|
||||
liveDataSize += updater.dataSize;
|
||||
currentOperations += update.operationCount();
|
||||
partitionKeysSize += updater.keySize;
|
||||
|
||||
columnsCollector.update(update.columns());
|
||||
statsCollector.update(update.stats());
|
||||
|
|
@ -526,6 +611,11 @@ public class TrieMemtable extends AbstractShardedMemtable
|
|||
return currentOperations;
|
||||
}
|
||||
|
||||
long partitionKeysSize()
|
||||
{
|
||||
return partitionKeysSize;
|
||||
}
|
||||
|
||||
long minLocalDeletionTime()
|
||||
{
|
||||
return minLocalDeletionTime;
|
||||
|
|
|
|||
|
|
@ -43,6 +43,9 @@ public class BTreePartitionUpdater implements UpdateFunction<Row, Row>, ColumnDa
|
|||
final Cloner cloner;
|
||||
final UpdateTransaction indexer;
|
||||
public long dataSize;
|
||||
|
||||
public long keySize;
|
||||
|
||||
long heapSize;
|
||||
public long colUpdateTimeDelta = Long.MAX_VALUE;
|
||||
|
||||
|
|
@ -54,12 +57,14 @@ public class BTreePartitionUpdater implements UpdateFunction<Row, Row>, ColumnDa
|
|||
this.indexer = indexer;
|
||||
this.heapSize = 0;
|
||||
this.dataSize = 0;
|
||||
this.keySize = 0;
|
||||
}
|
||||
|
||||
public BTreePartitionData mergePartitions(BTreePartitionData current, final PartitionUpdate update)
|
||||
{
|
||||
if (current == null)
|
||||
{
|
||||
keySize = update.partitionKey.getKeyLength();
|
||||
current = BTreePartitionData.EMPTY;
|
||||
onAllocatedOnHeap(BTreePartitionData.UNSHARED_HEAP_SIZE);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -303,6 +303,11 @@ public class ByteOrderedPartitioner implements IPartitioner
|
|||
return new BytesToken(ByteSourceInverse.getUnescapedBytes(comparableBytes));
|
||||
}
|
||||
|
||||
public void skipComparableBytes(ByteSource.Peekable comparableBytes, ByteComparable.Version version, IPartitioner partitioner)
|
||||
{
|
||||
ByteSourceInverse.skipUnescapedBytes(comparableBytes);
|
||||
}
|
||||
|
||||
public ByteBuffer toByteArray(Token token)
|
||||
{
|
||||
BytesToken bytesToken = (BytesToken) token;
|
||||
|
|
|
|||
|
|
@ -100,6 +100,12 @@ public class LocalPartitioner implements IPartitioner
|
|||
return new LocalToken(tokenData);
|
||||
}
|
||||
|
||||
public void skipComparableBytes(ByteSource.Peekable comparableBytes, ByteComparable.Version version, IPartitioner partitioner)
|
||||
{
|
||||
// read and ingore the result
|
||||
comparator.fromComparableBytes(ByteBufferAccessor.instance, comparableBytes, version);
|
||||
}
|
||||
|
||||
public ByteBuffer toByteArray(Token token)
|
||||
{
|
||||
return ((LocalToken)token).token;
|
||||
|
|
|
|||
|
|
@ -459,6 +459,11 @@ public class Murmur3Partitioner implements IPartitioner
|
|||
return new LongToken(tokenData);
|
||||
}
|
||||
|
||||
public void skipComparableBytes(ByteSource.Peekable comparableBytes, ByteComparable.Version version, IPartitioner partitioner)
|
||||
{
|
||||
ByteSourceInverse.skipBytes(comparableBytes, Long.BYTES);
|
||||
}
|
||||
|
||||
public ByteBuffer toByteArray(Token token)
|
||||
{
|
||||
LongToken longToken = (LongToken) token;
|
||||
|
|
|
|||
|
|
@ -165,6 +165,11 @@ public class OrderPreservingPartitioner implements IPartitioner
|
|||
return new StringToken(ByteSourceInverse.getString(comparableBytes));
|
||||
}
|
||||
|
||||
public void skipComparableBytes(ByteSource.Peekable comparableBytes, ByteComparable.Version version, IPartitioner partitioner)
|
||||
{
|
||||
ByteSourceInverse.skipUnescapedBytes(comparableBytes);
|
||||
}
|
||||
|
||||
public ByteBuffer toByteArray(Token token)
|
||||
{
|
||||
StringToken stringToken = (StringToken) token;
|
||||
|
|
|
|||
|
|
@ -196,6 +196,12 @@ public class RandomPartitioner implements IPartitioner
|
|||
return fromByteArray(IntegerType.instance.fromComparableBytes(ByteBufferAccessor.instance, comparableBytes, version));
|
||||
}
|
||||
|
||||
public void skipComparableBytes(ByteSource.Peekable comparableBytes, ByteComparable.Version version, IPartitioner partitioner)
|
||||
{
|
||||
// read and ignore the result
|
||||
IntegerType.instance.fromComparableBytes(ByteBufferAccessor.instance, comparableBytes, version);
|
||||
}
|
||||
|
||||
public ByteBuffer toByteArray(Token token)
|
||||
{
|
||||
BigIntegerToken bigIntegerToken = (BigIntegerToken) token;
|
||||
|
|
|
|||
|
|
@ -111,6 +111,11 @@ public class ReversedLongLocalPartitioner implements IPartitioner
|
|||
return new ReversedLongLocalToken(tokenData);
|
||||
}
|
||||
|
||||
public void skipComparableBytes(ByteSource.Peekable comparableBytes, ByteComparable.Version version, IPartitioner partitioner)
|
||||
{
|
||||
ByteSourceInverse.skipBytes(comparableBytes, Long.BYTES);
|
||||
}
|
||||
|
||||
public ByteBuffer toByteArray(Token token)
|
||||
{
|
||||
ReversedLongLocalToken longToken = (ReversedLongLocalToken) token;
|
||||
|
|
|
|||
|
|
@ -75,6 +75,8 @@ public abstract class Token implements RingPosition<Token>, Serializable
|
|||
*/
|
||||
public abstract Token fromComparableBytes(ByteSource.Peekable comparableBytes, ByteComparable.Version version);
|
||||
|
||||
public abstract void skipComparableBytes(ByteSource.Peekable comparableBytes, ByteComparable.Version version, IPartitioner partitioner);
|
||||
|
||||
public abstract String toString(Token token); // serialize as string, not necessarily human-readable
|
||||
public abstract Token fromString(String string); // deserialize
|
||||
|
||||
|
|
@ -330,6 +332,11 @@ public abstract class Token implements RingPosition<Token>, Serializable
|
|||
return this.equals(minValue());
|
||||
}
|
||||
|
||||
public boolean isMaximum()
|
||||
{
|
||||
return getPartitioner().supportsSplitting() && this.equals(getPartitioner().getMaximumTokenForSplitting());
|
||||
}
|
||||
|
||||
/*
|
||||
* A token corresponds to the range of all the keys having this token.
|
||||
* A token is thus not comparable directly to a key. But to be able to select
|
||||
|
|
@ -420,9 +427,15 @@ public abstract class Token implements RingPosition<Token>, Serializable
|
|||
|
||||
public boolean isMinimum()
|
||||
{
|
||||
// minimum token is reserved for boundaries, so there is no need for isMinimumBound check
|
||||
return getToken().isMinimum();
|
||||
}
|
||||
|
||||
public boolean isMaximum()
|
||||
{
|
||||
return getToken().isMaximum() && !isMinimumBound;
|
||||
}
|
||||
|
||||
public PartitionPosition.Kind kind()
|
||||
{
|
||||
return isMinimumBound ? PartitionPosition.Kind.MIN_BOUND : PartitionPosition.Kind.MAX_BOUND;
|
||||
|
|
|
|||
|
|
@ -841,6 +841,15 @@ public interface ByteSource
|
|||
peeked = wrapped.next();
|
||||
return peeked;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString()
|
||||
{
|
||||
return "Peekable{" +
|
||||
"wrapped=" + wrapped +
|
||||
", peeked=" + peeked +
|
||||
'}';
|
||||
}
|
||||
}
|
||||
|
||||
public static Peekable peekable(ByteSource p)
|
||||
|
|
|
|||
|
|
@ -190,6 +190,12 @@ public final class ByteSourceInverse
|
|||
return getUnsignedFixedLengthAsLong(byteSource, 8) ^ LONG_SIGN_BIT;
|
||||
}
|
||||
|
||||
public static void skipBytes(ByteSource byteSource, int bytesToSkip)
|
||||
{
|
||||
for (int i = 0; i < bytesToSkip; i++)
|
||||
getAndCheckByte(byteSource, i, bytesToSkip);
|
||||
}
|
||||
|
||||
/**
|
||||
* Converts the given {@link ByteSource} to a {@code byte}.
|
||||
*
|
||||
|
|
@ -297,6 +303,17 @@ public final class ByteSourceInverse
|
|||
return byteSource == null ? null : readBytes(unescape(byteSource));
|
||||
}
|
||||
|
||||
public static int getUnescapedBytesCount(ByteSource.Peekable byteSource)
|
||||
{
|
||||
return byteSource == null ? 0 : countBytes(unescape(byteSource));
|
||||
}
|
||||
|
||||
public static void skipUnescapedBytes(ByteSource.Peekable byteSource)
|
||||
{
|
||||
// read and ignore the result
|
||||
getUnescapedBytesCount(byteSource);
|
||||
}
|
||||
|
||||
/**
|
||||
* As above, but converts the result to a ByteSource.
|
||||
*/
|
||||
|
|
@ -385,6 +402,18 @@ public final class ByteSourceInverse
|
|||
return buf;
|
||||
}
|
||||
|
||||
public static int countBytes(ByteSource byteSource)
|
||||
{
|
||||
Preconditions.checkNotNull(byteSource);
|
||||
|
||||
int readBytes = 0;
|
||||
while (byteSource.next() != ByteSource.END_OF_STREAM)
|
||||
{
|
||||
readBytes++;
|
||||
}
|
||||
return readBytes;
|
||||
}
|
||||
|
||||
/**
|
||||
* Reads the bytes of the given source into a byte array. Doesn't do any transformation on the bytes, just reads
|
||||
* them until it reads an {@link ByteSource#END_OF_STREAM} byte, after which it returns an array of all the read
|
||||
|
|
|
|||
|
|
@ -0,0 +1,34 @@
|
|||
/*
|
||||
* 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.memtable;
|
||||
|
||||
import org.junit.BeforeClass;
|
||||
|
||||
import org.apache.cassandra.config.DatabaseDescriptor;
|
||||
import org.apache.cassandra.dht.ByteOrderedPartitioner;
|
||||
public class NonSplittablePartitionerTrieMemtableFlushSetTest extends TrieMemtableFlushSetTestBase
|
||||
{
|
||||
@BeforeClass
|
||||
public static void setUpClass()
|
||||
{
|
||||
configure();
|
||||
DatabaseDescriptor.setPartitionerUnsafe(ByteOrderedPartitioner.instance);
|
||||
prepareServer();
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,36 @@
|
|||
/*
|
||||
* 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.memtable;
|
||||
|
||||
import org.junit.BeforeClass;
|
||||
|
||||
import org.apache.cassandra.config.DatabaseDescriptor;
|
||||
import org.apache.cassandra.dht.Murmur3Partitioner;
|
||||
|
||||
public class SplittablePartitionerTrieMemtableFlushSetTest extends TrieMemtableFlushSetTestBase
|
||||
{
|
||||
|
||||
@BeforeClass
|
||||
public static void setUpClass()
|
||||
{
|
||||
configure();
|
||||
DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
|
||||
prepareServer();
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,193 @@
|
|||
/*
|
||||
* 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.memtable;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import com.google.common.collect.ImmutableList;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.junit.runners.Parameterized;
|
||||
|
||||
import org.apache.cassandra.config.Config;
|
||||
import org.apache.cassandra.config.DatabaseDescriptor;
|
||||
import org.apache.cassandra.config.InheritingClass;
|
||||
import org.apache.cassandra.cql3.CQLTester;
|
||||
import org.apache.cassandra.db.ColumnFamilyStore;
|
||||
import org.apache.cassandra.db.DecoratedKey;
|
||||
import org.apache.cassandra.db.Keyspace;
|
||||
import org.apache.cassandra.db.PartitionPosition;
|
||||
import org.apache.cassandra.db.marshal.LongType;
|
||||
import org.apache.cassandra.dht.Token;
|
||||
|
||||
import static org.apache.cassandra.db.memtable.AbstractShardedMemtable.SHARDS_OPTION;
|
||||
|
||||
@RunWith(Parameterized.class)
|
||||
public abstract class TrieMemtableFlushSetTestBase extends CQLTester
|
||||
{
|
||||
|
||||
static final int partitions = 1_000;
|
||||
static final int rowsPerPartition = 2;
|
||||
|
||||
@Parameterized.Parameter(0)
|
||||
public int shardsCount;
|
||||
|
||||
@Parameterized.Parameters(name = "shards={0}")
|
||||
public static List<Integer> parameters()
|
||||
{
|
||||
return ImmutableList.of(1, 2, 3, 16);
|
||||
}
|
||||
|
||||
|
||||
public static void configure()
|
||||
{
|
||||
prePrepareServer();
|
||||
|
||||
LinkedHashMap<String, InheritingClass> memtableConfig = new LinkedHashMap<>();
|
||||
for (int shardCount : parameters())
|
||||
memtableConfig.put("trie_" + shardCount,
|
||||
new InheritingClass(
|
||||
null,
|
||||
TrieMemtable.class.getName(),
|
||||
Map.of(SHARDS_OPTION, String.valueOf(shardCount))
|
||||
)
|
||||
);
|
||||
memtableConfig.put("skiplist", new InheritingClass(null, SkipListMemtable.class.getName(), Map.of()));
|
||||
DatabaseDescriptor.getRawConfig().memtable = new Config.MemtableOptions();
|
||||
DatabaseDescriptor.getRawConfig().memtable.configurations = memtableConfig;
|
||||
}
|
||||
|
||||
@Test
|
||||
public void checkPartitionCountAndPartitionKeysSizeReportedByFlushSet()
|
||||
{
|
||||
String keyspace = createKeyspace("CREATE KEYSPACE %s with replication = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 } and durable_writes = false");
|
||||
String table = createTable(keyspace, "CREATE TABLE %s ( userid bigint, picid bigint, commentid bigint, PRIMARY KEY(userid, picid))" +
|
||||
" WITH memtable = 'trie_" + shardsCount + "'");
|
||||
|
||||
String tableToCompare = createTable(keyspace, "CREATE TABLE %s ( userid bigint, picid bigint, commentid bigint, PRIMARY KEY(userid, picid))" +
|
||||
" WITH memtable = 'skiplist'");
|
||||
execute("use " + keyspace + ';');
|
||||
|
||||
String writeStatement = "INSERT INTO " + table + "(userid,picid,commentid) VALUES (?,?,?)";
|
||||
String writeStatementToCompare = "INSERT INTO " + tableToCompare + "(userid,picid,commentid) VALUES (?,?,?)";
|
||||
|
||||
|
||||
ColumnFamilyStore cfs = Keyspace.open(keyspace).getColumnFamilyStore(table);
|
||||
ColumnFamilyStore cfsToCompare = Keyspace.open(keyspace).getColumnFamilyStore(tableToCompare);
|
||||
|
||||
for (long i = 0; i < partitions; ++i)
|
||||
{
|
||||
for (long j = 0; j < rowsPerPartition; ++j)
|
||||
{
|
||||
execute(writeStatement, i, j, i + j);
|
||||
execute(writeStatementToCompare, i, j, i + j);
|
||||
}
|
||||
}
|
||||
|
||||
Memtable memtable = cfs.getCurrentMemtable();
|
||||
Memtable memtableToCompare = cfsToCompare.getCurrentMemtable();
|
||||
|
||||
{
|
||||
Memtable.FlushablePartitionSet<?> flushSetNull = memtable.getFlushSet(null, null);
|
||||
|
||||
Assert.assertEquals(partitions, flushSetNull.partitionCount());
|
||||
Assert.assertEquals(partitions * Long.BYTES, flushSetNull.partitionKeysSize());
|
||||
}
|
||||
|
||||
{
|
||||
Memtable.FlushablePartitionSet<?> flushSetFromMinToMax = memtable.getFlushSet(getMinPosition(cfs), getMaxPosition(cfs));
|
||||
|
||||
Assert.assertEquals(partitions, flushSetFromMinToMax.partitionCount());
|
||||
Assert.assertEquals(partitions * Long.BYTES, flushSetFromMinToMax.partitionKeysSize());
|
||||
}
|
||||
|
||||
ShardBoundaries shardBoundaries = cfs.localRangeSplits(shardsCount);
|
||||
testShardRanges(shardBoundaries, cfs, memtable, memtableToCompare);
|
||||
|
||||
// to test partial covered shards
|
||||
ShardBoundaries nonMatchingShardBoundaries1 = cfs.localRangeSplits(13); // a prime number to not match with shard boundaries
|
||||
testShardRanges(nonMatchingShardBoundaries1, cfs, memtable, memtableToCompare);
|
||||
|
||||
ShardBoundaries nonMatchingShardBoundaries2 = cfs.localRangeSplits(32); // to check small ranges within shard boundaries
|
||||
testShardRanges(nonMatchingShardBoundaries2, cfs, memtable, memtableToCompare);
|
||||
|
||||
// to test the case when a flush position is a partition key
|
||||
{
|
||||
ByteBuffer oneOfWrittenPartitionKeys = LongType.instance.decompose(0L);
|
||||
DecoratedKey decoratedKey = cfs.getPartitioner().decorateKey(oneOfWrittenPartitionKeys);
|
||||
testRange(memtable, memtableToCompare, getMinPosition(cfs), decoratedKey);
|
||||
testRange(memtable, memtableToCompare, decoratedKey, getMaxPosition(cfs));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private static void testShardRanges(ShardBoundaries shardBoundaries, ColumnFamilyStore cfs, Memtable memtable, Memtable memtableToCompare)
|
||||
{
|
||||
for (int fromShardIndex = 0; fromShardIndex < shardBoundaries.shardCount(); fromShardIndex++)
|
||||
{
|
||||
for (int toShardIndex = fromShardIndex; toShardIndex < shardBoundaries.shardCount(); toShardIndex++)
|
||||
{
|
||||
Token from = shardBoundaries.getShardStartBoundary(fromShardIndex);
|
||||
PartitionPosition positionFrom = from != null ? from.maxKeyBound() : getMinPosition(cfs);
|
||||
|
||||
Token to = shardBoundaries.getShardEndBoundary(toShardIndex);
|
||||
PartitionPosition positionTo = to != null ? to.maxKeyBound() : getMaxPosition(cfs);
|
||||
|
||||
Memtable.FlushablePartitionSet<?> flushSet = memtable.getFlushSet(positionFrom, positionTo);
|
||||
Memtable.FlushablePartitionSet<?> flushSetToCompare = memtableToCompare.getFlushSet(positionFrom, positionTo);
|
||||
|
||||
String rangeInfo = "shard id range = [" + fromShardIndex + ", " + toShardIndex + "]" +
|
||||
", token range = [" + positionFrom + ", " + positionTo + ")";
|
||||
|
||||
Assert.assertEquals("Number of partition keys does not match for " + rangeInfo,
|
||||
flushSetToCompare.partitionCount(), flushSet.partitionCount());
|
||||
Assert.assertEquals("Total size of partition keys does not match for " + rangeInfo,
|
||||
flushSetToCompare.partitionKeysSize(), flushSet.partitionKeysSize());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static void testRange(Memtable memtable, Memtable memtableToCompare, PartitionPosition positionFrom, PartitionPosition positionTo)
|
||||
{
|
||||
Memtable.FlushablePartitionSet<?> flushSet = memtable.getFlushSet(positionFrom, positionTo);
|
||||
Memtable.FlushablePartitionSet<?> flushSetToCompare = memtableToCompare.getFlushSet(positionFrom, positionTo);
|
||||
|
||||
String rangeInfo = "token range = [" + positionFrom + ", " + positionTo + ")";
|
||||
|
||||
Assert.assertEquals("Number of partition keys does not match for " + rangeInfo,
|
||||
flushSetToCompare.partitionCount(), flushSet.partitionCount());
|
||||
Assert.assertEquals("Total size of partition keys does not match for " + rangeInfo,
|
||||
flushSetToCompare.partitionKeysSize(), flushSet.partitionKeysSize());
|
||||
}
|
||||
|
||||
private static PartitionPosition getMinPosition(ColumnFamilyStore cfs)
|
||||
{
|
||||
return cfs.getPartitioner().getMinimumToken().minKeyBound();
|
||||
}
|
||||
|
||||
private static PartitionPosition getMaxPosition(ColumnFamilyStore cfs)
|
||||
{
|
||||
return cfs.getPartitioner().supportsSplitting() ? cfs.getPartitioner().getMaximumTokenForSplitting().maxKeyBound() : null;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -102,4 +102,17 @@ public class IPartitionerTest
|
|||
.isEqualTo(token);
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
public void skipTokenInComparableBytes()
|
||||
{
|
||||
qt().forAll(AccordGenerators.fromQT(CassandraGenerators.token())).check(token -> {
|
||||
var p = token.getPartitioner();
|
||||
var comparable = Objects.requireNonNull(ByteSource.peekable(p.getTokenFactory().asComparableBytes(token, ByteComparable.Version.OSS50)));
|
||||
p.getTokenFactory().skipComparableBytes(comparable, ByteComparable.Version.OSS50, p);
|
||||
Assertions.assertThat(comparable.next())
|
||||
.describedAs("skipComparableBytes should skip the whole encoded token value but %s is not ended, partitioner: %s", comparable, p)
|
||||
.isEqualTo(ByteSource.END_OF_STREAM);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
|
@ -108,6 +108,12 @@ public class LengthPartitioner extends AccordSplitter implements IPartitioner
|
|||
return fromByteArray(IntegerType.instance.fromComparableBytes(comparableBytes, version));
|
||||
}
|
||||
|
||||
public void skipComparableBytes(ByteSource.Peekable comparableBytes, ByteComparable.Version version, IPartitioner partitioner)
|
||||
{
|
||||
// read and ignore the result
|
||||
IntegerType.instance.fromComparableBytes(comparableBytes, version);
|
||||
}
|
||||
|
||||
public String toString(Token token)
|
||||
{
|
||||
BigIntegerToken bigIntegerToken = (BigIntegerToken) token;
|
||||
|
|
|
|||
|
|
@ -73,6 +73,11 @@ public class DecoratedKeyByteSourceTest
|
|||
byte[] keyBytes = DecoratedKey.keyFromByteSource(src, version, ByteOrderedPartitioner.instance);
|
||||
Assert.assertEquals(ByteSource.END_OF_STREAM, src.next());
|
||||
Assert.assertArrayEquals(initialBuffer.getKey().array(), keyBytes);
|
||||
|
||||
ByteSource.Peekable srcCount = ByteSource.peekable(initialBuffer.asComparableBytes(version));
|
||||
int keyBytesCount = DecoratedKey.keySizeFromByteSource(srcCount, version, ByteOrderedPartitioner.instance);
|
||||
Assert.assertEquals(ByteSource.END_OF_STREAM, srcCount.next());
|
||||
Assert.assertEquals(keyBytes.length, keyBytesCount);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue