This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 1f17ab0ede1 Reopen the mutable HNSW searcher on a background thread
instead of on the query path (#19467)
1f17ab0ede1 is described below
commit 1f17ab0ede12e2703e4a037837d4c641c4756d11
Author: Xiang Fu <[email protected]>
AuthorDate: Sat Sep 5 18:37:33 2026 -0700
Reopen the mutable HNSW searcher on a background thread instead of on the
query path (#19467)
* Reopen the mutable HNSW searcher on a background thread instead of on the
query path
A filtered vector search on a consuming segment must see every row the
query's visible-document set names,
including rows still in the writer's RAM buffer, so it reopened the
near-real-time searcher itself. That reopen
flushes the writer and builds an HNSW graph for the flushed rows, and under
continuous ingestion nearly every
filtered query found new rows -- so segment creation was driven by query
rate rather than by ingestion.
Queries now publish the writer generation they need and wait; one
background thread per index performs the
reopen. Concurrent queries needing the same generation share one flush
instead of forcing one each. Measured with
the workload added here at 2000 docs/s and 8 reader threads, against the
same harness on the previous behavior:
throughput 1104 -> 1645 QPS, p50 4.0 -> 3.2ms, p95 14.7 -> 9.5ms, p99 61.3
-> 34.4ms, with roughly a quarter of
the reopens.
Lucene's ControlledRealTimeReopenThread is deliberately not used. It
publishes its searching generation from a
RefreshListener that ReferenceManager invokes from a finally block, so a
reopen that threw still advertises the
generation it merely attempted -- a query would then search a stale
searcher and silently drop rows its filter
names. Its loop also reopens on a fixed cadence whether or not anyone is
waiting, which on a consuming segment
always flushes because rows are always buffered. The loop here publishes
only after a reopen returns normally,
and runs only while a query is waiting.
- Fail the query, rather than answer it from a searcher that may not hold
its rows, when a reopen fails or the
wait exceeds refreshWaitTimeoutMs.
- Space reopens by refreshMinIntervalMs, and back off with capped retries
after a failure so a persistently
failing writer cannot spin the loop or flood the log.
- Add refreshMinIntervalMs and refreshWaitTimeoutMs to
VectorIndexConfigValidator, parsing exactly as the
consumer does so a config it accepts cannot then halt ingestion on the
server.
- Cover the reopen-failure and timeout contracts, the generation handshake
under concurrency, and that close()
stops the thread and releases waiters.
* Hold the reopen spacing across wakeups
Every arriving query notifies the reopen thread, so the single timed wait
returned on that notification and
reopened early -- leaving refreshMinIntervalMs unenforced exactly under the
load it exists to bound. Measured
with the added test: 11 reopens inside a window that should hold 1.
Also read the requested generation after the wait rather than before, so a
reopen serves the newest request
rather than whichever one woke the loop, and cover the interrupted-wait
contract.
---
.../pinot/perf/BenchmarkVectorFilterWorkloads.java | 169 +++++++++
.../realtime/impl/vector/MutableVectorIndex.java | 307 ++++++++++++----
.../impl/vector/MutableVectorIndexTest.java | 395 ++++++++++++++++-----
.../index/creator/VectorIndexConfigValidator.java | 30 +-
.../creator/VectorIndexConfigValidatorTest.java | 46 +++
5 files changed, 788 insertions(+), 159 deletions(-)
diff --git
a/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkVectorFilterWorkloads.java
b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkVectorFilterWorkloads.java
index ab950a67099..a23e67a020f 100644
---
a/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkVectorFilterWorkloads.java
+++
b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkVectorFilterWorkloads.java
@@ -22,11 +22,22 @@ import java.io.File;
import java.io.IOException;
import java.io.PrintStream;
import java.nio.file.Files;
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
import java.util.Random;
import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.commons.io.FileUtils;
+import
org.apache.pinot.segment.local.realtime.impl.invertedindex.RealtimeLuceneTextIndexSearcherPool;
+import org.apache.pinot.segment.local.realtime.impl.vector.MutableVectorIndex;
import
org.apache.pinot.segment.local.segment.index.readers.vector.IvfFlatVectorIndexReader;
import
org.apache.pinot.segment.local.segment.index.readers.vector.IvfOnDiskVectorIndexReader;
import org.apache.pinot.segment.local.segment.index.vector.IvfCombinedBuffers;
@@ -89,6 +100,164 @@ public final class BenchmarkVectorFilterWorkloads {
benchmarkExactBaselines(out);
benchmarkFilterAwareAnn(out);
benchmarkApproximateRadius(out);
+ benchmarkConcurrentMutableIngestAndQuery(out);
+ }
+
+ /// Filtered search against a consuming segment while that segment is still
ingesting. This is the case the
+ /// near-real-time reopen exists for, and the one whose cost is invisible to
every other scenario here: the
+ /// others index everything up front, so their searcher is never stale and
never reopens.
+ ///
+ /// Reported alongside throughput is the reopen count. Queries do not reopen
the searcher themselves -- they wait
+ /// for a background thread -- so the gap between query count and reopen
count is the sharing that keeps a high
+ /// query rate from flushing the writer once per query.
+ private static void benchmarkConcurrentMutableIngestAndQuery(PrintStream out)
+ throws Exception {
+ out.println();
+ out.println("=== Concurrent Mutable Ingest + Filtered Query ===");
+ out.printf("%-28s %14s %10s %10s %10s %10s%n", "Workload",
"ingest(docs/s)", "QPS", "p50(us)", "p95(us)",
+ "p99(us)");
+
+ int seedDocs = 4000;
+ int readerThreads = 8;
+ long ingestPerSec = Long.getLong("pinot.perf.vector.filters.ingestPerSec",
2000L);
+ long runMs = Long.getLong("pinot.perf.vector.filters.concurrentRunMs",
10_000L);
+ long warmupMs =
Long.getLong("pinot.perf.vector.filters.concurrentWarmupMs", 3_000L);
+
+ RealtimeLuceneTextIndexSearcherPool.init(readerThreads * 2);
+ Map<String, String> properties = new HashMap<>();
+ properties.put("vectorIndexType", "HNSW");
+ properties.put("vectorDimension", String.valueOf(DIMENSION));
+ properties.put("commitDocs", String.valueOf(Integer.MAX_VALUE));
+ properties.put("commitIntervalMs",
String.valueOf(TimeUnit.DAYS.toMillis(1)));
+ VectorIndexConfig config = new VectorIndexConfig(false, "HNSW", DIMENSION,
1,
+ VectorIndexConfig.VectorDistanceFunction.EUCLIDEAN, properties);
+ MutableVectorIndex index =
+ new MutableVectorIndex("benchmarkConcurrentVector_" +
System.nanoTime(), COLUMN_NAME, config);
+ try {
+ Random seeder = new Random(SEED);
+ for (int i = 0; i < seedDocs; i++) {
+ index.add(boxed(BenchmarkVectorIndex.generateGaussianVectors(1,
DIMENSION, seeder.nextLong())[0]), null, i);
+ }
+ // ~1% of the seeded rows, which is the selectivity range where a
filtered scan is worth doing at all.
+ MutableRoaringBitmap allowed = new MutableRoaringBitmap();
+ for (int i = 0; i < seedDocs; i += 100) {
+ allowed.add(i);
+ }
+
+ AtomicBoolean stop = new AtomicBoolean();
+ AtomicLong ingested = new AtomicLong();
+ AtomicLong queried = new AtomicLong();
+ List<List<Long>> perReaderLatencies = new ArrayList<>();
+ Thread writer = new Thread(() -> {
+ Random wr = new Random(SEED + 1);
+ int docId = seedDocs;
+ long intervalNs = ingestPerSec > 0 ? 1_000_000_000L / ingestPerSec :
0L;
+ long next = System.nanoTime();
+ while (!stop.get()) {
+ index.add(boxed(BenchmarkVectorIndex.generateGaussianVectors(1,
DIMENSION, wr.nextLong())[0]), null,
+ docId++);
+ ingested.incrementAndGet();
+ if (intervalNs > 0) {
+ next += intervalNs;
+ long sleepNs = next - System.nanoTime();
+ if (sleepNs > 0) {
+ try {
+ Thread.sleep(sleepNs / 1_000_000L, (int) (sleepNs %
1_000_000L));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ }
+ }
+ }
+ }, "vector-bench-writer");
+
+ List<Thread> readers = new ArrayList<>(readerThreads);
+ CountDownLatch ready = new CountDownLatch(readerThreads);
+ for (int r = 0; r < readerThreads; r++) {
+ List<Long> latencies = Collections.synchronizedList(new ArrayList<>());
+ perReaderLatencies.add(latencies);
+ int readerId = r;
+ readers.add(new Thread(() -> {
+ Random qr = new Random(SEED + 100 + readerId);
+ ready.countDown();
+ while (!stop.get()) {
+ float[] query = BenchmarkVectorIndex.generateGaussianVectors(1,
DIMENSION, qr.nextLong())[0];
+ long start = System.nanoTime();
+ index.getDocIds(query, TOP_K, allowed);
+ latencies.add(System.nanoTime() - start);
+ queried.incrementAndGet();
+ }
+ }, "vector-bench-reader-" + readerId));
+ }
+
+ writer.start();
+ readers.forEach(Thread::start);
+ ready.await();
+ // Discard the warmup window: the first queries pay class loading, JIT,
and the first graph build.
+ Thread.sleep(warmupMs);
+ long queriesAtWarmupEnd = queried.get();
+ long ingestedAtWarmupEnd = ingested.get();
+ // Counters are cumulative from construction, so snapshot here: only the
delta over the measured window is
+ // comparable with the query count reported below.
+ Map<String, Object> debugAtWarmupEnd = index.getIndexDebugInfo();
+ long reopensAtWarmupEnd = ((Number)
debugAtWarmupEnd.get("searcherRefreshCount")).longValue();
+ long waitsAtWarmupEnd = ((Number)
debugAtWarmupEnd.get("searcherRefreshWaitCount")).longValue();
+ List<List<Long>> measured = new ArrayList<>();
+ for (List<Long> latencies : perReaderLatencies) {
+ synchronized (latencies) {
+ latencies.clear();
+ }
+ measured.add(latencies);
+ }
+ long startNs = System.nanoTime();
+ Thread.sleep(runMs);
+ long elapsedMs = (System.nanoTime() - startNs) / 1_000_000L;
+ long queries = queried.get() - queriesAtWarmupEnd;
+ long docs = ingested.get() - ingestedAtWarmupEnd;
+ stop.set(true);
+ writer.join(TimeUnit.SECONDS.toMillis(30));
+ for (Thread reader : readers) {
+ reader.join(TimeUnit.SECONDS.toMillis(30));
+ }
+ // Read after the threads stop, so the counters cover exactly the
measured window.
+ Map<String, Object> debugAtEnd = index.getIndexDebugInfo();
+ long reopens = ((Number)
debugAtEnd.get("searcherRefreshCount")).longValue() - reopensAtWarmupEnd;
+ long waits = ((Number)
debugAtEnd.get("searcherRefreshWaitCount")).longValue() - waitsAtWarmupEnd;
+
+ List<Long> all = new ArrayList<>();
+ for (List<Long> latencies : measured) {
+ synchronized (latencies) {
+ all.addAll(latencies);
+ }
+ }
+ Collections.sort(all);
+ out.printf("%-28s %14.1f %10.1f %10.1f %10.1f %10.1f%n",
+ ingestPerSec + "/s_" + readerThreads + "_readers", docs * 1000.0 /
elapsedMs, queries * 1000.0 / elapsedMs,
+ percentileMicros(all, 0.50), percentileMicros(all, 0.95),
percentileMicros(all, 0.99));
+ // Deltas over the measured window, not the cumulative counters: only
these are comparable with the query
+ // count above. waits/reopens is the sharing -- how many queries one
reopen served.
+ out.printf(" reopens=%d waits=%d luceneSegments=%s (waits per reopen is
the sharing)%n", reopens, waits,
+ debugAtEnd.get("luceneSegments"));
+ } finally {
+ index.close();
+ }
+ }
+
+ private static double percentileMicros(List<Long> sortedNanos, double
percentile) {
+ if (sortedNanos.isEmpty()) {
+ return 0.0;
+ }
+ int idx = Math.min(sortedNanos.size() - 1, (int) (sortedNanos.size() *
percentile));
+ return sortedNanos.get(idx) / 1000.0;
+ }
+
+ private static Float[] boxed(float[] values) {
+ Float[] boxedValues = new Float[values.length];
+ for (int i = 0; i < values.length; i++) {
+ boxedValues[i] = values[i];
+ }
+ return boxedValues;
}
private static void benchmarkExactBaselines(PrintStream out) {
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/vector/MutableVectorIndex.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/vector/MutableVectorIndex.java
index d7ecb1b5e04..f85b263a383 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/vector/MutableVectorIndex.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/vector/MutableVectorIndex.java
@@ -29,6 +29,8 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
import javax.annotation.Nullable;
import org.apache.commons.io.FileUtils;
import org.apache.lucene.document.Document;
@@ -71,7 +73,6 @@ import org.slf4j.LoggerFactory;
/// Every added document stores the supplied Pinot doc id as a
[NumericDocValuesField]. The same doc values
/// drive both filtered traversal and the translation of search hits, so no
assumption is made that
/// `ScoreDoc.doc == Pinot docId` (Lucene may renumber on merges).
-
///
/// Filtered search ([#getDocIds(float[], int, ImmutableRoaringBitmap)])
restricts HNSW candidate generation
/// to the given Pinot doc ids (used to enforce the upsert doc-ids snapshot).
It searches a near-real-time
@@ -80,16 +81,16 @@ import org.slf4j.LoggerFactory;
/// unfiltered path keeps searching the last committed generation (cheaper;
commit cadence is controlled by
/// `commitIntervalMs` / `commitDocs`).
///
-/// **Cost of that freshness.** A refresh flushes the writer's RAM buffer,
which for a vector field writes out
-/// the HNSW graph for the pending rows as a new Lucene segment. While a
segment is actively consuming, nearly
-/// every filtered query finds new rows and therefore triggers one, so segment
creation is driven by query rate
-/// rather than ingestion rate, and the resulting merges rebuild graphs.
Concurrent callers coalesce onto a
-/// single refresh rather than each forcing their own. Bounding this properly
means reopening on a background
-/// thread and having queries wait on a sequence number instead of driving the
flush themselves -- see
-/// `RealtimeLuceneIndexRefreshManager`, which does that for the text indexes,
and Lucene's
-/// `ControlledRealTimeReopenThread`. Until then, correctness is preserved at
the cost of that flush.
+/// **Cost of that freshness.** A reopen flushes the writer's RAM buffer,
which writes a new Lucene segment and
+/// builds an HNSW graph over the rows it contains. Query threads never pay
that cost directly: a filtered query
+/// publishes the writer generation it needs and waits, and a single
background thread performs the reopen, so
+/// concurrent queries needing the same generation share one flush rather than
forcing one each. Reopens happen
+/// only while a query is waiting -- never on a timer -- and are spaced at
least `refreshMinIntervalMs` apart, so
+/// segment creation stays bounded by ingestion rather than by query rate. A
query waits at most
+/// `refreshWaitTimeoutMs`, then fails rather than answering from a searcher
that may not hold its rows.
///
-/// This class is thread-safe for single writer multiple readers.
+/// This class is thread-safe for a single writer and multiple readers, plus
one background reopen thread that
+/// this instance owns and [#close()] stops.
public class MutableVectorIndex
implements FilterAwareVectorIndexReader, MutableIndex,
VectorIndexConfigProvider, EfSearchAware {
private static final Logger LOGGER =
LoggerFactory.getLogger(MutableVectorIndex.class);
@@ -99,6 +100,21 @@ public class MutableVectorIndex
public static final String VECTOR_INDEX_DOC_ID_COLUMN_NAME = "DocID";
public static final long DEFAULT_COMMIT_INTERVAL_MS = 10_000L;
public static final long DEFAULT_COMMIT_DOCS = 1000L;
+ public static final String REFRESH_MIN_INTERVAL_MS = "refreshMinIntervalMs";
+ public static final String REFRESH_WAIT_TIMEOUT_MS = "refreshWaitTimeoutMs";
+ /// Minimum spacing between reopens, and equally the freshness delay a
filtered query can pay. Raising it trades
+ /// query latency for fewer writer flushes; 0 disables the spacing.
+ ///
+ /// The default is deliberately small because most of the reduction comes
from sharing rather than from spacing:
+ /// one reopen serves every query waiting on it, so reopens fall roughly by
the number of concurrent readers
+ /// whatever this is set to. Measured at 2000 docs/s with 8 reader threads,
raising it from 1ms to 10ms cut
+ /// reopens a further 4x but cost 4x the throughput and 4.5x the p50,
landing well below the per-query-refresh
+ /// behaviour it replaces. At 1ms it measured better than that behaviour on
throughput, p50, p95 and p99 while
+ /// still performing about a quarter of its reopens.
+ public static final long DEFAULT_REFRESH_MIN_INTERVAL_MS = 1L;
+ /// How long a filtered query waits for a reopen before failing. Bounded on
purpose: an unbounded wait would let
+ /// a stalled reopen pin query threads of the shared searcher pool
indefinitely.
+ public static final long DEFAULT_REFRESH_WAIT_TIMEOUT_MS = 5_000L;
private final int _vectorDimension;
private final VectorIndexConfig _vectorIndexConfig;
private final VectorSimilarityFunction _vectorSimilarityFunction;
@@ -110,24 +126,47 @@ public class MutableVectorIndex
private final FSDirectory _indexDirectory;
private final IndexWriter _indexWriter;
// Near-real-time searcher over the writer, used by the filtered search path
(upsert doc-ids snapshot
- // enforcement) so uncommitted rows are visible; refreshed on demand, reused
across queries
+ // enforcement) so uncommitted rows are visible; reopened by _reopenThread,
reused across queries
private final SearcherManager _searcherManager;
- // Coordinates callers that need the same near-real-time generation. One
caller refreshes while the others wait
- // for its published watermark instead of serializing on
maybeRefreshBlocking or reopening the same generation.
- private final Object _searcherRefreshMonitor = new Object();
- // Guarded by _searcherRefreshMonitor.
- private boolean _searcherRefreshInProgress;
- private volatile long _searcherRefreshCount;
+ /// Performs every near-real-time reopen on one background thread. Query
threads never flush the writer
+ /// themselves: they publish the generation they need and block until this
thread has reopened past it, so N
+ /// concurrent queries share one reopen instead of forcing N.
+ ///
+ /// Deliberately hand-rolled rather than delegating to Lucene's
`ControlledRealTimeReopenThread`, for two reasons
+ /// that both bear on correctness here. That class publishes its searching
generation from a
+ /// `ReferenceManager.RefreshListener`, which
`ReferenceManager#doMaybeRefresh` invokes from a finally block --
+ /// so a reopen that *threw* still advertises the generation it merely
attempted, and a filtered query would then
+ /// search a stale searcher and silently drop rows its filter names. And its
reopen loop refreshes on a fixed
+ /// cadence whether or not anyone is waiting, which on a consuming segment
is never a no-op: rows are always
+ /// buffered, so every tick flushes and builds a graph even for a table that
runs no filtered query at all.
+ /// The loop below publishes only after a reopen returns normally, and runs
only when a query is waiting.
+ private final Thread _reopenThread;
+ private final long _refreshMinIntervalMs;
+ private final long _refreshWaitTimeoutMs;
+ /// Guards the reopen handshake: the requested and reached generations, the
failure record, and the closed flag.
+ private final Object _refreshMonitor = new Object();
+ /// Highest generation any waiting query has asked for. Guarded by
[#_refreshMonitor].
+ private long _requestedSequenceNumber = -1;
+ /// Highest generation the shared searcher is known to cover. Published only
after a reopen returns normally, so
+ /// a failed reopen can never make a query believe it is looking at rows the
searcher does not have.
+ private volatile long _refreshedThroughSequenceNumber = -1;
+ /// Last reopen failure and a count of them, so a query blocked across a
failure fails instead of waiting out its
+ /// timeout. Guarded by [#_refreshMonitor].
+ private Throwable _reopenFailure;
+ private long _reopenFailureCount;
+ private boolean _closed;
+ private final AtomicLong _searcherRefreshCount = new AtomicLong();
+ private final AtomicLong _searcherRefreshWaitCount = new AtomicLong();
// Number of documents added so far; used only for the commit cadence, never
as a doc id. Written by the indexing
// thread only; read cross-thread for debug output, where staleness is
acceptable.
private volatile int _numDocsAdded;
- /// Sequence number of the newest row handed to the writer, and the sequence
number the shared searcher was
- /// last refreshed through. Together they let a filtered search skip the
refresh when nothing has been added
- /// since. Doc ids cannot be used for this: [MutableIndex#add] allows rows
in arbitrary doc-id order, so a
- /// doc-id watermark would skip the refresh a lower-numbered but newer row
needs. Writer sequence numbers are
- /// monotonic by construction. Written by the single indexing thread, read
by query threads.
+ /// Sequence number of the newest row handed to the writer. A filtered
search waits for the reopen thread to
+ /// pass this value, and skips waiting entirely when the searcher is already
past it. Doc ids cannot be used
+ /// for this: [MutableIndex#add] allows rows in arbitrary doc-id order, so a
doc-id watermark would skip the
+ /// wait a lower-numbered but newer row needs. Writer sequence numbers are
monotonic by construction, and share
+ /// the space that `IndexWriter#getMaxCompletedSequenceNumber` reports,
which is what the reopen thread
+ /// publishes as its searching generation. Written by the single indexing
thread, read by query threads.
private volatile long _lastAddedSequenceNumber = -1;
- private volatile long _searcherRefreshedThroughSequenceNumber = -1;
private long _lastCommitTime;
private final ThreadLocal<Integer> _efSearchOverride = new ThreadLocal<>();
@@ -143,6 +182,14 @@ public class MutableVectorIndex
vectorIndexConfig.getProperties().getOrDefault("commitIntervalMs",
String.valueOf(DEFAULT_COMMIT_INTERVAL_MS)));
_commitDocs = Long.parseLong(
vectorIndexConfig.getProperties().getOrDefault("commitDocs",
String.valueOf(DEFAULT_COMMIT_DOCS)));
+ _refreshMinIntervalMs = Long.parseLong(vectorIndexConfig.getProperties()
+ .getOrDefault(REFRESH_MIN_INTERVAL_MS,
String.valueOf(DEFAULT_REFRESH_MIN_INTERVAL_MS)));
+ _refreshWaitTimeoutMs = Long.parseLong(vectorIndexConfig.getProperties()
+ .getOrDefault(REFRESH_WAIT_TIMEOUT_MS,
String.valueOf(DEFAULT_REFRESH_WAIT_TIMEOUT_MS)));
+ Preconditions.checkArgument(_refreshMinIntervalMs >= 0, "Require %s >= 0,
got %s for column: %s",
+ REFRESH_MIN_INTERVAL_MS, _refreshMinIntervalMs, vectorColumn);
+ Preconditions.checkArgument(_refreshWaitTimeoutMs > 0, "Require %s > 0,
got %s for column: %s",
+ REFRESH_WAIT_TIMEOUT_MS, _refreshWaitTimeoutMs, vectorColumn);
_vectorSimilarityFunction =
VectorIndexUtils.toSimilarityFunction(vectorIndexConfig.getVectorDistanceFunction());
// Each column of a segment gets its own directory, so that cleaning up
one column does not remove the index of
// another column of the same segment.
@@ -152,6 +199,7 @@ public class MutableVectorIndex
FSDirectory indexDirectory = null;
IndexWriter indexWriter = null;
SearcherManager searcherManager = null;
+ Thread reopenThread = null;
try {
// segment generation is always in V1 and later we convert (as part of
post creation processing)
// to V3 if segmentVersion is set to V3 in SegmentGeneratorConfig.
@@ -165,6 +213,8 @@ public class MutableVectorIndex
VectorIndexUtils.getIndexWriterConfig(vectorIndexConfig).setOpenMode(IndexWriterConfig.OpenMode.CREATE));
indexWriter.commit();
searcherManager = new SearcherManager(indexWriter, false, false, null);
+ reopenThread = new Thread(this::reopenLoop, "vector-nrt-reopen-" +
segmentName + "-" + vectorColumn);
+ reopenThread.setDaemon(true);
_lastCommitTime = System.currentTimeMillis();
} catch (Exception e) {
// IndexWriter does not close the Directory passed to it, so both need
to be closed.
@@ -180,6 +230,18 @@ public class MutableVectorIndex
_indexDirectory = indexDirectory;
_indexWriter = indexWriter;
_searcherManager = searcherManager;
+ _reopenThread = reopenThread;
+ // Started last: the loop reads final fields assigned above, so publishing
`this` to another thread any
+ // earlier would race construction. Guarded because this is where
OutOfMemoryError: unable to create native
+ // thread lands, and by now nothing else holds a reference that could
close the writer -- which would keep
+ // write.lock on a directory whose name is deterministic, so every retry
of this segment would then fail.
+ try {
+ _reopenThread.start();
+ } catch (Throwable t) {
+ IOUtils.closeWhileHandlingException(_searcherManager, _indexWriter,
_indexDirectory);
+ deleteIndexDir();
+ throw t;
+ }
}
@Override
@@ -314,6 +376,11 @@ public class MutableVectorIndex
info.put("effectiveHnswUseRelativeDistance",
getEffectiveUseRelativeDistance());
info.put("effectiveHnswUseBoundedQueue", getEffectiveUseBoundedQueue());
info.put("supportsPreFilter", supportsPreFilter());
+ info.put(REFRESH_MIN_INTERVAL_MS, _refreshMinIntervalMs);
+ info.put(REFRESH_WAIT_TIMEOUT_MS, _refreshWaitTimeoutMs);
+ // Reopens vs. the queries that had to wait for one: the gap between them
is the sharing this path relies on.
+ info.put("searcherRefreshCount", _searcherRefreshCount.get());
+ info.put("searcherRefreshWaitCount", _searcherRefreshWaitCount.get());
try (DirectoryReader directoryReader =
DirectoryReader.open(_indexDirectory)) {
info.put("numDocs", directoryReader.numDocs());
info.put("numDeletedDocs", directoryReader.numDeletedDocs());
@@ -333,14 +400,9 @@ public class MutableVectorIndex
throws IOException {
if (preFilterBitmap != null) {
// Filtered search enforces the query's visible-document set, so it must
see every row that set names --
- // including rows still in the writer's RAM buffer. Refreshing flushes
the writer, so only refresh when this
- // query can actually see past the last refresh, and coalesce callers
waiting for the same generation. The
- // added-doc watermark is read BEFORE refreshing so rows arriving during
the refresh are not wrongly claimed
- // as visible.
- long lastAdded = _lastAddedSequenceNumber;
- if (lastAdded > _searcherRefreshedThroughSequenceNumber) {
- refreshSearcherThrough(lastAdded);
- }
+ // including rows still in the writer's RAM buffer. The watermark is
read BEFORE waiting, so rows arriving
+ // during the reopen are not wrongly claimed as visible.
+ awaitSearcherGeneration(_lastAddedSequenceNumber);
IndexSearcher indexSearcher = _searcherManager.acquire();
try {
return search(indexSearcher, vector, topK, efSearch,
useRelativeDistance, useBoundedQueue,
@@ -356,64 +418,166 @@ public class MutableVectorIndex
}
}
- /// Refreshes the shared near-real-time searcher through
`targetSequenceNumber`, coalescing concurrent callers.
+ /// Blocks until the shared searcher is known to cover
`targetSequenceNumber`.
///
- /// The winning caller publishes the exact generation it captured before
refreshing and wakes the waiters. A
- /// waiter whose target is newer then performs the next refresh. This is
necessary because rows added during a
- /// refresh are not guaranteed to be visible in the reopened reader.
- private void refreshSearcherThrough(long targetSequenceNumber)
+ /// The reopen itself runs on [#_reopenThread]; this only publishes the
generation needed and waits. That is the
+ /// point of the indirection -- every concurrent caller needing the same or
an older generation is satisfied by
+ /// one reopen, and no query thread ever flushes the writer.
+ ///
+ /// Fails rather than waits when the reopen cannot deliver: a reopen that
threw, a closed index, or a wait past
+ /// [#_refreshWaitTimeoutMs]. Returning normally without the generation
would mean searching a stale searcher and
+ /// silently dropping rows the query's filter names, which is the failure
this whole path exists to prevent.
+ @VisibleForTesting
+ void awaitSearcherGeneration(long targetSequenceNumber)
throws IOException {
- synchronized (_searcherRefreshMonitor) {
- while (_searcherRefreshInProgress
- && _searcherRefreshedThroughSequenceNumber < targetSequenceNumber) {
- onSearcherRefreshWait();
+ // Plain volatile read, so a query that is already covered adds no
synchronization at all.
+ if (targetSequenceNumber <= _refreshedThroughSequenceNumber) {
+ return;
+ }
+ _searcherRefreshWaitCount.incrementAndGet();
+ long deadlineMs = System.currentTimeMillis() + _refreshWaitTimeoutMs;
+ synchronized (_refreshMonitor) {
+ long failuresBefore = _reopenFailureCount;
+ if (targetSequenceNumber > _requestedSequenceNumber) {
+ _requestedSequenceNumber = targetSequenceNumber;
+ }
+ _refreshMonitor.notifyAll();
+ while (targetSequenceNumber > _refreshedThroughSequenceNumber) {
+ if (_closed) {
+ throw new IOException(describe("Vector index closed while waiting
for the searcher to reopen"));
+ }
+ if (_reopenFailureCount != failuresBefore) {
+ throw new IOException(describe("Vector searcher reopen failed"),
_reopenFailure);
+ }
+ long remainingMs = deadlineMs - System.currentTimeMillis();
+ if (remainingMs <= 0) {
+ throw new IOException(describe(
+ "Timed out after " + _refreshWaitTimeoutMs + "ms waiting for the
vector searcher to reopen through "
+ + "generation " + targetSequenceNumber));
+ }
try {
- _searcherRefreshMonitor.wait();
+ _refreshMonitor.wait(remainingMs);
} catch (InterruptedException e) {
- throw new IOException("Interrupted while waiting to refresh vector
searcher for segment " + _segmentName
- + " column " + _vectorColumn, e);
+ // Not re-arming the interrupt flag: the cause is preserved on the
IOException, and this runs on a pooled
+ // Lucene searcher thread the pool deliberately keeps un-interrupted
(see submitSearch).
+ throw new IOException(describe("Interrupted while waiting for the
vector searcher to reopen"), e);
}
}
- if (_searcherRefreshedThroughSequenceNumber >= targetSequenceNumber) {
- return;
- }
- _searcherRefreshInProgress = true;
}
+ }
- try {
- beforeSearcherRefresh();
- // A false return means another caller already owns SearcherManager's
refresh lock. This is benign: use the
- // blocking form so this call still returns a searcher covering the
requested writer generation.
- if (!_searcherManager.maybeRefresh()) {
- _searcherManager.maybeRefreshBlocking();
+ /// Reopens the shared searcher whenever a query is waiting for a generation
it does not yet cover.
+ ///
+ /// Runs only on demand. An idle cadence would be pure cost here: a
consuming segment always has buffered rows,
+ /// so a timed reopen is never the cheap no-op it is for a settled index --
it flushes the writer and builds an
+ /// HNSW graph for the flushed rows, even for a table that issues no
filtered query at all.
+ private void reopenLoop() {
+ long lastReopenMs = 0L;
+ long failedRequest = Long.MIN_VALUE;
+ int consecutiveFailures = 0;
+ while (true) {
+ long request;
+ synchronized (_refreshMonitor) {
+ // Park unless someone needs a generation we have not reached AND have
not already failed on. Without the
+ // failedRequest half, a request whose waiters have all given up would
drive an unbounded retry-and-log
+ // loop, because a failed reopen never advances
_refreshedThroughSequenceNumber.
+ while (!_closed && (_requestedSequenceNumber <=
_refreshedThroughSequenceNumber
+ || _requestedSequenceNumber <= failedRequest)) {
+ try {
+ _refreshMonitor.wait();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ }
+ if (_closed) {
+ return;
+ }
+ // Spacing waits on the monitor rather than sleeping, so close() can
wake it immediately. It has to loop:
+ // every arriving query calls notifyAll, and a single timed wait would
return on that notification and
+ // reopen early, leaving the interval unenforced exactly when load
makes it matter.
+ long reopenNotBeforeMs = lastReopenMs + Math.max(_refreshMinIntervalMs,
+ consecutiveFailures == 0 ? 0L : Math.min(1000L <<
Math.min(consecutiveFailures - 1, 5), 30_000L));
+ while (!_closed) {
+ long waitMs = reopenNotBeforeMs - System.currentTimeMillis();
+ if (waitMs <= 0) {
+ break;
+ }
+ try {
+ _refreshMonitor.wait(waitMs);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ }
+ if (_closed) {
+ return;
+ }
+ // Read after the wait, so this reopen serves the newest request
rather than the one that woke us.
+ request = _requestedSequenceNumber;
}
- synchronized (_searcherRefreshMonitor) {
- _searcherRefreshedThroughSequenceNumber =
- Math.max(_searcherRefreshedThroughSequenceNumber,
targetSequenceNumber);
- _searcherRefreshCount++;
+ // Read the writer's generation BEFORE reopening, so rows arriving
during the reopen are not claimed as
+ // visible by a reader that may not contain them.
+ long generation;
+ try {
+ generation = _indexWriter.getMaxCompletedSequenceNumber();
+ doReopen();
+ } catch (Throwable t) {
+ consecutiveFailures++;
+ // Log the first failure of a run in full; after that only
periodically, so a persistently failing writer
+ // cannot turn an already-degraded server into a log flood.
+ if (consecutiveFailures == 1 || consecutiveFailures % 100 == 0) {
+ LOGGER.error("Failed to reopen the vector searcher ({} consecutive)
for segment: {}, column: {}",
+ consecutiveFailures, _segmentName, _vectorColumn, t);
+ }
+ lastReopenMs = System.currentTimeMillis();
+ synchronized (_refreshMonitor) {
+ _reopenFailure = t;
+ _reopenFailureCount++;
+ // Remember what failed so an abandoned request cannot drive a retry
loop. A newer request still gets a
+ // fresh attempt, so a transient failure (disk pressure, a slow
flush) recovers on the next query.
+ failedRequest = request;
+ _refreshMonitor.notifyAll();
+ }
+ continue;
}
- } finally {
- synchronized (_searcherRefreshMonitor) {
- _searcherRefreshInProgress = false;
- _searcherRefreshMonitor.notifyAll();
+ lastReopenMs = System.currentTimeMillis();
+ consecutiveFailures = 0;
+ _searcherRefreshCount.incrementAndGet();
+ synchronized (_refreshMonitor) {
+ // Published only here, after a reopen that returned normally.
+ _refreshedThroughSequenceNumber =
Math.max(_refreshedThroughSequenceNumber, generation);
+ _refreshMonitor.notifyAll();
}
}
}
- /// Test hook invoked by the winning refresher after it publishes the
in-progress state.
+ /// Seam for tests that need a reopen to fail; the failure handling around
it is the part worth covering.
@VisibleForTesting
- void beforeSearcherRefresh()
+ void doReopen()
throws IOException {
+ _searcherManager.maybeRefreshBlocking();
+ }
+
+ private String describe(String message) {
+ return message + " for segment: " + _segmentName + ", column: " +
_vectorColumn;
}
- /// Test hook invoked when another caller is about to wait for the winning
refresher.
@VisibleForTesting
- void onSearcherRefreshWait() {
+ long getLastAddedSequenceNumber() {
+ return _lastAddedSequenceNumber;
}
@VisibleForTesting
long getSearcherRefreshCount() {
- return _searcherRefreshCount;
+ return _searcherRefreshCount.get();
+ }
+
+ /// Cumulative number of queries that had to wait for a reopen. Compared
against the reopen count, this is what
+ /// shows sharing: many waits against one reopen.
+ @VisibleForTesting
+ long getSearcherRefreshWaitCount() {
+ return _searcherRefreshWaitCount.get();
}
private MutableRoaringBitmap search(IndexSearcher indexSearcher, float[]
vector, int topK, int efSearch,
@@ -527,6 +691,17 @@ public class MutableVectorIndex
@Override
public void close() {
+ // Stop the reopen loop and release every waiting query before the
searcher manager goes away: the loop
+ // refreshes through it, and a query blocked on a generation that will now
never arrive must not hang.
+ synchronized (_refreshMonitor) {
+ _closed = true;
+ _refreshMonitor.notifyAll();
+ }
+ try {
+ _reopenThread.join(TimeUnit.SECONDS.toMillis(30));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
try {
_indexWriter.commit();
// IndexWriter does not close the Directory passed to it, so both need
to be closed.
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/impl/vector/MutableVectorIndexTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/impl/vector/MutableVectorIndexTest.java
index b2814b4398a..23a6c59cbfc 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/impl/vector/MutableVectorIndexTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/impl/vector/MutableVectorIndexTest.java
@@ -25,13 +25,14 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.Phaser;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import
org.apache.pinot.segment.local.realtime.impl.invertedindex.RealtimeLuceneTextIndexSearcherPool;
import org.apache.pinot.segment.spi.index.creator.VectorIndexConfig;
@@ -39,6 +40,7 @@ import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
import org.roaringbitmap.buffer.MutableRoaringBitmap;
import org.testng.Assert;
import org.testng.annotations.BeforeClass;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@@ -289,8 +291,8 @@ public class MutableVectorIndexTest {
() -> index.getDocIds(new float[]{1.0F, 0.0F}, 2,
ImmutableRoaringBitmap.bitmapOf()));
ImmutableRoaringBitmap filtered = emptySearch.get(10, TimeUnit.SECONDS);
Assert.assertEquals(filtered.getCardinality(), 0);
- Assert.assertEquals(index.getSearcherRefreshCount(), 0L,
- "An empty filter must return before search submission or NRT
refresh");
+ Assert.assertEquals(index.getSearcherRefreshWaitCount(), 0L,
+ "An empty filter must return before search submission or entering
the NRT wait path");
} finally {
releaseBlocker.countDown();
try {
@@ -312,45 +314,314 @@ public class MutableVectorIndexTest {
}
@Test(timeOut = 60_000)
- public void testConcurrentFilteredSearchesCoalesceRefresh()
+ public void testConcurrentFilteredSearchesShareOneReopen()
throws Exception {
- int numCallers = SEARCHER_POOL_SIZE;
+ // Exercises the generation handshake directly rather than through
getDocIds. The searcher pool is a shared,
+ // scaling singleton that serializes blocking searches in this JVM, so
routing through it would only ever put
+ // one caller in the wait path at a time and the assertion would measure
the pool, not the sharing.
+ //
+ // A long refreshMinIntervalMs makes the overlap deterministic: the first
waiter's reopen cannot start until
+ // the limiter elapses, so both callers are guaranteed to be parked before
it runs, and one reopen must serve
+ // them both.
+ int numCallers = 2;
ExecutorService callers = Executors.newFixedThreadPool(numCallers);
- CoordinatedRefreshMutableVectorIndex index =
createCoordinatedIndexWithoutCommits(1);
+ MutableVectorIndex index = createIndexWithRefreshTuning("1000");
try (index) {
float[] query = {1.0F, 0.0F, 0.0F, 0.0F, 0.0F};
Assert.assertEquals(index.getDocIds(query, 1, bitmapOf(0)).toArray(),
new int[]{0});
long initialRefreshCount = index.getSearcherRefreshCount();
+ long initialWaits = index.getSearcherRefreshWaitCount();
addVector(index, query, 10);
- index.coordinateNextRefresh();
+ long target = index.getLastAddedSequenceNumber();
CyclicBarrier startTogether = new CyclicBarrier(numCallers);
- List<Future<ImmutableRoaringBitmap>> searches = new
ArrayList<>(numCallers);
+ List<Future<?>> waits = new ArrayList<>(numCallers);
for (int i = 0; i < numCallers; i++) {
- searches.add(callers.submit(() -> {
+ waits.add(callers.submit(() -> {
startTogether.await(10, TimeUnit.SECONDS);
- return index.getDocIds(query, 1, bitmapOf(10));
+ index.awaitSearcherGeneration(target);
+ return null;
}));
}
- index.awaitWinningRefresher();
- index.awaitWaitingCallers();
- index.releaseWinningRefresher();
- for (Future<ImmutableRoaringBitmap> search : searches) {
- Assert.assertEquals(search.get(10, TimeUnit.SECONDS).toArray(), new
int[]{10});
+ for (Future<?> wait : waits) {
+ wait.get(30, TimeUnit.SECONDS);
}
+ Assert.assertEquals(index.getSearcherRefreshWaitCount() - initialWaits,
numCallers,
+ "Every caller must have blocked on the shared reopen rather than
being served without waiting");
Assert.assertEquals(index.getSearcherRefreshCount(), initialRefreshCount
+ 1,
- "Concurrent readers targeting one writer generation must share
exactly one refresh");
- // The refresh count alone does not prove coalescing: a caller arriving
after publication would skip refresh
- // and still leave the count at one. Require actual participation in the
production waiter path.
- Assert.assertTrue(index.getObservedRefreshWaiters() > 0,
- "Expected concurrent callers to coalesce onto the in-flight refresh,
but none entered the waiter path");
+ "Callers targeting one writer generation must cost exactly one
reopen between them");
+ // And the searcher really does cover the row, not merely claim the
generation.
+ Assert.assertEquals(index.getDocIds(query, 1, bitmapOf(10)).toArray(),
new int[]{10});
} finally {
- index.releaseWinningRefresher();
callers.shutdownNow();
Assert.assertTrue(callers.awaitTermination(10, TimeUnit.SECONDS),
"Concurrent callers did not terminate");
}
}
+ /// The reason this path does not delegate to Lucene's
ControlledRealTimeReopenThread: a reopen that threw must
+ /// not publish the generation it merely attempted. Publishing it would let
the next query search a searcher
+ /// that does not hold the rows its filter names, and return fewer results
with no error at all.
+ @Test(timeOut = 60_000)
+ public void testFailedReopenNeitherPublishesItsGenerationNorHangsTheQuery()
+ throws Exception {
+ AtomicBoolean failReopen = new AtomicBoolean(true);
+ Map<String, String> properties = new HashMap<>();
+ properties.put("commitDocs", String.valueOf(Integer.MAX_VALUE));
+ properties.put("commitIntervalMs",
String.valueOf(TimeUnit.DAYS.toMillis(1)));
+ properties.put("vectorIndexType", "HNSW");
+ properties.put("vectorDimension", "5");
+ properties.put(MutableVectorIndex.REFRESH_MIN_INTERVAL_MS, "0");
+ VectorIndexConfig config = new VectorIndexConfig(false, "HNSW", 5, 1,
+ VectorIndexConfig.VectorDistanceFunction.EUCLIDEAN, properties);
+ try (MutableVectorIndex index =
+ new MutableVectorIndex("mutableVectorIndexReopenFailureTest_" +
System.nanoTime(), COLUMN_NAME, config) {
+ @Override
+ void doReopen()
+ throws IOException {
+ if (failReopen.get()) {
+ throw new IOException("injected reopen failure");
+ }
+ super.doReopen();
+ }
+ }) {
+ float[] query = {1.0F, 0.0F, 0.0F, 0.0F, 0.0F};
+ addVector(index, query, 10);
+ long generation = index.getLastAddedSequenceNumber();
+
+ IOException thrown = Assert.expectThrows(IOException.class, () ->
index.awaitSearcherGeneration(generation));
+ Assert.assertTrue(thrown.getMessage().contains("reopen failed"),
+ "A query must fail when the reopen it needs failed, got: " +
thrown.getMessage());
+ Assert.assertEquals(index.getSearcherRefreshCount(), 0L,
+ "A reopen that threw must not be counted as having produced a
searcher");
+ // The decisive assertion: the failed attempt must not have advanced the
published generation, or a later
+ // query would take the fast path and search a searcher that never
received the row.
+ IOException second = Assert.expectThrows(IOException.class, () ->
index.awaitSearcherGeneration(generation));
+ Assert.assertNotNull(second.getMessage());
+
+ // And a transient failure recovers on the next request rather than
disabling the segment.
+ failReopen.set(false);
+ addVector(index, query, 11);
+ Assert.assertEquals(index.getDocIds(query, 1, bitmapOf(11)).toArray(),
new int[]{11});
+ }
+ }
+
+ /// A reopen that never completes must surface as a failed query rather than
an unbounded wait holding a thread
+ /// of the shared searcher pool.
+ @Test(timeOut = 60_000)
+ public void testFilteredSearchTimesOutRatherThanWaitingForever()
+ throws Exception {
+ Map<String, String> properties = new HashMap<>();
+ properties.put("commitDocs", String.valueOf(Integer.MAX_VALUE));
+ properties.put("commitIntervalMs",
String.valueOf(TimeUnit.DAYS.toMillis(1)));
+ properties.put("vectorIndexType", "HNSW");
+ properties.put("vectorDimension", "5");
+ // The spacing outlives the timeout, so the reopen this query needs cannot
start before it gives up.
+ properties.put(MutableVectorIndex.REFRESH_MIN_INTERVAL_MS,
String.valueOf(TimeUnit.SECONDS.toMillis(30)));
+ properties.put(MutableVectorIndex.REFRESH_WAIT_TIMEOUT_MS, "50");
+ VectorIndexConfig config = new VectorIndexConfig(false, "HNSW", 5, 1,
+ VectorIndexConfig.VectorDistanceFunction.EUCLIDEAN, properties);
+ try (MutableVectorIndex index =
+ new MutableVectorIndex("mutableVectorIndexTimeoutTest_" +
System.nanoTime(), COLUMN_NAME, config)) {
+ addVector(index, new float[]{1.0F, 0.0F, 0.0F, 0.0F, 0.0F}, 0);
+ // Spend the first reopen, which is never delayed: only from the second
on does the spacing apply, and it is
+ // the spacing that keeps the awaited reopen from starting before the
wait gives up.
+ Assert.assertEquals(index.getDocIds(new float[]{1.0F, 0.0F, 0.0F, 0.0F,
0.0F}, 1, bitmapOf(0)).toArray(),
+ new int[]{0});
+ addVector(index, new float[]{1.0F, 0.0F, 0.0F, 0.0F, 0.0F}, 10);
+ IOException thrown = Assert.expectThrows(IOException.class,
+ () ->
index.awaitSearcherGeneration(index.getLastAddedSequenceNumber()));
+ Assert.assertTrue(thrown.getMessage().contains("Timed out after 50ms"),
+ "Expected the bounded-wait failure, got: " + thrown.getMessage());
+ }
+ }
+
+ /// The spacing must hold even while queries keep arriving. Each arriving
query notifies the reopen thread, so a
+ /// single timed wait would return on that notification and reopen early --
leaving the interval unenforced
+ /// exactly under the load it exists to bound.
+ @Test(timeOut = 60_000)
+ public void testReopenSpacingHoldsWhileQueriesKeepArriving()
+ throws Exception {
+ long spacingMs = 1000L;
+ try (MutableVectorIndex index =
createIndexWithRefreshTuning(String.valueOf(spacingMs))) {
+ float[] query = {1.0F, 0.0F, 0.0F, 0.0F, 0.0F};
+ Assert.assertEquals(index.getDocIds(query, 1, bitmapOf(0)).toArray(),
new int[]{0});
+ long reopensAfterFirst = index.getSearcherRefreshCount();
+
+ // Keep registering fresh requests for most of one spacing window; every
one of them notifies the loop.
+ long deadline = System.nanoTime() +
TimeUnit.MILLISECONDS.toNanos(spacingMs / 2);
+ int docId = 100;
+ while (System.nanoTime() < deadline) {
+ addVector(index, query, docId++);
+ long generation = index.getLastAddedSequenceNumber();
+ Thread nudge = new Thread(() -> {
+ try {
+ index.awaitSearcherGeneration(generation);
+ } catch (IOException e) {
+ // The wait may outlive the window; the notification it sent is
what this test is about.
+ }
+ });
+ nudge.setDaemon(true);
+ nudge.start();
+ Thread.sleep(20);
+ }
+ Assert.assertEquals(index.getSearcherRefreshCount(), reopensAfterFirst,
+ "No reopen may run inside the spacing window, however many queries
notify the loop");
+ }
+ }
+
+ /// A query interrupted while waiting must surface that as a failure
carrying the cause, not return as though
+ /// the generation had arrived.
+ @Test(timeOut = 60_000)
+ public void testInterruptedWaitFailsRatherThanReturningEarly()
+ throws Exception {
+ try (MutableVectorIndex index =
createIndexWithRefreshTuning(String.valueOf(TimeUnit.SECONDS.toMillis(30)))) {
+ float[] query = {1.0F, 0.0F, 0.0F, 0.0F, 0.0F};
+ Assert.assertEquals(index.getDocIds(query, 1, bitmapOf(0)).toArray(),
new int[]{0});
+ addVector(index, query, 10);
+ long generation = index.getLastAddedSequenceNumber();
+
+ AtomicReference<Throwable> failure = new AtomicReference<>();
+ CountDownLatch waiting = new CountDownLatch(1);
+ Thread waiter = new Thread(() -> {
+ waiting.countDown();
+ try {
+ index.awaitSearcherGeneration(generation);
+ failure.set(new AssertionError("Interrupted wait returned
normally"));
+ } catch (Throwable t) {
+ failure.set(t);
+ }
+ });
+ waiter.setDaemon(true);
+ waiter.start();
+ Assert.assertTrue(waiting.await(10, TimeUnit.SECONDS), "Waiter did not
start");
+ awaitWaitCount(index, 2);
+ waiter.interrupt();
+ waiter.join(TimeUnit.SECONDS.toMillis(10));
+
+ Assert.assertTrue(failure.get() instanceof IOException,
+ "An interrupted wait must fail, got: " + failure.get());
+ Assert.assertTrue(failure.get().getMessage().contains("Interrupted while
waiting"),
+ "Expected the interrupt failure, got: " +
failure.get().getMessage());
+ }
+ }
+
+ /// One reopen thread exists per consuming segment per vector column, so a
close that failed to stop it would
+ /// leak a thread per partition -- invisible at runtime because the thread
is a daemon.
+ @Test(timeOut = 60_000)
+ public void testCloseStopsTheReopenThread()
+ throws Exception {
+ String segmentName = "mutableVectorIndexCloseTest_" + System.nanoTime();
+ String threadName = "vector-nrt-reopen-" + segmentName + "-" + COLUMN_NAME;
+ Map<String, String> properties = new HashMap<>();
+ properties.put("vectorIndexType", "HNSW");
+ properties.put("vectorDimension", "5");
+ MutableVectorIndex index = new MutableVectorIndex(segmentName, COLUMN_NAME,
+ new VectorIndexConfig(false, "HNSW", 5, 1,
VectorIndexConfig.VectorDistanceFunction.EUCLIDEAN, properties));
+ Assert.assertTrue(hasLiveThread(threadName), "The reopen thread must run
while the index is open");
+ index.close();
+ Assert.assertFalse(hasLiveThread(threadName),
+ "close() must stop the reopen thread, or every consuming segment leaks
one");
+ }
+
+ /// A query must never be told a generation is available when the reopen
that would have produced it failed:
+ /// answering from the stale searcher would silently drop rows the filter
names. Closing the index mid-wait is
+ /// the reachable version of that -- the awaited generation can no longer
arrive, so the query must fail.
+ @Test(timeOut = 60_000)
+ public void testFilteredSearchFailsWhenTheAwaitedGenerationCannotArrive()
+ throws Exception {
+ MutableVectorIndex index =
createIndexWithRefreshTuning(String.valueOf(TimeUnit.SECONDS.toMillis(5)));
+ float[] query = {1.0F, 0.0F, 0.0F, 0.0F, 0.0F};
+ Assert.assertEquals(index.getDocIds(query, 1, bitmapOf(0)).toArray(), new
int[]{0});
+ addVector(index, query, 10);
+ ExecutorService caller = Executors.newSingleThreadExecutor();
+ try {
+ // The 30s limiter keeps the reopen pending, so this call is parked in
the wait path when close() lands.
+ Future<ImmutableRoaringBitmap> search = caller.submit(() ->
index.getDocIds(query, 1, bitmapOf(10)));
+ awaitWaitCount(index, 2);
+ index.close();
+ ExecutionException thrown = Assert.expectThrows(ExecutionException.class,
+ () -> search.get(30, TimeUnit.SECONDS));
+ Assert.assertTrue(hasCauseContaining(thrown, "closed while waiting for
the searcher to reopen"),
+ "A query whose generation can never arrive must fail for that
reason: " + thrown.getCause());
+ } finally {
+ caller.shutdownNow();
+ Assert.assertTrue(caller.awaitTermination(10, TimeUnit.SECONDS), "Caller
did not terminate");
+ }
+ }
+
+ @Test(dataProvider = "invalidRefreshTuning")
+ public void testRejectsInvalidRefreshTuning(String key, String value) {
+ Map<String, String> properties = new HashMap<>();
+ properties.put("vectorIndexType", "HNSW");
+ properties.put("vectorDimension", "5");
+ properties.put(key, value);
+ VectorIndexConfig config = new VectorIndexConfig(false, "HNSW", 5, 1,
+ VectorIndexConfig.VectorDistanceFunction.EUCLIDEAN, properties);
+ Assert.expectThrows(IllegalArgumentException.class,
+ () -> new MutableVectorIndex("mutableVectorIndexConfigTest_" +
System.nanoTime(), COLUMN_NAME, config));
+ }
+
+ @DataProvider(name = "invalidRefreshTuning")
+ public Object[][] invalidRefreshTuning() {
+ return new Object[][]{
+ {MutableVectorIndex.REFRESH_MIN_INTERVAL_MS, "-1"},
+ {MutableVectorIndex.REFRESH_MIN_INTERVAL_MS, "abc"},
+ {MutableVectorIndex.REFRESH_WAIT_TIMEOUT_MS, "0"},
+ {MutableVectorIndex.REFRESH_WAIT_TIMEOUT_MS, "-5"}
+ };
+ }
+
+ private static boolean hasCauseContaining(Throwable thrown, String fragment)
{
+ for (Throwable cause = thrown; cause != null; cause = cause.getCause()) {
+ if (cause.getMessage() != null && cause.getMessage().contains(fragment))
{
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static boolean hasLiveThread(String threadName) {
+ return Thread.getAllStackTraces().keySet().stream()
+ .anyMatch(thread -> thread.isAlive() &&
threadName.equals(thread.getName()));
+ }
+
+ private static void awaitWaitCount(MutableVectorIndex index, long expected)
+ throws InterruptedException, TimeoutException {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+ while (index.getSearcherRefreshWaitCount() < expected) {
+ if (System.nanoTime() > deadline) {
+ throw new TimeoutException("Only " +
index.getSearcherRefreshWaitCount() + " of " + expected + " waits seen");
+ }
+ Thread.sleep(5);
+ }
+ }
+
+ /// Queries already covered by the searcher's current generation must
neither reopen nor enter the wait path.
+ /// This guards the regression where a filtered query drives a reopen
unconditionally, which is what makes the
+ /// per-query flush cost unbounded. It does not on its own distinguish
publishing the true generation reached
+ /// from the previous per-caller watermark: with no rows added between these
queries, both skip the refresh.
+ @Test
+ public void testFilteredSearchesAfterOneReopenDoNotRefreshAgain()
+ throws Exception {
+ try (MutableVectorIndex index = createIndexWithRefreshTuning()) {
+ float[] query = {1.0F, 0.0F, 0.0F, 0.0F, 0.0F};
+ Assert.assertEquals(index.getDocIds(query, 1, bitmapOf(0)).toArray(),
new int[]{0});
+ addVector(index, query, 10);
+
+ Assert.assertEquals(index.getDocIds(query, 1, bitmapOf(10)).toArray(),
new int[]{10});
+ long refreshesAfterFirstSearch = index.getSearcherRefreshCount();
+ long waitersAfterFirstSearch = index.getSearcherRefreshWaitCount();
+
+ for (int i = 0; i < 20; i++) {
+ Assert.assertEquals(index.getDocIds(query, 1, bitmapOf(10)).toArray(),
new int[]{10});
+ }
+ Assert.assertEquals(index.getSearcherRefreshCount(),
refreshesAfterFirstSearch,
+ "Queries already covered by the published generation must not
trigger another reopen");
+ Assert.assertEquals(index.getSearcherRefreshWaitCount(),
waitersAfterFirstSearch,
+ "Queries already covered by the published generation must not even
enter the waiter path");
+ }
+ }
+
@Test
public void testUnfilteredSearchTranslatesSuppliedPinotDocIds() {
// commitDocs=4 commits on the 4th add, so the committed-view unfiltered
path sees all rows; with doc ids
@@ -440,74 +711,6 @@ public class MutableVectorIndexTest {
return phaser.awaitAdvanceInterruptibly(phase, 10, TimeUnit.SECONDS) >= 0;
}
- /// Holds the winning refresher after it publishes
`_searcherRefreshInProgress`, so at least one concurrent search
- /// is proven to enter the production waiter path before the refresh is
allowed to finish.
- private static class CoordinatedRefreshMutableVectorIndex extends
MutableVectorIndex {
- private final CountDownLatch _winningRefresherEntered = new
CountDownLatch(1);
- private final CountDownLatch _releaseWinningRefresher = new
CountDownLatch(1);
- private final AtomicInteger _observedRefreshWaiters = new AtomicInteger();
- private final CountDownLatch _waitingCallers;
- private volatile boolean _coordinateRefresh;
-
- CoordinatedRefreshMutableVectorIndex(String segmentName, VectorIndexConfig
config, int expectedWaitingCallers) {
- super(segmentName, COLUMN_NAME, config);
- _waitingCallers = new CountDownLatch(expectedWaitingCallers);
- }
-
- void coordinateNextRefresh() {
- _coordinateRefresh = true;
- }
-
- void awaitWinningRefresher()
- throws InterruptedException, TimeoutException {
- if (!_winningRefresherEntered.await(10, TimeUnit.SECONDS)) {
- throw new TimeoutException("No filtered-search caller became the
winning refresher");
- }
- }
-
- void awaitWaitingCallers()
- throws InterruptedException, TimeoutException {
- if (!_waitingCallers.await(10, TimeUnit.SECONDS)) {
- throw new TimeoutException("No concurrent caller entered the refresh
waiter path");
- }
- }
-
- void releaseWinningRefresher() {
- _releaseWinningRefresher.countDown();
- }
-
- @Override
- void beforeSearcherRefresh()
- throws IOException {
- if (!_coordinateRefresh) {
- return;
- }
- _winningRefresherEntered.countDown();
- try {
- if (!_releaseWinningRefresher.await(10, TimeUnit.SECONDS)) {
- throw new IOException("Timed out waiting to release the winning
refresher");
- }
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new IOException("Interrupted while holding the winning
refresher", e);
- } finally {
- _coordinateRefresh = false;
- }
- }
-
- @Override
- void onSearcherRefreshWait() {
- _observedRefreshWaiters.incrementAndGet();
- if (_coordinateRefresh) {
- _waitingCallers.countDown();
- }
- }
-
- int getObservedRefreshWaiters() {
- return _observedRefreshWaiters.get();
- }
- }
-
/// Blocks the first filter membership check until the writer has added its
concurrent rows. This puts the
/// synchronization point inside Lucene's actual filtered search, rather
than merely racing two caller threads.
private static class ConcurrentWriteCoordinatingBitmap extends
MutableRoaringBitmap {
@@ -550,16 +753,24 @@ public class MutableVectorIndexTest {
return index;
}
- private static CoordinatedRefreshMutableVectorIndex
createCoordinatedIndexWithoutCommits(int expectedWaiters) {
+ /// An index whose reopen cadence is pinned for assertions: no commits, no
idle reopen inside the test
+ /// window, and a waiting query allowed to trigger its reopen immediately.
+ private static MutableVectorIndex createIndexWithRefreshTuning() {
+ return createIndexWithRefreshTuning("0");
+ }
+
+ private static MutableVectorIndex createIndexWithRefreshTuning(String
refreshMinIntervalMs) {
Map<String, String> properties = new HashMap<>();
properties.put("commitDocs", String.valueOf(Integer.MAX_VALUE));
properties.put("commitIntervalMs",
String.valueOf(TimeUnit.DAYS.toMillis(1)));
properties.put("vectorIndexType", "HNSW");
properties.put("vectorDimension", "5");
+ // No rate limiting, so a waiting query's reopen starts immediately and
the test never sits on the limiter.
+ properties.put(MutableVectorIndex.REFRESH_MIN_INTERVAL_MS,
refreshMinIntervalMs);
VectorIndexConfig config = new VectorIndexConfig(false, "HNSW", 5, 1,
VectorIndexConfig.VectorDistanceFunction.EUCLIDEAN, properties);
- CoordinatedRefreshMutableVectorIndex index = new
CoordinatedRefreshMutableVectorIndex(
- "mutableVectorIndexCoalescingTest_" + System.nanoTime(), config,
expectedWaiters);
+ MutableVectorIndex index =
+ new MutableVectorIndex("mutableVectorIndexCoalescingTest_" +
System.nanoTime(), COLUMN_NAME, config);
addVector(index, new float[]{1.0F, 0.0F, 0.0F, 0.0F, 0.0F}, 0);
addVector(index, new float[]{0.0F, 1.0F, 0.0F, 0.0F, 0.0F}, 1);
return index;
diff --git
a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/creator/VectorIndexConfigValidator.java
b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/creator/VectorIndexConfigValidator.java
index b21eb6aa335..cdcb70a803f 100644
---
a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/creator/VectorIndexConfigValidator.java
+++
b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/creator/VectorIndexConfigValidator.java
@@ -42,7 +42,8 @@ public final class VectorIndexConfigValidator {
// HNSW-specific property keys
static final Set<String> HNSW_PROPERTIES = Collections.unmodifiableSet(new
HashSet<>(
Arrays.asList("maxCon", "beamWidth", "maxDimensions", "maxBufferSizeMB",
- "useCompoundFile", "mode", "commit", "commitIntervalMs",
"commitDocs")));
+ "useCompoundFile", "mode", "commit", "commitIntervalMs",
"commitDocs",
+ "refreshMinIntervalMs", "refreshWaitTimeoutMs")));
// IVF_FLAT-specific property keys
static final Set<String> IVF_FLAT_PROPERTIES =
Collections.unmodifiableSet(new HashSet<>(
@@ -260,6 +261,33 @@ public final class VectorIndexConfigValidator {
validatePositiveIntProperty(properties, "beamWidth", "HNSW beamWidth");
validatePositiveIntProperty(properties, "maxDimensions", "HNSW
maxDimensions");
validatePositiveDoubleProperty(properties, "maxBufferSizeMB", "HNSW
maxBufferSizeMB");
+ // Rejected here rather than only in MutableVectorIndex: the constructor
guard runs when a consuming segment is
+ // created on the server, which would stop ingestion for the partition
instead of failing the table config.
+ validateLongProperty(properties, "refreshMinIntervalMs", "HNSW
refreshMinIntervalMs", 0L);
+ validateLongProperty(properties, "refreshWaitTimeoutMs", "HNSW
refreshWaitTimeoutMs", 1L);
+ }
+
+ /// Absent means "use the default"; anything present must parse and satisfy
the bound.
+ ///
+ /// Deliberately parses exactly as the consumer does -- no trimming -- so a
value this accepts cannot then fail
+ /// on the server when a consuming segment is created, which would stop
ingestion for the partition rather than
+ /// rejecting the table config. A key present with a null value is rejected
for the same reason: `getOrDefault`
+ /// does not substitute the default for it.
+ private static void validateLongProperty(Map<String, String> properties,
String key, String displayName,
+ long minInclusive) {
+ if (!properties.containsKey(key)) {
+ return;
+ }
+ String value = properties.get(key);
+ long longValue;
+ try {
+ longValue = Long.parseLong(value);
+ } catch (NumberFormatException | NullPointerException e) {
+ throw new IllegalArgumentException(displayName + " must be a valid long,
got: '" + value + "'");
+ }
+ if (longValue < minInclusive) {
+ throw new IllegalArgumentException(displayName + " must be >= " +
minInclusive + ", got: " + longValue);
+ }
}
/// Validates IVF_FLAT-specific property values.
diff --git
a/pinot-segment-spi/src/test/java/org/apache/pinot/segment/spi/index/creator/VectorIndexConfigValidatorTest.java
b/pinot-segment-spi/src/test/java/org/apache/pinot/segment/spi/index/creator/VectorIndexConfigValidatorTest.java
index 89261761a12..95242e3d44b 100644
---
a/pinot-segment-spi/src/test/java/org/apache/pinot/segment/spi/index/creator/VectorIndexConfigValidatorTest.java
+++
b/pinot-segment-spi/src/test/java/org/apache/pinot/segment/spi/index/creator/VectorIndexConfigValidatorTest.java
@@ -393,6 +393,52 @@ public class VectorIndexConfigValidatorTest {
// Property value validation tests
// ============================================================
+ @Test
+ public void testAcceptRefreshTuning() {
+ Map<String, String> properties = new HashMap<>();
+ properties.put("vectorIndexType", "HNSW");
+ properties.put("vectorDimension", "768");
+ properties.put("refreshMinIntervalMs", "0");
+ properties.put("refreshWaitTimeoutMs", "5000");
+
+ VectorIndexConfigValidator.validate(new VectorIndexConfig(properties));
+ }
+
+ @Test(expectedExceptions = IllegalArgumentException.class,
+ expectedExceptionsMessageRegExp = ".*HNSW refreshMinIntervalMs must be
>= 0.*")
+ public void testRejectNegativeRefreshMinIntervalMs() {
+ Map<String, String> properties = new HashMap<>();
+ properties.put("vectorIndexType", "HNSW");
+ properties.put("vectorDimension", "768");
+ properties.put("refreshMinIntervalMs", "-1");
+
+ VectorIndexConfigValidator.validate(new VectorIndexConfig(properties));
+ }
+
+ @Test(expectedExceptions = IllegalArgumentException.class,
+ expectedExceptionsMessageRegExp = ".*HNSW refreshWaitTimeoutMs must be
>= 1.*")
+ public void testRejectZeroRefreshWaitTimeoutMs() {
+ Map<String, String> properties = new HashMap<>();
+ properties.put("vectorIndexType", "HNSW");
+ properties.put("vectorDimension", "768");
+ properties.put("refreshWaitTimeoutMs", "0");
+
+ VectorIndexConfigValidator.validate(new VectorIndexConfig(properties));
+ }
+
+ /// The validator must be exactly as strict as MutableVectorIndex's own
parse, including whitespace: a value it
+ /// accepts here but the server rejects would stop ingestion instead of
failing the table config.
+ @Test(expectedExceptions = IllegalArgumentException.class,
+ expectedExceptionsMessageRegExp = ".*HNSW refreshMinIntervalMs must be a
valid long.*")
+ public void testRejectNonNumericRefreshMinIntervalMs() {
+ Map<String, String> properties = new HashMap<>();
+ properties.put("vectorIndexType", "HNSW");
+ properties.put("vectorDimension", "768");
+ properties.put("refreshMinIntervalMs", " 100 ");
+
+ VectorIndexConfigValidator.validate(new VectorIndexConfig(properties));
+ }
+
@Test(expectedExceptions = IllegalArgumentException.class,
expectedExceptionsMessageRegExp = ".*HNSW maxCon must be a positive
integer.*")
public void testRejectNegativeMaxCon() {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
