maedhroz commented on code in PR #4905:
URL: https://github.com/apache/cassandra/pull/4905#discussion_r3501394194
##########
test/unit/org/apache/cassandra/index/sai/memory/VectorMemoryIndexTest.java:
##########
@@ -182,18 +198,489 @@ public void randomQueryTest() throws Exception
expectedVectorsReturned += expectedResults;
if (foundKeys.size() < expectedResults)
assertTrue("Expected at least " + expectedResults + "
results but got " + foundKeys.size(),
- foundKeys.size() >= expectedResults *
expectedRecall);
+ foundKeys.size() >= expectedResults *
RECALL_THRESHOLD);
}
}
assertTrue("Expected at least " + expectedVectorsReturned + " results
but got " + actualVectorsReturned,
- actualVectorsReturned >= expectedVectorsReturned *
expectedRecall);
+ actualVectorsReturned >= expectedVectorsReturned *
RECALL_THRESHOLD);
+ }
+
+ /**
+ * Verifies that concurrent calls to add() do not corrupt the graph or
lose data.
+ * <p>
+ * GraphIndexBuilder.addGraphNode() is designed for concurrent use:
insertionsInProgress
+ * is a ConcurrentSkipListSet, and PoolingSupport gives each thread its
own GraphSearcher
+ * and scratch arrays. This test validates the full stack from
VectorMemoryIndex.index()
+ * through OnHeapGraph.add() through GraphIndexBuilder.addGraphNode().
+ * <p>
+ * Each thread owns a disjoint range of partition key integers so every
insert is a
+ * pure add with no PK collisions, isolating add() from update() semantics.
+ * <p>
+ * 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.
+ */
+ @Test
+ public void testConcurrentAddsProduceConsistentFinalState() throws
Exception
+ {
+ Memtable memtable = Mockito.mock(Memtable.class);
+ memtableIndex = new VectorMemoryIndex(index, memtable);
+
+ int numThreads = Runtime.getRuntime().availableProcessors();
+ int vectorsPerThread = 2000;
+ int totalInserted = numThreads * vectorsPerThread;
+
+ ExecutorService executor = Executors.newFixedThreadPool(numThreads);
+ // CyclicBarrier ensures all threads begin inserting simultaneously,
+ // maximizing contention on GraphIndexBuilder and
ConcurrentVectorValues.
+ CyclicBarrier barrier = new CyclicBarrier(numThreads);
+ List<Future<?>> futures = new ArrayList<>();
+
+ for (int t = 0; t < numThreads; t++)
+ {
+ final int threadId = t;
+ futures.add(executor.submit(() -> {
+ try
+ {
+ barrier.await();
+ for (int i = 0; i < vectorsPerThread; i++)
+ {
+ // Partition key is globally unique across all threads
+ int pk = threadId * vectorsPerThread + i;
+ addRow(pk, randomVectorFromThreadLocal());
Review Comment:
I think having the threads share a vector sequence would make a lot of
sense...let's try it...
##########
src/java/org/apache/cassandra/index/sai/memory/VectorMemoryIndex.java:
##########
@@ -98,11 +97,11 @@ public synchronized long add(DecoratedKey key,
Clustering<?> clustering, ByteBuf
private long index(PrimaryKey primaryKey, ByteBuffer value)
{
- updateKeyBounds(primaryKey);
-
writeCount.increment();
primaryKeys.add(primaryKey);
- return graph.add(value, primaryKey,
OnHeapGraph.InvalidVectorBehavior.FAIL);
+ long bytesUsed = graph.add(value, primaryKey,
OnHeapGraph.InvalidVectorBehavior.FAIL);
Review Comment:
Agree
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
