diff --git a/CHANGES.txt b/CHANGES.txt index 10fe1993a8..e97e1d71bf 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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) diff --git a/src/java/org/apache/cassandra/db/DecoratedKey.java b/src/java/org/apache/cassandra/db/DecoratedKey.java index a974de81d6..309f764a96 100644 --- a/src/java/org/apache/cassandra/db/DecoratedKey.java +++ b/src/java/org/apache/cassandra/db/DecoratedKey.java @@ -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; + } + + } \ No newline at end of file diff --git a/src/java/org/apache/cassandra/db/PartitionPosition.java b/src/java/org/apache/cassandra/db/PartitionPosition.java index d936e566a7..fd7a1e948f 100644 --- a/src/java/org/apache/cassandra/db/PartitionPosition.java +++ b/src/java/org/apache/cassandra/db/PartitionPosition.java @@ -56,6 +56,8 @@ public interface PartitionPosition extends RingPosition, 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), diff --git a/src/java/org/apache/cassandra/db/memtable/ShardBoundaries.java b/src/java/org/apache/cassandra/db/memtable/ShardBoundaries.java index 5412c2cf62..102d087bb2 100644 --- a/src/java/org/apache/cassandra/db/memtable/ShardBoundaries.java +++ b/src/java/org/apache/cassandra/db/memtable/ShardBoundaries.java @@ -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. diff --git a/src/java/org/apache/cassandra/db/memtable/TrieMemtable.java b/src/java/org/apache/cassandra/db/memtable/TrieMemtable.java index 2a2813d0ad..3e183c572e 100644 --- a/src/java/org/apache/cassandra/db/memtable/TrieMemtable.java +++ b/src/java/org/apache/cassandra/db/memtable/TrieMemtable.java @@ -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 getFlushSet(PartitionPosition from, PartitionPosition to) { - Trie 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 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> 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> it = toFlush.entryIterator(); it.hasNext(); ) + Trie partialShardsTrie = Trie.mergeDistinct(partialShardTries).subtrie(from, true, to, false); + IPartitioner partitioner = metadata().partitioner; + for (Iterator> it = partialShardsTrie.entryIterator(); it.hasNext(); ) { Map.Entry 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() { @@ -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; diff --git a/src/java/org/apache/cassandra/db/partitions/BTreePartitionUpdater.java b/src/java/org/apache/cassandra/db/partitions/BTreePartitionUpdater.java index 023019242d..bce379d98a 100644 --- a/src/java/org/apache/cassandra/db/partitions/BTreePartitionUpdater.java +++ b/src/java/org/apache/cassandra/db/partitions/BTreePartitionUpdater.java @@ -43,6 +43,9 @@ public class BTreePartitionUpdater implements UpdateFunction, 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, 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); } diff --git a/src/java/org/apache/cassandra/dht/ByteOrderedPartitioner.java b/src/java/org/apache/cassandra/dht/ByteOrderedPartitioner.java index 527b5113ba..551963ecec 100644 --- a/src/java/org/apache/cassandra/dht/ByteOrderedPartitioner.java +++ b/src/java/org/apache/cassandra/dht/ByteOrderedPartitioner.java @@ -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; diff --git a/src/java/org/apache/cassandra/dht/LocalPartitioner.java b/src/java/org/apache/cassandra/dht/LocalPartitioner.java index 4c45887dc4..1fec57065a 100644 --- a/src/java/org/apache/cassandra/dht/LocalPartitioner.java +++ b/src/java/org/apache/cassandra/dht/LocalPartitioner.java @@ -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; diff --git a/src/java/org/apache/cassandra/dht/Murmur3Partitioner.java b/src/java/org/apache/cassandra/dht/Murmur3Partitioner.java index 35122bc249..74a605946a 100644 --- a/src/java/org/apache/cassandra/dht/Murmur3Partitioner.java +++ b/src/java/org/apache/cassandra/dht/Murmur3Partitioner.java @@ -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; diff --git a/src/java/org/apache/cassandra/dht/OrderPreservingPartitioner.java b/src/java/org/apache/cassandra/dht/OrderPreservingPartitioner.java index fe76e0eace..db60e0f046 100644 --- a/src/java/org/apache/cassandra/dht/OrderPreservingPartitioner.java +++ b/src/java/org/apache/cassandra/dht/OrderPreservingPartitioner.java @@ -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; diff --git a/src/java/org/apache/cassandra/dht/RandomPartitioner.java b/src/java/org/apache/cassandra/dht/RandomPartitioner.java index 45aff607c4..40dfbbe614 100644 --- a/src/java/org/apache/cassandra/dht/RandomPartitioner.java +++ b/src/java/org/apache/cassandra/dht/RandomPartitioner.java @@ -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; diff --git a/src/java/org/apache/cassandra/dht/ReversedLongLocalPartitioner.java b/src/java/org/apache/cassandra/dht/ReversedLongLocalPartitioner.java index f95f1e7764..c39b920864 100644 --- a/src/java/org/apache/cassandra/dht/ReversedLongLocalPartitioner.java +++ b/src/java/org/apache/cassandra/dht/ReversedLongLocalPartitioner.java @@ -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; diff --git a/src/java/org/apache/cassandra/dht/Token.java b/src/java/org/apache/cassandra/dht/Token.java index edee034aea..ae99b2b37d 100644 --- a/src/java/org/apache/cassandra/dht/Token.java +++ b/src/java/org/apache/cassandra/dht/Token.java @@ -75,6 +75,8 @@ public abstract class Token implements RingPosition, 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, 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, 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; diff --git a/src/java/org/apache/cassandra/utils/bytecomparable/ByteSource.java b/src/java/org/apache/cassandra/utils/bytecomparable/ByteSource.java index 83bb828f30..ade383dc52 100644 --- a/src/java/org/apache/cassandra/utils/bytecomparable/ByteSource.java +++ b/src/java/org/apache/cassandra/utils/bytecomparable/ByteSource.java @@ -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) diff --git a/src/java/org/apache/cassandra/utils/bytecomparable/ByteSourceInverse.java b/src/java/org/apache/cassandra/utils/bytecomparable/ByteSourceInverse.java index 4bf9d8c365..005b859669 100644 --- a/src/java/org/apache/cassandra/utils/bytecomparable/ByteSourceInverse.java +++ b/src/java/org/apache/cassandra/utils/bytecomparable/ByteSourceInverse.java @@ -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 diff --git a/test/unit/org/apache/cassandra/db/memtable/NonSplittablePartitionerTrieMemtableFlushSetTest.java b/test/unit/org/apache/cassandra/db/memtable/NonSplittablePartitionerTrieMemtableFlushSetTest.java new file mode 100644 index 0000000000..2869e50703 --- /dev/null +++ b/test/unit/org/apache/cassandra/db/memtable/NonSplittablePartitionerTrieMemtableFlushSetTest.java @@ -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(); + } +} diff --git a/test/unit/org/apache/cassandra/db/memtable/SplittablePartitionerTrieMemtableFlushSetTest.java b/test/unit/org/apache/cassandra/db/memtable/SplittablePartitionerTrieMemtableFlushSetTest.java new file mode 100644 index 0000000000..fd08e34027 --- /dev/null +++ b/test/unit/org/apache/cassandra/db/memtable/SplittablePartitionerTrieMemtableFlushSetTest.java @@ -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(); + } +} diff --git a/test/unit/org/apache/cassandra/db/memtable/TrieMemtableFlushSetTestBase.java b/test/unit/org/apache/cassandra/db/memtable/TrieMemtableFlushSetTestBase.java new file mode 100644 index 0000000000..d5149ca855 --- /dev/null +++ b/test/unit/org/apache/cassandra/db/memtable/TrieMemtableFlushSetTestBase.java @@ -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 parameters() + { + return ImmutableList.of(1, 2, 3, 16); + } + + + public static void configure() + { + prePrepareServer(); + + LinkedHashMap 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; + } + +} \ No newline at end of file diff --git a/test/unit/org/apache/cassandra/dht/IPartitionerTest.java b/test/unit/org/apache/cassandra/dht/IPartitionerTest.java index e531e25da5..14c8154e59 100644 --- a/test/unit/org/apache/cassandra/dht/IPartitionerTest.java +++ b/test/unit/org/apache/cassandra/dht/IPartitionerTest.java @@ -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); + }); + } } \ No newline at end of file diff --git a/test/unit/org/apache/cassandra/dht/LengthPartitioner.java b/test/unit/org/apache/cassandra/dht/LengthPartitioner.java index 1fcfa14ee1..401c0e0cd4 100644 --- a/test/unit/org/apache/cassandra/dht/LengthPartitioner.java +++ b/test/unit/org/apache/cassandra/dht/LengthPartitioner.java @@ -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; diff --git a/test/unit/org/apache/cassandra/utils/bytecomparable/DecoratedKeyByteSourceTest.java b/test/unit/org/apache/cassandra/utils/bytecomparable/DecoratedKeyByteSourceTest.java index 6c3e72efbe..aab6316e2c 100644 --- a/test/unit/org/apache/cassandra/utils/bytecomparable/DecoratedKeyByteSourceTest.java +++ b/test/unit/org/apache/cassandra/utils/bytecomparable/DecoratedKeyByteSourceTest.java @@ -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); } }