This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch cassandra-5.0
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/cassandra-5.0 by this push:
new bf06dbdd9a Fix concurrent SAI vector inserts failing once jvector's
per-graph pool limit is exceeded
bf06dbdd9a is described below
commit bf06dbdd9aafb6b1d4ef17f1e06ec2b720d40d2b
Author: Jeremiah Jordan <[email protected]>
AuthorDate: Tue Sep 8 15:25:54 2026 -0500
Fix concurrent SAI vector inserts failing once jvector's per-graph pool
limit is exceeded
CASSANDRA-21160 removed the synchronized keyword from
VectorMemoryIndex.add(). Concurrent inserts into a table with a
Storage-Attached Index (SAI) vector index now reach jvector's
GraphIndexBuilder.addGraphNode() in parallel.
jvector 1.0.2 pools the scratch objects that addGraphNode() uses. Each
pool holds at most Runtime.availableProcessors() + 1 objects. When more
threads than that are inside addGraphNode() at the same time, the pool
throws IllegalStateException. The mutation fails with
WriteFailureException. The failed insert also leaves a node with no
neighbors in the graph. A retry of the same row finds the existing
postings entry and does not link the node.
The default configuration reaches this limit. SkipListMemtable locks
each partition only, and concurrent_writes defaults to 32.
Bound the number of threads inside addGraphNode() in OnHeapGraph with a
semaphore of availableProcessors() + 1 permits. Writers beyond the limit
wait for a permit. Graph insertion is processor bound, so the throughput
gain from CASSANDRA-21160 stays.
Add Memtable.limitsConcurrentWritesTo(). TrieMemtable and the locking
ShardedSkipListMemtable serialize writes per shard. They return true
when their shard count is within the limit, and OnHeapGraph then creates
no semaphore.
Add a regression test that inserts from more writers than the limit. Add
a test for the memtable implementations.
patch by Jeremiah Jordan; reviewed by Caleb Rackliffe for CASSANDRA-21644
---
CHANGES.txt | 1 +
.../org/apache/cassandra/db/memtable/Memtable.java | 11 +++
.../apache/cassandra/db/memtable/Memtable_API.md | 4 +
.../db/memtable/ShardedSkipListMemtable.java | 5 +
.../apache/cassandra/db/memtable/TrieMemtable.java | 7 ++
.../index/sai/disk/v1/vector/OnHeapGraph.java | 31 ++++++-
.../memtable/MemtableConcurrentWriteLimitTest.java | 102 +++++++++++++++++++++
.../index/sai/memory/VectorMemoryIndexTest.java | 54 +++++++----
8 files changed, 197 insertions(+), 18 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 83979af9a4..8b9c00f041 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
5.0.10
+ * Fix concurrent SAI vector inserts failing once jvector's per-graph pool
limit is exceeded (CASSANDRA-21644)
* Let the commit log allocator observe the shutdown state (CASSANDRA-21616)
* Unwrap LongType properly when calculating min/max terms in V1SSTableIndex
(CASSANDRA-21635)
* Force repair should ignore min_repair_interval (CASSANDRA-21552)
diff --git a/src/java/org/apache/cassandra/db/memtable/Memtable.java
b/src/java/org/apache/cassandra/db/memtable/Memtable.java
index d4f0dedb99..dac1cd5ccd 100644
--- a/src/java/org/apache/cassandra/db/memtable/Memtable.java
+++ b/src/java/org/apache/cassandra/db/memtable/Memtable.java
@@ -196,6 +196,17 @@ public interface Memtable extends Comparable<Memtable>,
UnfilteredSource, CellSo
*/
long put(PartitionUpdate update, UpdateTransaction indexer, OpOrder.Group
opGroup);
+ /**
+ * Whether this memtable guarantees that no more than {@code maxWriters}
threads are inside {@link #put} at once,
+ * and therefore that no more than that many threads concurrently apply
the update's effect to the {@code indexer}.
+ * Memtables that serialize writes per shard can return true when they
have at most {@code maxWriters} shards.
+ * Indexes whose implementation has a limit on concurrent writers can skip
their own bounding when this is true.
+ */
+ default boolean limitsConcurrentWritesTo(int maxWriters)
+ {
+ return false;
+ }
+
// Read operations are provided by the UnfilteredSource interface.
// Statistics
diff --git a/src/java/org/apache/cassandra/db/memtable/Memtable_API.md
b/src/java/org/apache/cassandra/db/memtable/Memtable_API.md
index 70f8f0b605..ae1eda1ca7 100644
--- a/src/java/org/apache/cassandra/db/memtable/Memtable_API.md
+++ b/src/java/org/apache/cassandra/db/memtable/Memtable_API.md
@@ -135,6 +135,10 @@ implementations) are provided as the `AbstractMemtable`
(statistics tracking), `
commit log span tracking) and `AbstractAllocatorMemtable` (adds memory
management via the `Allocator` class, together
with flush triggering on memory use and time interval expiration).
+A memtable that serializes writes, for example per shard, can say so through
`limitsConcurrentWritesTo()`. Secondary
+indexes whose implementation limits the number of concurrent writers use this
to skip their own bounding when the
+memtable already provides it. The default answer is false.
+
The memtable API also gives the memtable some control over flushing and the
functioning of the commit log. The former
is there to permit memtables that operate long-term and/or can handle some
events internally, without a need to flush.
The latter enables memtables that have an internal durability mechanism, such
as ones using persistent memory or a
diff --git
a/src/java/org/apache/cassandra/db/memtable/ShardedSkipListMemtable.java
b/src/java/org/apache/cassandra/db/memtable/ShardedSkipListMemtable.java
index 92cdbbad9f..955e6b0d52 100644
--- a/src/java/org/apache/cassandra/db/memtable/ShardedSkipListMemtable.java
+++ b/src/java/org/apache/cassandra/db/memtable/ShardedSkipListMemtable.java
@@ -494,6 +494,11 @@ public class ShardedSkipListMemtable extends
AbstractShardedMemtable
}
}
+ @Override
+ public boolean limitsConcurrentWritesTo(int maxWriters)
+ {
+ return boundaries.shardCount() <= maxWriters;
+ }
}
public static Factory factory(Map<String, String> optionsCopy)
diff --git a/src/java/org/apache/cassandra/db/memtable/TrieMemtable.java
b/src/java/org/apache/cassandra/db/memtable/TrieMemtable.java
index 83b02db06a..f652c4675f 100644
--- a/src/java/org/apache/cassandra/db/memtable/TrieMemtable.java
+++ b/src/java/org/apache/cassandra/db/memtable/TrieMemtable.java
@@ -174,6 +174,13 @@ public class TrieMemtable extends AbstractShardedMemtable
}
}
+ @Override
+ public boolean limitsConcurrentWritesTo(int maxWriters)
+ {
+ // each shard applies one update at a time under its write lock
+ return boundaries.shardCount() <= maxWriters;
+ }
+
/**
* Should only be called by ColumnFamilyStore.apply via Keyspace.apply,
which supplies the appropriate
* OpOrdering.
diff --git
a/src/java/org/apache/cassandra/index/sai/disk/v1/vector/OnHeapGraph.java
b/src/java/org/apache/cassandra/index/sai/disk/v1/vector/OnHeapGraph.java
index a823943a4e..62b7c4dfa4 100644
--- a/src/java/org/apache/cassandra/index/sai/disk/v1/vector/OnHeapGraph.java
+++ b/src/java/org/apache/cassandra/index/sai/disk/v1/vector/OnHeapGraph.java
@@ -31,6 +31,10 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.stream.IntStream;
+import javax.annotation.Nullable;
+
+import com.google.common.annotations.VisibleForTesting;
+
import org.cliffc.high_scale_lib.NonBlockingHashMap;
import org.cliffc.high_scale_lib.NonBlockingHashMapLong;
import org.slf4j.Logger;
@@ -63,6 +67,7 @@ import
org.apache.cassandra.index.sai.disk.v1.segment.SegmentMetadata;
import org.apache.cassandra.io.util.SequentialWriter;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.CloseableIterator;
+import org.apache.cassandra.utils.concurrent.Semaphore;
import org.apache.lucene.util.StringHelper;
public class OnHeapGraph<T>
@@ -71,6 +76,13 @@ public class OnHeapGraph<T>
public static final int MIN_PQ_ROWS = 1024;
+ /**
+ * jvector's {@link GraphIndexBuilder#addGraphNode} throws if more than
this many threads are inside it at once
+ * (see {@code PoolingSupport}). Uses {@link Runtime} rather than {@code
DatabaseDescriptor} because that is what jvector reads.
+ */
+ @VisibleForTesting
+ public static final int MAX_CONCURRENT_GRAPH_INSERTS =
Runtime.getRuntime().availableProcessors() + 1;
+
private final RamAwareVectorValues vectorValues;
private final GraphIndexBuilder<float[]> builder;
private final VectorType<?> vectorType;
@@ -79,6 +91,10 @@ public class OnHeapGraph<T>
private final NonBlockingHashMapLong<VectorPostings<T>> postingsByOrdinal;
private final NonBlockingHashMap<T, float[]> vectorsByKey;
private final AtomicInteger nextOrdinal = new AtomicInteger();
+ // memtable inserts are concurrent (CASSANDRA-21160); excess writers wait
here rather than fail in jvector.
+ // Null when the memtable itself already limits writers to that many, e.g.
a TrieMemtable with few enough shards.
+ @Nullable
+ private final Semaphore graphInsertPermits;
private volatile boolean hasDeletions;
private String source;
@@ -107,6 +123,9 @@ public class OnHeapGraph<T>
postingsMap = new ConcurrentSkipListMap<>(Arrays::compare);
postingsByOrdinal = new NonBlockingHashMapLong<>();
vectorsByKey = memtable != null ? new NonBlockingHashMap<>() : null;
+ graphInsertPermits = memtable != null &&
memtable.limitsConcurrentWritesTo(MAX_CONCURRENT_GRAPH_INSERTS)
+ ? null
+ :
Semaphore.newSemaphore(MAX_CONCURRENT_GRAPH_INSERTS);
builder = new GraphIndexBuilder<>(vectorValues,
VectorEncoding.FLOAT32,
@@ -187,7 +206,17 @@ public class OnHeapGraph<T>
: ((CompactionVectorValues)
vectorValues).add(ordinal, term);
bytesUsed += VectorPostings.emptyBytesUsed() +
VectorPostings.bytesPerPosting();
postingsByOrdinal.put(ordinal, postings);
- bytesUsed += builder.addGraphNode(ordinal, vectorValues);
+ if (graphInsertPermits != null)
+ graphInsertPermits.acquireThrowUncheckedOnInterrupt(1);
+ try
+ {
+ bytesUsed += builder.addGraphNode(ordinal, vectorValues);
+ }
+ finally
+ {
+ if (graphInsertPermits != null)
+ graphInsertPermits.release(1);
+ }
return bytesUsed;
}
else
diff --git
a/test/unit/org/apache/cassandra/db/memtable/MemtableConcurrentWriteLimitTest.java
b/test/unit/org/apache/cassandra/db/memtable/MemtableConcurrentWriteLimitTest.java
new file mode 100644
index 0000000000..76ce3d01e1
--- /dev/null
+++
b/test/unit/org/apache/cassandra/db/memtable/MemtableConcurrentWriteLimitTest.java
@@ -0,0 +1,102 @@
+/*
+ * 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.LinkedHashMap;
+import java.util.Map;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.InheritingClass;
+import org.apache.cassandra.cql3.CQLTester;
+import org.apache.cassandra.db.ColumnFamilyStore;
+
+import static
org.apache.cassandra.db.memtable.AbstractShardedMemtable.SHARDS_OPTION;
+import static
org.apache.cassandra.db.memtable.ShardedSkipListMemtable.LOCKING_OPTION;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * {@link Memtable#limitsConcurrentWritesTo} must be true only for memtables
that serialize writes per shard, and
+ * only when the shard count is within the limit asked about.
+ */
+public class MemtableConcurrentWriteLimitTest extends CQLTester
+{
+ private static final int SHARDS = 4;
+
+ // Overrides CQLTester.setUpClass so the memtable configurations are
registered before the server is prepared
+ @BeforeClass
+ public static void setUpClass()
+ {
+ prePrepareServer();
+
+ LinkedHashMap<String, InheritingClass> memtableConfig = new
LinkedHashMap<>();
+ memtableConfig.put("skiplist", new InheritingClass(null,
SkipListMemtable.class.getName(), Map.of()));
+ memtableConfig.put("trie", new InheritingClass(null,
TrieMemtable.class.getName(), Map.of(SHARDS_OPTION, String.valueOf(SHARDS))));
+ memtableConfig.put("sharded", new InheritingClass(null,
ShardedSkipListMemtable.class.getName(), Map.of(SHARDS_OPTION,
String.valueOf(SHARDS))));
+ memtableConfig.put("sharded_locking", new InheritingClass(null,
ShardedSkipListMemtable.class.getName(), Map.of(SHARDS_OPTION,
String.valueOf(SHARDS), LOCKING_OPTION, "true")));
+ DatabaseDescriptor.getRawConfig().memtable = new
Config.MemtableOptions();
+ DatabaseDescriptor.getRawConfig().memtable.configurations =
memtableConfig;
+
+ prepareServer();
+ }
+
+ @Test
+ public void unshardedMemtableDoesNotLimitWriters()
+ {
+
assertFalse(memtableFor("skiplist").limitsConcurrentWritesTo(Integer.MAX_VALUE));
+ }
+
+ @Test
+ public void shardedSkipListWithoutLockingDoesNotLimitWriters()
+ {
+
assertFalse(memtableFor("sharded").limitsConcurrentWritesTo(Integer.MAX_VALUE));
+ }
+
+ @Test
+ public void trieMemtableLimitsWritersToShardCount()
+ {
+ assertLimitsToShardCount(memtableFor("trie"));
+ }
+
+ @Test
+ public void lockingShardedSkipListLimitsWritersToShardCount()
+ {
+ assertLimitsToShardCount(memtableFor("sharded_locking"));
+ }
+
+ private void assertLimitsToShardCount(Memtable memtable)
+ {
+ // the memtable may get fewer shards than requested, e.g. if local
ranges cannot be split that finely
+ int shards =
getCurrentColumnFamilyStore().localRangeSplits(SHARDS).shardCount();
+ assertTrue(memtable.limitsConcurrentWritesTo(shards));
+ assertTrue(memtable.limitsConcurrentWritesTo(shards + 1));
+ assertFalse(memtable.limitsConcurrentWritesTo(shards - 1));
+ }
+
+ private Memtable memtableFor(String memtableConfig)
+ {
+ createTable("CREATE TABLE %s (pk int PRIMARY KEY, v int) WITH memtable
= '" + memtableConfig + '\'');
+ ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
+ return cfs.getTracker().getView().getCurrentMemtable();
+ }
+}
diff --git
a/test/unit/org/apache/cassandra/index/sai/memory/VectorMemoryIndexTest.java
b/test/unit/org/apache/cassandra/index/sai/memory/VectorMemoryIndexTest.java
index 3aa10bdd6b..91ea00ad75 100644
--- a/test/unit/org/apache/cassandra/index/sai/memory/VectorMemoryIndexTest.java
+++ b/test/unit/org/apache/cassandra/index/sai/memory/VectorMemoryIndexTest.java
@@ -74,6 +74,7 @@ import org.apache.cassandra.dht.Range;
import org.apache.cassandra.index.sai.QueryContext;
import org.apache.cassandra.index.sai.SAITester;
import org.apache.cassandra.index.sai.StorageAttachedIndex;
+import org.apache.cassandra.index.sai.disk.v1.vector.OnHeapGraph;
import org.apache.cassandra.index.sai.disk.v1.vector.PrimaryKeyWithScore;
import org.apache.cassandra.index.sai.plan.Expression;
import org.apache.cassandra.index.sai.utils.PrimaryKey;
@@ -102,6 +103,7 @@ public class VectorMemoryIndexTest extends SAITester
private static final double RECALL_THRESHOLD = 0.9;
private static final int VECTORS_PER_THREAD = 2000;
+ private static final int VECTORS_PER_THREAD_BEYOND_POOL_CAP = 500; // more
writers, so fewer vectors each
private ColumnFamilyStore cfs;
private StorageAttachedIndex index;
@@ -138,8 +140,7 @@ public class VectorMemoryIndexTest extends SAITester
public void randomQueryTest() throws Exception
{
// A non-null memtable tells it to track the mapping from primary key
to vector, needed for brute force search
- Memtable memtable = Mockito.mock(Memtable.class);
- memtableIndex = new VectorMemoryIndex(index, memtable);
+ memtableIndex = new VectorMemoryIndex(index, mockMemtable(1));
for (int row = 0; row < getRandom().nextIntBetween(1000, 5000); row++)
{
@@ -242,13 +243,26 @@ public class VectorMemoryIndexTest extends SAITester
@Test
public void testConcurrentAddsWithRandomVectors() throws Exception
{
- testConcurrentAddsAreEventuallyConsistent((threadId, i) ->
randomVectorFromThreadLocal());
+
testConcurrentAddsAreEventuallyConsistent(Runtime.getRuntime().availableProcessors(),
VECTORS_PER_THREAD, (threadId, i) -> randomVectorFromThreadLocal());
}
@Test
public void testConcurrentAddsWithSharedVectors() throws Exception
{
- testConcurrentAddsAreEventuallyConsistent((threadId, i) ->
makeSharedVector(i));
+
testConcurrentAddsAreEventuallyConsistent(Runtime.getRuntime().availableProcessors(),
VECTORS_PER_THREAD, (threadId, i) -> makeSharedVector(i));
+ }
+
+ /**
+ * More writers than jvector's GraphIndexBuilder can serve at once must
wait, not fail the insert. The other
+ * concurrent tests use exactly availableProcessors writers, within
jvector's limit, so their memtable reports that
+ * it bounds writers and OnHeapGraph skips its semaphore; this test's
memtable cannot make that promise, so
+ * OnHeapGraph bounds the writers itself.
+ */
+ @Test
+ public void testConcurrentAddsExceedingJVectorPoolCap() throws Exception
+ {
+ int numThreads = 2 * OnHeapGraph.MAX_CONCURRENT_GRAPH_INSERTS;
+ testConcurrentAddsAreEventuallyConsistent(numThreads,
VECTORS_PER_THREAD_BEYOND_POOL_CAP, (threadId, i) ->
randomVectorFromThreadLocal());
}
/**
@@ -262,13 +276,11 @@ public class VectorMemoryIndexTest extends SAITester
* After all writers complete, a full-ring search must return the vast
majority of
* inserted keys with valid scores, confirming no data was lost or
corrupted.
*/
- private void testConcurrentAddsAreEventuallyConsistent(BiFunction<Integer,
Integer, ByteBuffer> vectorFactory) throws Exception
+ private void testConcurrentAddsAreEventuallyConsistent(int numThreads, int
vectorsPerThread, BiFunction<Integer, Integer, ByteBuffer> vectorFactory)
throws Exception
{
- Memtable memtable = Mockito.mock(Memtable.class);
- memtableIndex = new VectorMemoryIndex(index, memtable);
+ memtableIndex = new VectorMemoryIndex(index, mockMemtable(numThreads));
- int numThreads = Runtime.getRuntime().availableProcessors();
- int totalInserted = numThreads * VECTORS_PER_THREAD;
+ int totalInserted = numThreads * vectorsPerThread;
ExecutorService executor = Executors.newFixedThreadPool(numThreads);
@@ -284,9 +296,9 @@ public class VectorMemoryIndexTest extends SAITester
try
{
barrier.await();
- for (int i = 0; i < VECTORS_PER_THREAD; i++)
+ for (int i = 0; i < vectorsPerThread; i++)
{
- int pk = threadId * VECTORS_PER_THREAD + i;
+ int pk = threadId * vectorsPerThread + i;
addRow(pk, vectorFactory.apply(threadId, i));
}
}
@@ -376,11 +388,9 @@ public class VectorMemoryIndexTest extends SAITester
*/
public void testConcurrentAddsAndOrderByNeverThrow(BiFunction<Integer,
Integer, ByteBuffer> vectorFactory) throws Exception
{
- Memtable memtable = Mockito.mock(Memtable.class);
- memtableIndex = new VectorMemoryIndex(index, memtable);
-
int numWriterThreads = Runtime.getRuntime().availableProcessors();
int numReaderThreads = Runtime.getRuntime().availableProcessors();
+ memtableIndex = new VectorMemoryIndex(index,
mockMemtable(numWriterThreads));
int totalInserted = numWriterThreads * VECTORS_PER_THREAD;
// Pre-seed enough rows that orderBy() always has a non-empty graph to
search,
@@ -540,11 +550,9 @@ public class VectorMemoryIndexTest extends SAITester
*/
private void
testConcurrentAddsAndOrderResultsByNeverThrow(BiFunction<Integer, Integer,
ByteBuffer> vectorFactory) throws Exception
{
- Memtable memtable = Mockito.mock(Memtable.class);
- memtableIndex = new VectorMemoryIndex(index, memtable);
-
int numWriterThreads = Runtime.getRuntime().availableProcessors();
int numReaderThreads = Runtime.getRuntime().availableProcessors();
+ memtableIndex = new VectorMemoryIndex(index,
mockMemtable(numWriterThreads));
int totalInserted = numWriterThreads * VECTORS_PER_THREAD;
// Pre-seed rows so orderResultsBy() always has a non-empty
[minimumKey, maximumKey]
@@ -757,6 +765,18 @@ public class VectorMemoryIndexTest extends SAITester
keyMap.put(key, pk);
}
+ /**
+ * A memtable that answers {@link Memtable#limitsConcurrentWritesTo}
truthfully for the number of writer threads the
+ * test will use. With that many writers at or under {@link
OnHeapGraph#MAX_CONCURRENT_GRAPH_INSERTS}, OnHeapGraph
+ * relies on the memtable and creates no semaphore; with more, it bounds
the writers itself.
+ */
+ private static Memtable mockMemtable(int writers)
+ {
+ Memtable memtable = Mockito.mock(Memtable.class);
+
Mockito.when(memtable.limitsConcurrentWritesTo(Mockito.anyInt())).thenAnswer(invocation
-> writers <= (int) invocation.getArgument(0));
+ return memtable;
+ }
+
private DecoratedKey makeKey(TableMetadata table, Integer partitionKey)
{
ByteBuffer key =
table.partitionKeyType.fromString(partitionKey.toString());
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
