dcapwell commented on code in PR #4905:
URL: https://github.com/apache/cassandra/pull/4905#discussion_r3501253903
##########
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 feel that `randomVectorFromThreadLocal` isn't the best option here. We
now allow `add` to overlap which means all protection from that is now lost.
So we should also include *distinct* PKs that haver the *same* vectors... In
the common case all vectors are not equal so `OnHeapGraph.add()` almost never
tests the branch where a conflict happens.
what if we ran this test twice, but this logic is pluggable? one run does
random like you do and the other creates vectors like this
```
[0.5, 05, ..., <i/vectorsPerThread>]
```
with that encoding each thread *will* produce the same vectors, allowing
this race to happen?
--
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]
