This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang 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 88e4204cfe5 Fix resource leaks and cross-column cleanup in HNSW vector
index (#19175)
88e4204cfe5 is described below
commit 88e4204cfe5e5d2c80b457b79d71dbe4759e2e5e
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Fri Aug 7 11:49:11 2026 -0700
Fix resource leaks and cross-column cleanup in HNSW vector index (#19175)
---
.../realtime/impl/vector/MutableVectorIndex.java | 55 ++++++++++++----------
.../impl/vector/HnswVectorIndexCreator.java | 40 ++++++++--------
.../impl/vector/MutableVectorIndexTest.java | 22 ++++++++-
3 files changed, 68 insertions(+), 49 deletions(-)
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 6f47afb24c4..27d77ad37cf 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
@@ -36,6 +36,7 @@ import org.apache.lucene.search.IndexSearcher;
import org.apache.lucene.search.KnnFloatVectorQuery;
import org.apache.lucene.search.TopDocs;
import org.apache.lucene.store.FSDirectory;
+import org.apache.lucene.util.IOUtils;
import
org.apache.pinot.segment.local.realtime.impl.invertedindex.RealtimeLuceneTextIndexSearcherPool;
import
org.apache.pinot.segment.local.segment.creator.impl.vector.XKnnFloatVectorField;
import
org.apache.pinot.segment.local.segment.index.readers.vector.LuceneHnswRuntimeControlUtils;
@@ -65,14 +66,13 @@ public class MutableVectorIndex implements
VectorIndexReader, MutableIndex, Vect
private final int _vectorDimension;
private final VectorIndexConfig _vectorIndexConfig;
private final VectorSimilarityFunction _vectorSimilarityFunction;
- private final IndexWriter _indexWriter;
private final String _vectorColumn;
private final String _segmentName;
private final long _commitIntervalMs;
private final long _commitDocs;
private final File _indexDir;
-
private final FSDirectory _indexDirectory;
+ private final IndexWriter _indexWriter;
private int _nextDocId;
private long _lastCommitTime;
@@ -90,40 +90,33 @@ public class MutableVectorIndex implements
VectorIndexReader, MutableIndex, Vect
_commitDocs = Long.parseLong(
vectorIndexConfig.getProperties().getOrDefault("commitDocs",
String.valueOf(DEFAULT_COMMIT_DOCS)));
_vectorSimilarityFunction =
VectorIndexUtils.toSimilarityFunction(vectorIndexConfig.getVectorDistanceFunction());
- // Use local variables so that resources opened before a failure can be
closed before rethrowing,
- // preventing file-descriptor and temp-directory leaks.
- File indexDir = new File(FileUtils.getTempDirectory(), segmentName);
+ // 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);
+
FSDirectory indexDirectory = null;
IndexWriter indexWriter = 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.
- indexDirectory = FSDirectory.open(
- new File(indexDir, _vectorColumn +
V1Constants.Indexes.VECTOR_V912_HNSW_INDEX_FILE_EXTENSION).toPath());
+ indexDirectory = FSDirectory.open(_indexDir.toPath());
LOGGER.info("Creating mutable HNSW index for segment: {}, column: {} at
path: {} with {}", segmentName,
- vectorColumn, indexDir.getAbsolutePath(),
vectorIndexConfig.getProperties());
+ vectorColumn, _indexDir.getAbsolutePath(),
vectorIndexConfig.getProperties());
indexWriter = new IndexWriter(indexDirectory,
VectorIndexUtils.getIndexWriterConfig(vectorIndexConfig));
indexWriter.commit();
_lastCommitTime = System.currentTimeMillis();
} catch (Exception e) {
- if (indexWriter != null) {
- try {
- indexWriter.close();
- } catch (IOException closeEx) {
- e.addSuppressed(closeEx);
- }
- } else if (indexDirectory != null) {
- try {
- indexDirectory.close();
- } catch (IOException closeEx) {
- e.addSuppressed(closeEx);
- }
+ // IndexWriter does not close the Directory passed to it, so both need
to be closed.
+ try {
+ IOUtils.close(indexWriter, indexDirectory);
+ } catch (Exception closeEx) {
+ e.addSuppressed(closeEx);
}
- FileUtils.deleteQuietly(indexDir);
+ deleteIndexDir();
throw new RuntimeException(
"Caught exception while instantiating the LuceneTextIndexCreator for
column: " + vectorColumn, e);
}
- _indexDir = indexDir;
_indexDirectory = indexDirectory;
_indexWriter = indexWriter;
}
@@ -285,13 +278,23 @@ public class MutableVectorIndex implements
VectorIndexReader, MutableIndex, Vect
public void close() {
try {
_indexWriter.commit();
- _indexWriter.close();
- _indexDirectory.close();
+ // IndexWriter does not close the Directory passed to it, so both need
to be closed.
+ IOUtils.close(_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);
throw new RuntimeException(e);
} finally {
- // Delete the temporary index directory.
- FileUtils.deleteQuietly(_indexDir);
+ 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.
+ private void deleteIndexDir() {
+ FileUtils.deleteQuietly(_indexDir);
+ // Only succeeds when no other column of the same segment still has an
index directory under it.
+ //noinspection ResultOfMethodCallIgnored
+ _indexDir.getParentFile().delete();
+ }
}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/vector/HnswVectorIndexCreator.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/vector/HnswVectorIndexCreator.java
index 415e20602a1..9703ca7b5b9 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/vector/HnswVectorIndexCreator.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/vector/HnswVectorIndexCreator.java
@@ -28,6 +28,7 @@ import org.apache.lucene.index.IndexWriter;
import org.apache.lucene.index.VectorSimilarityFunction;
import org.apache.lucene.store.Directory;
import org.apache.lucene.store.FSDirectory;
+import org.apache.lucene.util.IOUtils;
import
org.apache.pinot.segment.local.segment.creator.impl.vector.lucene99.HnswVectorIndexCombined;
import org.apache.pinot.segment.local.segment.store.VectorIndexUtils;
import org.apache.pinot.segment.spi.V1Constants;
@@ -44,8 +45,6 @@ public class HnswVectorIndexCreator implements
VectorIndexCreator {
private static final Logger LOGGER =
LoggerFactory.getLogger(HnswVectorIndexCreator.class);
public static final String VECTOR_INDEX_DOC_ID_COLUMN_NAME = "DocID";
- private final Directory _indexDirectory;
- private final IndexWriter _indexWriter;
private final String _vectorColumn;
private final VectorSimilarityFunction _vectorSimilarityFunction;
private final int _vectorDimension;
@@ -53,6 +52,8 @@ public class HnswVectorIndexCreator implements
VectorIndexCreator {
private final boolean _storeInSegmentFile;
private final File _segmentIndexDir;
private final File _hnswIndexDir;
+ private final Directory _indexDirectory;
+ private final IndexWriter _indexWriter;
private int _nextDocId = 0;
@@ -62,27 +63,24 @@ public class HnswVectorIndexCreator implements
VectorIndexCreator {
_vectorSimilarityFunction =
VectorIndexUtils.toSimilarityFunction(vectorIndexConfig.getVectorDistanceFunction());
_storeInSegmentFile = vectorIndexConfig.isStoreInSegmentFile();
_segmentIndexDir = segmentIndexDir;
- // Use local variables so that, if IndexWriter construction fails after
FSDirectory is already
- // open, we can close the directory before rethrowing — preventing a
file-descriptor leak.
+ // 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.
+ _hnswIndexDir =
+ new File(segmentIndexDir, _vectorColumn +
V1Constants.Indexes.VECTOR_V912_HNSW_INDEX_FILE_EXTENSION);
+
Directory indexDirectory = null;
- IndexWriter indexWriter;
+ IndexWriter indexWriter = 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.
- File indexFile = new File(segmentIndexDir, _vectorColumn
- + V1Constants.Indexes.VECTOR_V912_HNSW_INDEX_FILE_EXTENSION);
- _hnswIndexDir = indexFile;
- indexDirectory = FSDirectory.open(indexFile.toPath());
+ indexDirectory = FSDirectory.open(_hnswIndexDir.toPath());
LOGGER.info("Creating HNSW index for column: {} at path: {} with {} for
segment: {}", column,
- indexFile.getAbsolutePath(), vectorIndexConfig.getProperties(),
segmentIndexDir.getAbsolutePath());
+ _hnswIndexDir.getAbsolutePath(), vectorIndexConfig.getProperties(),
segmentIndexDir.getAbsolutePath());
indexWriter = new IndexWriter(indexDirectory,
VectorIndexUtils.getIndexWriterConfig(vectorIndexConfig));
} catch (Exception e) {
- if (indexDirectory != null) {
- try {
- indexDirectory.close();
- } catch (IOException closeEx) {
- e.addSuppressed(closeEx);
- }
+ // IndexWriter does not close the Directory passed to it, so both need
to be closed.
+ try {
+ IOUtils.close(indexWriter, indexDirectory);
+ } catch (Exception closeEx) {
+ e.addSuppressed(closeEx);
}
throw new RuntimeException(
"Caught exception while instantiating the HnswVectorIndexCreator for
column: " + column, e);
@@ -129,9 +127,9 @@ public class HnswVectorIndexCreator implements
VectorIndexCreator {
public void close()
throws IOException {
try {
- // based on the commit flag set in IndexWriterConfig, this will decide
to commit or not
- _indexWriter.close();
- _indexDirectory.close();
+ // Based on the commit flag set in IndexWriterConfig, closing the writer
will decide to commit or not.
+ // IndexWriter does not close the Directory passed to it, so both need
to be closed.
+ IOUtils.close(_indexWriter, _indexDirectory);
} catch (Exception e) {
throw new RuntimeException("Caught exception while closing the HNSW
index for column: " + _vectorColumn, e);
}
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 74eaae5b76a..16b18b8509d 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
@@ -29,6 +29,7 @@ import org.testng.annotations.Test;
public class MutableVectorIndexTest {
private static final String COLUMN_NAME = "embedding";
+ private static final String OTHER_COLUMN_NAME = "otherEmbedding";
@BeforeClass
public void setUpSearcherPool() {
@@ -85,9 +86,26 @@ public class MutableVectorIndexTest {
}
}
+ @Test
+ public void testCloseOnlyRemovesIndexOfClosedColumn() {
+ String segmentName = "mutableVectorIndexTest_" + System.nanoTime();
+ MutableVectorIndex index = createIndex(segmentName, COLUMN_NAME);
+ MutableVectorIndex otherIndex = createIndex(segmentName,
OTHER_COLUMN_NAME);
+ try {
+ index.close();
+ int[] matches = otherIndex.getDocIds(new float[]{5.0F, 42.0F, 54.33333F,
42.24F, 3413.4F}, 3).toArray();
+ Assert.assertEquals(matches.length, 3);
+ } finally {
+ otherIndex.close();
+ }
+ }
+
private static MutableVectorIndex createIndex() {
- MutableVectorIndex index =
- new MutableVectorIndex("mutableVectorIndexTest_" + System.nanoTime(),
COLUMN_NAME, createConfig());
+ return createIndex("mutableVectorIndexTest_" + System.nanoTime(),
COLUMN_NAME);
+ }
+
+ private static MutableVectorIndex createIndex(String segmentName, String
column) {
+ MutableVectorIndex index = new MutableVectorIndex(segmentName, column,
createConfig());
addVector(index, new float[]{5.0F, 42.0F, 54.33333F, 42.24F, 1001.045F},
0);
addVector(index, new float[]{42.0F, 23423.0F, 42431.32532F, 6785676.3242F,
42.3F}, 1);
addVector(index, new float[]{1.0F, 2.0F, 3.0F, 4.0F, 5.0F}, 2);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
