Jackie-Jiang commented on code in PR #19303:
URL: https://github.com/apache/pinot/pull/19303#discussion_r3890217425
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/vector/MutableVectorIndex.java:
##########
@@ -90,13 +127,11 @@ public MutableVectorIndex(String segmentName, String
vectorColumn, VectorIndexCo
_commitDocs = Long.parseLong(
vectorIndexConfig.getProperties().getOrDefault("commitDocs",
String.valueOf(DEFAULT_COMMIT_DOCS)));
_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.
- _indexDir = new File(new File(FileUtils.getTempDirectory(), segmentName),
- _vectorColumn +
V1Constants.Indexes.VECTOR_V912_HNSW_INDEX_FILE_EXTENSION);
+ _indexDir = createIndexDir(segmentName, _vectorColumn);
Review Comment:
The per-instance directory isolation change fixes an independent
write-lock/lifecycle problem and has its own dedicated test. This PR is already
a large correctness-sensitive filtered-ANN change; please move directory
isolation into a focused follow-up PR unless it is required for the mutable
pre-filter capability.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/vector/MutableVectorIndex.java:
##########
@@ -279,18 +450,35 @@ public void close() {
try {
_indexWriter.commit();
// IndexWriter does not close the Directory passed to it, so both need
to be closed.
- IOUtils.close(_indexWriter, _indexDirectory);
+ IOUtils.close(_searcherManager, _indexWriter, _indexDirectory);
} catch (IOException e) {
- // Both close() implementations are idempotent, so this is a no-op for
whatever was already closed above.
- IOUtils.closeWhileHandlingException(_indexWriter, _indexDirectory);
+ // All close() implementations are idempotent, so this is a no-op for
whatever was already closed above.
+ IOUtils.closeWhileHandlingException(_searcherManager, _indexWriter,
_indexDirectory);
throw new RuntimeException(e);
} finally {
deleteIndexDir();
}
}
- /// Deletes the temporary index directory of this column, then the segment
directory holding it if this was the last
- /// column with an index under it.
+ /// Creates a private directory for this index instance, under a directory
named for the segment.
+ ///
+ /// The directory is unique per instance rather than per segment and column,
so two replicas of the same segment
+ /// hosted in one JVM cannot collide on the Lucene write lock. Segment and
column stay in the path so a leaked
+ /// directory can still be attributed to its owner.
+ private static File createIndexDir(String segmentName, String column) {
+ File segmentDir = new File(FileUtils.getTempDirectory(), segmentName);
+ try {
+ Files.createDirectories(segmentDir.toPath());
+ return Files.createTempDirectory(segmentDir.toPath(),
+ column + V1Constants.Indexes.VECTOR_V912_HNSW_INDEX_FILE_EXTENSION +
'_').toFile();
+ } catch (IOException e) {
+ throw new RuntimeException(
+ "Failed to create mutable vector index directory for column: " +
column + ", segment: " + segmentName, e);
+ }
+ }
+
+ /// Deletes the temporary index directory of this instance, then the segment
directory holding it if this was the
Review Comment:
Deleting the shared segment parent races with construction of another index
for the same segment. T2 can create/observe the parent, T1 can delete it after
removing its private child, and then T2's `createTempDirectory(segmentDir,
...)` fails with `NoSuchFileException`. Please delete only this instance's
private `_indexDir`, or coordinate parent creation/deletion atomically.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/readers/vector/IvfPqVectorIndexReader.java:
##########
@@ -217,6 +217,13 @@ public MutableRoaringBitmap getDocIds(float[] searchQuery,
int topK) {
return result;
}
+ @Override
+ public boolean supportsPreFilter() {
Review Comment:
This explicit pre-filter opt-in is behaviorally significant now that the
interface default is `false`, but IVF_PQ has no test covering the override or
filtered path. Please add coverage asserting `supportsPreFilter()` and
verifying that filtered search excludes nearer disallowed documents, or an
integration query whose EXPLAIN reports `FILTER_THEN_ANN`.
##########
pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/impl/vector/MutableVectorIndexTest.java:
##########
@@ -80,12 +84,180 @@ public void testRuntimeControlDebugInfoReflectsOverrides()
{
Assert.assertEquals(debugInfo.get("effectiveEfSearch"), 6);
Assert.assertEquals(debugInfo.get("effectiveHnswUseRelativeDistance"),
Boolean.FALSE);
Assert.assertEquals(debugInfo.get("effectiveHnswUseBoundedQueue"),
Boolean.FALSE);
- Assert.assertEquals(debugInfo.get("supportsPreFilter"), Boolean.FALSE);
+ Assert.assertEquals(debugInfo.get("supportsPreFilter"), Boolean.TRUE);
+ Assert.assertTrue(index.supportsPreFilter(),
+ "The reader must advertise filtered search: that is what makes the
planner choose it over an exact scan");
} finally {
index.close();
}
}
+ // -----------------------------------------------------------------------
+ // Filtered search (upsert doc-ids snapshot enforcement)
+ // -----------------------------------------------------------------------
+
+ /// 2-D corpus with distinct distances from the query vector {1, 0}:
+ /// docs 0 and 1 are nearest (the "upsert-obsoleted" rows), docs 2 and 3 are
the valid rows.
+ private static MutableVectorIndex create2DIndex(long commitDocs, int
docIdOffset) {
+ Map<String, String> properties = new HashMap<>();
+ properties.put("commitDocs", String.valueOf(commitDocs));
+ properties.put("vectorIndexType", "HNSW");
+ properties.put("vectorDimension", "2");
+ VectorIndexConfig config = new VectorIndexConfig(false, "HNSW", 2, 1,
+ VectorIndexConfig.VectorDistanceFunction.EUCLIDEAN, properties);
+ MutableVectorIndex index =
+ new MutableVectorIndex("mutableVectorIndexFilterTest_" +
System.nanoTime(), COLUMN_NAME, config);
+ addVector(index, new float[]{1.0F, 0.0F}, docIdOffset);
+ addVector(index, new float[]{0.99F, 0.01F}, docIdOffset + 1);
+ addVector(index, new float[]{0.0F, 1.0F}, docIdOffset + 2);
+ addVector(index, new float[]{0.0F, -1.0F}, docIdOffset + 3);
+ return index;
+ }
+
+ @Test
+ public void testFilteredSearchExcludesNearestDisallowedDocs() {
+ // commitDocs=4 commits on the 4th add, so the unfiltered committed-view
sanity check below sees all rows
+ MutableVectorIndex index = create2DIndex(4, 0);
+ try {
+ // Sanity: unfiltered top-2 returns the physically nearest ("obsolete")
docs 0 and 1
+ ImmutableRoaringBitmap unfiltered = index.getDocIds(new float[]{1.0F,
0.0F}, 2);
+ Assert.assertEquals(unfiltered, ImmutableRoaringBitmap.bitmapOf(0, 1));
+
+ // Filtered top-2 restricted to docs 2 and 3 must return exactly those
docs. A post-intersection
+ // implementation would return empty here (the unfiltered top-2 has no
overlap with the allowed set),
+ // so this assertion genuinely discriminates filtered candidate
generation.
+ ImmutableRoaringBitmap filtered =
+ index.getDocIds(new float[]{1.0F, 0.0F}, 2,
ImmutableRoaringBitmap.bitmapOf(2, 3));
+ Assert.assertEquals(filtered, ImmutableRoaringBitmap.bitmapOf(2, 3),
+ "Filtered search must return the allowed docs, not the nearest
disallowed ones");
+ } finally {
+ index.close();
+ }
+ }
+
+ @Test
+ public void testConcurrentFilteredSearchWithLiveWriter()
+ throws Exception {
+ // Single writer, concurrent reader: filtered searches must stay correct
(results always a subset of
+ // the filter bitmap) while rows are being added and commits fire mid-run
+ Map<String, String> properties = new HashMap<>();
+ properties.put("commitDocs", "7");
+ properties.put("vectorIndexType", "HNSW");
+ properties.put("vectorDimension", "2");
+ VectorIndexConfig config = new VectorIndexConfig(false, "HNSW", 2, 1,
+ VectorIndexConfig.VectorDistanceFunction.EUCLIDEAN, properties);
+ MutableVectorIndex index =
+ new MutableVectorIndex("mutableVectorIndexConcurrentTest_" +
System.nanoTime(), COLUMN_NAME, config);
+ int numDocs = 200;
+ ImmutableRoaringBitmap allowed = ImmutableRoaringBitmap.bitmapOf(2, 3);
+ AtomicReference<Throwable> failure =
+ new AtomicReference<>();
+ try {
+ addVector(index, new float[]{0.0F, 1.0F}, 0);
+ addVector(index, new float[]{0.0F, -1.0F}, 1);
+ addVector(index, new float[]{1.0F, 0.0F}, 2);
+ addVector(index, new float[]{0.99F, 0.01F}, 3);
+
+ Thread writer = new Thread(() -> {
+ try {
+ for (int docId = 4; docId < numDocs; docId++) {
+ addVector(index, new float[]{-1.0F, 0.0F}, docId);
+ }
+ } catch (Throwable t) {
+ failure.compareAndSet(null, t);
+ }
+ });
+ writer.start();
+ while (writer.isAlive() && failure.get() == null) {
Review Comment:
This test does not guarantee concurrent reader/writer execution: the writer
can finish all additions before the main thread evaluates `writer.isAlive()`,
leaving only the post-join assertion. Please use latches or a `Phaser` to force
at least one interleaved add/search phase and assert that a filtered read
occurred before allowing the writer to finish.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/readers/vector/BaseDocIdBitmapFilterQuery.java:
##########
@@ -0,0 +1,136 @@
+/**
+ * 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.pinot.segment.local.segment.index.readers.vector;
+
+import java.io.IOException;
+import javax.annotation.Nullable;
+import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.search.ConstantScoreWeight;
+import org.apache.lucene.search.DocIdSetIterator;
+import org.apache.lucene.search.IndexSearcher;
+import org.apache.lucene.search.Query;
+import org.apache.lucene.search.QueryVisitor;
+import org.apache.lucene.search.ScoreMode;
+import org.apache.lucene.search.Scorer;
+import org.apache.lucene.search.Weight;
+import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
+
+
+/// Base class for Lucene [Query] implementations that accept only documents
whose Pinot doc id is present
+/// in a [ImmutableRoaringBitmap]. Used to implement pre-filter ANN search by
restricting HNSW graph
+/// traversal to the filtered document set.
+///
+/// Because Lucene uses its own internal doc ids (which differ from Pinot doc
ids), subclasses supply the
+/// per-leaf iterator that maps Lucene doc ids to Pinot doc ids before testing
membership in the bitmap
+/// (via a doc-id translator, doc values, etc.). This class owns the
constant-score weight/scorer
+/// scaffolding, identity-based equality, and cache opt-out, so
filter-correctness fixes apply to every
+/// implementation at once.
+///
+/// Instances are single-use per search and must never be cached by Lucene
([Weight#isCacheable] returns
+/// false), since the accepted docs depend on the bitmap instance.
+public abstract class BaseDocIdBitmapFilterQuery extends Query {
Review Comment:
Please document bitmap ownership and thread safety on this new public base
class. `MutableRoaringBitmap` is an `ImmutableRoaringBitmap` subtype, so a
caller can mutate the retained instance during search and change matching
behavior. State that the bitmap must remain unchanged for the query lifetime
and whether query instances may be shared across threads (or snapshot
defensively if mutation is allowed).
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/vector/MutableVectorIndex.java:
##########
@@ -230,33 +294,140 @@ public Map<String, Object> getIndexDebugInfo() {
info.put("effectiveEfSearch", getEffectiveEfSearch());
info.put("effectiveHnswUseRelativeDistance",
getEffectiveUseRelativeDistance());
info.put("effectiveHnswUseBoundedQueue", getEffectiveUseBoundedQueue());
- info.put("supportsPreFilter", false);
+ info.put("supportsPreFilter", true);
try (DirectoryReader directoryReader =
DirectoryReader.open(_indexDirectory)) {
info.put("numDocs", directoryReader.numDocs());
info.put("numDeletedDocs", directoryReader.numDeletedDocs());
info.put("luceneSegments", directoryReader.leaves().size());
} catch (IOException e) {
LOGGER.warn("Failed to load mutable HNSW debug stats for segment: {},
column: {}", _segmentName, _vectorColumn,
e);
- info.put("numDocs", _nextDocId);
+ info.put("numDocs", _numDocsAdded);
info.put("numDeletedDocs", 0);
info.put("luceneSegments", 0);
}
return info;
}
private MutableRoaringBitmap executeVectorSearch(float[] vector, int topK,
int efSearch,
- boolean useRelativeDistance, boolean useBoundedQueue) throws IOException
{
+ boolean useRelativeDistance, boolean useBoundedQueue, @Nullable
ImmutableRoaringBitmap preFilterBitmap)
+ 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 is
expensive: maybeRefreshBlocking takes an
+ // exclusive lock (it does not coalesce; that is maybeRefresh), and on
an actively consuming segment the
+ // reopen always flushes the writer, which stalls indexing. So only
refresh when this query can actually
+ // see past the last refresh. The added-doc watermark is read BEFORE
refreshing so rows arriving during
+ // the refresh are not wrongly claimed as visible.
+ if (!preFilterBitmap.isEmpty() && preFilterBitmap.last() >
_searcherRefreshedThroughDocId) {
Review Comment:
**Blocking correctness issue:** this freshness check assumes Pinot doc IDs
arrive monotonically, but `MutableIndex` explicitly allows additions in
arbitrary doc-ID order. For example, after refreshing through doc 10, adding
uncommitted doc 5 and searching with bitmap `{5}` skips refresh because `5 <=
_searcherRefreshedThroughDocId`; the stale searcher silently omits doc 5.
Please track an insertion/writer generation rather than the maximum doc ID, and
add an out-of-order regression test.
This path can also call `maybeRefreshBlocking()` once per interleaved
ingested row, forcing tiny Lucene flushes and serializing query callers. Please
coalesce generation-aware refreshes (or merge a stable ANN generation with an
exact-scanned unrefreshed tail) and validate concurrent ingest/query
performance.
--
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]
