This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch cassandra-6.0
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/cassandra-6.0 by this push:
new b45f90cec4 Sharding support for non-vector SAI in-memory indexes
b45f90cec4 is described below
commit b45f90cec4b74f0dadec1ced92d7d5921645b279
Author: Yash Nasery <[email protected]>
AuthorDate: Sat Mar 14 11:45:12 2026 +0530
Sharding support for non-vector SAI in-memory indexes
patch by Yash Nasery, reviewed by Caleb Rackliffe and Francisco Guerrero
for CASSANDRA-18216
---
CHANGES.txt | 1 +
NEWS.txt | 1 +
.../developing/cql/indexing/sai/_sai-create.adoc | 1 +
.../pages/reference/cql-commands/create-index.adoc | 8 +-
.../cassandra/db/memtable/ShardBoundaries.java | 58 +++-
src/java/org/apache/cassandra/dht/Range.java | 16 +-
.../index/accord/RouteMemtableIndexManager.java | 9 +-
.../cassandra/index/sai/StorageAttachedIndex.java | 50 +++-
.../cassandra/index/sai/disk/RowMapping.java | 12 +-
.../index/sai/disk/v1/MemtableIndexWriter.java | 2 +-
.../cassandra/index/sai/memory/MemtableIndex.java | 103 +++----
.../index/sai/memory/MemtableIndexManager.java | 16 +-
.../index/sai/memory/ShardedMemtableIndex.java | 272 +++++++++++++++++++
...tableIndex.java => UnshardedMemtableIndex.java} | 38 ++-
.../test/cql3/SingleNodeTableWalkTest.java | 17 +-
.../microbench/sai/AbstractMemtableIndexBench.java | 118 ++++++++
.../microbench/sai/MemtableIndexFlushBench.java | 150 +++++++++++
.../sai/MemtableIndexPartitionReadBench.java | 174 ++++++++++++
.../microbench/sai/MemtableIndexReadBench.java | 135 ++++++++++
.../microbench/sai/MemtableIndexWriteBench.java | 105 ++++++++
.../cassandra/db/memtable/ShardBoundariesTest.java | 117 ++++++++
.../cassandra/dht/RangeIntersectsBoundsTest.java | 296 +++++++++++++++++++++
.../index/sai/cql/StorageAttachedIndexDDLTest.java | 75 ++++++
.../sai/memory/ShardedMemtableIndexFlushTest.java | 55 ++++
.../index/sai/memory/ShardedMemtableIndexTest.java | 255 ++++++++++++++++++
.../index/sai/memory/TrieMemoryIndexTest.java | 65 +----
.../index/sai/utils/SAIRandomizedTester.java | 69 +++++
27 files changed, 2056 insertions(+), 162 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 80603ef454..015680a337 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
6.0-alpha3
+ * Sharding support for non-vector SAI in-memory indexes (CASSANDRA-18216)
* Index hints serialization rejects a count that the validation accepts
(CASSANDRA-21700)
* Do not expose system_metrics virtual keyspace by default (CASSANDRA-21720)
* Tighten up NoSpamLogger changes after migration to Caffeine cache
(CASSANDRA-21721)
diff --git a/NEWS.txt b/NEWS.txt
index 7cb208f520..af526ab092 100644
--- a/NEWS.txt
+++ b/NEWS.txt
@@ -149,6 +149,7 @@ New features
cassandra.data_response_in_memory_max_rows_cl_one
andcassandra.data_response_in_memory_max_size_cl_one.
A limit is disabled by setting it to 0, and setting both limits of a
tier to 0 disables the in-memory path for that tier.
Note that the limits are not enforced in the middle of a range tombstone
span, so a response may exceed them in that case.
+ - SAI Memtable-attached indexes now support sharding via WITH OPTIONS. See
CQL documentation for details.
Upgrading
---------
- memtable_offheap_space: 0 is now rejected at startup when
memtable_allocation_type is
diff --git
a/doc/modules/cassandra/pages/developing/cql/indexing/sai/_sai-create.adoc
b/doc/modules/cassandra/pages/developing/cql/indexing/sai/_sai-create.adoc
index 45e9362837..b5627c9065 100644
--- a/doc/modules/cassandra/pages/developing/cql/indexing/sai/_sai-create.adoc
+++ b/doc/modules/cassandra/pages/developing/cql/indexing/sai/_sai-create.adoc
@@ -17,6 +17,7 @@ The SAI index options are defined in the `WITH OPTIONS`
clause.
The `case_sensitive` option is set to `false` to allow case-insensitive
searches.
The `normalize` option is set to `true` to allow searches to be normalized for
Unicode characters.
The `ascii_only` option is set to `true` to allow searches to be limited to
ASCII characters.
+The `shards` option sets the number of in-memory index shards per Memtable.
(Set to `'auto'` for `cassandra.memtable.shard.count` shards. Not supported on
vector column indexes.)
The `map` collection data type is the one exception, as shown in the
xref:#map-collection-in-sai-index[example below].
diff --git
a/doc/modules/cassandra/pages/reference/cql-commands/create-index.adoc
b/doc/modules/cassandra/pages/reference/cql-commands/create-index.adoc
index 5b191709e5..5a2802f45d 100644
--- a/doc/modules/cassandra/pages/reference/cql-commands/create-index.adoc
+++ b/doc/modules/cassandra/pages/reference/cql-commands/create-index.adoc
@@ -90,7 +90,7 @@ The regular column syntax applies for collection types `list`
and `set`.
[cols=",,"]
|============================
|option_map 2+<|Define options in JSON simple format.
-.4+||`case_sensitive` |Ignore case in matching string values. Default: `true`
+.5+||`case_sensitive` |Ignore case in matching string values. Default: `true`
|`normalize`|When set to `true`, perform
https://unicode.org/faq/normalization.html[Unicode normalization] on indexed
strings.
SAI supports Normalization Form C (NFC) Unicode.
When set to `true`, SAI normalizes the different versions of a given Unicode
character to a single version, retaining all the marks and symbols in the index.
@@ -107,6 +107,12 @@ Default: `false`.
The similarity function is used to compute the similarity between two vectors.
Valid options are: `EUCLIDEAN`, `DOT_PRODUCT`, `COSINE`
Default: `COSINE`
+|`shards` |Number of in-memory index shards per memtable. Set to `'auto'` to
use `cassandra.memtable.shard.count` shards.
+
+Not supported on vector column indexes.
+
+Default: `1`
+Maximum: `256`
|============================
== Usage notes
diff --git a/src/java/org/apache/cassandra/db/memtable/ShardBoundaries.java
b/src/java/org/apache/cassandra/db/memtable/ShardBoundaries.java
index 102d087bb2..b09687fa32 100644
--- a/src/java/org/apache/cassandra/db/memtable/ShardBoundaries.java
+++ b/src/java/org/apache/cassandra/db/memtable/ShardBoundaries.java
@@ -22,8 +22,13 @@ import java.util.List;
import com.google.common.annotations.VisibleForTesting;
+import org.agrona.collections.IntArrayList;
+
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.PartitionPosition;
+import org.apache.cassandra.dht.AbstractBounds;
+import org.apache.cassandra.dht.IPartitioner;
+import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.tcm.Epoch;
@@ -39,14 +44,18 @@ public class ShardBoundaries
{
private static final Token[] EMPTY_TOKEN_ARRAY = new Token[0];
- // Special boundaries that map all tokens to one shard.
- // These boundaries will be used in either of these cases:
+ // Special boundaries that map all tokens to one shard. These boundaries
will be used in either of these cases:
// - there is only 1 shard configured
// - the default partitioner doesn't support splitting
// - the keyspace is local system keyspace
+ private static final Range<PartitionPosition>[] EMPTY_RANGE_ARRAY = new
Range[0];
+ private static final IntArrayList EMPTY_BOUNDARIES_SHARDS = new
IntArrayList(new int[] { 0 }, 1, IntArrayList.DEFAULT_NULL_VALUE);
+
public static final ShardBoundaries NONE = new
ShardBoundaries(EMPTY_TOKEN_ARRAY, Epoch.EMPTY);
private final Token[] boundaries;
+ private final Range<PartitionPosition>[] ranges;
+ private final IntArrayList allShards;
public final Epoch epoch;
@VisibleForTesting
@@ -54,6 +63,34 @@ public class ShardBoundaries
{
this.boundaries = boundaries;
this.epoch = epoch;
+ this.ranges = precomputeRanges();
+
+ IntArrayList shards = new IntArrayList(boundaries.length + 1,
IntArrayList.DEFAULT_NULL_VALUE);
+ for (int i = 0; i <= boundaries.length; i++)
+ shards.addInt(i);
+ this.allShards = shards;
+ }
+
+ private Range<PartitionPosition>[] precomputeRanges()
+ {
+ if (boundaries.length == 0)
+ return EMPTY_RANGE_ARRAY;
+
+ IPartitioner partitioner = boundaries[0].getPartitioner();
+ Range<PartitionPosition>[] ranges = new Range[boundaries.length + 1];
+ int rangeIndex = 0;
+ PartitionPosition minimum =
partitioner.getMinimumToken().minKeyBound();
+
+ for (Token boundary : boundaries)
+ {
+ PartitionPosition boundaryPosition = boundary.maxKeyBound();
+ ranges[rangeIndex++] = new Range<>(minimum, boundaryPosition);
+ minimum = boundaryPosition;
+ }
+
+ ranges[rangeIndex] = new Range<>(minimum,
partitioner.getMaximumTokenForSplitting().maxKeyBound());
+
+ return ranges;
}
public ShardBoundaries(List<Token> boundaries, Epoch epoch)
@@ -87,6 +124,23 @@ public class ShardBoundaries
return getShardForToken(key.getToken());
}
+ public IntArrayList getShardsForRange(AbstractBounds<PartitionPosition>
keyRange)
+ {
+ if (boundaries.length == 0)
+ return EMPTY_BOUNDARIES_SHARDS;
+
+ // If the keyRange tokens match and are minimum then it represents the
entire token ring
+ // then we need to return all the shards.
+ if (keyRange.right.isMinimum() &&
keyRange.left.compareTo(keyRange.right) == 0)
+ return allShards;
+
+ IntArrayList inRange = new IntArrayList(ranges.length,
IntArrayList.DEFAULT_NULL_VALUE);
+ for (int shard = 0; shard < ranges.length; shard++)
+ if (ranges[shard].intersects(keyRange))
+ inRange.addInt(shard);
+ return inRange;
+ }
+
public Token getShardStartBoundary(int shardId)
{
if (shardId <= 0 || shardId >= boundaries.length)
diff --git a/src/java/org/apache/cassandra/dht/Range.java
b/src/java/org/apache/cassandra/dht/Range.java
index a95249d426..910a3a1539 100644
--- a/src/java/org/apache/cassandra/dht/Range.java
+++ b/src/java/org/apache/cassandra/dht/Range.java
@@ -178,7 +178,21 @@ public class Range<T extends RingPosition<T>> extends
AbstractBounds<T> implemen
return intersects((Range<T>) that);
if (that instanceof Bounds)
return intersects((Bounds<T>) that);
- throw new UnsupportedOperationException("Intersection is only
supported for Bounds and Range objects; found " + that.getClass());
+ if (that instanceof ExcludingBounds)
+ return intersects((ExcludingBounds<T>) that);
+ if (that instanceof IncludingExcludingBounds)
+ return intersects((IncludingExcludingBounds<T>) that);
+ throw new UnsupportedOperationException("Intersection is not supported
for " + that.getClass());
+ }
+
+ public boolean intersects(IncludingExcludingBounds<T> that)
+ {
+ return contains(that.left) || intersects(new Range<>(that.left,
that.right));
+ }
+
+ public boolean intersects(ExcludingBounds<T> that)
+ {
+ return intersects(new Range<>(that.left, that.right));
}
/**
diff --git
a/src/java/org/apache/cassandra/index/accord/RouteMemtableIndexManager.java
b/src/java/org/apache/cassandra/index/accord/RouteMemtableIndexManager.java
index 7c9adb6c3d..ce39300d22 100644
--- a/src/java/org/apache/cassandra/index/accord/RouteMemtableIndexManager.java
+++ b/src/java/org/apache/cassandra/index/accord/RouteMemtableIndexManager.java
@@ -58,17 +58,10 @@ public class RouteMemtableIndexManager implements
MemtableIndexManager
JournalKey journalKey =
AccordKeyspace.JournalColumns.getJournalKey(key);
if (!RouteJournalIndex.allowed(journalKey))
return 0;
- //TODO (performance): we dropped jdk8 and this was fixed in jdk8... so
do we need to do this still?
- MemtableIndex current = liveMemtableIndexMap.get(mt);
- // We expect the relevant IndexMemtable to be present most of the
time, so only make the
- // call to computeIfAbsent() if it's not. (see
https://bugs.openjdk.java.net/browse/JDK-8161372)
- MemtableIndex target = (current != null)
- ? current
- : liveMemtableIndexMap.computeIfAbsent(mt,
memtable -> new MemtableIndex());
+ MemtableIndex target = liveMemtableIndexMap.computeIfAbsent(mt,
memtable -> new MemtableIndex());
long start = nanoTime();
-
long bytes = 0;
ByteBuffer value = RouteIndexFormat.extractParticipants(index,
journalKey.id, row);
diff --git a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndex.java
b/src/java/org/apache/cassandra/index/sai/StorageAttachedIndex.java
index 881ca81911..9f93e0bf95 100644
--- a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndex.java
+++ b/src/java/org/apache/cassandra/index/sai/StorageAttachedIndex.java
@@ -88,6 +88,7 @@ import
org.apache.cassandra.index.sai.disk.format.IndexDescriptor;
import org.apache.cassandra.index.sai.disk.format.Version;
import org.apache.cassandra.index.sai.disk.v1.IndexWriterConfig;
import org.apache.cassandra.index.sai.memory.MemtableIndexManager;
+import org.apache.cassandra.index.sai.memory.ShardedMemtableIndex;
import org.apache.cassandra.index.sai.metrics.ColumnQueryMetrics;
import org.apache.cassandra.index.sai.metrics.IndexMetrics;
import org.apache.cassandra.index.sai.utils.IndexIdentifier;
@@ -117,6 +118,7 @@ import org.apache.cassandra.utils.concurrent.OpOrder;
import io.github.jbellis.jvector.vector.VectorSimilarityFunction;
+import static
org.apache.cassandra.config.CassandraRelevantProperties.MEMTABLE_SHARD_COUNT;
import static
org.apache.cassandra.index.sai.disk.v1.IndexWriterConfig.MAX_TOP_K;
public class StorageAttachedIndex implements Index
@@ -140,6 +142,9 @@ public class StorageAttachedIndex implements Index
@VisibleForTesting
public static final String ANALYSIS_ON_KEY_COLUMNS_MESSAGE = "Analysis
options are not supported on primary key columns, but found ";
+ @VisibleForTesting
+ public static final String VECTOR_SHARDING_SUPPORT_MESSAGE = "A
storage-attached index on a vector column does not support sharding";
+
public static final String ANN_LIMIT_ERROR = "Use of ANN OF in an ORDER BY
clause requires a LIMIT that is not greater than %s. LIMIT was %s";
private static final Logger logger =
LoggerFactory.getLogger(StorageAttachedIndex.class);
@@ -159,7 +164,8 @@ public class StorageAttachedIndex implements Index
IndexWriterConfig.OPTIMIZE_FOR,
NonTokenizingOptions.CASE_SENSITIVE,
NonTokenizingOptions.NORMALIZE,
-
NonTokenizingOptions.ASCII);
+
NonTokenizingOptions.ASCII,
+
ShardedMemtableIndex.SHARDS_OPTION);
public static final Set<CQL3Type> SUPPORTED_TYPES =
ImmutableSet.of(CQL3Type.Native.ASCII, CQL3Type.Native.BIGINT,
CQL3Type.Native.DATE,
CQL3Type.Native.DOUBLE, CQL3Type.Native.FLOAT, CQL3Type.Native.INT,
@@ -169,8 +175,13 @@ public class StorageAttachedIndex implements Index
CQL3Type.Native.VARINT, CQL3Type.Native.DECIMAL, CQL3Type.Native.BOOLEAN,
CQL3Type.Native.BLOB);
+ // Upper bound on the shards option to prevent runaway per-shard
allocations at first-write.
+ public static final int MAX_SHARD_COUNT = 256;
+
private static final Set<Class<? extends IPartitioner>>
ILLEGAL_PARTITIONERS =
ImmutableSet.of(OrderPreservingPartitioner.class,
LocalPartitioner.class, ByteOrderedPartitioner.class, RandomPartitioner.class);
+ private static final int DEFAULT_SHARD_COUNT =
MEMTABLE_SHARD_COUNT.getInt(FBUtilities.getAvailableProcessors());
+ private static final String AUTO_SHARDS_OPTION = "auto";
private final ColumnFamilyStore baseCfs;
private final IndexMetadata indexMetadata;
@@ -184,6 +195,7 @@ public class StorageAttachedIndex implements Index
private final MemtableIndexManager memtableIndexManager;
private final IndexMetrics indexMetrics;
private final MaxThreshold maxTermSizeGuardrail;
+ private final int shardCount;
// Tracks whether we've started the index build on initialization.
private volatile boolean initBuildStarted = false;
@@ -207,6 +219,7 @@ public class StorageAttachedIndex implements Index
analyzerFactory = AbstractAnalyzer.fromOptions(indexTermType,
indexMetadata.options);
memtableIndexManager = new MemtableIndexManager(this);
indexMetrics = new IndexMetrics(this, memtableIndexManager);
+
if (indexTermType.isVector())
maxTermSizeGuardrail = Guardrails.saiVectorTermSize;
else if (indexTermType.isFrozen())
@@ -215,6 +228,14 @@ public class StorageAttachedIndex implements Index
maxTermSizeGuardrail = Guardrails.saiBlobTermSize;
else
maxTermSizeGuardrail = Guardrails.saiStringTermSize;
+
+ String shardsOption =
indexMetadata.options.get(ShardedMemtableIndex.SHARDS_OPTION);
+ if (shardsOption == null)
+ shardCount = 1;
+ else if (shardsOption.equalsIgnoreCase(AUTO_SHARDS_OPTION))
+ shardCount = DEFAULT_SHARD_COUNT;
+ else
+ shardCount = Integer.parseInt(shardsOption);
}
/**
@@ -281,6 +302,28 @@ public class StorageAttachedIndex implements Index
}
IndexTermType indexTermType = IndexTermType.create(target.left,
metadata.partitionKeyColumns(), target.right);
+ String shardsOption = options.get(ShardedMemtableIndex.SHARDS_OPTION);
+ if (shardsOption != null)
+ {
+ if (indexTermType.isVector())
+ throw new
InvalidRequestException(VECTOR_SHARDING_SUPPORT_MESSAGE);
+
+ if (!shardsOption.equalsIgnoreCase(AUTO_SHARDS_OPTION))
+ {
+ try
+ {
+ int shardCount = Integer.parseInt(shardsOption);
+ if (shardCount <= 0)
+ throw new InvalidRequestException("Shard count for a
storage-attached index must be a positive integer, was " + shardCount);
+ if (shardCount > MAX_SHARD_COUNT)
+ throw new InvalidRequestException("Shard count for a
storage-attached index must not exceed " + MAX_SHARD_COUNT + ", was " +
shardCount);
+ }
+ catch (NumberFormatException e)
+ {
+ throw new InvalidRequestException("Shard count for a
storage-attached index must be a valid integer, got '" + shardsOption + "'");
+ }
+ }
+ }
AbstractAnalyzer.fromOptions(indexTermType, analysisOptions);
IndexWriterConfig config = IndexWriterConfig.fromOptions(null,
indexTermType, options);
@@ -334,6 +377,11 @@ public class StorageAttachedIndex implements Index
return indexMetadata;
}
+ public int shardCount()
+ {
+ return shardCount;
+ }
+
@Override
public Callable<?> getInitializationTask()
{
diff --git a/src/java/org/apache/cassandra/index/sai/disk/RowMapping.java
b/src/java/org/apache/cassandra/index/sai/disk/RowMapping.java
index 5bf64360af..19cedd0b93 100644
--- a/src/java/org/apache/cassandra/index/sai/disk/RowMapping.java
+++ b/src/java/org/apache/cassandra/index/sai/disk/RowMapping.java
@@ -52,7 +52,7 @@ public class RowMapping
public static final RowMapping DUMMY = new RowMapping()
{
@Override
- public Iterator<Pair<ByteComparable, LongArrayList>>
merge(MemtableIndex index) { return Collections.emptyIterator(); }
+ public Iterator<Pair<ByteComparable, LongArrayList>>
merge(MemtableIndex index, PrimaryKey minKey, PrimaryKey maxKey) { return
Collections.emptyIterator(); }
@Override
public void complete() {}
@@ -99,11 +99,13 @@ public class RowMapping
*
* @return an iterator of term -> postings list {@link Pair}s
*/
- public Iterator<Pair<ByteComparable, LongArrayList>> merge(MemtableIndex
index)
+ public Iterator<Pair<ByteComparable, LongArrayList>> merge(MemtableIndex
index,
+ PrimaryKey
minKey,
+ PrimaryKey
maxKey)
{
assert complete : "RowMapping is not built.";
- Iterator<Pair<ByteComparable, PrimaryKeys>> iterator =
index.iterator();
+ Iterator<Pair<ByteComparable, Iterator<PrimaryKey>>> iterator =
index.iterator(minKey.partitionKey(), maxKey.partitionKey());
return new AbstractGuavaIterator<>()
{
@Override
@@ -111,10 +113,10 @@ public class RowMapping
{
while (iterator.hasNext())
{
- Pair<ByteComparable, PrimaryKeys> pair = iterator.next();
+ Pair<ByteComparable, Iterator<PrimaryKey>> pair =
iterator.next();
LongArrayList postings = null;
- Iterator<PrimaryKey> primaryKeys = pair.right.iterator();
+ Iterator<PrimaryKey> primaryKeys = pair.right;
while (primaryKeys.hasNext())
{
diff --git
a/src/java/org/apache/cassandra/index/sai/disk/v1/MemtableIndexWriter.java
b/src/java/org/apache/cassandra/index/sai/disk/v1/MemtableIndexWriter.java
index 795d5c1a0d..e8201c800f 100644
--- a/src/java/org/apache/cassandra/index/sai/disk/v1/MemtableIndexWriter.java
+++ b/src/java/org/apache/cassandra/index/sai/disk/v1/MemtableIndexWriter.java
@@ -145,7 +145,7 @@ public class MemtableIndexWriter implements
PerColumnIndexWriter
}
else
{
- final Iterator<Pair<ByteComparable, LongArrayList>> iterator =
rowMapping.merge(memtable);
+ final Iterator<Pair<ByteComparable, LongArrayList>> iterator =
rowMapping.merge(memtable, minKey, maxKey);
long cellCount = 0;
if (iterator.hasNext())
diff --git a/src/java/org/apache/cassandra/index/sai/memory/MemtableIndex.java
b/src/java/org/apache/cassandra/index/sai/memory/MemtableIndex.java
index ab24fff087..cd4cb76da1 100644
--- a/src/java/org/apache/cassandra/index/sai/memory/MemtableIndex.java
+++ b/src/java/org/apache/cassandra/index/sai/memory/MemtableIndex.java
@@ -22,7 +22,6 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.List;
-import java.util.concurrent.atomic.LongAdder;
import java.util.function.Function;
import org.apache.cassandra.db.Clustering;
@@ -31,7 +30,6 @@ import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.db.memtable.Memtable;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.index.sai.QueryContext;
-import org.apache.cassandra.index.sai.StorageAttachedIndex;
import org.apache.cassandra.index.sai.disk.format.IndexDescriptor;
import org.apache.cassandra.index.sai.disk.v1.segment.SegmentMetadata;
import org.apache.cassandra.index.sai.disk.v1.vector.PrimaryKeyWithScore;
@@ -39,96 +37,63 @@ import
org.apache.cassandra.index.sai.iterators.KeyRangeIterator;
import org.apache.cassandra.index.sai.plan.Expression;
import org.apache.cassandra.index.sai.utils.IndexIdentifier;
import org.apache.cassandra.index.sai.utils.PrimaryKey;
-import org.apache.cassandra.index.sai.utils.PrimaryKeys;
import org.apache.cassandra.utils.CloseableIterator;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.bytecomparable.ByteComparable;
-public class MemtableIndex implements MemtableOrdering
+public interface MemtableIndex extends MemtableOrdering
{
- private final MemoryIndex memoryIndex;
- private final LongAdder writeCount = new LongAdder();
- private final LongAdder estimatedMemoryUsed = new LongAdder();
- private final Memtable memtable;
+ long writeCount();
- public MemtableIndex(StorageAttachedIndex index, Memtable memtable)
- {
- this.memoryIndex = index.termType().isVector() ? new
VectorMemoryIndex(index, memtable) : new TrieMemoryIndex(index);
- this.memtable = memtable;
- }
+ long estimatedMemoryUsed();
- public long writeCount()
- {
- return writeCount.sum();
- }
+ boolean isEmpty();
- public long estimatedMemoryUsed()
- {
- return estimatedMemoryUsed.sum();
- }
+ Memtable getMemtable();
- public boolean isEmpty()
- {
- return memoryIndex.isEmpty();
- }
-
- public Memtable getMemtable()
- {
- return memtable;
- }
+ ByteBuffer getMinTerm();
- public ByteBuffer getMinTerm()
- {
- return memoryIndex.getMinTerm();
- }
+ ByteBuffer getMaxTerm();
- public ByteBuffer getMaxTerm()
- {
- return memoryIndex.getMaxTerm();
- }
-
- public long index(DecoratedKey key, Clustering<?> clustering, ByteBuffer
value)
- {
- if (value == null || (value.remaining() == 0 &&
memoryIndex.index.termType().skipsEmptyValue()))
- return 0;
-
- long ram = memoryIndex.add(key, clustering, value);
- writeCount.increment();
- estimatedMemoryUsed.add(ram);
- return ram;
- }
-
- public long update(DecoratedKey key, Clustering<?> clustering, ByteBuffer
oldValue, ByteBuffer newValue)
- {
- return memoryIndex.update(key, clustering, oldValue, newValue);
- }
-
- public KeyRangeIterator search(QueryContext queryContext, Expression
expression, AbstractBounds<PartitionPosition> keyRange)
- {
- return memoryIndex.search(queryContext, expression, keyRange);
- }
+ long index(DecoratedKey key, Clustering<?> clustering, ByteBuffer value);
- public Iterator<Pair<ByteComparable, PrimaryKeys>> iterator()
+ /** Implementation only for Vector Indexes */
+ default long update(DecoratedKey key, Clustering<?> clustering, ByteBuffer
oldValue, ByteBuffer newValue)
{
- return memoryIndex.iterator();
+ throw new UnsupportedOperationException();
}
- public SegmentMetadata.ComponentMetadataMap writeDirect(IndexDescriptor
indexDescriptor,
- IndexIdentifier
indexIdentifier,
-
Function<PrimaryKey, Integer> postingTransformer) throws IOException
+ KeyRangeIterator search(QueryContext queryContext, Expression expression,
AbstractBounds<PartitionPosition> keyRange);
+
+ /**
+ * Returns an iterator over the in-memory index in ascending term order.
+ * <p>
+ * Returned data may contain keys outside [minKey, maxKey]. The bounds
+ * are used for shard selection only; implementations without sharding may
+ * ignore them.
+ * <p>
+ * Callers should invoke this method only when the source Memtable is no
longer accepting
+ * writes. Otherwise, concurrent inserts may be observed partway through
iteration.
+ */
+ Iterator<Pair<ByteComparable, Iterator<PrimaryKey>>> iterator(DecoratedKey
min, DecoratedKey max);
+
+ /** Implementation only for Vector Indexes */
+ default SegmentMetadata.ComponentMetadataMap writeDirect(IndexDescriptor
indexDescriptor,
+ IndexIdentifier
indexIdentifier,
+
Function<PrimaryKey, Integer> postingTransformer) throws IOException
{
- return memoryIndex.writeDirect(indexDescriptor, indexIdentifier,
postingTransformer);
+ throw new UnsupportedOperationException();
}
@Override
- public CloseableIterator<PrimaryKeyWithScore> orderBy(QueryContext
queryContext, Expression orderer, AbstractBounds<PartitionPosition> keyRange)
+ default CloseableIterator<PrimaryKeyWithScore> orderBy(QueryContext
queryContext, Expression orderer, AbstractBounds<PartitionPosition> keyRange)
{
- return memoryIndex.orderBy(queryContext, orderer, keyRange);
+ throw new UnsupportedOperationException();
}
@Override
- public CloseableIterator<PrimaryKeyWithScore> orderResultsBy(QueryContext
queryContext, List<PrimaryKey> results, Expression orderer)
+ default CloseableIterator<PrimaryKeyWithScore> orderResultsBy(QueryContext
queryContext, List<PrimaryKey> results, Expression orderer)
{
- return memoryIndex.orderResultsBy(queryContext, results, orderer);
+ throw new UnsupportedOperationException();
}
}
diff --git
a/src/java/org/apache/cassandra/index/sai/memory/MemtableIndexManager.java
b/src/java/org/apache/cassandra/index/sai/memory/MemtableIndexManager.java
index c2f7087fc9..e6e9342ac2 100644
--- a/src/java/org/apache/cassandra/index/sai/memory/MemtableIndexManager.java
+++ b/src/java/org/apache/cassandra/index/sai/memory/MemtableIndexManager.java
@@ -35,6 +35,7 @@ import
org.apache.cassandra.cql3.statements.schema.IndexTarget;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.lifecycle.ILifecycleTransaction;
import org.apache.cassandra.db.memtable.Memtable;
+import org.apache.cassandra.db.memtable.ShardBoundaries;
import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.index.sai.StorageAttachedIndex;
import org.apache.cassandra.utils.Clock;
@@ -59,12 +60,15 @@ public class MemtableIndexManager
private MemtableIndex initializeMemtableIndex(Memtable mt)
{
- MemtableIndex current = liveMemtableIndexMap.get(mt);
-
- // We expect the relevant IndexMemtable to be present most of the
time, so only make the
- // call to computeIfAbsent() if it's not. (see
https://bugs.openjdk.java.net/browse/JDK-8161372)
- return current != null ? current
- : liveMemtableIndexMap.computeIfAbsent(mt,
memtable -> new MemtableIndex(index, memtable));
+ return liveMemtableIndexMap.computeIfAbsent(mt, memtable -> {
+ int shardCount = index.shardCount();
+ if (shardCount > 1)
+ {
+ ShardBoundaries boundaries =
index.baseCfs().localRangeSplits(shardCount);
+ return new ShardedMemtableIndex(index, boundaries, memtable);
+ }
+ return new UnshardedMemtableIndex(index, memtable);
+ });
}
public long index(DecoratedKey key, Row row, Memtable mt)
diff --git
a/src/java/org/apache/cassandra/index/sai/memory/ShardedMemtableIndex.java
b/src/java/org/apache/cassandra/index/sai/memory/ShardedMemtableIndex.java
new file mode 100644
index 0000000000..48bd107b23
--- /dev/null
+++ b/src/java/org/apache/cassandra/index/sai/memory/ShardedMemtableIndex.java
@@ -0,0 +1,272 @@
+/*
+ * 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.index.sai.memory;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.atomic.LongAdder;
+
+import javax.annotation.Nullable;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Iterators;
+
+import org.agrona.collections.IntArrayList;
+
+import org.apache.cassandra.db.Clustering;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.PartitionPosition;
+import org.apache.cassandra.db.memtable.Memtable;
+import org.apache.cassandra.db.memtable.ShardBoundaries;
+import org.apache.cassandra.dht.AbstractBounds;
+import org.apache.cassandra.index.sai.QueryContext;
+import org.apache.cassandra.index.sai.StorageAttachedIndex;
+import org.apache.cassandra.index.sai.iterators.KeyRangeConcatIterator;
+import org.apache.cassandra.index.sai.iterators.KeyRangeIterator;
+import org.apache.cassandra.index.sai.plan.Expression;
+import org.apache.cassandra.index.sai.utils.PrimaryKey;
+import org.apache.cassandra.index.sai.utils.PrimaryKeys;
+import org.apache.cassandra.utils.MergeIterator;
+import org.apache.cassandra.utils.Pair;
+import org.apache.cassandra.utils.bytecomparable.ByteComparable;
+
+public class ShardedMemtableIndex implements MemtableIndex
+{
+ private final ShardBoundaries boundaries;
+ private final MemoryIndex[] shards;
+ private final StorageAttachedIndex index;
+ private final LongAdder writeCount = new LongAdder();
+ private final LongAdder estimatedMemoryUsed = new LongAdder();
+ private final Memtable memtable;
+
+ public static final String SHARDS_OPTION = "shards";
+
+ public ShardedMemtableIndex(StorageAttachedIndex index, ShardBoundaries
boundaries, Memtable memtable)
+ {
+ this.index = index;
+ this.boundaries = boundaries;
+ this.shards = generateShards(boundaries.shardCount(), index);
+ this.memtable = memtable;
+ }
+
+ private MemoryIndex[] generateShards(int splits, StorageAttachedIndex
index)
+ {
+ MemoryIndex[] generatedShards = new MemoryIndex[splits];
+
+ for (int shard = 0; shard < boundaries.shardCount(); shard++)
+ {
+ generatedShards[shard] = new TrieMemoryIndex(index);
+ }
+
+ return generatedShards;
+ }
+
+ @VisibleForTesting
+ public int shardCount()
+ {
+ return shards.length;
+ }
+
+ @Override
+ public long writeCount()
+ {
+ return writeCount.sum();
+ }
+
+ @Override
+ public boolean isEmpty()
+ {
+ return getMinTerm() == null;
+ }
+
+ @Override
+ public Memtable getMemtable()
+ {
+ return memtable;
+ }
+
+ @Override
+ public long estimatedMemoryUsed()
+ {
+ return estimatedMemoryUsed.sum();
+ }
+
+ /**
+ * Returns the minimum indexed term in the combined memory indexes.
+ * This can be {@code null} if the indexed memtable was empty. Users of the
+ * {@code MemtableIndex} requiring a non-null minimum term should
+ * use the {@link MemtableIndex#isEmpty} method.
+ *
+ * <p>
+ * <b>Note:</b> Individual index shards can return {@code null} here if
the index
+ * didn't receive any terms within the token range of the shard
+ * </p>
+ *
+ * @return the minimum indexed term across all shards, or {@code null} if
the index is empty.
+ */
+ @Override
+ @Nullable
+ public ByteBuffer getMinTerm()
+ {
+ ByteBuffer result = null;
+ for (MemoryIndex shard : shards)
+ result = index.termType().min(shard.getMinTerm(), result);
+ return result;
+ }
+
+ /**
+ * Returns the maximum indexed term in the combined memory indexes.
+ * This can be {@code null} if the indexed memtable was empty. Users of
the
+ * {@code MemtableIndex} requiring a non-null maximum term should
+ * use the {@link MemtableIndex#isEmpty} method.
+ *
+ * <p>
+ * <b>Note:</b> Individual index shards can return {@code null} here if
the index
+ * didn't receive any terms within the token range of the shard
+ * </p>
+ *
+ * @return the maximum indexed term across all shards, or {@code null} if
the index is empty
+ */
+ @Override
+ @Nullable
+ public ByteBuffer getMaxTerm()
+ {
+ ByteBuffer result = null;
+ for (MemoryIndex shard : shards)
+ result = index.termType().max(shard.getMaxTerm(), result);
+ return result;
+ }
+
+ @Override
+ public long index(DecoratedKey key, Clustering<?> clustering, ByteBuffer
value)
+ {
+ if (value == null || (value.remaining() == 0 &&
index.termType().skipsEmptyValue()))
+ return 0;
+
+ long ram = shards[boundaries.getShardForKey(key)].add(key, clustering,
value);
+ writeCount.increment();
+ estimatedMemoryUsed.add(ram);
+ return ram;
+ }
+
+ @Override
+ public KeyRangeIterator search(QueryContext queryContext, Expression
expression, AbstractBounds<PartitionPosition> keyRange)
+ {
+ IntArrayList shardsForRange = boundaries.getShardsForRange(keyRange);
+ KeyRangeConcatIterator.Builder builder =
KeyRangeConcatIterator.builder(shardsForRange.size());
+
+ // InMemoryKeyRangeIterator holds no closeable resources; so a builder
cleanup is not needed on exceptions.
+ for (int i = 0; i < shardsForRange.size(); i++)
+ {
+ int shard = shardsForRange.getInt(i);
+ builder.add(shards[shard].search(queryContext, expression,
keyRange));
+ }
+
+ return builder.build();
+ }
+
+ @Override
+ public Iterator<Pair<ByteComparable, Iterator<PrimaryKey>>>
iterator(DecoratedKey min, DecoratedKey max)
+ {
+ int minSubrange = min == null ? 0 : boundaries.getShardForKey(min);
+ int maxSubrange = max == null ? shards.length - 1 :
boundaries.getShardForKey(max);
+
+ Preconditions.checkArgument(minSubrange <= maxSubrange,
+ "iterator(min, max) requires min <= max
but got min shard %s > max shard %s",
+ minSubrange, maxSubrange);
+
+ List<Iterator<Pair<ByteComparable, PrimaryKeys>>> shardIterators = new
ArrayList<>(maxSubrange - minSubrange + 1);
+
+ for (int i = minSubrange; i <= maxSubrange; i++)
+ shardIterators.add(shards[i].iterator());
+
+ return MergeIterator.get(shardIterators,
+ (o1, o2) -> ByteComparable.compare(o1.left,
o2.left, ByteComparable.Version.OSS50),
+ new
PrimaryKeysMergeReducer(shardIterators.size()));
+ }
+
+ /**
+ * The PrimaryKeysMergeReducer receives the range iterators from each of
the shards selected based on the
+ * min and max keys passed to the iterator method. It doesn't strictly do
any reduction because the terms in each
+ * shard are unique. It will receive at most one shard entry per selected
shard before {@link #getReduced}
+ * is called.
+ */
+ private static class PrimaryKeysMergeReducer extends
MergeIterator.Reducer<Pair<ByteComparable, PrimaryKeys>, Pair<ByteComparable,
Iterator<PrimaryKey>>>
+ {
+ private final Pair<ByteComparable, PrimaryKeys>[] perShardTermToKeys;
+ private final List<Iterator<PrimaryKey>> currentTermKeys;
+
+ private ByteComparable currentTerm;
+ private Iterator<PrimaryKey> lastReturnedIterator;
+
+ @SuppressWarnings("unchecked")
+ PrimaryKeysMergeReducer(int shards)
+ {
+ this.perShardTermToKeys = new Pair[shards];
+ this.currentTermKeys = new ArrayList<>(shards);
+ }
+
+ /**
+ * Receive the term entry for a shard. This should only be called once
for each
+ * shard before reduction.
+ *
+ * @param idx the index of the shard contributing this entry
+ * @param current the term and its associated primary keys from the
shard
+ */
+ @Override
+ public void reduce(int idx, Pair<ByteComparable, PrimaryKeys> current)
+ {
+ Preconditions.checkArgument(perShardTermToKeys[idx] == null,
"Terms should be unique in the memory index");
+
+ perShardTermToKeys[idx] = current;
+
+ if (current != null && currentTerm == null)
+ currentTerm = current.left;
+ }
+
+ @Override
+ protected Pair<ByteComparable, Iterator<PrimaryKey>> getReduced()
+ {
+ Preconditions.checkArgument(currentTerm != null, "The term must
exist in memory index");
+
+ for (Pair<ByteComparable, PrimaryKeys> p : perShardTermToKeys)
+ if (p != null && p.right != null && !p.right.isEmpty())
+ currentTermKeys.add(p.right.iterator());
+
+ lastReturnedIterator =
Iterators.concat(currentTermKeys.iterator());
+ return Pair.create(currentTerm, lastReturnedIterator);
+ }
+
+ @Override
+ protected void onKeyChange()
+ {
+ Preconditions.checkState(lastReturnedIterator == null ||
!lastReturnedIterator.hasNext(),
+ "Previous per-term Iterator<PrimaryKey>
was not drained before the outer iterator advanced");
+
+ Arrays.fill(perShardTermToKeys, null);
+ currentTermKeys.clear();
+ currentTerm = null;
+ lastReturnedIterator = null;
+ }
+ }
+}
diff --git a/src/java/org/apache/cassandra/index/sai/memory/MemtableIndex.java
b/src/java/org/apache/cassandra/index/sai/memory/UnshardedMemtableIndex.java
similarity index 81%
copy from src/java/org/apache/cassandra/index/sai/memory/MemtableIndex.java
copy to
src/java/org/apache/cassandra/index/sai/memory/UnshardedMemtableIndex.java
index ab24fff087..c40ab93125 100644
--- a/src/java/org/apache/cassandra/index/sai/memory/MemtableIndex.java
+++ b/src/java/org/apache/cassandra/index/sai/memory/UnshardedMemtableIndex.java
@@ -44,49 +44,56 @@ import org.apache.cassandra.utils.CloseableIterator;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.bytecomparable.ByteComparable;
-public class MemtableIndex implements MemtableOrdering
+public class UnshardedMemtableIndex implements MemtableIndex
{
private final MemoryIndex memoryIndex;
private final LongAdder writeCount = new LongAdder();
private final LongAdder estimatedMemoryUsed = new LongAdder();
private final Memtable memtable;
- public MemtableIndex(StorageAttachedIndex index, Memtable memtable)
+ public UnshardedMemtableIndex(StorageAttachedIndex index, Memtable
memtable)
{
this.memoryIndex = index.termType().isVector() ? new
VectorMemoryIndex(index, memtable) : new TrieMemoryIndex(index);
this.memtable = memtable;
}
+ @Override
public long writeCount()
{
return writeCount.sum();
}
+ @Override
public long estimatedMemoryUsed()
{
return estimatedMemoryUsed.sum();
}
+ @Override
public boolean isEmpty()
{
return memoryIndex.isEmpty();
}
+ @Override
public Memtable getMemtable()
{
return memtable;
}
+ @Override
public ByteBuffer getMinTerm()
{
return memoryIndex.getMinTerm();
}
+ @Override
public ByteBuffer getMaxTerm()
{
return memoryIndex.getMaxTerm();
}
+ @Override
public long index(DecoratedKey key, Clustering<?> clustering, ByteBuffer
value)
{
if (value == null || (value.remaining() == 0 &&
memoryIndex.index.termType().skipsEmptyValue()))
@@ -98,21 +105,43 @@ public class MemtableIndex implements MemtableOrdering
return ram;
}
+ /** Used only for Vector Indexes */
+ @Override
public long update(DecoratedKey key, Clustering<?> clustering, ByteBuffer
oldValue, ByteBuffer newValue)
{
return memoryIndex.update(key, clustering, oldValue, newValue);
}
+ @Override
public KeyRangeIterator search(QueryContext queryContext, Expression
expression, AbstractBounds<PartitionPosition> keyRange)
{
return memoryIndex.search(queryContext, expression, keyRange);
}
- public Iterator<Pair<ByteComparable, PrimaryKeys>> iterator()
+ @Override
+ public Iterator<Pair<ByteComparable, Iterator<PrimaryKey>>>
iterator(DecoratedKey min, DecoratedKey max)
{
- return memoryIndex.iterator();
+ Iterator<Pair<ByteComparable, PrimaryKeys>> memoryIndexIterator =
memoryIndex.iterator();
+
+ return new Iterator<>()
+ {
+ @Override
+ public boolean hasNext()
+ {
+ return memoryIndexIterator.hasNext();
+ }
+
+ @Override
+ public Pair<ByteComparable, Iterator<PrimaryKey>> next()
+ {
+ Pair<ByteComparable, PrimaryKeys> p =
memoryIndexIterator.next();
+ return Pair.create(p.left, p.right.iterator());
+ }
+ };
}
+ /** Used only for Vector Indexes */
+ @Override
public SegmentMetadata.ComponentMetadataMap writeDirect(IndexDescriptor
indexDescriptor,
IndexIdentifier
indexIdentifier,
Function<PrimaryKey, Integer> postingTransformer) throws IOException
@@ -120,6 +149,7 @@ public class MemtableIndex implements MemtableOrdering
return memoryIndex.writeDirect(indexDescriptor, indexIdentifier,
postingTransformer);
}
+ /** Used only for Vector Indexes */
@Override
public CloseableIterator<PrimaryKeyWithScore> orderBy(QueryContext
queryContext, Expression orderer, AbstractBounds<PartitionPosition> keyRange)
{
diff --git
a/test/distributed/org/apache/cassandra/distributed/test/cql3/SingleNodeTableWalkTest.java
b/test/distributed/org/apache/cassandra/distributed/test/cql3/SingleNodeTableWalkTest.java
index adba049f90..7eb72ebce9 100644
---
a/test/distributed/org/apache/cassandra/distributed/test/cql3/SingleNodeTableWalkTest.java
+++
b/test/distributed/org/apache/cassandra/distributed/test/cql3/SingleNodeTableWalkTest.java
@@ -25,6 +25,7 @@ import java.util.Collections;
import java.util.EnumSet;
import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
import java.util.NavigableSet;
import java.util.Optional;
import java.util.TreeMap;
@@ -95,6 +96,14 @@ public class SingleNodeTableWalkTest extends StatefulASTBase
.stream()
.filter(t ->
!AbstractTypeGenerators.isUnsafeEquality(t))
.collect(Collectors.toList()));
+ private static final String SHARDS = "shards";
+ @SuppressWarnings("unchecked")
+ private static final Map<String, String>[] SHARD_OPTIONS = new Map[] {
+ Collections.emptyMap(),
+ Collections.singletonMap(SHARDS, "1"),
+ Collections.singletonMap(SHARDS, "4"),
+ Collections.singletonMap(SHARDS, "auto")
+ };
private static final Logger logger =
LoggerFactory.getLogger(SingleNodeTableWalkTest.class);
protected static boolean READ_AFTER_WRITE = false;
@@ -695,12 +704,18 @@ public class SingleNodeTableWalkTest extends
StatefulASTBase
colExpression = new
CreateIndexDDL.CollectionReference(CreateIndexDDL.CollectionReference.Kind.FULL,
colExpression);
String name = "tbl_" + col.name;
+
+ Map<String, String> indexOptions = Collections.emptyMap();
+ // If indexer is of Kind SAI, we want to randomize shards for
testing the ShardedMemtableIndex implementation.
+ if (indexer.kind() == CreateIndexDDL.Indexer.Kind.sai )
+ indexOptions = rs.pick(SHARD_OPTIONS);
+
CreateIndexDDL ddl = new
CreateIndexDDL(rs.pick(CreateIndexDDL.Version.values()),
indexer,
Optional.of(new
Symbol(name, UTF8Type.instance)),
TableReference.from(metadata),
Collections.singletonList(colExpression),
-
Collections.emptyMap());
+ indexOptions);
String stmt = ddl.toCQL();
logger.info(stmt);
cluster.schemaChange(stmt);
diff --git
a/test/microbench/org/apache/cassandra/test/microbench/sai/AbstractMemtableIndexBench.java
b/test/microbench/org/apache/cassandra/test/microbench/sai/AbstractMemtableIndexBench.java
new file mode 100644
index 0000000000..4632ea6eab
--- /dev/null
+++
b/test/microbench/org/apache/cassandra/test/microbench/sai/AbstractMemtableIndexBench.java
@@ -0,0 +1,118 @@
+/*
+ * 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.sai;
+
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Random;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.cql3.CQLTester;
+import org.apache.cassandra.cql3.statements.schema.IndexTarget;
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.marshal.UTF8Type;
+import org.apache.cassandra.index.sai.StorageAttachedIndex;
+import org.apache.cassandra.index.sai.memory.MemtableIndex;
+import org.apache.cassandra.schema.IndexMetadata;
+import org.apache.cassandra.schema.TableMetadata;
+
+public abstract class AbstractMemtableIndexBench extends CQLTester
+{
+ private static final int RANDOM_STRING_SIZE = 64 * 1024 * 1024;
+ private static String keyspace;
+
+ protected MemtableIndex memtableIndex;
+ protected DecoratedKey[] partitionKeys;
+ protected ByteBuffer[] terms;
+ protected StorageAttachedIndex index;
+
+ protected ColumnFamilyStore cfs;
+ protected String table;
+ private char[] randomChars = new char[RANDOM_STRING_SIZE];
+
+ public void setup(int numberOfTerms, int rowsPerPartition)
+ {
+ setupServer();
+ setupTableAndKeyspace();
+ setupCfsAndIndex();
+ setupPartitionKeys(numberOfTerms, rowsPerPartition);
+ setupTerms(numberOfTerms);
+ }
+
+ public void setupServer()
+ {
+ CQLTester.setUpClass();
+ DatabaseDescriptor.setAutoSnapshot(false);
+ }
+
+ public void setupTableAndKeyspace()
+ {
+ keyspace = createKeyspace("CREATE KEYSPACE %s with replication = {
'class' : 'SimpleStrategy', 'replication_factor' : 1 } and durable_writes =
false");
+ table = createTable(keyspace,
+ "CREATE TABLE %s ( partition_id text, value_text
text, PRIMARY KEY(partition_id)) with compression = {'enabled': false}",
+ "memtable_index");
+ execute("use " + keyspace + ";");
+ }
+
+ public void setupCfsAndIndex()
+ {
+ cfs = Keyspace.open(keyspace).getColumnFamilyStore(table);
+ cfs.disableAutoCompaction();
+
+ Map<String, String> options = new HashMap<>();
+ options.put(IndexTarget.CUSTOM_INDEX_OPTION_NAME,
+ StorageAttachedIndex.class.getCanonicalName());
+ options.put("target", "value_text");
+
+ IndexMetadata indexMetadata =
IndexMetadata.fromSchemaMetadata("value_text_idx", IndexMetadata.Kind.CUSTOM,
options);
+
+ index = new StorageAttachedIndex(cfs, indexMetadata);
+ }
+
+ public void setupPartitionKeys(int numberOfTerms, int rowsPerPartition)
+ {
+ TableMetadata tableMetadata = cfs.metadata();
+
+ int numberOfKeys = numberOfTerms / rowsPerPartition;
+
+ partitionKeys = new DecoratedKey[numberOfKeys];
+ for (int i = 0; i < numberOfKeys; i++)
+ partitionKeys[i] =
tableMetadata.partitioner.decorateKey(tableMetadata.partitionKeyType.fromString("partition_"
+ i));
+ }
+
+ public void setupTerms(int numberOfTerms)
+ {
+ Random random = new Random();
+ for (int i = 0; i < RANDOM_STRING_SIZE; i++)
+ randomChars[i] = (char)('a' + random.nextInt(26));
+
+ int length = 64;
+ terms = new ByteBuffer[numberOfTerms];
+ for (int i = 0; i < numberOfTerms; i++)
+ terms[i] =
UTF8Type.instance.decompose(generateRandomString(random, length));
+ }
+
+ private String generateRandomString(Random random, int length)
+ {
+ return new String(randomChars, random.nextInt(RANDOM_STRING_SIZE -
length), length);
+ }
+}
diff --git
a/test/microbench/org/apache/cassandra/test/microbench/sai/MemtableIndexFlushBench.java
b/test/microbench/org/apache/cassandra/test/microbench/sai/MemtableIndexFlushBench.java
new file mode 100644
index 0000000000..aa57522ae1
--- /dev/null
+++
b/test/microbench/org/apache/cassandra/test/microbench/sai/MemtableIndexFlushBench.java
@@ -0,0 +1,150 @@
+/*
+ * 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.sai;
+
+import java.nio.ByteBuffer;
+import java.util.Iterator;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+
+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.Warmup;
+
+import org.apache.cassandra.db.Clustering;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.marshal.UTF8Type;
+import org.apache.cassandra.db.memtable.Memtable;
+import org.apache.cassandra.db.memtable.ShardBoundaries;
+import org.apache.cassandra.index.sai.memory.ShardedMemtableIndex;
+import org.apache.cassandra.index.sai.memory.UnshardedMemtableIndex;
+import org.apache.cassandra.index.sai.utils.PrimaryKey;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.utils.Pair;
+import org.apache.cassandra.utils.bytecomparable.ByteComparable;
+
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@Warmup(iterations = 3, time = 2)
+@Measurement(iterations = 5, time = 5)
+@Fork(value = 1, jvmArgsAppend = { "-Xmx4G", "-Xms4G",
"-Djmh.executor=CUSTOM",
"-Djmh.executor.class=org.apache.cassandra.test.microbench.FastThreadExecutor"})
+@State(Scope.Benchmark)
+public class MemtableIndexFlushBench extends AbstractMemtableIndexBench
+{
+ // 1 Million partitionKeys
+ private static final int NUM_PARTITION_KEYS = 1000000;
+ private static final int SMALL_POOL_SIZE = 8 * 1024;
+
+ @Param({ "1", "4", "8"})
+ int shardCount;
+
+ @Param({"50", "100"})
+ int numberOfTerms;
+
+ private char[] smallPool = new char[SMALL_POOL_SIZE];
+
+ @Setup(Level.Trial)
+ public void setup()
+ {
+ setupServer();
+ setupTableAndKeyspace();
+ setupCfsAndIndex();
+ setupPartitionKeys();
+ setupIndexesExpressionsAndTerms();
+ }
+
+ public void setupPartitionKeys()
+ {
+ TableMetadata tableMetadata = cfs.metadata();
+
+ partitionKeys = new DecoratedKey[NUM_PARTITION_KEYS];
+ for (int i = 0; i < NUM_PARTITION_KEYS; i++)
+ partitionKeys[i] =
tableMetadata.partitioner.decorateKey(tableMetadata.partitionKeyType.fromString("partition_"
+ i));
+ }
+
+ public void setupIndexesExpressionsAndTerms() {
+ Memtable memtable = cfs.getCurrentMemtable();
+ if (shardCount > 1)
+ {
+ ShardBoundaries boundaries = cfs.localRangeSplits(shardCount);
+ memtableIndex = new ShardedMemtableIndex(index, boundaries,
memtable);
+ }
+ else
+ memtableIndex = new UnshardedMemtableIndex(index, memtable);
+
+ setupTerms(numberOfTerms);
+ populateIndexData();
+ }
+
+ @Override
+ public void setupTerms(int numberOfTerms)
+ {
+ Random random = new Random();
+ for (int i = 0; i < SMALL_POOL_SIZE; i++)
+ smallPool[i] = (char)('a' + random.nextInt(26));
+
+ int length = 64;
+ terms = new ByteBuffer[numberOfTerms];
+ for (int i = 0; i < numberOfTerms; i++)
+ terms[i] = UTF8Type.instance.decompose(
+ new String(smallPool, random.nextInt(SMALL_POOL_SIZE - length),
length));
+ }
+
+ private void populateIndexData()
+ {
+ int termCount = 0;
+
+ for (int i = 0; i < NUM_PARTITION_KEYS; i++)
+ {
+ DecoratedKey partitionKey = partitionKeys[i];
+ memtableIndex.index(partitionKey, Clustering.EMPTY,
terms[termCount]);
+
+ if (++termCount == numberOfTerms)
+ {
+ termCount = 0;
+ }
+ }
+ }
+
+ @Benchmark
+ public long flushBench()
+ {
+ Iterator<Pair<ByteComparable, Iterator<PrimaryKey>>> it =
memtableIndex.iterator(null, null);
+ long count = 0;
+ while (it.hasNext())
+ {
+ Iterator<PrimaryKey> primaryKeyIterator = it.next().right;
+ while (primaryKeyIterator.hasNext())
+ {
+ primaryKeyIterator.next();
+ count++;
+ }
+ }
+ return count;
+ }
+}
diff --git
a/test/microbench/org/apache/cassandra/test/microbench/sai/MemtableIndexPartitionReadBench.java
b/test/microbench/org/apache/cassandra/test/microbench/sai/MemtableIndexPartitionReadBench.java
new file mode 100644
index 0000000000..b77e1f4c66
--- /dev/null
+++
b/test/microbench/org/apache/cassandra/test/microbench/sai/MemtableIndexPartitionReadBench.java
@@ -0,0 +1,174 @@
+/*
+ * 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.sai;
+
+import java.nio.ByteBuffer;
+import java.util.Random;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+
+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.Warmup;
+
+import org.apache.cassandra.cql3.Operator;
+import org.apache.cassandra.db.Clustering;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.PartitionPosition;
+import org.apache.cassandra.db.marshal.UTF8Type;
+import org.apache.cassandra.db.memtable.Memtable;
+import org.apache.cassandra.db.memtable.ShardBoundaries;
+import org.apache.cassandra.dht.AbstractBounds;
+import org.apache.cassandra.dht.Bounds;
+import org.apache.cassandra.index.sai.QueryContext;
+import org.apache.cassandra.index.sai.memory.ShardedMemtableIndex;
+import org.apache.cassandra.index.sai.memory.UnshardedMemtableIndex;
+import org.apache.cassandra.index.sai.plan.Expression;
+import org.apache.cassandra.schema.TableMetadata;
+
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@Warmup(iterations = 6, time = 3)
+@Measurement(iterations = 5, time = 5)
+@Fork(value = 1, jvmArgsAppend = { "-Xmx4G", "-Xms4G",
"-Djmh.executor=CUSTOM",
"-Djmh.executor.class=org.apache.cassandra.test.microbench.FastThreadExecutor"})
+@State(Scope.Benchmark)
+public class MemtableIndexPartitionReadBench extends AbstractMemtableIndexBench
+{
+ // 1 Million partitionKeys
+ private static final int NUM_PARTITION_KEYS = 1000000;
+ private static final int NUMBER_OF_SEARCHES = 1000;
+ private static final int SMALL_POOL_SIZE = 8 * 1024;
+
+ @Param({ "1", "4", "8"})
+ int shardCount;
+
+ @Param({"50", "100"})
+ int numberOfTerms;
+
+ private char[] smallPool = new char[SMALL_POOL_SIZE];
+ private Expression[] stringEqualityExpressions;
+ private QueryContext queryContext;
+ private AbstractBounds<PartitionPosition>[] keyRanges;
+
+ @State(Scope.Thread)
+ public static class ThreadState
+ {
+ ThreadLocalRandom random = ThreadLocalRandom.current();
+ }
+
+ @Setup(Level.Trial)
+ public void setup()
+ {
+ setupServer();
+ setupTableAndKeyspace();
+ setupCfsAndIndex();
+ setupPartitionKeys();
+ setupQueryContext();
+ setupIndexesExpressionsAndTerms();
+ }
+
+ private void setupQueryContext()
+ {
+ // setup a dummy query context, interface signature needs it.
+ queryContext = new QueryContext(null, Long.MAX_VALUE);
+ }
+
+ public void setupPartitionKeys()
+ {
+ TableMetadata tableMetadata = cfs.metadata();
+
+ partitionKeys = new DecoratedKey[NUM_PARTITION_KEYS];
+ for (int i = 0; i < NUM_PARTITION_KEYS; i++)
+ partitionKeys[i] =
tableMetadata.partitioner.decorateKey(tableMetadata.partitionKeyType.fromString("partition_"
+ i));
+ }
+
+ public void setupIndexesExpressionsAndTerms() {
+ Memtable memtable = cfs.getCurrentMemtable();
+ if (shardCount > 1)
+ {
+ ShardBoundaries boundaries = cfs.localRangeSplits(shardCount);
+ memtableIndex = new ShardedMemtableIndex(index, boundaries,
memtable);
+ }
+ else
+ memtableIndex = new UnshardedMemtableIndex(index, memtable);
+
+ setupTerms(numberOfTerms);
+ populateIndexDataAndKeyRanges();
+ populateExpressions();
+ }
+
+ @Override
+ public void setupTerms(int numberOfTerms)
+ {
+ Random random = new Random();
+ for (int i = 0; i < SMALL_POOL_SIZE; i++)
+ smallPool[i] = (char)('a' + random.nextInt(26));
+
+ int length = 64;
+ terms = new ByteBuffer[numberOfTerms];
+ for (int i = 0; i < numberOfTerms; i++)
+ terms[i] = UTF8Type.instance.decompose(
+ new String(smallPool, random.nextInt(SMALL_POOL_SIZE -
length), length));
+ }
+
+ private void populateIndexDataAndKeyRanges()
+ {
+ int termCount = 0;
+ keyRanges = new AbstractBounds[NUM_PARTITION_KEYS];
+
+ for (int i = 0; i < NUM_PARTITION_KEYS; i++)
+ {
+ DecoratedKey partitionKey = partitionKeys[i];
+ memtableIndex.index(partitionKey, Clustering.EMPTY,
terms[termCount]);
+ keyRanges[i] = new Bounds<>(partitionKey, partitionKey);
+
+ if (++termCount == numberOfTerms)
+ {
+ termCount = 0;
+ }
+ }
+ }
+
+ private void populateExpressions()
+ {
+ Random random = new Random();
+ stringEqualityExpressions = new Expression[NUMBER_OF_SEARCHES];
+ for (int i = 0; i < NUMBER_OF_SEARCHES; i++)
+ stringEqualityExpressions[i] =
Expression.create(index).add(Operator.EQ, terms[random.nextInt(terms.length)]);
+ }
+
+ @Benchmark
+ public long stringEqualityPartitionRestrictedRangeSearch(ThreadState state)
+ {
+ long size = 0;
+ memtableIndex.search(queryContext,
+
stringEqualityExpressions[state.random.nextInt(stringEqualityExpressions.length)],
+
keyRanges[state.random.nextInt(keyRanges.length)]);
+ return size;
+ }
+}
diff --git
a/test/microbench/org/apache/cassandra/test/microbench/sai/MemtableIndexReadBench.java
b/test/microbench/org/apache/cassandra/test/microbench/sai/MemtableIndexReadBench.java
new file mode 100644
index 0000000000..560b348f58
--- /dev/null
+++
b/test/microbench/org/apache/cassandra/test/microbench/sai/MemtableIndexReadBench.java
@@ -0,0 +1,135 @@
+/*
+ * 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.sai;
+
+import java.util.Random;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+
+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.Warmup;
+
+import org.apache.cassandra.cql3.Operator;
+import org.apache.cassandra.db.Clustering;
+import org.apache.cassandra.db.DataRange;
+import org.apache.cassandra.db.PartitionPosition;
+import org.apache.cassandra.db.memtable.Memtable;
+import org.apache.cassandra.db.memtable.ShardBoundaries;
+import org.apache.cassandra.dht.AbstractBounds;
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.index.sai.QueryContext;
+import org.apache.cassandra.index.sai.memory.ShardedMemtableIndex;
+import org.apache.cassandra.index.sai.memory.UnshardedMemtableIndex;
+import org.apache.cassandra.index.sai.plan.Expression;
+
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@Warmup(iterations = 3, time = 2)
+@Measurement(iterations = 5, time = 5)
+@Fork(value = 1, jvmArgsAppend = { "-Xmx4G", "-Xms4G",
"-Djmh.executor=CUSTOM",
"-Djmh.executor.class=org.apache.cassandra.test.microbench.FastThreadExecutor"})
+@State(Scope.Benchmark)
+public class MemtableIndexReadBench extends AbstractMemtableIndexBench
+{
+ private static final int NUMBER_OF_SEARCHES = 1000;
+ private static final AbstractBounds<PartitionPosition> ALL_DATA_RANGE =
DataRange.allData(Murmur3Partitioner.instance).keyRange();
+
+ @Param({ "1", "4", "8"})
+ int shardCount;
+
+ @Param({"1000000" })
+ protected int numberOfTerms;
+
+ @Param({ "1", "10", "100"})
+ protected int rowsPerPartition;
+
+ private Expression[] stringEqualityExpressions;
+ private QueryContext queryContext;
+
+ @State(Scope.Thread)
+ public static class ThreadState
+ {
+ ThreadLocalRandom random = ThreadLocalRandom.current();
+ }
+
+ @Setup(Level.Trial)
+ public void setup()
+ {
+ super.setup(numberOfTerms, rowsPerPartition);
+ }
+
+ @Setup(Level.Iteration)
+ public void setupIndexesAndExpressions() {
+ Memtable memtable = cfs.getCurrentMemtable();
+ if (shardCount > 1)
+ {
+ ShardBoundaries boundaries = cfs.localRangeSplits(shardCount);
+ memtableIndex = new ShardedMemtableIndex(index, boundaries,
memtable);
+ }
+ else
+ memtableIndex = new UnshardedMemtableIndex(index, memtable);
+
+ populateIndexData();
+ populateExpressions();
+ // setup a dummy query context, interface signature needs it.
+ queryContext = new QueryContext(null, Long.MAX_VALUE);
+ }
+
+ private void populateIndexData()
+ {
+ int rowCount = 0;
+ int keyCount = 0;
+ for (int i = 0; i < numberOfTerms; i++)
+ {
+ memtableIndex.index(partitionKeys[keyCount], Clustering.EMPTY,
terms[i]);
+ if (++rowCount == rowsPerPartition)
+ {
+ rowCount = 0;
+ keyCount++;
+ }
+ }
+ }
+
+ private void populateExpressions()
+ {
+ Random random = new Random();
+ stringEqualityExpressions = new Expression[NUMBER_OF_SEARCHES];
+ for (int i = 0; i < NUMBER_OF_SEARCHES; i++)
+ stringEqualityExpressions[i] =
Expression.create(index).add(Operator.EQ, terms[random.nextInt(terms.length)]);
+ }
+
+ @Benchmark
+ public long stringEqualitySearch(ThreadState state)
+ {
+ long size = 0;
+ memtableIndex.search(queryContext,
+
stringEqualityExpressions[state.random.nextInt(stringEqualityExpressions.length)],
+ ALL_DATA_RANGE);
+ return size;
+ }
+}
diff --git
a/test/microbench/org/apache/cassandra/test/microbench/sai/MemtableIndexWriteBench.java
b/test/microbench/org/apache/cassandra/test/microbench/sai/MemtableIndexWriteBench.java
new file mode 100644
index 0000000000..3cdb9ce9c4
--- /dev/null
+++
b/test/microbench/org/apache/cassandra/test/microbench/sai/MemtableIndexWriteBench.java
@@ -0,0 +1,105 @@
+/*
+ * 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.sai;
+
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+
+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.TearDown;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+
+import org.apache.cassandra.cql3.CQLTester;
+import org.apache.cassandra.db.Clustering;
+import org.apache.cassandra.db.commitlog.CommitLog;
+import org.apache.cassandra.db.memtable.Memtable;
+import org.apache.cassandra.db.memtable.ShardBoundaries;
+import org.apache.cassandra.index.sai.memory.ShardedMemtableIndex;
+import org.apache.cassandra.index.sai.memory.UnshardedMemtableIndex;
+
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@Warmup(iterations = 3, time = 2)
+@Measurement(iterations = 5, time = 5)
+@Fork(value = 1, jvmArgsAppend = { "-Xmx4G", "-Xms4G",
"-Djmh.executor=CUSTOM",
"-Djmh.executor.class=org.apache.cassandra.test.microbench.FastThreadExecutor"})
+@Threads(8)
+@State(Scope.Benchmark)
+public class MemtableIndexWriteBench extends AbstractMemtableIndexBench
+{
+ @Param({ "1", "4", "8"})
+ int shardCount;
+
+ @Param({"1000000" })
+ protected int numberOfTerms;
+
+ @Param({ "1", "10", "100"})
+ protected int rowsPerPartition;
+
+ @State(Scope.Thread)
+ public static class ThreadState
+ {
+ ThreadLocalRandom random = ThreadLocalRandom.current();
+ }
+
+ @Setup(Level.Trial)
+ public void setup()
+ {
+ super.setup(numberOfTerms, rowsPerPartition);
+ }
+
+ @Setup(Level.Iteration)
+ public void setupIndexes()
+ {
+ Memtable memtable = cfs.getCurrentMemtable();
+ if (shardCount > 1)
+ {
+ ShardBoundaries boundaries = cfs.localRangeSplits(shardCount);
+ memtableIndex = new ShardedMemtableIndex(index, boundaries,
memtable);
+ }
+ else
+ memtableIndex = new UnshardedMemtableIndex(index, memtable);
+ }
+
+ @Benchmark
+ public long write(ThreadState state)
+ {
+ return
memtableIndex.index(partitionKeys[state.random.nextInt(partitionKeys.length)],
+ Clustering.EMPTY,
+ terms[state.random.nextInt(terms.length)]);
+ }
+
+ @TearDown(Level.Trial)
+ public void teardown() throws InterruptedException
+ {
+ CommitLog.instance.shutdownBlocking();
+ CQLTester.tearDownClass();
+ CQLTester.cleanup();
+ }
+}
diff --git
a/test/unit/org/apache/cassandra/db/memtable/ShardBoundariesTest.java
b/test/unit/org/apache/cassandra/db/memtable/ShardBoundariesTest.java
new file mode 100644
index 0000000000..813dbb793e
--- /dev/null
+++ b/test/unit/org/apache/cassandra/db/memtable/ShardBoundariesTest.java
@@ -0,0 +1,117 @@
+/*
+ * 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.util.List;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import org.apache.cassandra.ServerTestUtils;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.BufferDecoratedKey;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.PartitionPosition;
+import org.apache.cassandra.dht.Bounds;
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.tcm.Epoch;
+import org.apache.cassandra.utils.ByteBufferUtil;
+
+import static org.junit.Assert.assertEquals;
+
+public class ShardBoundariesTest
+{
+ private ShardBoundaries boundaries;
+
+ @Before
+ public void setUp()
+ {
+ DatabaseDescriptor.daemonInitialization();
+ DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
+ ServerTestUtils.prepareServerNoRegister();
+ boundaries = new ShardBoundaries(new Token[]{ token(10), token(20),
token(30) }, Epoch.EMPTY);
+ }
+
+ private static Murmur3Partitioner.LongToken token(long value)
+ {
+ return new Murmur3Partitioner.LongToken(value);
+ }
+
+ private static DecoratedKey key(long tokenValue)
+ {
+ return new BufferDecoratedKey(token(tokenValue),
ByteBufferUtil.bytes(tokenValue));
+ }
+
+ @Test
+ public void testGetShardForToken()
+ {
+ assertEquals(0, boundaries.getShardForToken(token(5)));
+ assertEquals(0, boundaries.getShardForToken(token(10)));
+ assertEquals(1, boundaries.getShardForToken(token(15)));
+ assertEquals(1, boundaries.getShardForToken(token(20)));
+ assertEquals(2, boundaries.getShardForToken(token(30)));
+ assertEquals(3, boundaries.getShardForToken(token(35)));
+ }
+
+ @Test
+ public void testGetShardsForRangeNoBoundaries()
+ {
+ List<Integer> shards = ShardBoundaries.NONE.getShardsForRange(new
Bounds<>(key(1), key(100)));
+ assertEquals(List.of(0), shards);
+ }
+
+ @Test
+ public void testGetShardsForRangeWholeRing()
+ {
+ PartitionPosition minimum =
Murmur3Partitioner.instance.getMinimumToken().minKeyBound();
+ List<Integer> shards = boundaries.getShardsForRange(new
Range<>(minimum, minimum));
+ assertEquals(List.of(0, 1, 2, 3), shards);
+ }
+
+ @Test
+ public void testGetShardsForRangeSingleShard()
+ {
+ List<Integer> shards = boundaries.getShardsForRange(new
Bounds<>(key(12), key(15)));
+ assertEquals(List.of(1), shards);
+ }
+
+ @Test
+ public void testGetShardsForRangeMultipleContiguousShards()
+ {
+ List<Integer> shards = boundaries.getShardsForRange(new
Bounds<>(key(5), key(25)));
+ assertEquals(List.of(0, 1, 2), shards);
+ }
+
+ @Test
+ public void testGetShardsForRangeOpenEndedRight()
+ {
+ PartitionPosition minimum =
Murmur3Partitioner.instance.getMinimumToken().minKeyBound();
+ List<Integer> shards = boundaries.getShardsForRange(new
Bounds<>(key(15), minimum));
+ assertEquals(List.of(1, 2, 3), shards);
+ }
+
+ @Test
+ public void testGetShardsForRangeWrappingRange()
+ {
+ List<Integer> shards = boundaries.getShardsForRange(new
Range<>(key(25), key(5)));
+ assertEquals(List.of(0, 2, 3), shards);
+ }
+}
diff --git a/test/unit/org/apache/cassandra/dht/RangeIntersectsBoundsTest.java
b/test/unit/org/apache/cassandra/dht/RangeIntersectsBoundsTest.java
new file mode 100644
index 0000000000..daca3787f7
--- /dev/null
+++ b/test/unit/org/apache/cassandra/dht/RangeIntersectsBoundsTest.java
@@ -0,0 +1,296 @@
+/*
+ * 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.dht;
+
+import org.junit.Test;
+
+import org.apache.cassandra.CassandraTestBase;
+import org.apache.cassandra.CassandraTestBase.DDDaemonInitialization;
+import org.apache.cassandra.dht.Murmur3Partitioner.LongToken;
+import org.apache.cassandra.dht.RandomPartitioner.BigIntegerToken;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+@DDDaemonInitialization
+public class RangeIntersectsBoundsTest extends CassandraTestBase
+{
+ @Test
+ public void rangeIntersectsBounds() throws Exception
+ {
+ Range<Token> all = new Range<>(new BigIntegerToken("0"), new
BigIntegerToken("0"));
+ Range<Token> some = new Range<>(new BigIntegerToken("4"), new
BigIntegerToken("8"));
+ Range<Token> someWrapped = new Range<>(some.right, some.left);
+
+ // Coda:
+ // l - matches left token of some range
+ // r - matches right token of some range
+ // b - below left token of some range
+ // a - above right token of some range
+ // i - inside some range (above left and below right
+ Bounds<Token> lr = new Bounds<>(new BigIntegerToken("4"), new
BigIntegerToken("8"));
+ Bounds<Token> br = new Bounds<>(new BigIntegerToken("3"), new
BigIntegerToken("8"));
+ Bounds<Token> bi = new Bounds<>(new BigIntegerToken("3"), new
BigIntegerToken("7"));
+ Bounds<Token> ba = new Bounds<>(new BigIntegerToken("3"), new
BigIntegerToken("9"));
+ Bounds<Token> la = new Bounds<>(new BigIntegerToken("4"), new
BigIntegerToken("9"));
+ Bounds<Token> li = new Bounds<>(new BigIntegerToken("4"), new
BigIntegerToken("7"));
+ Bounds<Token> ii = new Bounds<>(new BigIntegerToken("5"), new
BigIntegerToken("7"));
+ Bounds<Token> ir = new Bounds<>(new BigIntegerToken("5"), new
BigIntegerToken("8"));
+ Bounds<Token> bb = new Bounds<>(new BigIntegerToken("2"), new
BigIntegerToken("3"));
+ Bounds<Token> aa = new Bounds<>(new BigIntegerToken("9"), new
BigIntegerToken("10"));
+ Bounds<Token> bl = new Bounds<>(new BigIntegerToken("3"), new
BigIntegerToken("4"));
+ Bounds<Token> ra = new Bounds<>(new BigIntegerToken("8"), new
BigIntegerToken("9"));
+
+ assertTrue(all.intersects(lr));
+ assertTrue(all.intersects(br));
+ assertTrue(all.intersects(bi));
+ assertTrue(all.intersects(ba));
+ assertTrue(all.intersects(la));
+ assertTrue(all.intersects(li));
+ assertTrue(all.intersects(ii));
+ assertTrue(all.intersects(ir));
+ assertTrue(all.intersects(bb));
+ assertTrue(all.intersects(aa));
+ assertTrue(all.intersects(bl));
+ assertTrue(all.intersects(ra));
+
+ assertTrue(some.intersects(lr));
+ assertTrue(some.intersects(br));
+ assertTrue(some.intersects(bi));
+ assertTrue(some.intersects(ba));
+ assertTrue(some.intersects(la));
+ assertTrue(some.intersects(li));
+ assertTrue(some.intersects(ii));
+ assertTrue(some.intersects(ir));
+ assertFalse(some.intersects(bb));
+ assertFalse(some.intersects(aa));
+ assertFalse(some.intersects(bl));
+ assertTrue(some.intersects(ra));
+
+ assertTrue(someWrapped.intersects(lr));
+ assertTrue(someWrapped.intersects(br));
+ assertTrue(someWrapped.intersects(bi));
+ assertTrue(someWrapped.intersects(ba));
+ assertTrue(someWrapped.intersects(la));
+ assertTrue(someWrapped.intersects(li));
+ assertFalse(someWrapped.intersects(ii));
+ assertFalse(someWrapped.intersects(ir));
+ assertTrue(someWrapped.intersects(bb));
+ assertTrue(someWrapped.intersects(aa));
+ assertTrue(someWrapped.intersects(bl));
+ assertTrue(someWrapped.intersects(ra));
+ }
+
+ @Test
+ public void rangeIntersectsExcludingBounds() throws Exception
+ {
+ Range<Token> all = new Range<>(new BigIntegerToken("0"), new
BigIntegerToken("0"));
+ Range<Token> some = new Range<>(new BigIntegerToken("4"), new
BigIntegerToken("8"));
+ Range<Token> someWrapped = new Range<>(some.right, some.left);
+
+ // Coda:
+ // l - matches left token of some range
+ // r - matches right token of some range
+ // b - below left token of some range
+ // a - above right token of some range
+ // i - inside some range (above left and below right
+ ExcludingBounds<Token> lr = new ExcludingBounds<>(new
BigIntegerToken("4"), new BigIntegerToken("8"));
+ ExcludingBounds<Token> br = new ExcludingBounds<>(new
BigIntegerToken("3"), new BigIntegerToken("8"));
+ ExcludingBounds<Token> bi = new ExcludingBounds<>(new
BigIntegerToken("3"), new BigIntegerToken("7"));
+ ExcludingBounds<Token> ba = new ExcludingBounds<>(new
BigIntegerToken("3"), new BigIntegerToken("9"));
+ ExcludingBounds<Token> la = new ExcludingBounds<>(new
BigIntegerToken("4"), new BigIntegerToken("9"));
+ ExcludingBounds<Token> li = new ExcludingBounds<>(new
BigIntegerToken("4"), new BigIntegerToken("7"));
+ ExcludingBounds<Token> ii = new ExcludingBounds<>(new
BigIntegerToken("5"), new BigIntegerToken("7"));
+ ExcludingBounds<Token> ir = new ExcludingBounds<>(new
BigIntegerToken("5"), new BigIntegerToken("8"));
+ ExcludingBounds<Token> bb = new ExcludingBounds<>(new
BigIntegerToken("2"), new BigIntegerToken("3"));
+ ExcludingBounds<Token> aa = new ExcludingBounds<>(new
BigIntegerToken("9"), new BigIntegerToken("10"));
+ ExcludingBounds<Token> bl = new ExcludingBounds<>(new
BigIntegerToken("3"), new BigIntegerToken("4"));
+ ExcludingBounds<Token> ra = new ExcludingBounds<>(new
BigIntegerToken("8"), new BigIntegerToken("9"));
+
+ assertTrue(all.intersects(lr));
+ assertTrue(all.intersects(br));
+ assertTrue(all.intersects(bi));
+ assertTrue(all.intersects(ba));
+ assertTrue(all.intersects(la));
+ assertTrue(all.intersects(li));
+ assertTrue(all.intersects(ii));
+ assertTrue(all.intersects(ir));
+ assertTrue(all.intersects(bb));
+ assertTrue(all.intersects(aa));
+ assertTrue(all.intersects(bl));
+ assertTrue(all.intersects(ra));
+
+ assertTrue(some.intersects(lr));
+ assertTrue(some.intersects(br));
+ assertTrue(some.intersects(bi));
+ assertTrue(some.intersects(ba));
+ assertTrue(some.intersects(la));
+ assertTrue(some.intersects(li));
+ assertTrue(some.intersects(ii));
+ assertTrue(some.intersects(ir));
+ assertFalse(some.intersects(bb));
+ assertFalse(some.intersects(aa));
+ assertFalse(some.intersects(bl));
+ assertFalse(some.intersects(ra));
+
+ assertFalse(someWrapped.intersects(lr));
+ assertTrue(someWrapped.intersects(br));
+ assertTrue(someWrapped.intersects(bi));
+ assertTrue(someWrapped.intersects(ba));
+ assertTrue(someWrapped.intersects(la));
+ assertFalse(someWrapped.intersects(li));
+ assertFalse(someWrapped.intersects(ii));
+ assertFalse(someWrapped.intersects(ir));
+ assertTrue(someWrapped.intersects(bb));
+ assertTrue(someWrapped.intersects(aa));
+ assertTrue(someWrapped.intersects(bl));
+ assertTrue(someWrapped.intersects(ra));
+
+ Range<Token> range = new Range<>(Murmur3Partitioner.MINIMUM, new
Murmur3Partitioner.LongToken(-1));
+ ExcludingBounds<Token> bounds = new ExcludingBounds<>(new
Murmur3Partitioner.LongToken(-3248873570005575792L),
Murmur3Partitioner.MINIMUM);
+
+ assertTrue(range.intersects(bounds));
+
+ range = new Range<>(new Murmur3Partitioner.LongToken(-1),
Murmur3Partitioner.MINIMUM);
+
+ assertTrue(range.intersects(bounds));
+
+ }
+
+ @Test
+ public void rangeIntersectsIncludingExcludingBounds()
+ {
+ Range<Token> all = new Range<>(new BigIntegerToken("0"), new
BigIntegerToken("0"));
+ Range<Token> some = new Range<>(new BigIntegerToken("4"), new
BigIntegerToken("8"));
+ Range<Token> someWrapped = new Range<>(some.right, some.left);
+
+ // Coda:
+ // l - matches left token of some range
+ // r - matches right token of some range
+ // b - below left token of some range
+ // a - above right token of some range
+ // i - inside some range (above left and below right
+ IncludingExcludingBounds<Token> lr = new
IncludingExcludingBounds<>(new BigIntegerToken("4"), new BigIntegerToken("8"));
+ IncludingExcludingBounds<Token> br = new
IncludingExcludingBounds<>(new BigIntegerToken("3"), new BigIntegerToken("8"));
+ IncludingExcludingBounds<Token> bi = new
IncludingExcludingBounds<>(new BigIntegerToken("3"), new BigIntegerToken("7"));
+ IncludingExcludingBounds<Token> ba = new
IncludingExcludingBounds<>(new BigIntegerToken("3"), new BigIntegerToken("9"));
+ IncludingExcludingBounds<Token> la = new
IncludingExcludingBounds<>(new BigIntegerToken("4"), new BigIntegerToken("9"));
+ IncludingExcludingBounds<Token> li = new
IncludingExcludingBounds<>(new BigIntegerToken("4"), new BigIntegerToken("7"));
+ IncludingExcludingBounds<Token> ii = new
IncludingExcludingBounds<>(new BigIntegerToken("5"), new BigIntegerToken("7"));
+ IncludingExcludingBounds<Token> ir = new
IncludingExcludingBounds<>(new BigIntegerToken("5"), new BigIntegerToken("8"));
+ IncludingExcludingBounds<Token> bb = new
IncludingExcludingBounds<>(new BigIntegerToken("2"), new BigIntegerToken("3"));
+ IncludingExcludingBounds<Token> aa = new
IncludingExcludingBounds<>(new BigIntegerToken("9"), new BigIntegerToken("10"));
+ IncludingExcludingBounds<Token> bl = new
IncludingExcludingBounds<>(new BigIntegerToken("3"), new BigIntegerToken("4"));
+ IncludingExcludingBounds<Token> ra = new
IncludingExcludingBounds<>(new BigIntegerToken("8"), new BigIntegerToken("9"));
+
+ assertTrue(all.intersects(lr));
+ assertTrue(all.intersects(br));
+ assertTrue(all.intersects(bi));
+ assertTrue(all.intersects(ba));
+ assertTrue(all.intersects(la));
+ assertTrue(all.intersects(li));
+ assertTrue(all.intersects(ii));
+ assertTrue(all.intersects(ir));
+ assertTrue(all.intersects(bb));
+ assertTrue(all.intersects(aa));
+ assertTrue(all.intersects(bl));
+ assertTrue(all.intersects(ra));
+
+ assertTrue(some.intersects(lr));
+ assertTrue(some.intersects(br));
+ assertTrue(some.intersects(bi));
+ assertTrue(some.intersects(ba));
+ assertTrue(some.intersects(la));
+ assertTrue(some.intersects(li));
+ assertTrue(some.intersects(ii));
+ assertTrue(some.intersects(ir));
+ assertFalse(some.intersects(bb));
+ assertFalse(some.intersects(aa));
+ assertFalse(some.intersects(bl));
+ assertTrue(some.intersects(ra));
+
+ assertTrue(someWrapped.intersects(lr));
+ assertTrue(someWrapped.intersects(br));
+ assertTrue(someWrapped.intersects(bi));
+ assertTrue(someWrapped.intersects(ba));
+ assertTrue(someWrapped.intersects(la));
+ assertTrue(someWrapped.intersects(li));
+ assertFalse(someWrapped.intersects(ii));
+ assertFalse(someWrapped.intersects(ir));
+ assertTrue(someWrapped.intersects(bb));
+ assertTrue(someWrapped.intersects(aa));
+ assertTrue(someWrapped.intersects(bl));
+ assertTrue(someWrapped.intersects(ra));
+ }
+
+ /**
+ * Test that we handle partial bounds of the type x > n or x >= n which
specifically have
+ * their right value as minimum.
+ */
+ @Test
+ public void rangeIntersectsPartialBounds()
+ {
+ Range<Token> range = new Range<>(Murmur3Partitioner.MINIMUM, new
Murmur3Partitioner.LongToken(-1L));
+
+ Bounds<Token> boundsMatch = new Bounds<>(new
Murmur3Partitioner.LongToken(-2L), Murmur3Partitioner.MINIMUM);
+ Bounds<Token> boundsNoMatch = new Bounds<>(new
Murmur3Partitioner.LongToken(0L), Murmur3Partitioner.MINIMUM);
+
+ assertTrue(range.intersects(boundsMatch));
+ assertFalse(range.intersects(boundsNoMatch));
+
+ ExcludingBounds<Token> excBoundsMatch = new ExcludingBounds<>(new
Murmur3Partitioner.LongToken(-2L), Murmur3Partitioner.MINIMUM);
+ ExcludingBounds<Token> excBoundsNoMatch = new ExcludingBounds<>(new
Murmur3Partitioner.LongToken(-1L), Murmur3Partitioner.MINIMUM);
+
+ assertTrue(range.intersects(excBoundsMatch));
+ assertFalse(range.intersects(excBoundsNoMatch));
+
+ IncludingExcludingBounds<Token> incExcBoundsMatch = new
IncludingExcludingBounds<>(new Murmur3Partitioner.LongToken(-2L),
Murmur3Partitioner.MINIMUM);
+ IncludingExcludingBounds<Token> incExcBoundsNoMatch = new
IncludingExcludingBounds<>(new Murmur3Partitioner.LongToken(0L),
Murmur3Partitioner.MINIMUM);
+
+ assertTrue(range.intersects(incExcBoundsMatch));
+ assertFalse(range.intersects(incExcBoundsNoMatch));
+ }
+
+ @Test
+ public void wrappingRangeIntersectsTest()
+ {
+ // (100, max] + [min, -100]
+ Range<Token> wrapping = new Range<>(new LongToken(100), new
LongToken(-100));
+
+ // inside the upper segment, sharing its left endpoint with the range
+ assertFalse(wrapping.intersects(new ExcludingBounds<>(new
LongToken(50), new LongToken(100))));
+ assertTrue(wrapping.intersects(new ExcludingBounds<>(new
LongToken(50), new LongToken(101))));
+ assertTrue(wrapping.intersects(new ExcludingBounds<>(new
LongToken(100), new LongToken(200))));
+ assertTrue(wrapping.intersects(new IncludingExcludingBounds<>(new
LongToken(100), new LongToken(200))));
+
+ // inside the lower segment, sharing its right endpoint with the range
+ assertTrue(wrapping.intersects(new ExcludingBounds<>(new
LongToken(-200), new LongToken(-100))));
+ assertTrue(wrapping.intersects(new IncludingExcludingBounds<>(new
LongToken(-200), new LongToken(-100))));
+
+ // in the gap between the two segments; only the included left
endpoint of [-100, 100) is in the range
+ assertFalse(wrapping.intersects(new ExcludingBounds<>(new
LongToken(-100), new LongToken(100))));
+ assertTrue(wrapping.intersects(new IncludingExcludingBounds<>(new
LongToken(-100), new LongToken(100))));
+ assertFalse(wrapping.intersects(new ExcludingBounds<>(new
LongToken(-50), new LongToken(50))));
+ assertFalse(wrapping.intersects(new IncludingExcludingBounds<>(new
LongToken(-50), new LongToken(50))));
+
+ // right endpoint at the minimum, i.e. unbounded on the right
+ assertTrue(wrapping.intersects(new ExcludingBounds<>(new
LongToken(200), Murmur3Partitioner.MINIMUM)));
+ assertTrue(wrapping.intersects(new IncludingExcludingBounds<>(new
LongToken(200), Murmur3Partitioner.MINIMUM)));
+ }
+}
diff --git
a/test/unit/org/apache/cassandra/index/sai/cql/StorageAttachedIndexDDLTest.java
b/test/unit/org/apache/cassandra/index/sai/cql/StorageAttachedIndexDDLTest.java
index 8e82d15a27..d179a2512c 100644
---
a/test/unit/org/apache/cassandra/index/sai/cql/StorageAttachedIndexDDLTest.java
+++
b/test/unit/org/apache/cassandra/index/sai/cql/StorageAttachedIndexDDLTest.java
@@ -82,9 +82,12 @@ import org.apache.cassandra.inject.InvokePointBuilder;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Throwables;
import static java.util.Collections.singletonList;
+import static
org.apache.cassandra.config.CassandraRelevantProperties.MEMTABLE_SHARD_COUNT;
+import static
org.apache.cassandra.index.sai.StorageAttachedIndex.VECTOR_SHARDING_SUPPORT_MESSAGE;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -269,6 +272,78 @@ public class StorageAttachedIndexDDLTest extends SAITester
.hasMessage(String.format(CreateIndexStatement.INVALID_CHARS_CUSTOM_INDEX_TARGET,
invalidColumn));
}
+ @Test
+ public void shouldRejectShardsOptionOnVectorIndex()
+ {
+ createTable("CREATE TABLE %s (id text PRIMARY KEY, val vector<float,
3>)");
+
+ assertThatThrownBy(() -> executeNet("CREATE INDEX ON %s(val) USING
'sai' WITH OPTIONS = { 'shards' : '4' }"))
+ .isInstanceOf(InvalidQueryException.class)
+ .hasMessageContaining(VECTOR_SHARDING_SUPPORT_MESSAGE);
+
+ assertThatThrownBy(() -> executeNet("CREATE INDEX ON %s(val) USING
'sai' WITH OPTIONS = { 'shards' : 'auto' }"))
+ .isInstanceOf(InvalidQueryException.class)
+ .hasMessageContaining(VECTOR_SHARDING_SUPPORT_MESSAGE);
+ }
+
+ @Test
+ public void shouldCreateIndexWithValidShardsOption()
+ {
+ createTable("CREATE TABLE %s (id text PRIMARY KEY, val1 text, val2
text)");
+ createIndex("CREATE INDEX ON %s(val1) USING 'sai' WITH OPTIONS = {
'shards' : '1' }");
+ createIndex("CREATE INDEX ON %s(val2) USING 'sai' WITH OPTIONS = {
'shards' : '4' }");
+ assertEquals(2, saiCreationCounter.get());
+ }
+
+ @Test
+ public void shouldRejectNonIntegerShardsOption()
+ {
+ createTable("CREATE TABLE %s (id text PRIMARY KEY, val text)");
+
+ assertThatThrownBy(() -> executeNet("CREATE INDEX ON %s(val) USING
'sai' WITH OPTIONS = { 'shards' : 'abc' }"))
+ .isInstanceOf(InvalidQueryException.class)
+ .hasMessageContaining("Shard count for a storage-attached index must
be a valid integer");
+ }
+
+ @Test
+ public void shouldCreateIndexWithAutoShardsOption()
+ {
+ createTable("CREATE TABLE %s (id text PRIMARY KEY, val1 text, val2
text, val3 text)");
+ createIndex("CREATE INDEX auto_idx ON %s(val1) USING 'sai' WITH
OPTIONS = { 'shards' : 'auto' }");
+ createIndex("CREATE INDEX ON %s(val2) USING 'sai' WITH OPTIONS = {
'shards' : 'Auto' }");
+ createIndex("CREATE INDEX ON %s(val3) USING 'sai' WITH OPTIONS = {
'shards' : 'AUTO' }");
+ assertEquals(3, saiCreationCounter.get());
+
+ SecondaryIndexManager sim = getCurrentColumnFamilyStore().indexManager;
+ StorageAttachedIndex index = (StorageAttachedIndex)
sim.getIndexByName("auto_idx");
+ int expectedShardCount =
MEMTABLE_SHARD_COUNT.getInt(FBUtilities.getAvailableProcessors());
+ assertEquals(expectedShardCount, index.shardCount());
+ }
+
+ @Test
+ public void shouldRejectNonPositiveShardsOption()
+ {
+ createTable("CREATE TABLE %s (id text PRIMARY KEY, val text)");
+
+ assertThatThrownBy(() -> executeNet("CREATE INDEX ON %s(val) USING
'sai' WITH OPTIONS = { 'shards' : '0' }"))
+ .isInstanceOf(InvalidQueryException.class)
+ .hasMessageContaining("Shard count for a storage-attached index must
be a positive integer");
+
+ assertThatThrownBy(() -> executeNet("CREATE INDEX ON %s(val) USING
'sai' WITH OPTIONS = { 'shards' : '-5' }"))
+ .isInstanceOf(InvalidQueryException.class)
+ .hasMessageContaining("Shard count for a storage-attached index must
be a positive integer");
+ }
+
+ @Test
+ public void shouldRejectShardsOptionGreaterThanMaxShardCount()
+ {
+ createTable("CREATE TABLE %s (id text PRIMARY KEY, val text)");
+
+ assertThatThrownBy(() -> executeNet("CREATE INDEX ON %s(val) USING
'sai' WITH OPTIONS = { 'shards' : '257' }"))
+ .isInstanceOf(InvalidQueryException.class)
+ .hasMessageContaining("Shard count for a storage-attached index must
not exceed");
+ }
+
@Test
public void shouldCreateIndexIfExists()
{
diff --git
a/test/unit/org/apache/cassandra/index/sai/memory/ShardedMemtableIndexFlushTest.java
b/test/unit/org/apache/cassandra/index/sai/memory/ShardedMemtableIndexFlushTest.java
new file mode 100644
index 0000000000..d2d41a301c
--- /dev/null
+++
b/test/unit/org/apache/cassandra/index/sai/memory/ShardedMemtableIndexFlushTest.java
@@ -0,0 +1,55 @@
+/*
+ * 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.index.sai.memory;
+
+import org.junit.Test;
+
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.index.sai.utils.SAIRandomizedTester;
+
+public class ShardedMemtableIndexFlushTest extends SAIRandomizedTester
+{
+ @Test
+ public void flushShardedIndexWithReversedClustering()
+ {
+ createTable("CREATE TABLE %s (pk0 varint, ck0 bigint, ck1 text, v0
uuid, PRIMARY KEY (pk0, ck0, ck1)) WITH CLUSTERING ORDER BY (ck0 DESC, ck1
DESC)");
+
+ createIndex("CREATE CUSTOM INDEX ON %s(ck0) USING
'StorageAttachedIndex' WITH OPTIONS = {'shards': '4'}");
+ createIndex("CREATE CUSTOM INDEX ON %s(ck1) USING
'StorageAttachedIndex'");
+
+ execute("INSERT INTO %s (pk0, ck0, ck1, v0) VALUES (0,
3796795943589367651, 'apple', uuid()) USING TIMESTAMP 2");
+ execute("INSERT INTO %s (pk0, ck0, ck1, v0) VALUES (0,
5390896686766209114, 'banana', uuid()) USING TIMESTAMP 4");
+
+ execute("UPDATE %s USING TIMESTAMP 5 SET v0 = uuid() WHERE pk0 =
13271891 AND ck0 = 7466188966332780245 AND ck1 = 'cherry'");
+ execute("UPDATE %s USING TIMESTAMP 5 SET v0 = uuid() WHERE pk0 =
2081887771 AND ck0 = 7466188966332780245 AND ck1 = 'cherry'");
+
+ execute("UPDATE %s USING TIMESTAMP 6 SET v0 = uuid() WHERE pk0 = 0 AND
ck0 = 8977613850235134686 AND ck1 = 'date'");
+ execute("UPDATE %s USING TIMESTAMP 6 SET v0 = uuid() WHERE pk0 = 4982
AND ck0 = 8977613850235134686 AND ck1 = 'date'");
+
+ // Verify data is visible before flush
+ UntypedResultSet beforeFlush = execute("SELECT * FROM %s WHERE ck0 =
8977613850235134686 AND ck1 = 'date' ALLOW FILTERING");
+ assertRowCount(beforeFlush, 2);
+
+ flush();
+
+ // After flush, same query should return same results
+ UntypedResultSet afterFlush = execute("SELECT * FROM %s WHERE ck0 =
8977613850235134686 AND ck1 = 'date' ALLOW FILTERING");
+ assertRowCount(afterFlush, 2);
+ }
+}
\ No newline at end of file
diff --git
a/test/unit/org/apache/cassandra/index/sai/memory/ShardedMemtableIndexTest.java
b/test/unit/org/apache/cassandra/index/sai/memory/ShardedMemtableIndexTest.java
new file mode 100644
index 0000000000..9241e1da8d
--- /dev/null
+++
b/test/unit/org/apache/cassandra/index/sai/memory/ShardedMemtableIndexTest.java
@@ -0,0 +1,255 @@
+/*
+ * 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.index.sai.memory;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.stream.Collectors;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import org.apache.cassandra.cql3.statements.schema.IndexTarget;
+import org.apache.cassandra.db.Clustering;
+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.Int32Type;
+import org.apache.cassandra.db.memtable.ShardBoundaries;
+import org.apache.cassandra.dht.AbstractBounds;
+import org.apache.cassandra.dht.IPartitioner;
+import org.apache.cassandra.index.sai.StorageAttachedIndex;
+import org.apache.cassandra.index.sai.iterators.KeyRangeIterator;
+import org.apache.cassandra.index.sai.plan.Expression;
+import org.apache.cassandra.index.sai.utils.PrimaryKey;
+import org.apache.cassandra.index.sai.utils.SAIRandomizedTester;
+import org.apache.cassandra.schema.IndexMetadata;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.utils.Pair;
+import org.apache.cassandra.utils.bytecomparable.ByteComparable;
+import org.apache.cassandra.utils.bytecomparable.ByteSource;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class ShardedMemtableIndexTest extends SAIRandomizedTester
+{
+ private ColumnFamilyStore cfs;
+ private IPartitioner partitioner;
+ private StorageAttachedIndex index;
+ private ShardBoundaries boundaries;
+ private ShardedMemtableIndex memtableIndex;
+ private Map<DecoratedKey, Integer> keyMap;
+ private Map<Integer, Integer> rowMap;
+
+ @Before
+ public void setup() throws Throwable
+ {
+ // CQLTester @BeforeClass already sets up server.
+ // Set up the keyspace and the table.
+ String keyspace = createKeyspace("CREATE KEYSPACE %s with replication
= { 'class' : 'SimpleStrategy', 'replication_factor' : 1 } and durable_writes =
false");
+ String table = createTable(keyspace, "CREATE TABLE %s (pk int PRIMARY
KEY, val int)", "memtable_index");
+
+ setupCfsAndIndex(keyspace, table);
+
+ partitioner = cfs.getPartitioner();
+ keyMap = new TreeMap<>();
+ rowMap = new HashMap<>();
+ }
+
+ public void setupCfsAndIndex(String keyspace, String table)
+ {
+ cfs = Keyspace.open(keyspace).getColumnFamilyStore(table);
+ cfs.disableAutoCompaction();
+
+ Map<String, String> options = new HashMap<>();
+ options.put(IndexTarget.CUSTOM_INDEX_OPTION_NAME,
StorageAttachedIndex.class.getCanonicalName());
+ options.put("target", "val");
+
+ IndexMetadata indexMetadata =
IndexMetadata.fromSchemaMetadata("val_idx", IndexMetadata.Kind.CUSTOM, options);
+
+ index = new StorageAttachedIndex(cfs, indexMetadata);
+ boundaries = cfs.localRangeSplits(8);
+ }
+
+ @Test
+ public void onHeapAllocationTest()
+ {
+ memtableIndex = new ShardedMemtableIndex(index, boundaries,
cfs.getCurrentMemtable());
+ assertEquals(8, memtableIndex.shardCount());
+
+ assertEquals(0L, memtableIndex.writeCount());
+
+ for (int row = 0; row < 100; row++)
+ {
+ addRow(row, row);
+ }
+
+ assertTrue(memtableIndex.writeCount() > 0);
+ }
+
+ @Test
+ public void randomQueryTest()
+ {
+ memtableIndex = new ShardedMemtableIndex(index, boundaries,
cfs.getCurrentMemtable());
+ assertEquals(8, memtableIndex.shardCount());
+
+ for (int row = 0; row < getRandom().nextIntBetween(1000, 5000); row++)
+ {
+ int pk = getRandom().nextIntBetween(0, 10000);
+ while (rowMap.containsKey(pk))
+ pk = getRandom().nextIntBetween(0, 10000);
+ int value = getRandom().nextIntBetween(0, 100);
+ rowMap.put(pk, value);
+ addRow(pk, value);
+ }
+
+ List<DecoratedKey> keys = new ArrayList<>(keyMap.keySet());
+
+ for (int executionCount = 0; executionCount < 1000; executionCount++)
+ {
+ Expression expression = generateRandomExpression(index);
+
+ AbstractBounds<PartitionPosition> keyRange =
generateRandomBounds(keys, partitioner);
+
+ Set<Integer> expectedKeys = keyMap.keySet()
+ .stream()
+ .filter(keyRange::contains)
+ .map(keyMap::get)
+ .filter(pk ->
expression.isSatisfiedBy(Int32Type.instance.decompose(rowMap.get(pk))))
+ .collect(Collectors.toSet());
+
+ Set<Integer> foundKeys = new HashSet<>();
+
+ try (KeyRangeIterator iterator = memtableIndex.search(null,
expression, keyRange))
+ {
+ while (iterator.hasNext())
+ {
+ int key =
Int32Type.instance.compose(iterator.next().partitionKey().getKey());
+ assertFalse(foundKeys.contains(key));
+ foundKeys.add(key);
+ }
+ }
+
+ assertEquals(expectedKeys, foundKeys);
+ }
+ }
+
+ @Test
+ public void indexIteratorTest()
+ {
+ memtableIndex = new ShardedMemtableIndex(index, boundaries,
cfs.getCurrentMemtable());
+ assertEquals(8, memtableIndex.shardCount());
+
+ Map<Integer, Set<DecoratedKey>> terms = buildTermMap();
+
+ terms.entrySet()
+ .stream()
+ .forEach(entry -> entry.getValue()
+ .forEach(pk ->
addRow(Int32Type.instance.compose(pk.getKey()), entry.getKey())));
+
+ for (int executionCount = 0; executionCount < 1000; executionCount++)
+ {
+ // These keys have midrange tokens that select 3 of the 8 shards
+ DecoratedKey first = makeKey(cfs.metadata(),
getRandom().nextIntBetween(0, 20000));
+ DecoratedKey second = makeKey(cfs.metadata(),
getRandom().nextIntBetween(0, 20000));
+ final DecoratedKey minimum = first.compareTo(second) <= 0 ? first
: second;
+ final DecoratedKey maximum = first.compareTo(second) <= 0 ? second
: first;
+
+ Iterator<Pair<ByteComparable, Iterator<PrimaryKey>>> iterator =
memtableIndex.iterator(minimum, maximum);
+
+ while (iterator.hasNext())
+ {
+ Pair<ByteComparable, Iterator<PrimaryKey>> termPair =
iterator.next();
+ int term = termFromComparable(termPair.left);
+
+ // The iterator will return keys outside the range of min/max
so we need to filter here to
+ // get the correct keys
+ List<DecoratedKey> expectedPks = terms.get(term)
+ .stream()
+ .filter(pk ->
pk.compareTo(minimum) >= 0 && pk.compareTo(maximum) <= 0)
+ .sorted()
+
.collect(Collectors.toList());
+
+ List<DecoratedKey> termPks = new ArrayList<>();
+
+ while (termPair.right.hasNext())
+ {
+ DecoratedKey pk = termPair.right.next().partitionKey();
+ if (pk.compareTo(minimum) >= 0 && pk.compareTo(maximum) <=
0)
+ termPks.add(pk);
+ }
+
+ assertEquals(expectedPks, termPks);
+ }
+ }
+ }
+
+ private int termFromComparable(ByteComparable comparable)
+ {
+ ByteSource.Peekable peekable =
ByteSource.peekable(comparable.asComparableBytes(ByteComparable.Version.OSS50));
+ return
Int32Type.instance.compose(Int32Type.instance.fromComparableBytes(peekable,
ByteComparable.Version.OSS50));
+ }
+
+ private Map<Integer, Set<DecoratedKey>> buildTermMap()
+ {
+ Map<Integer, Set<DecoratedKey>> terms = new HashMap<>();
+
+ for (int count = 0; count < 10000; count++)
+ {
+ int term = getRandom().nextIntBetween(0, 100);
+ Set<DecoratedKey> pks;
+ if (terms.containsKey(term))
+ pks = terms.get(term);
+ else
+ {
+ pks = new HashSet<>();
+ terms.put(term, pks);
+ }
+ DecoratedKey key = makeKey(cfs.metadata(),
getRandom().nextIntBetween(0, 20000));
+ while (pks.contains(key))
+ key = makeKey(cfs.metadata(), getRandom().nextIntBetween(0,
20000));
+ pks.add(key);
+ }
+
+ return terms;
+ }
+
+ private void addRow(int pk, int value)
+ {
+ DecoratedKey key = makeKey(cfs.metadata(), pk);
+ memtableIndex.index(key, Clustering.EMPTY,
Int32Type.instance.decompose(value));
+ keyMap.put(key, pk);
+ }
+
+ private DecoratedKey makeKey(TableMetadata table, Integer partitionKey)
+ {
+ ByteBuffer key =
table.partitionKeyType.fromString(partitionKey.toString());
+ return table.partitioner.decorateKey(key);
+ }
+}
diff --git
a/test/unit/org/apache/cassandra/index/sai/memory/TrieMemoryIndexTest.java
b/test/unit/org/apache/cassandra/index/sai/memory/TrieMemoryIndexTest.java
index 3e512e9e23..0c396f8747 100644
--- a/test/unit/org/apache/cassandra/index/sai/memory/TrieMemoryIndexTest.java
+++ b/test/unit/org/apache/cassandra/index/sai/memory/TrieMemoryIndexTest.java
@@ -33,7 +33,6 @@ import org.HdrHistogram.Histogram;
import org.junit.Ignore;
import org.junit.Test;
-import org.apache.cassandra.cql3.Operator;
import org.apache.cassandra.cql3.statements.schema.IndexTarget;
import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.ColumnFamilyStore;
@@ -43,11 +42,7 @@ import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.dht.AbstractBounds;
-import org.apache.cassandra.dht.Bounds;
-import org.apache.cassandra.dht.ExcludingBounds;
-import org.apache.cassandra.dht.IncludingExcludingBounds;
import org.apache.cassandra.dht.Murmur3Partitioner;
-import org.apache.cassandra.dht.Range;
import org.apache.cassandra.index.sai.StorageAttachedIndex;
import org.apache.cassandra.index.sai.iterators.KeyRangeIterator;
import org.apache.cassandra.index.sai.plan.Expression;
@@ -110,9 +105,9 @@ public class TrieMemoryIndexTest extends SAIRandomizedTester
for (int executionCount = 0; executionCount < 1000; executionCount++)
{
- Expression expression = generateRandomExpression();
+ Expression expression = generateRandomExpression(this.index);
- AbstractBounds<PartitionPosition> keyRange =
generateRandomBounds(keys);
+ AbstractBounds<PartitionPosition> keyRange =
generateRandomBounds(keys, Murmur3Partitioner.instance);
Set<Integer> expectedKeys = keyMap.keySet()
.stream()
@@ -137,62 +132,6 @@ public class TrieMemoryIndexTest extends
SAIRandomizedTester
}
}
- private AbstractBounds<PartitionPosition>
generateRandomBounds(List<DecoratedKey> keys)
- {
- PartitionPosition leftBound = getRandom().nextBoolean() ?
Murmur3Partitioner.instance.getMinimumToken().minKeyBound()
- :
keys.get(getRandom().nextIntBetween(0, keys.size() -
1)).getToken().minKeyBound();
-
- PartitionPosition rightBound = getRandom().nextBoolean() ?
Murmur3Partitioner.instance.getMinimumToken().minKeyBound()
- :
keys.get(getRandom().nextIntBetween(0, keys.size() -
1)).getToken().maxKeyBound();
-
- AbstractBounds<PartitionPosition> keyRange;
-
- if (leftBound.isMinimum() && rightBound.isMinimum())
- keyRange = new Range<>(leftBound, rightBound);
- else
- {
- if (AbstractBounds.strictlyWrapsAround(leftBound, rightBound))
- {
- PartitionPosition temp = leftBound;
- leftBound = rightBound;
- rightBound = temp;
- }
- if (getRandom().nextBoolean())
- keyRange = new Bounds<>(leftBound, rightBound);
- else if (getRandom().nextBoolean())
- keyRange = new ExcludingBounds<>(leftBound, rightBound);
- else
- keyRange = new IncludingExcludingBounds<>(leftBound,
rightBound);
- }
- return keyRange;
- }
-
- private Expression generateRandomExpression()
- {
- Expression expression = Expression.create(index);
-
- int equality = getRandom().nextIntBetween(0, 100);
- int lower = getRandom().nextIntBetween(0, 75);
- int upper = getRandom().nextIntBetween(25, 100);
- while (upper <= lower)
- upper = getRandom().nextIntBetween(0, 100);
-
- if (getRandom().nextBoolean())
- expression.add(Operator.EQ,
Int32Type.instance.decompose(equality));
- else
- {
- boolean useLower = getRandom().nextBoolean();
- boolean useUpper = getRandom().nextBoolean();
- if (!useLower && !useUpper)
- useLower = useUpper = true;
- if (useLower)
- expression.add(getRandom().nextBoolean() ? Operator.GT :
Operator.GTE, Int32Type.instance.decompose(lower));
- if (useUpper)
- expression.add(getRandom().nextBoolean() ? Operator.LT :
Operator.LTE, Int32Type.instance.decompose(upper));
- }
- return expression;
- }
-
@Test
public void shouldAcceptPrefixValuesTest()
{
diff --git
a/test/unit/org/apache/cassandra/index/sai/utils/SAIRandomizedTester.java
b/test/unit/org/apache/cassandra/index/sai/utils/SAIRandomizedTester.java
index 7cff65acd8..b321280af2 100644
--- a/test/unit/org/apache/cassandra/index/sai/utils/SAIRandomizedTester.java
+++ b/test/unit/org/apache/cassandra/index/sai/utils/SAIRandomizedTester.java
@@ -18,6 +18,7 @@
package org.apache.cassandra.index.sai.utils;
import java.io.IOException;
+import java.util.List;
import com.google.common.base.Preconditions;
@@ -28,11 +29,22 @@ import org.junit.rules.TemporaryFolder;
import org.junit.rules.TestRule;
import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.cql3.Operator;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.db.marshal.Int32Type;
+import org.apache.cassandra.dht.AbstractBounds;
+import org.apache.cassandra.dht.Bounds;
+import org.apache.cassandra.dht.ExcludingBounds;
+import org.apache.cassandra.dht.IPartitioner;
+import org.apache.cassandra.dht.IncludingExcludingBounds;
import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.dht.Range;
import org.apache.cassandra.index.sai.SAITester;
+import org.apache.cassandra.index.sai.StorageAttachedIndex;
import org.apache.cassandra.index.sai.disk.format.IndexDescriptor;
import org.apache.cassandra.index.sai.disk.io.IndexFileUtils;
+import org.apache.cassandra.index.sai.plan.Expression;
import org.apache.cassandra.index.sai.postings.PostingList;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.SequenceBasedSSTableId;
@@ -199,4 +211,61 @@ public class SAIRandomizedTester extends SAITester
array[randomPosition] = temp;
}
}
+
+ public static Expression generateRandomExpression(StorageAttachedIndex
index)
+ {
+ Expression expression = Expression.create(index);
+
+ int equality = getRandom().nextIntBetween(0, 100);
+ int lower = getRandom().nextIntBetween(0, 75);
+ int upper = getRandom().nextIntBetween(25, 100);
+ while (upper <= lower)
+ upper = getRandom().nextIntBetween(0, 100);
+
+ if (getRandom().nextBoolean())
+ expression.add(Operator.EQ,
Int32Type.instance.decompose(equality));
+ else
+ {
+ boolean useLower = getRandom().nextBoolean();
+ boolean useUpper = getRandom().nextBoolean();
+ if (!useLower && !useUpper)
+ useLower = useUpper = true;
+ if (useLower)
+ expression.add(getRandom().nextBoolean() ? Operator.GT :
Operator.GTE, Int32Type.instance.decompose(lower));
+ if (useUpper)
+ expression.add(getRandom().nextBoolean() ? Operator.LT :
Operator.LTE, Int32Type.instance.decompose(upper));
+ }
+ return expression;
+ }
+
+ public static AbstractBounds<PartitionPosition>
generateRandomBounds(List<DecoratedKey> keys,
+
IPartitioner partitioner)
+ {
+ PartitionPosition leftBound = getRandom().nextBoolean() ?
partitioner.getMinimumToken().minKeyBound()
+ :
keys.get(getRandom().nextIntBetween(0, keys.size() -
1)).getToken().minKeyBound();
+
+ PartitionPosition rightBound = getRandom().nextBoolean() ?
partitioner.getMinimumToken().minKeyBound()
+ :
keys.get(getRandom().nextIntBetween(0, keys.size() -
1)).getToken().maxKeyBound();
+
+ AbstractBounds<PartitionPosition> keyRange;
+
+ if (leftBound.isMinimum() && rightBound.isMinimum())
+ keyRange = new Range<>(leftBound, rightBound);
+ else
+ {
+ if (AbstractBounds.strictlyWrapsAround(leftBound, rightBound))
+ {
+ PartitionPosition temp = leftBound;
+ leftBound = rightBound;
+ rightBound = temp;
+ }
+ if (getRandom().nextBoolean())
+ keyRange = new Bounds<>(leftBound, rightBound);
+ else if (getRandom().nextBoolean())
+ keyRange = new ExcludingBounds<>(leftBound, rightBound);
+ else
+ keyRange = new IncludingExcludingBounds<>(leftBound,
rightBound);
+ }
+ return keyRange;
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
