You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "raghavyadav01 (via GitHub)" <gi...@apache.org> on 2026/06/25 21:36:40 UTC
[PR] Consolidate IVF and HNSW vector indexes into segment columns.psf (opt-in via storeInSegmentFile) [pinot]
raghavyadav01 opened a new pull request, #18852:
URL: https://github.com/apache/pinot/pull/18852
## Motivation
Today a vector index is written as a per-column sidecar in the segment directory: a flat
`.vector.ivfflat.index` / `.vector.ivfpq.index` file for IVF, or a Lucene HNSW directory for
HNSW (plus the lazily-built `.vector.hnsw.mapping` file). Every vector column adds extra
inodes per segment, and on large clusters with many vector columns the file count balloons.
The text index already solved this by packing its Lucene directory into `columns.psf` (the
segment's single-file index buffer) via the V2→V3 converter. This PR adds the same pattern
to IVF and HNSW vector indexes — opt-in via a new `storeInSegmentFile` flag on
`VectorIndexConfig`, default `false` so behaviour is unchanged.
End-to-end verification (against MinIO via an S3-tier offline build, same vector data
indexed twice — once with the flag, once without):
| Form | Files per segment on S3 | Cold-mmap query | HNSW prune (numDocsScanned/total) |
|---|---|---|---|
| Sidecar (`storeInSegmentFile=false`) | 9 | ~22 ms | 10 / 400 |
| Consolidated (`storeInSegmentFile=true`) | 4 | ~6 ms | 10 / 400 |
The 5 Lucene HNSW files (`_0.cfe`, `_0.cfs`, `_0.si`, `segments_1`, `write.lock`) and the
mapping file are gone in the consolidated form — those bytes are a typed entry inside
`columns.psf` (verified via `index_map`: `embedding.vector_index.startOffset=48, size=12034`).
## What changes
### IVF (flat file)
- **Readers consume `PinotDataBuffer`.** `IvfFlatVectorIndexReader`, `IvfOnDiskVectorIndexReader`,
and `IvfPqVectorIndexReader` now read from a `PinotDataBuffer` slice. New `IvfCombinedBuffers`
helper opens the buffer for both the consolidated path (slice owned by the segment directory)
and the legacy path (mmap of the sidecar). Readers honour an `ownsBuffer` flag so borrowed
slices (`columns.psf` views) are not closed by the reader's `close()`.
- **Build path.** `IvfFlatVectorIndexCreator`, `IvfPqVectorIndexCreator` write a transient
`.combined.index` file when the flag is on. The V2→V3 converter picks it up via the standard
`copyIndexIfExists` loop and packs it into `columns.psf`.
### HNSW (Lucene directory)
- **Build-time consolidation** via new `HnswVectorIndexCombined`: packs the Lucene HNSW
directory's files plus the docId mapping (when present) into a single
`.vector.hnsw.combined.index` file using the same `LUCENE_V2` layout the text index uses.
Reuses `LuceneCombinedTextIndexConstants` — no duplicate format definition.
- **Read path** via new `HnswVectorIndexBufferReader`: parses the combined buffer and returns
a Lucene `Directory` built on `PinotBufferLuceneDirectory` + `PinotBufferIndexInput` (reused
from the text-index package — `KnnVectorsReader` uses the same `Directory` interface, so no
HNSW-specific Directory adapter is needed).
- `HnswVectorIndexReader` gains a `PinotDataBuffer`-based constructor. The `DocIdTranslator`
reads its mapping from inside the combined buffer when available, or rebuilds it in heap
when not.
### Shared
- **Bidirectional migration** in `VectorIndexHandler`: forward (sidecar → `columns.psf`) and
reverse (`columns.psf` → sidecar) on a flag toggle, without rebuilding the index. For HNSW
the absorb step first packs the Lucene directory into a transient combined file; the extract
step unpacks the combined entry back into a Lucene directory.
- **Crash recovery.** `absorbCombinedIntoColumnsPsf` detects a previously-crashed absorb via
`hasIndexFor` + size check on the typed entry — no exception-message matching. Size mismatch
refuses with `IOException` rather than guessing which copy is authoritative.
- **Orphan cleanup.** A crash with `storeInSegmentFile=true` followed by a retry with the
flag off leaves a stale combined file + `.inprogress` marker; the next build sweeps both
before starting. Symmetric in the other direction. Works for both IVF (flat orphan) and HNSW
(legacy Lucene directory orphan, recursive delete).
- **HNSW extract ordering.** Unpack into Lucene directory runs BEFORE `removeIndex`. A crash
mid-unpack leaves the typed entry in `columns.psf` and the next handler run retries — no
permanent loss.
- **Mixed-state convert.** If a V1/V2 segment somehow has both a legacy sidecar and a
combined file for the same column, the V2→V3 converter packs the combined file and drops
the legacy sibling — combined wins.
### SPI / utility plumbing
- `V1Constants.Indexes` gains `VECTOR_IVF_FLAT_COMBINED_INDEX_FILE_EXTENSION`,
`VECTOR_IVF_PQ_COMBINED_INDEX_FILE_EXTENSION`, `VECTOR_HNSW_COMBINED_INDEX_FILE_EXTENSION`.
- `SegmentDirectoryPaths.findVectorIndexIndexFile` probes both legacy and combined
extensions for IVF and HNSW. Legacy wins on tie (stable behaviour for unchanged segments).
- `VectorIndexUtils.hasCombinedFormVectorIndex` is a new predicate the converter uses;
`hasVectorIndex` deliberately excludes the combined form so the converter knows to pack it
rather than preserve it as a sibling.
- `SingleFileIndexDirectory` symmetry fixes in `hasIndexFor`, `getColumnsWithIndex`,
`removeIndex` — vector entries are now treated uniformly whether they live as a sidecar or
as a `_columnEntries` slot.
## Backward compatibility
- Default behaviour unchanged. `storeInSegmentFile=false` preserves the legacy sidecar /
Lucene-directory layout exactly. Existing segments load and serve identically.
- No segment-format or wire-protocol break. The `.combined.index` extensions are build-time
transient and never appear in a V3 segment after the converter runs.
- Mixed-version safe: a server running this code reads a segment built without the flag
exactly as before; the new code only fires when the flag is set in the table config.
## Tests
- **132 tests green** (130 unit + 2 integration). New classes:
- `IvfFlatConsolidatedVectorTest` — end-to-end integration: build with flag=true, load,
run `vectorSimilarity` query, including a `MAP<STRING,STRING>` column to exercise the
S3A read path.
- `IvfDiskFormatInspectionTest` — byte-level header dumps validate creator output
matches the documented IVF format unchanged in both extensions.
- `IvfReaderFailurePathTest` — mmap-count tracking proves borrowed buffers are not
closed by the reader (`ownsBuffer=false`) and that constructor failures still release
owned buffers.
- `HnswVectorIndexCombinedTest` — pack/unpack round-trip plus an end-to-end HNSW build
read via the buffer-backed `Directory`.
- `VectorIndexHandlerTest` — bidirectional migration for both IVF and HNSW,
crash-recovery, size-mismatch refusal, extract-refusal-when-sidecar-exists, orphan
sweep in both directions.
- Spotless / checkstyle / license clean.
## Related
The buffer-backed Lucene `Directory` plumbing (`PinotBufferLuceneDirectory`,
`PinotBufferIndexInput`, `LuceneCombinedTextIndexConstants`) is reused as-is from the
text-index implementation — this PR adds zero new code in those generic helpers.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Consolidate IVF and HNSW vector indexes into segment columns.psf (opt-in via storeInSegmentFile) [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 merged PR #18852:
URL: https://github.com/apache/pinot/pull/18852
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Consolidate IVF and HNSW vector indexes into segment columns.psf (opt-in via storeInSegmentFile) [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18852:
URL: https://github.com/apache/pinot/pull/18852#discussion_r3523751104
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/VectorIndexHandler.java:
##########
@@ -78,6 +85,31 @@ public boolean needUpdateIndices(SegmentDirectory.Reader segmentReader) {
segmentName, column, existingBackend, desiredBackend);
return true;
}
+ // Migration: sidecar -> columns.psf. Fires either when the offline writer just wrote a
+ // sidecar (build with flag on) or when the operator just flipped the flag from off to on
+ // for an existing legacy segment. Same handler step covers both.
+ // For HNSW the "sidecar" is the combined packed file (.vector.hnsw.combined.index);
+ // for IVF it is the legacy combined extension.
+ if (desiredConfig.isStoreInSegmentFile()
+ && _segmentDirectory.getSegmentMetadata().getVersion() == SegmentVersion.v3
+ && existingBackend != null
+ && hasCombinedFile(indexDir, column, desiredBackend)) {
+ LOGGER.info("Need to consolidate Vector sidecar file into columns.psf for segment: {}, column: {}",
+ segmentName, column);
+ return true;
+ }
+ // Migration: columns.psf -> sidecar. Fires when the operator flipped the flag from on to
+ // off for a segment whose vector payload was previously absorbed. We detect this from the
+ // SegmentDirectory: the column has a vector index ({@code existingColumns} contains it via
+ // {@code SingleFileIndexDirectory._columnEntries}) but no sidecar file is on disk.
+ if (!desiredConfig.isStoreInSegmentFile()
+ && _segmentDirectory.getSegmentMetadata().getVersion() == SegmentVersion.v3
+ && existingBackend == null
Review Comment:
Fixed in 277ff1d02a (helpers in d8396fdcb5). `needUpdateIndices` and `updateIndices` now share `getDriftedConsolidatedBackend`, which sniffs the consolidated payload's leading magic (`LUCENE_V2` / `IVFF` / `IVPQ`, via the new `VectorIndexUtils.detectConsolidatedVectorBackend`) when `existingBackend == null` on a V3 segment. On a mismatch the column is rebuilt (`removeIndex` + re-add) instead of running the absorb/extract migrations — so a consolidated HNSW payload with the config flipped to IVF_PQ can no longer be extracted under `.vector.ivfpq.index` and parsed by the wrong reader.
One subtlety the fix had to handle: `IVF_ON_DISK` shares the IVF_FLAT file format, so its payloads sniff as `IVF_FLAT`. All four drift comparisons (sidecar + consolidated, check + act) therefore compare via `storageFormatOf` (IVF_ON_DISK → IVF_FLAT) — a raw enum comparison would have flagged a healthy IVF_ON_DISK column as drifted and destroyed/rebuilt it on every segment load, forever. This also fixes the same latent loop in the pre-existing sidecar drift check. Unknown-magic payloads deliberately do NOT count as drift (extract preserves the bytes; rebuilding would destroy an unrecognized-but-possibly-valid payload).
Regression tests: `testNeedUpdateIndicesReturnsTrueWhenConsolidatedBackendDiffers`, `testUpdateIndicesRebuildsInsteadOfExtractingWhenConsolidatedBackendDiffers` (asserts no `.vector.ivfpq.index` is written and the column reaches the rebuild loop), `testUpdateIndicesExtractsWhenConsolidatedBackendMatches`, plus IVF_ON_DISK no-drift tests for both consolidated and sidecar layouts, and magic-sniff unit tests.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/vector/VectorIndexType.java:
##########
@@ -210,25 +211,85 @@ public VectorIndexReader createIndexReader(SegmentDirectory.Reader segmentReader
return null;
}
VectorBackendType backendType = indexConfig.resolveBackendType();
- File configuredIndexFile =
- SegmentDirectoryPaths.findVectorIndexIndexFile(segmentDir, metadata.getColumnName(), indexConfig);
- if (configuredIndexFile == null || !configuredIndexFile.exists()) {
- LOGGER.warn("Skipping vector index reader for column: {} because configured backend {} does not have a "
- + "matching on-disk artifact in segment: {}",
- metadata.getColumnName(), backendType, segmentDir);
- return null;
+ String column = metadata.getColumnName();
+
+ if (backendType == VectorBackendType.HNSW) {
+ // Combined form: load the HNSW index from the typed entry inside columns.psf when one
+ // actually exists. getConsolidatedVectorEntry (unlike hasIndexFor) does not mistake the
+ // legacy Lucene directory for a packed entry, so a segment whose handler has not yet
+ // migrated falls through to the legacy path below instead of failing the whole load. The
+ // buffer is owned by the segment directory — this reader must not close it.
+ if (indexConfig.isStoreInSegmentFile()) {
+ PinotDataBuffer buffer;
+ try {
+ buffer = VectorIndexUtils.getConsolidatedVectorEntry(segmentReader, column);
Review Comment:
Fixed in 4cef636d3a (utility hardening in d8396fdcb5), two layers:
1. **Reader factory — legacy-directory-first**: with `storeInSegmentFile=true`, the HNSW branch now checks the on-disk artifact before probing columns.psf. If `findVectorIndexIndexFile(HNSW)` resolves a Lucene *directory* (unmigrated V3, or any V1/V2 `FilePerIndexDirectory` segment), it reads the directory directly and never calls `getIndexFor` — so the `mapForReads` "must be a regular file" rejection can no longer kill the load. Tests cover both the V1 root layout and the `v3/` subdirectory layout, asserting via `verify(never())` that the probe does not run.
2. **Utility hardening**: `getConsolidatedVectorEntry` itself now maps `FilePerIndexDirectory`'s directory-resolution `IllegalArgumentException` (matched via a shared message-suffix constant) to `null` — a directory can never be a packed typed entry — so any future ungated caller is protected too, and its contract Javadoc now states the per-store behavior honestly (on V1/V2, an IVF sidecar *file* is returned as-is; callers needing "truly consolidated" must gate on version or on-disk artifacts).
Note: `TextIndexType.ReaderFactory` (which this vector machinery was modeled on) has the identical unguarded probe on master for legacy `.lucene.index` directories — filing that as a separate follow-up rather than expanding this PR.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Consolidate IVF and HNSW vector indexes into segment columns.psf (opt-in via storeInSegmentFile) [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #18852:
URL: https://github.com/apache/pinot/pull/18852#discussion_r3486025461
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/store/SingleFileIndexDirectory.java:
##########
@@ -525,12 +531,14 @@ public Set<String> getColumnsWithIndex(IndexType<?, ?, ?> type) {
}
}
if (type == StandardIndexes.vector()) {
+ // Vector may live as a combined file (legacy / storeInSegmentFile=false) or as a typed
+ // entry in columns.psf (storeInSegmentFile=true). Collect both. Removed the early-return
+ // that previously hid consolidated entries from this view.
for (String column : _segmentMetadata.getAllColumns()) {
if (VectorIndexUtils.hasVectorIndex(_segmentDirectory, column)) {
columns.add(column);
}
}
Review Comment:
`getColumnsWithIndex(StandardIndexes.vector())` only adds columns when `hasVectorIndex(...)` is true, so columns that only have the transient `*.combined.index` form will be omitted. That can break handler/loader logic that relies on this set during migrations or crash recovery.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/store/SingleFileIndexDirectory.java:
##########
@@ -160,8 +160,11 @@ public boolean hasIndexFor(String column, IndexType<?, ?, ?> type) {
if (type == StandardIndexes.text() && TextIndexUtils.hasTextIndex(_segmentDirectory, column)) {
return true;
}
- if (type == StandardIndexes.vector()) {
- return VectorIndexUtils.hasVectorIndex(_segmentDirectory, column);
+ // Vector index may live either as a combined file (legacy / storeInSegmentFile=false) or as
+ // a typed entry inside columns.psf (storeInSegmentFile=true). Check both — mirror the text
+ // pattern of "combined OR _columnEntries".
+ if (type == StandardIndexes.vector() && VectorIndexUtils.hasVectorIndex(_segmentDirectory, column)) {
+ return true;
}
Review Comment:
`hasIndexFor(..., StandardIndexes.vector())` only checks `VectorIndexUtils.hasVectorIndex`, which deliberately excludes the new `*.combined.index` transient form. If a segment directory temporarily contains only the combined-form vector index file (e.g. crash/rollback before absorb), `hasIndexFor` will return false and the vector reader will never be constructed even though `SegmentDirectoryPaths.findVectorIndexIndexFile(...)` can now resolve the combined file.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/store/VectorIndexUtils.java:
##########
@@ -76,6 +95,39 @@ static boolean hasVectorIndex(File segDir, String column) {
|| new File(segDir, column + Indexes.VECTOR_IVF_PQ_INDEX_FILE_EXTENSION).exists();
}
+ /// Returns {@code true} when the V1/V2 segment directory holds an IVF vector index in the
+ /// combined-form extension ({@code .vector.ivfflat.combined.index} or
+ /// {@code .vector.ivfpq.combined.index}). The combined form is written by an IVF creator run
+ /// with {@code storeInSegmentFile=true} and is meant to be packed into {@code columns.psf} by
+ /// the V2→V3 converter, not preserved as a sibling.
+ public static boolean hasCombinedFormVectorIndex(File segDir, String column) {
+ return new File(segDir, column + Indexes.VECTOR_IVF_FLAT_COMBINED_INDEX_FILE_EXTENSION).exists()
+ || new File(segDir, column + Indexes.VECTOR_IVF_PQ_COMBINED_INDEX_FILE_EXTENSION).exists()
+ || new File(segDir, column + Indexes.VECTOR_HNSW_COMBINED_INDEX_FILE_EXTENSION).exists();
+ }
+
+ /// Returns the {@code columns.psf} typed-entry buffer holding the column's consolidated vector
+ /// index, or {@code null} when no such entry has been packed into {@code columns.psf} yet.
+ ///
+ /// Unlike {@link SegmentDirectory.Reader#hasIndexFor}, this does NOT report a legacy on-disk
+ /// sidecar (an IVF flat file or an HNSW Lucene directory) as a match — only a real packed
+ /// `_columnEntries` slot counts. {@code SingleFileIndexDirectory} signals an absent typed slot by
+ /// throwing an unchecked exception from {@code getIndexFor}; that is mapped to {@code null} here,
+ /// while genuine I/O failures propagate. Callers use this to tell "a prior absorb already
+ /// committed bytes" (crash recovery) apart from "first absorb of an existing sidecar", and to
+ /// select the {@code columns.psf} read path only when the consolidated entry truly exists.
+ ///
+ /// The returned buffer is owned by the segment directory and must NOT be closed by the caller.
+ @Nullable
+ public static PinotDataBuffer getConsolidatedVectorEntry(SegmentDirectory.Reader reader, String column)
+ throws IOException {
+ try {
+ return reader.getIndexFor(column, StandardIndexes.vector());
+ } catch (RuntimeException e) {
+ return null;
+ }
Review Comment:
`getConsolidatedVectorEntry` catches *all* `RuntimeException` from `reader.getIndexFor(...)` and converts it to `null`. In `SingleFileIndexDirectory`, `getIndexFor` throws `RuntimeException` not only for "missing typed entry" but also for corruption scenarios (e.g. missing magic marker). Swallowing those will make real corruption look like "absent entry" and can cause migration/crash-recovery to proceed on a broken segment without surfacing the underlying problem.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/vector/VectorIndexType.java:
##########
@@ -210,25 +211,81 @@ public VectorIndexReader createIndexReader(SegmentDirectory.Reader segmentReader
return null;
}
VectorBackendType backendType = indexConfig.resolveBackendType();
- File configuredIndexFile =
- SegmentDirectoryPaths.findVectorIndexIndexFile(segmentDir, metadata.getColumnName(), indexConfig);
- if (configuredIndexFile == null || !configuredIndexFile.exists()) {
- LOGGER.warn("Skipping vector index reader for column: {} because configured backend {} does not have a "
- + "matching on-disk artifact in segment: {}",
- metadata.getColumnName(), backendType, segmentDir);
- return null;
+ String column = metadata.getColumnName();
+
+ if (backendType == VectorBackendType.HNSW) {
+ // Combined form: load the HNSW index from the typed entry inside columns.psf when one
+ // actually exists. getConsolidatedVectorEntry (unlike hasIndexFor) does not mistake the
+ // legacy Lucene directory for a packed entry, so a segment whose handler has not yet
+ // migrated falls through to the legacy path below instead of failing the whole load. The
+ // buffer is owned by the segment directory — this reader must not close it.
+ if (indexConfig.isStoreInSegmentFile()) {
+ PinotDataBuffer buffer;
+ try {
+ buffer = VectorIndexUtils.getConsolidatedVectorEntry(segmentReader, column);
+ } catch (IOException e) {
+ throw new RuntimeException(
+ "Failed to read consolidated HNSW vector index from columns.psf for column: " + column, e);
+ }
+ if (buffer != null) {
+ return new HnswVectorIndexReader(column, buffer, metadata.getTotalDocs(), indexConfig);
+ }
+ LOGGER.warn("storeInSegmentFile=true but no consolidated HNSW entry found in columns.psf for column: {} "
+ + "in segment: {}; falling back to the on-disk Lucene directory", column, segmentDir);
+ }
+ // Legacy path: load the HNSW index from the Lucene directory on disk.
+ return new HnswVectorIndexReader(column, segmentDir, metadata.getTotalDocs(), indexConfig);
+ }
+
+ // IVF backends accept a PinotDataBuffer; that buffer either comes from the consolidated
+ // typed entry inside columns.psf (when storeInSegmentFile=true) or from the legacy combined
+ // file. The chosen reader takes ownership of the buffer and is responsible for closing it
+ // (including the constructor's own failure path).
+ PinotDataBuffer buffer;
+ if (indexConfig.isStoreInSegmentFile()) {
+ try {
+ buffer = VectorIndexUtils.getConsolidatedVectorEntry(segmentReader, column);
+ } catch (IOException e) {
+ throw new RuntimeException(
+ "Failed to read consolidated vector index from columns.psf for column: " + column, e);
+ }
+ if (buffer == null) {
+ LOGGER.warn("Skipping vector index reader for column: {} because storeInSegmentFile=true "
+ + "but no consolidated entry was found in columns.psf in segment: {}", column, segmentDir);
+ return null;
+ }
+ } else {
+ File configuredIndexFile = SegmentDirectoryPaths.findVectorIndexIndexFile(segmentDir, column, indexConfig);
+ if (configuredIndexFile == null || !configuredIndexFile.exists()) {
+ LOGGER.warn("Skipping vector index reader for column: {} because configured backend {} does not have a "
+ + "matching on-disk artifact in segment: {}", column, backendType, segmentDir);
+ return null;
+ }
+ buffer = IvfCombinedBuffers.mapCombinedFile(configuredIndexFile, column,
+ "vector-" + backendType.name().toLowerCase());
}
+ // Buffer ownership: when reading from columns.psf, the segment directory owns the buffer
+ // and the reader must not close it. Combined mmap buffers are owned by the reader.
+ boolean ownsBuffer = !indexConfig.isStoreInSegmentFile();
Review Comment:
After adding IVF fallback to an on-disk sidecar, buffer ownership should be derived from where the buffer came from (columns.psf vs mmap sidecar), not from the config flag alone. Otherwise a sidecar-mapped buffer used as a fallback would be treated as "borrowed" and leaked.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/vector/VectorIndexType.java:
##########
@@ -210,25 +211,81 @@ public VectorIndexReader createIndexReader(SegmentDirectory.Reader segmentReader
return null;
}
VectorBackendType backendType = indexConfig.resolveBackendType();
- File configuredIndexFile =
- SegmentDirectoryPaths.findVectorIndexIndexFile(segmentDir, metadata.getColumnName(), indexConfig);
- if (configuredIndexFile == null || !configuredIndexFile.exists()) {
- LOGGER.warn("Skipping vector index reader for column: {} because configured backend {} does not have a "
- + "matching on-disk artifact in segment: {}",
- metadata.getColumnName(), backendType, segmentDir);
- return null;
+ String column = metadata.getColumnName();
+
+ if (backendType == VectorBackendType.HNSW) {
+ // Combined form: load the HNSW index from the typed entry inside columns.psf when one
+ // actually exists. getConsolidatedVectorEntry (unlike hasIndexFor) does not mistake the
+ // legacy Lucene directory for a packed entry, so a segment whose handler has not yet
+ // migrated falls through to the legacy path below instead of failing the whole load. The
+ // buffer is owned by the segment directory — this reader must not close it.
+ if (indexConfig.isStoreInSegmentFile()) {
+ PinotDataBuffer buffer;
+ try {
+ buffer = VectorIndexUtils.getConsolidatedVectorEntry(segmentReader, column);
+ } catch (IOException e) {
+ throw new RuntimeException(
+ "Failed to read consolidated HNSW vector index from columns.psf for column: " + column, e);
+ }
+ if (buffer != null) {
+ return new HnswVectorIndexReader(column, buffer, metadata.getTotalDocs(), indexConfig);
+ }
+ LOGGER.warn("storeInSegmentFile=true but no consolidated HNSW entry found in columns.psf for column: {} "
+ + "in segment: {}; falling back to the on-disk Lucene directory", column, segmentDir);
+ }
+ // Legacy path: load the HNSW index from the Lucene directory on disk.
+ return new HnswVectorIndexReader(column, segmentDir, metadata.getTotalDocs(), indexConfig);
+ }
+
+ // IVF backends accept a PinotDataBuffer; that buffer either comes from the consolidated
+ // typed entry inside columns.psf (when storeInSegmentFile=true) or from the legacy combined
+ // file. The chosen reader takes ownership of the buffer and is responsible for closing it
+ // (including the constructor's own failure path).
+ PinotDataBuffer buffer;
+ if (indexConfig.isStoreInSegmentFile()) {
+ try {
+ buffer = VectorIndexUtils.getConsolidatedVectorEntry(segmentReader, column);
+ } catch (IOException e) {
+ throw new RuntimeException(
+ "Failed to read consolidated vector index from columns.psf for column: " + column, e);
+ }
+ if (buffer == null) {
+ LOGGER.warn("Skipping vector index reader for column: {} because storeInSegmentFile=true "
+ + "but no consolidated entry was found in columns.psf in segment: {}", column, segmentDir);
+ return null;
+ }
+ } else {
+ File configuredIndexFile = SegmentDirectoryPaths.findVectorIndexIndexFile(segmentDir, column, indexConfig);
+ if (configuredIndexFile == null || !configuredIndexFile.exists()) {
+ LOGGER.warn("Skipping vector index reader for column: {} because configured backend {} does not have a "
+ + "matching on-disk artifact in segment: {}", column, backendType, segmentDir);
+ return null;
+ }
+ buffer = IvfCombinedBuffers.mapCombinedFile(configuredIndexFile, column,
+ "vector-" + backendType.name().toLowerCase());
}
Review Comment:
For IVF backends, when `storeInSegmentFile=true` but the consolidated typed entry is missing (e.g. before/after a failed absorb), the reader factory returns null and silently disables the vector index even if a usable on-disk artifact exists. HNSW already falls back to legacy in this situation; IVF should do the same so the system can still use the index (and avoid unexpected exact-scan fallback) until migration completes.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/readers/vector/HnswVectorIndexReader.java:
##########
@@ -86,6 +91,36 @@ public HnswVectorIndexReader(String column, File indexDir, int numDocs, VectorIn
}
}
+ /**
+ * Buffer-backed constructor: reads the HNSW index from a combined {@link PinotDataBuffer}
+ * (the {@code LUCENE_V2} packed form produced by {@code HnswVectorIndexCombined}).
+ *
+ * <p>The buffer is <em>not</em> owned by this reader — closing this reader does not close the
+ * buffer. The buffer's lifetime must exceed this reader's lifetime; the segment directory is
+ * responsible for closing it.</p>
+ *
+ * @param column column name
+ * @param indexBuffer combined buffer in LUCENE_V2 format; not owned by this reader
+ * @param numDocs number of documents in the segment
+ * @param config vector index configuration
+ */
+ public HnswVectorIndexReader(String column, PinotDataBuffer indexBuffer, int numDocs, VectorIndexConfig config) {
+ _column = column;
+ try {
+ _indexDirectory = HnswVectorIndexBufferReader.createLuceneDirectory(indexBuffer, column);
+ _indexReader = DirectoryReader.open(_indexDirectory);
+ _indexSearcher = new IndexSearcher(_indexReader);
+
+ // Try to extract the mapping from the packed buffer first; build from the Lucene index if absent.
+ PinotDataBuffer mappingBuffer = HnswVectorIndexBufferReader.extractDocIdMappingBuffer(indexBuffer, column);
+ _docIdTranslator = new DocIdTranslator(mappingBuffer, numDocs, _indexSearcher);
+ } catch (Exception e) {
+ LOGGER.error("Failed to instantiate buffer-backed HNSW index reader for column {}, exception {}", column,
+ e.getMessage());
+ throw new RuntimeException(e);
+ }
Review Comment:
The new buffer-backed constructor logs only `e.getMessage()` and drops the throwable, which loses stack traces in server logs (making segment-load failures hard to debug). Log the exception itself via the SLF4J throwable parameter.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/vector/lucene99/HnswVectorIndexCombined.java:
##########
@@ -0,0 +1,347 @@
+/**
+ * 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.creator.impl.vector.lucene99;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.channels.FileChannel;
+import java.nio.file.StandardOpenOption;
+import java.util.Map;
+import java.util.TreeMap;
+import javax.annotation.Nullable;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.segment.local.segment.creator.impl.text.LuceneCombinedTextIndexConstants;
+import org.apache.pinot.segment.spi.V1Constants;
+import org.apache.pinot.segment.spi.store.SegmentDirectoryPaths;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Utility class to pack a Lucene HNSW index directory (and its optional docId mapping file) into
+/// a single combined file using the {@link LuceneCombinedTextIndexConstants#MAGIC_NUMBER LUCENE_V2}
+/// layout. Mirrors {@code LuceneTextIndexCombined} — reuses the same format constants to keep a
+/// single on-disk format shared across text and HNSW vector indexes.
+///
+/// Layout (identical to the text-index LUCENE_V2 format):
+/// ```
+/// [Header]
+/// Magic "LUCENE_V2" 9 bytes
+/// Version 4 bytes (little-endian int)
+/// Total buffer size 8 bytes (little-endian long)
+/// File count 4 bytes (little-endian int)
+/// Reserved 4 bytes
+///
+/// [File metadata, one entry per file]
+/// Name length 2 bytes (little-endian short)
+/// Name variable
+/// File offset 8 bytes (little-endian long)
+/// File size 8 bytes (little-endian long)
+///
+/// [File data]
+/// Raw bytes of each file concatenated in metadata order
+/// ```
+public final class HnswVectorIndexCombined {
+ private static final Logger LOGGER = LoggerFactory.getLogger(HnswVectorIndexCombined.class);
+
+ private HnswVectorIndexCombined() {
+ }
+
+ /// Packs all files in {@code hnswIndexDir} (plus the optional docId mapping file) into a single
+ /// combined file at {@code outputFilePath}.
+ ///
+ /// @param hnswIndexDir the Lucene HNSW index directory to pack
+ /// @param outputFilePath destination path for the combined file
+ /// @param segmentIndexDir when non-null, the segment's top-level index directory; used to
+ /// locate the docId mapping file for inclusion in the packed output
+ /// @param column column name; used to locate the docId mapping file when
+ /// {@code segmentIndexDir} is provided
+ /// @throws IOException if any file operations fail
+ public static void combineHnswIndexFiles(File hnswIndexDir, String outputFilePath,
+ @Nullable File segmentIndexDir, @Nullable String column)
+ throws IOException {
+ if (!hnswIndexDir.exists() || !hnswIndexDir.isDirectory()) {
+ throw new IllegalArgumentException(
+ "HNSW index directory does not exist or is not a directory: " + hnswIndexDir);
+ }
+
+ LOGGER.info("Combining HNSW index files from directory: {}", hnswIndexDir.getAbsolutePath());
+
+ Map<String, FileInfo> fileInfoMap = collectFiles(hnswIndexDir, segmentIndexDir, column);
+ int fileCount = fileInfoMap.size();
+
+ if (fileCount == 0) {
+ throw new IOException("No files found in HNSW index directory: " + hnswIndexDir);
+ }
+
+ long totalSize = calculateTotalBufferSize(fileInfoMap);
+ if (totalSize > Integer.MAX_VALUE) {
+ throw new IOException("Combined HNSW index size too large: " + totalSize + " bytes");
+ }
+
+ File outputFile = new File(outputFilePath);
+ // TRUNCATE_EXISTING so a leftover (larger) file from a previously-crashed pack does not leave
+ // stale trailing bytes past the new payload — that would inflate the file length and break the
+ // size-based crash-recovery check in VectorIndexHandler that compares the combined file length
+ // to the columns.psf typed-entry size.
+ try (FileChannel outputChannel = FileChannel.open(outputFile.toPath(), StandardOpenOption.CREATE,
+ StandardOpenOption.WRITE, StandardOpenOption.TRUNCATE_EXISTING)) {
+ writeHeader(outputChannel, fileCount, (int) totalSize);
+ long dataOffset = LuceneCombinedTextIndexConstants.getHeaderSize() + calculateMetadataSize(fileInfoMap);
+ writeFileMetadata(outputChannel, fileInfoMap, dataOffset);
+ writeFileData(outputChannel, fileInfoMap);
+ }
+
+ LOGGER.info("Combined {} HNSW index files into: {} ({} bytes)", fileCount, outputFilePath, totalSize);
+ }
+
+ /// Collects all regular files from {@code hnswIndexDir} and optionally the docId mapping file
+ /// from the segment's flat directory. Uses a {@link TreeMap} for deterministic ordering.
+ private static Map<String, FileInfo> collectFiles(File hnswIndexDir, @Nullable File segmentIndexDir,
+ @Nullable String column)
+ throws IOException {
+ Map<String, FileInfo> fileInfoMap = new TreeMap<>();
+
+ File[] files = hnswIndexDir.listFiles();
+ if (files != null) {
+ for (File file : files) {
+ if (file.isFile()) {
+ fileInfoMap.put(file.getName(), new FileInfo(file, file.getName(), file.length()));
+ }
+ }
+ }
+
+ // Include the docId mapping file when available. It lives beside the HNSW directory, not
+ // inside it (same convention as text-index).
+ if (segmentIndexDir != null && column != null) {
+ File segmentDir = SegmentDirectoryPaths.findSegmentDirectory(segmentIndexDir);
+ File mappingFile = new File(segmentDir,
+ column + V1Constants.Indexes.VECTOR_HNSW_INDEX_DOCID_MAPPING_FILE_EXTENSION);
+ if (mappingFile.exists() && mappingFile.isFile()) {
+ fileInfoMap.put(mappingFile.getName(), new FileInfo(mappingFile, mappingFile.getName(), mappingFile.length()));
+ LOGGER.info("Including docId mapping file: {} ({} bytes)", mappingFile.getName(), mappingFile.length());
+ }
+ }
+
+ return fileInfoMap;
+ }
+
+ /// Extracts files packed inside a combined HNSW file back into a Lucene directory.
+ ///
+ /// This is the inverse of {@link #combineHnswIndexFiles}. The docId mapping file (if present)
+ /// is extracted into {@code targetDir} alongside the Lucene index files; the caller is
+ /// responsible for moving it to its canonical location if needed.
+ ///
+ /// @param combinedFile source combined file (LUCENE_V2 layout)
+ /// @param targetDir destination directory; created if absent
+ /// @throws IOException if any file operations fail
+ public static void extractHnswIndexFiles(File combinedFile, File targetDir)
+ throws IOException {
+ if (!combinedFile.exists() || !combinedFile.isFile()) {
+ throw new IllegalArgumentException("Combined file does not exist or is not a file: " + combinedFile);
+ }
+ FileUtils.forceMkdir(targetDir);
+
+ try (FileChannel inputChannel = FileChannel.open(combinedFile.toPath(), StandardOpenOption.READ)) {
+ // Parse header
+ byte[] magicBytes = new byte[LuceneCombinedTextIndexConstants.MAGIC_NUMBER_LENGTH];
+ readFully(inputChannel, ByteBuffer.wrap(magicBytes));
+ String magic = new String(magicBytes);
+ if (!LuceneCombinedTextIndexConstants.MAGIC_NUMBER.equals(magic)) {
+ throw new IOException("Invalid magic number in combined HNSW file: " + magic);
+ }
+
+ ByteBuffer intBuf = ByteBuffer.allocate(Integer.BYTES).order(ByteOrder.LITTLE_ENDIAN);
+ readFully(inputChannel, intBuf);
+ intBuf.flip();
+ int version = intBuf.getInt();
+ if (version != LuceneCombinedTextIndexConstants.VERSION) {
+ throw new IOException("Unsupported version in combined HNSW file: " + version);
+ }
+
+ // Skip total size (8 bytes) and file count (4 bytes) header fields
+ ByteBuffer longBuf = ByteBuffer.allocate(Long.BYTES).order(ByteOrder.LITTLE_ENDIAN);
+ readFully(inputChannel, longBuf); // totalSize
+ longBuf.flip();
+ // (unused, but we advance the channel position past it)
+
+ intBuf = ByteBuffer.allocate(Integer.BYTES).order(ByteOrder.LITTLE_ENDIAN);
+ readFully(inputChannel, intBuf);
+ intBuf.flip();
+ int fileCount = intBuf.getInt();
+
+ // Skip reserved field
+ intBuf = ByteBuffer.allocate(Integer.BYTES).order(ByteOrder.LITTLE_ENDIAN);
+ readFully(inputChannel, intBuf);
+
+ // Parse file metadata
+ String[] fileNames = new String[fileCount];
+ long[] fileOffsets = new long[fileCount];
+ long[] fileSizes = new long[fileCount];
+ for (int i = 0; i < fileCount; i++) {
+ ByteBuffer shortBuf = ByteBuffer.allocate(Short.BYTES).order(ByteOrder.LITTLE_ENDIAN);
+ readFully(inputChannel, shortBuf);
+ shortBuf.flip();
+ short nameLength = shortBuf.getShort();
+
+ byte[] nameBytes = new byte[nameLength];
+ readFully(inputChannel, ByteBuffer.wrap(nameBytes));
+ fileNames[i] = new String(nameBytes);
+
+ longBuf = ByteBuffer.allocate(Long.BYTES).order(ByteOrder.LITTLE_ENDIAN);
+ readFully(inputChannel, longBuf);
+ longBuf.flip();
+ fileOffsets[i] = longBuf.getLong();
+
+ longBuf = ByteBuffer.allocate(Long.BYTES).order(ByteOrder.LITTLE_ENDIAN);
+ readFully(inputChannel, longBuf);
+ longBuf.flip();
+ fileSizes[i] = longBuf.getLong();
+ }
+
+ // Extract each file by seeking to its offset and copying bytes
+ for (int i = 0; i < fileCount; i++) {
+ File outFile = new File(targetDir, fileNames[i]);
+ long fileSize = fileSizes[i];
+ try (FileChannel outChannel = FileChannel.open(outFile.toPath(), StandardOpenOption.CREATE,
+ StandardOpenOption.WRITE)) {
+ long remaining = fileSize;
+ long srcOffset = fileOffsets[i];
+ while (remaining > 0) {
+ long transferred = inputChannel.transferTo(srcOffset, remaining, outChannel);
+ srcOffset += transferred;
+ remaining -= transferred;
+ }
Review Comment:
`extractHnswIndexFiles` loops on `FileChannel.transferTo` without a progress check. If the combined file is corrupt (bad offsets) or `transferTo` returns 0, this can become an infinite loop during segment load/migration.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Consolidate IVF and HNSW vector indexes into segment columns.psf (opt-in via storeInSegmentFile) [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18852:
URL: https://github.com/apache/pinot/pull/18852#discussion_r3481296172
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/readers/vector/HnswVectorIndexBufferReader.java:
##########
@@ -0,0 +1,119 @@
+/**
+ * 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 java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import javax.annotation.Nullable;
+import org.apache.lucene.store.Directory;
+import org.apache.pinot.segment.local.segment.index.readers.text.LuceneTextIndexBufferReader;
+import org.apache.pinot.segment.local.segment.index.readers.text.LuceneTextIndexHeader;
+import org.apache.pinot.segment.local.segment.index.readers.text.PinotBufferLuceneDirectory;
+import org.apache.pinot.segment.spi.V1Constants;
+import org.apache.pinot.segment.spi.memory.PinotDataBuffer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Utility for reading an HNSW vector index from a combined buffer ({@link PinotDataBuffer}).
+ *
+ * <p>The combined buffer uses the same {@code LUCENE_V2} layout as the text index (written by
+ * {@code HnswVectorIndexCombined}). This class delegates header and metadata parsing to
+ * {@link LuceneTextIndexBufferReader}, then constructs a {@link PinotBufferLuceneDirectory} that
+ * Lucene's {@code DirectoryReader} can open without any filesystem access.</p>
+ *
+ * <p>Two entry-points are provided:</p>
+ * <ul>
+ * <li>{@link #createLuceneDirectory} — a read-only {@link Directory} over the Lucene index
+ * files embedded in the buffer.</li>
+ * <li>{@link #extractDocIdMappingBuffer} — a sub-buffer view covering just the docId mapping
+ * bytes, or {@code null} if the mapping was not packed.</li>
+ * </ul>
+ */
+public final class HnswVectorIndexBufferReader {
+ private static final Logger LOGGER = LoggerFactory.getLogger(HnswVectorIndexBufferReader.class);
+
+ private HnswVectorIndexBufferReader() {
+ }
+
+ /**
+ * Creates a Lucene {@link Directory} that reads all Lucene index files from a combined HNSW
+ * buffer. The docId mapping file (a Pinot-only artifact with the
+ * {@link V1Constants.Indexes#VECTOR_HNSW_INDEX_DOCID_MAPPING_FILE_EXTENSION} suffix) is
+ * excluded from the directory view so that {@code DirectoryReader} does not see an unknown file.
+ *
+ * <p>The directory does <em>not</em> own the buffer — closing it is a no-op. The caller must
+ * keep the buffer alive for as long as the directory (and any reader opened on it) is in use.</p>
+ *
+ * @param indexBuffer combined buffer in LUCENE_V2 format
+ * @param column column name; used to identify and exclude the mapping file
+ * @return a read-only {@link Directory} backed by the buffer
+ * @throws IOException if the buffer is malformed or the magic/version check fails
+ */
+ public static Directory createLuceneDirectory(PinotDataBuffer indexBuffer, String column)
+ throws IOException {
+ List<String> fileNames = LuceneTextIndexBufferReader.listFiles(indexBuffer);
+ Map<String, LuceneTextIndexHeader.FileInfo> fileMap = new HashMap<>(fileNames.size());
+
+ String mappingFileName = column + V1Constants.Indexes.VECTOR_HNSW_INDEX_DOCID_MAPPING_FILE_EXTENSION;
+ for (String name : fileNames) {
+ if (name.equals(mappingFileName)) {
+ // Skip: not a Lucene index file; presenting it to DirectoryReader triggers an error.
+ continue;
+ }
+ LuceneTextIndexHeader.FileInfo info = LuceneTextIndexBufferReader.getFileInfo(indexBuffer, name);
+ if (info != null) {
+ fileMap.put(name, info);
+ }
+ }
+
+ LOGGER.debug("Creating buffer-backed Lucene directory for HNSW column '{}' with {} file(s)", column,
+ fileMap.size());
+ return new PinotBufferLuceneDirectory(indexBuffer, fileMap, column);
+ }
+
+ /**
+ * Extracts a sub-buffer covering the docId mapping bytes packed inside the combined buffer.
+ *
+ * <p>Returns {@code null} when the mapping was not included (e.g. the index was built from an
+ * in-memory segment that had not yet materialised a mapping file). The caller must build the
+ * mapping from the Lucene index in that case.</p>
+ *
+ * <p>The returned sub-buffer is a {@link PinotDataBuffer#view view} of the original buffer and
+ * shares its lifetime. The caller must not close it independently.</p>
+ *
+ * @param indexBuffer combined buffer in LUCENE_V2 format
+ * @param column column name; used to identify the mapping file entry
+ * @return sub-buffer for the mapping bytes, or {@code null} if not present
+ * @throws IOException if header parsing fails
+ */
+ @Nullable
+ public static PinotDataBuffer extractDocIdMappingBuffer(PinotDataBuffer indexBuffer, String column)
+ throws IOException {
+ String mappingFileName = column + V1Constants.Indexes.VECTOR_HNSW_INDEX_DOCID_MAPPING_FILE_EXTENSION;
+ LuceneTextIndexHeader.FileInfo fileInfo = LuceneTextIndexBufferReader.getFileInfo(indexBuffer, mappingFileName);
+ if (fileInfo == null) {
+ return null;
+ }
+ return indexBuffer.view(fileInfo.getOffset(), fileInfo.getOffset() + fileInfo.getSize());
Review Comment:
The packed HNSW docId-mapping bytes are written/read as little-endian in the file-backed path, but this view inherits BIG_ENDIAN from `columns.psf`. `DocIdTranslator.getInt()` will therefore byte-swap Lucene-to-Pinot doc IDs for consolidated HNSW segments, which can corrupt ANN result translation and pre-filter matching whenever the packed mapping is present. Please return a `LITTLE_ENDIAN` view here, or normalize the buffer before `DocIdTranslator` consumes it.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Consolidate IVF and HNSW vector indexes into segment columns.psf (opt-in via storeInSegmentFile) [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18852:
URL: https://github.com/apache/pinot/pull/18852#discussion_r3487855199
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/VectorIndexHandler.java:
##########
@@ -90,6 +122,287 @@ public boolean needUpdateIndices(SegmentDirectory.Reader segmentReader) {
return false;
}
+ /// Absorbs a vector index sidecar into {@code columns.psf} as a typed entry, then deletes the
+ /// sibling. For IVF backends the bytes are copied verbatim; the on-disk IVF header is the
+ /// contract and the reader handles version dispatch. For HNSW, if a combined-form file already
+ /// exists (written by a creator run with the flag on), it is absorbed directly; if only the
+ /// Lucene directory remains (operator just flipped the flag on an existing segment), it is first
+ /// packed into a transient combined file, which is then absorbed and removed.
+ ///
+ /// **Crash recovery:** if a prior absorb crashed between {@code newIndexFor} (which committed
+ /// bytes into {@code columns.psf} and added a {@code _columnEntries} entry) and the sidecar file
+ /// deletion, the next load will see both forms. We detect that state upfront via the columns.psf
+ /// typed entry ({@link VectorIndexUtils#getConsolidatedVectorEntry}), NOT via {@code hasIndexFor}
+ /// — which for vector also reports the on-disk sidecar and would therefore flag a normal
+ /// first-absorb as already-absorbed. We then verify the typed entry's size matches the sidecar's
+ /// length (so we do not delete a sidecar that happens to coexist with an unrelated typed entry),
+ /// and clean up the orphan sidecar instead of re-running the absorb.
+ private void absorbCombinedIntoColumnsPsf(SegmentDirectory.Writer segmentWriter, String column,
+ VectorBackendType backendType, File indexDir)
+ throws Exception {
+ File v3Dir = SegmentDirectoryPaths.segmentDirectoryFor(indexDir,
+ _segmentDirectory.getSegmentMetadata().getVersion());
+ String segmentName = _segmentDirectory.getSegmentMetadata().getName();
+
+ if (backendType == VectorBackendType.HNSW) {
+ absorbHnswIntoColumnsPsf(segmentWriter, column, v3Dir, segmentName);
+ return;
+ }
+
+ // IVF path: prefer the combined-form file (freshly-written by a creator run with flag=on);
+ // fall back to the legacy sidecar (segments built with flag=off that the operator now wants
+ // consolidated).
+ File combinedFile = new File(v3Dir,
+ column + VectorIndexUtils.getIndexFileExtension(backendType, /* combined */ true));
+ File legacyFile = new File(v3Dir,
+ column + VectorIndexUtils.getIndexFileExtension(backendType, /* combined */ false));
+ File picked = combinedFile.exists() ? combinedFile : legacyFile;
+ if (!picked.exists()) {
+ LOGGER.warn("Expected vector index file {} not found during vector consolidation; skipping", picked);
+ return;
+ }
+ // Crash-recovery: if a prior absorb committed bytes to columns.psf but crashed before
+ // deleting the sidecar, the typed entry is already present. Detect that via the columns.psf
+ // typed entry directly — NOT hasIndexFor, which for vector also reports the on-disk sidecar we
+ // are about to absorb. Using hasIndexFor here would make a normal first-absorb look
+ // already-absorbed and the subsequent getIndexFor().size() would throw.
+ PinotDataBuffer existingEntry = VectorIndexUtils.getConsolidatedVectorEntry(segmentWriter, column);
+ if (existingEntry != null) {
+ long existingSize = existingEntry.size();
+ if (existingSize == picked.length()) {
+ LOGGER.warn("Vector index already present in columns.psf for segment: {}, column: {}; "
+ + "deleting orphan sidecar file from a previously-crashed absorb run", segmentName, column);
+ FileUtils.deleteQuietly(picked);
+ return;
+ }
+ // Size mismatch — the typed entry is from a different build than the sidecar. Refuse to
+ // proceed; an operator must reconcile manually rather than have us guess.
+ throw new IOException("Vector index already present in columns.psf for column: " + column
+ + " (size=" + existingSize + ") but sidecar file " + picked.getName()
+ + " has different size " + picked.length() + ". Refusing to overwrite — please remove "
+ + "the conflicting sidecar or rebuild the segment.");
+ }
+ LOGGER.info("Absorbing vector sidecar file into columns.psf for segment: {}, column: {} (backend={})",
+ segmentName, column, backendType);
+ LoaderUtils.writeIndexToV3Format(segmentWriter, column, picked, StandardIndexes.vector());
+ }
+
+ /// HNSW-specific absorb: packs the Lucene directory (or uses an existing combined-form file) into
+ /// a single combined file and absorbs it into {@code columns.psf}.
+ ///
+ /// If the combined-form file already exists (creator ran with the flag on and produced it),
+ /// it is absorbed directly. If only the Lucene directory exists (operator just flipped the flag
+ /// on an existing segment), the directory is first packed into a transient combined file, which is
+ /// then absorbed and cleaned up.
+ private void absorbHnswIntoColumnsPsf(SegmentDirectory.Writer segmentWriter, String column, File v3Dir,
+ String segmentName)
+ throws Exception {
+ File combinedFile = new File(v3Dir,
+ column + VectorIndexUtils.getIndexFileExtension(VectorBackendType.HNSW, /* combined */ true));
+ File hnswDir = new File(v3Dir,
+ column + VectorIndexUtils.getIndexFileExtension(VectorBackendType.HNSW, /* combined */ false));
+ if (!combinedFile.exists()) {
+ if (!hnswDir.exists() || !hnswDir.isDirectory()) {
+ LOGGER.warn("Expected HNSW directory or combined file for column {} not found; skipping absorb", column);
+ return;
+ }
+ // Pack the Lucene directory into a transient combined file for absorption. The directory
+ // and the combined file are both cleaned up below after a successful absorb (or by the
+ // crash-recovery branch above on a subsequent retry).
+ HnswVectorIndexCombined.combineHnswIndexFiles(hnswDir, combinedFile.getAbsolutePath(), v3Dir.getParentFile(),
+ column);
+ }
+
+ // Crash-recovery: check whether the typed entry already exists. Detect it via the columns.psf
+ // typed entry directly — NOT hasIndexFor, which for vector also reports the still-present
+ // legacy Lucene directory (we pack but do not delete it above), so hasIndexFor would be true on
+ // a normal first-absorb and the subsequent getIndexFor().size() would throw.
+ PinotDataBuffer existingEntry = VectorIndexUtils.getConsolidatedVectorEntry(segmentWriter, column);
+ if (existingEntry != null) {
+ long existingSize = existingEntry.size();
+ if (existingSize == combinedFile.length()) {
+ LOGGER.warn("HNSW vector index already present in columns.psf for segment: {}, column: {}; "
+ + "deleting orphan files from a previously-crashed absorb run", segmentName, column);
+ FileUtils.deleteQuietly(combinedFile);
+ // Clean up the Lucene directory regardless of createdTransient: if we packed the directory
+ // into a transient combined file in this very run, the directory still exists and must be
+ // removed to prevent hasCombinedFile from re-triggering on the next segment load.
+ if (hnswDir.exists()) {
+ FileUtils.deleteDirectory(hnswDir);
+ }
+ return;
+ }
+ throw new IOException("HNSW vector index already present in columns.psf for column: " + column
+ + " (size=" + existingSize + ") but combined file " + combinedFile.getName()
+ + " has different size " + combinedFile.length() + ". Refusing to overwrite — please remove "
+ + "the conflicting sidecar or rebuild the segment.");
+ }
+
+ LOGGER.info("Absorbing HNSW combined file into columns.psf for segment: {}, column: {}", segmentName, column);
+ LoaderUtils.writeIndexToV3Format(segmentWriter, column, combinedFile, StandardIndexes.vector());
+ // writeIndexToV3Format already force-deletes the absorbed combined file (like the IVF path above),
+ // so only the Lucene directory needs explicit cleanup here.
+ if (hnswDir.exists()) {
+ FileUtils.deleteDirectory(hnswDir);
+ }
+ }
+
+ /// Extracts the consolidated vector payload from {@code columns.psf} back into the legacy on-disk
+ /// form and drops the typed entry. Used when the operator flips {@code storeInSegmentFile} from
+ /// {@code true} to {@code false}.
+ ///
+ /// For IVF backends the bytes are streamed verbatim to a sidecar file. For HNSW, the packed
+ /// combined file is streamed out first, then unpacked into a Lucene directory — the inverse of
+ /// the absorb path.
+ ///
+ /// **Ordering:** bytes are streamed to a temp file *before* {@code removeIndex}
+ /// is called, because {@code SingleFileIndexDirectory.removeIndex} for vector also runs
+ /// {@link VectorIndexUtils#cleanupVectorIndex(File, String)}, which deletes any file with a
+ /// recognised extension. The temp extension ({@code .vector.extract-tmp}) is not in that list,
+ /// so the bytes survive until the consolidated entry is safely removed.
+ private void extractConsolidatedToLegacyFile(SegmentDirectory.Writer segmentWriter, String column,
+ VectorBackendType backendType, File indexDir)
+ throws IOException {
+ File v3Dir = SegmentDirectoryPaths.segmentDirectoryFor(indexDir,
+ _segmentDirectory.getSegmentMetadata().getVersion());
+ String segmentName = _segmentDirectory.getSegmentMetadata().getName();
+
+ if (backendType == VectorBackendType.HNSW) {
+ extractHnswConsolidatedToDirectory(segmentWriter, column, v3Dir, segmentName);
+ return;
+ }
+
+ // IVF path: stream bytes to the legacy combined extension.
+ File finalCombined = new File(v3Dir, column + VectorIndexUtils.getIndexFileExtension(backendType));
+ File combinedFormFile = new File(v3Dir,
+ column + VectorIndexUtils.getIndexFileExtension(backendType, /* combined */ true));
+ // Defensive: extract is destructive (removeIndex calls cleanupVectorIndex, which deletes any
+ // sidecar with a recognised IVF extension). If a sidecar already exists on disk we'd silently
+ // clobber it. Refuse to proceed and force the operator to reconcile — this state should not
+ // arise from the handler's own paths (see hasCombinedFile in needUpdateIndices) but a
+ // manually-edited segment dir or a half-completed migration could produce it.
+ if (finalCombined.exists()) {
+ throw new IOException("Extract path expected no on-disk sidecar but found: " + finalCombined
+ + ". Refusing to proceed — please remove the conflicting file manually.");
+ }
+ if (combinedFormFile.exists()) {
+ throw new IOException("Extract path expected no on-disk sidecar but found: " + combinedFormFile
+ + ". Refusing to proceed — please remove the conflicting file manually.");
+ }
+ File tempCombined = new File(v3Dir, column + ".vector.extract-tmp");
+ // Clean any leftover from a previously-crashed extract.
+ FileUtils.deleteQuietly(tempCombined);
+
+ LOGGER.info("Extracting vector consolidated entry to sidecar file for segment: {}, column: {} (backend={})",
+ segmentName, column, backendType);
+ PinotDataBuffer buffer = segmentWriter.getIndexFor(column, StandardIndexes.vector());
+ long size = buffer.size();
+ streamBufferToFile(buffer, size, tempCombined);
+
+ // Remove the consolidated entry. {@code cleanupVectorIndex} runs as a side effect; it will
+ // not touch our temp file because the temp extension is not in its recognised list.
+ segmentWriter.removeIndex(column, StandardIndexes.vector());
+
+ if (!tempCombined.renameTo(finalCombined)) {
+ // Best-effort copy-and-delete fallback if rename fails (e.g. cross-FS in tests).
+ FileUtils.copyFile(tempCombined, finalCombined);
+ FileUtils.deleteQuietly(tempCombined);
+ }
+ }
+
+ /// HNSW-specific extract: streams the combined HNSW payload from {@code columns.psf} to a temp
+ /// file, unpacks it into a Lucene directory, and only then removes the consolidated typed entry.
+ ///
+ /// **Ordering rationale:** unpack runs *before* {@code removeIndex}. If the unpack
+ /// fails mid-way, the typed entry is still present in {@code columns.psf} and the next load
+ /// retries the extract — the operator does not lose the index. The reverse order would leave a
+ /// crash window in which the typed entry is gone but the Lucene directory does not yet exist,
+ /// leaving the segment permanently without its HNSW index until a full rebuild.
+ private void extractHnswConsolidatedToDirectory(SegmentDirectory.Writer segmentWriter, String column,
+ File v3Dir, String segmentName)
+ throws IOException {
+ File hnswDir = new File(v3Dir,
+ column + VectorIndexUtils.getIndexFileExtension(VectorBackendType.HNSW, /* combined */ false));
+ File combinedFormFile = new File(v3Dir,
+ column + VectorIndexUtils.getIndexFileExtension(VectorBackendType.HNSW, /* combined */ true));
+ if (hnswDir.exists()) {
+ throw new IOException("Extract path expected no on-disk HNSW directory but found: " + hnswDir
+ + ". Refusing to proceed — please remove the conflicting directory manually.");
+ }
+ if (combinedFormFile.exists()) {
+ throw new IOException("Extract path expected no on-disk combined HNSW file but found: " + combinedFormFile
+ + ". Refusing to proceed — please remove the conflicting file manually.");
+ }
+ File tempCombined = new File(v3Dir, column + ".vector.extract-tmp");
+ FileUtils.deleteQuietly(tempCombined);
+
+ LOGGER.info(
+ "Extracting HNSW consolidated entry to Lucene directory for segment: {}, column: {}", segmentName, column);
+ PinotDataBuffer buffer = segmentWriter.getIndexFor(column, StandardIndexes.vector());
+ streamBufferToFile(buffer, buffer.size(), tempCombined);
+
+ // Unpack the combined file back into a Lucene directory BEFORE removing the typed entry. If
+ // unpack throws, the typed entry remains in columns.psf and the next handler run retries —
+ // no permanent loss. Best-effort clean up the half-unpacked directory so the retry's
+ // pre-flight directory-exists check still fires.
+ try {
+ HnswVectorIndexCombined.extractHnswIndexFiles(tempCombined, hnswDir);
+ } catch (IOException | RuntimeException e) {
+ if (hnswDir.exists()) {
+ FileUtils.deleteQuietly(hnswDir);
+ }
+ FileUtils.deleteQuietly(tempCombined);
+ throw e;
+ }
+
+ // Unpack succeeded — the bytes are now on disk in legacy form. Remove the consolidated entry.
+ // If this throws, the segment ends up with BOTH forms; the reader factory's file-backed path
+ // (flag=false) uses the Lucene directory and ignores the orphan typed entry until the next
+ // handler run cleans it.
+ segmentWriter.removeIndex(column, StandardIndexes.vector());
Review Comment:
`removeIndex()` now runs `VectorIndexUtils.cleanupVectorIndex()`, and that helper recursively `deleteQuietly()`s the legacy HNSW directory and mapping file. So the directory unpacked at line 349 gets deleted immediately here, and the segment finishes the `storeInSegmentFile=true -> false` migration without a usable legacy HNSW artifact. Please unpack into a temp location that vector cleanup will not match, drop only the typed `columns.psf` entry, then move the Lucene directory into place after cleanup.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Consolidate IVF and HNSW vector indexes into segment columns.psf (opt-in via storeInSegmentFile) [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #18852:
URL: https://github.com/apache/pinot/pull/18852#issuecomment-4804721135
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/18852?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `3.50000%` with `579 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 56.82%. Comparing base ([`302783f`](https://app.codecov.io/gh/apache/pinot/commit/302783f42ca608394100e1b0062654a76e265880?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`561240c`](https://app.codecov.io/gh/apache/pinot/commit/561240c3463369495739738634979ed24e59f48c?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 12 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/pinot/pull/18852?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...index/loader/invertedindex/VectorIndexHandler.java](https://app.codecov.io/gh/apache/pinot/pull/18852?src=pr&el=tree&filepath=pinot-segment-local%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fsegment%2Flocal%2Fsegment%2Findex%2Floader%2Finvertedindex%2FVectorIndexHandler.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc2VnbWVudC1sb2NhbC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VnbWVudC9sb2NhbC9zZWdtZW50L2luZGV4L2xvYWRlci9pbnZlcnRlZGluZGV4L1ZlY3RvckluZGV4SGFuZGxlci5qYXZh) | 0.00% | [157 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18852?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../impl/vector/lucene99/HnswVectorIndexCombined.java](https://app.codecov.io/gh/apache/pinot/pull/18852?src=pr&el=tree&filepath=pinot-segment-local%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fsegment%2Flocal%2Fsegment%2Fcreator%2Fimpl%2Fvector%2Flucene99%2FHnswVectorIndexCombined.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc2VnbWVudC1sb2NhbC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VnbWVudC9sb2NhbC9zZWdtZW50L2NyZWF0b3IvaW1wbC92ZWN0b3IvbHVjZW5lOTkvSG5zd1ZlY3RvckluZGV4Q29tYmluZWQuamF2YQ==) | 0.00% | [159 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18852?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...dex/readers/vector/IvfOnDiskVectorIndexReader.java](https://app.codecov.io/gh/apache/pinot/pull/18852?src=pr&el=tree&filepath=pinot-segment-local%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fsegment%2Flocal%2Fsegment%2Findex%2Freaders%2Fvector%2FIvfOnDiskVectorIndexReader.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc2VnbWVudC1sb2NhbC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VnbWVudC9sb2NhbC9zZWdtZW50L2luZGV4L3JlYWRlcnMvdmVjdG9yL0l2Zk9uRGlza1ZlY3RvckluZGV4UmVhZGVyLmphdmE=) | 0.00% | [39 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18852?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...nt/index/readers/vector/HnswVectorIndexReader.java](https://app.codecov.io/gh/apache/pinot/pull/18852?src=pr&el=tree&filepath=pinot-segment-local%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fsegment%2Flocal%2Fsegment%2Findex%2Freaders%2Fvector%2FHnswVectorIndexReader.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc2VnbWVudC1sb2NhbC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VnbWVudC9sb2NhbC9zZWdtZW50L2luZGV4L3JlYWRlcnMvdmVjdG9yL0huc3dWZWN0b3JJbmRleFJlYWRlci5qYXZh) | 8.33% | [32 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18852?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...nt/local/segment/index/vector/VectorIndexType.java](https://app.codecov.io/gh/apache/pinot/pull/18852?src=pr&el=tree&filepath=pinot-segment-local%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fsegment%2Flocal%2Fsegment%2Findex%2Fvector%2FVectorIndexType.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc2VnbWVudC1sb2NhbC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VnbWVudC9sb2NhbC9zZWdtZW50L2luZGV4L3ZlY3Rvci9WZWN0b3JJbmRleFR5cGUuamF2YQ==) | 5.88% | [30 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18852?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...local/segment/index/vector/IvfCombinedBuffers.java](https://app.codecov.io/gh/apache/pinot/pull/18852?src=pr&el=tree&filepath=pinot-segment-local%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fsegment%2Flocal%2Fsegment%2Findex%2Fvector%2FIvfCombinedBuffers.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc2VnbWVudC1sb2NhbC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VnbWVudC9sb2NhbC9zZWdtZW50L2luZGV4L3ZlY3Rvci9JdmZDb21iaW5lZEJ1ZmZlcnMuamF2YQ==) | 0.00% | [24 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18852?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../segment/local/segment/store/VectorIndexUtils.java](https://app.codecov.io/gh/apache/pinot/pull/18852?src=pr&el=tree&filepath=pinot-segment-local%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fsegment%2Flocal%2Fsegment%2Fstore%2FVectorIndexUtils.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc2VnbWVudC1sb2NhbC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VnbWVudC9sb2NhbC9zZWdtZW50L3N0b3JlL1ZlY3RvckluZGV4VXRpbHMuamF2YQ==) | 0.00% | [19 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18852?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ex/readers/vector/HnswVectorIndexBufferReader.java](https://app.codecov.io/gh/apache/pinot/pull/18852?src=pr&el=tree&filepath=pinot-segment-local%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fsegment%2Flocal%2Fsegment%2Findex%2Freaders%2Fvector%2FHnswVectorIndexBufferReader.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc2VnbWVudC1sb2NhbC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VnbWVudC9sb2NhbC9zZWdtZW50L2luZGV4L3JlYWRlcnMvdmVjdG9yL0huc3dWZWN0b3JJbmRleEJ1ZmZlclJlYWRlci5qYXZh) | 0.00% | [19 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18852?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...t/local/segment/index/vector/IvfPqIndexFormat.java](https://app.codecov.io/gh/apache/pinot/pull/18852?src=pr&el=tree&filepath=pinot-segment-local%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fsegment%2Flocal%2Fsegment%2Findex%2Fvector%2FIvfPqIndexFormat.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc2VnbWVudC1sb2NhbC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VnbWVudC9sb2NhbC9zZWdtZW50L2luZGV4L3ZlY3Rvci9JdmZQcUluZGV4Rm9ybWF0LmphdmE=) | 0.00% | [18 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18852?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...index/readers/vector/IvfFlatVectorIndexReader.java](https://app.codecov.io/gh/apache/pinot/pull/18852?src=pr&el=tree&filepath=pinot-segment-local%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fsegment%2Flocal%2Fsegment%2Findex%2Freaders%2Fvector%2FIvfFlatVectorIndexReader.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc2VnbWVudC1sb2NhbC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VnbWVudC9sb2NhbC9zZWdtZW50L2luZGV4L3JlYWRlcnMvdmVjdG9yL0l2ZkZsYXRWZWN0b3JJbmRleFJlYWRlci5qYXZh) | 0.00% | [14 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18852?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| ... and [7 more](https://app.codecov.io/gh/apache/pinot/pull/18852?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | |
> :exclamation: There is a different number of reports uploaded between BASE (302783f) and HEAD (561240c). Click for more details.
>
> <details><summary>HEAD has 16 uploads less than BASE</summary>
>
>| Flag | BASE (302783f) | HEAD (561240c) |
>|------|------|------|
>|unittests2|1|0|
>|unittests|2|1|
>|java-21|5|1|
>|temurin|5|1|
>|integration|3|0|
>|integration1|1|0|
>|integration2|1|0|
>|custom-integration1|1|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #18852 +/- ##
============================================
- Coverage 64.77% 56.82% -7.95%
+ Complexity 1322 1 -1321
============================================
Files 3393 2607 -786
Lines 211022 153109 -57913
Branches 33135 24907 -8228
============================================
- Hits 136687 87009 -49678
+ Misses 63322 58679 -4643
+ Partials 11013 7421 -3592
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/18852/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [custom-integration1](https://app.codecov.io/gh/apache/pinot/pull/18852/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration](https://app.codecov.io/gh/apache/pinot/pull/18852/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/18852/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/18852/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-21](https://app.codecov.io/gh/apache/pinot/pull/18852/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.82% <3.50%> (-7.95%)` | :arrow_down: |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/18852/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.82% <3.50%> (-7.95%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/18852/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.82% <3.50%> (-7.95%)` | :arrow_down: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/18852/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.82% <3.50%> (-0.17%)` | :arrow_down: |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/18852/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/pinot/pull/18852?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Consolidate IVF and HNSW vector indexes into segment columns.psf (opt-in via storeInSegmentFile) [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #18852:
URL: https://github.com/apache/pinot/pull/18852#discussion_r3478417134
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/store/SingleFileIndexDirectory.java:
##########
@@ -160,8 +160,11 @@ public boolean hasIndexFor(String column, IndexType<?, ?, ?> type) {
if (type == StandardIndexes.text() && TextIndexUtils.hasTextIndex(_segmentDirectory, column)) {
return true;
}
- if (type == StandardIndexes.vector()) {
- return VectorIndexUtils.hasVectorIndex(_segmentDirectory, column);
+ // Vector index may live either as a combined file (legacy / storeInSegmentFile=false) or as
+ // a typed entry inside columns.psf (storeInSegmentFile=true). Check both — mirror the text
+ // pattern of "combined OR _columnEntries".
+ if (type == StandardIndexes.vector() && VectorIndexUtils.hasVectorIndex(_segmentDirectory, column)) {
+ return true;
}
IndexKey key = new IndexKey(column, type);
return _columnEntries.containsKey(key);
Review Comment:
`hasIndexFor(column, StandardIndexes.vector())` currently returns true when a legacy vector sidecar exists, even if there is no typed-entry in `columns.psf`. However, `getIndexFor(column, StandardIndexes.vector())` in `SingleFileIndexDirectory` only serves typed entries and will throw if the entry is absent. This breaks the `hasIndexFor`/`getIndexFor` contract and also causes the new migration/crash-recovery logic in `VectorIndexHandler` to mis-detect normal legacy segments as “already absorbed”.
Consider limiting `hasIndexFor(..., vector)` to `_columnEntries` presence only (typed entry), and rely on `VectorIndexUtils.hasVectorIndex()` / `getColumnsWithIndex()` when you need to detect legacy sidecars on disk.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/VectorIndexHandler.java:
##########
@@ -90,6 +122,290 @@ public boolean needUpdateIndices(SegmentDirectory.Reader segmentReader) {
return false;
}
+ /**
+ * Absorbs a vector index sidecar into {@code columns.psf} as a typed entry, then deletes the
+ * sibling. For IVF backends the bytes are copied verbatim; the on-disk IVF header is the
+ * contract and the reader handles version dispatch. For HNSW, if a combined-form file already
+ * exists (written by a creator run with the flag on), it is absorbed directly; if only the
+ * Lucene directory remains (operator just flipped the flag on an existing segment), it is first
+ * packed into a transient combined file, which is then absorbed and removed.
+ *
+ * <p><b>Crash recovery:</b> if a prior absorb crashed between {@code newIndexFor} (which
+ * committed bytes into {@code columns.psf} and added a {@code _columnEntries} entry) and the
+ * sidecar file deletion, the next load will see both forms. We detect that state upfront via
+ * {@code hasIndexFor}, verify the typed entry's size matches the sidecar's length (so we do not
+ * delete a sidecar that happens to coexist with an unrelated typed entry), and clean up the
+ * orphan sidecar instead of re-running the absorb.</p>
+ */
+ private void absorbCombinedIntoColumnsPsf(SegmentDirectory.Writer segmentWriter, String column,
+ VectorBackendType backendType, File indexDir)
+ throws Exception {
+ File v3Dir = SegmentDirectoryPaths.segmentDirectoryFor(indexDir,
+ _segmentDirectory.getSegmentMetadata().getVersion());
+ String segmentName = _segmentDirectory.getSegmentMetadata().getName();
+
+ if (backendType == VectorBackendType.HNSW) {
+ absorbHnswIntoColumnsPsf(segmentWriter, column, v3Dir, segmentName);
+ return;
+ }
+
+ // IVF path: prefer the combined-form file (freshly-written by a creator run with flag=on);
+ // fall back to the legacy sidecar (segments built with flag=off that the operator now wants
+ // consolidated).
+ File combinedFile = new File(v3Dir,
+ column + VectorIndexUtils.getIndexFileExtension(backendType, /* combined */ true));
+ File legacyFile = new File(v3Dir,
+ column + VectorIndexUtils.getIndexFileExtension(backendType, /* combined */ false));
+ File picked = combinedFile.exists() ? combinedFile : legacyFile;
+ if (!picked.exists()) {
+ LOGGER.warn("Expected vector index file {} not found during vector consolidation; skipping", picked);
+ return;
+ }
+ // Crash-recovery: if a prior absorb committed bytes to columns.psf but crashed before
+ // deleting the sidecar, the typed entry is already present. Detect that directly via
+ // hasIndexFor + size check, rather than catching the duplicate-key exception by message.
+ if (segmentWriter.hasIndexFor(column, StandardIndexes.vector())) {
+ long existingSize = segmentWriter.getIndexFor(column, StandardIndexes.vector()).size();
+ if (existingSize == picked.length()) {
+ LOGGER.warn("Vector index already present in columns.psf for segment: {}, column: {}; "
+ + "deleting orphan sidecar file from a previously-crashed absorb run", segmentName, column);
+ FileUtils.deleteQuietly(picked);
+ return;
+ }
+ // Size mismatch — the typed entry is from a different build than the sidecar. Refuse to
+ // proceed; an operator must reconcile manually rather than have us guess.
+ throw new IOException("Vector index already present in columns.psf for column: " + column
+ + " (size=" + existingSize + ") but sidecar file " + picked.getName()
+ + " has different size " + picked.length() + ". Refusing to overwrite — please remove "
+ + "the conflicting sidecar or rebuild the segment.");
+ }
Review Comment:
Crash-recovery in `absorbCombinedIntoColumnsPsf` uses `segmentWriter.hasIndexFor(column, StandardIndexes.vector())` to infer that a typed entry already exists in `columns.psf`. With the updated `SingleFileIndexDirectory.hasIndexFor`, this can be true just because a legacy sidecar exists on disk, even when no typed entry has been created yet. In that case, the subsequent `getIndexFor(...).size()` will throw and block the legacy→columns.psf migration.
Safer pattern: attempt `segmentWriter.getIndexFor(...)` and treat “missing typed entry” as the normal absorb path; only run the size-check recovery when `getIndexFor` succeeds.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/converter/SegmentV1V2ToV3FormatConverter.java:
##########
@@ -147,16 +148,31 @@ private void copyIndexData(File v2Directory, SegmentMetadataImpl v2Metadata, Fil
SegmentDirectory.Writer v3DataWriter = v3Segment.createWriter()) {
for (String column : v2Metadata.getAllColumns()) {
for (IndexType<?, ?, ?> indexType : IndexService.getInstance().getAllIndexes()) {
- //If Text index files are combined merge into columns.psf else no-op
+ // Text index: skip the standard copy when a legacy Lucene directory is present
+ // ({@code copyLuceneTextIndexIfExists} below handles that as a sibling copy).
+ // Combined .text.index files fall through to the standard path and get packed.
if (indexType == StandardIndexes.text()) {
if (!TextIndexUtils.hasTextIndex(v2Directory, column)) {
copyIndexIfExists(v2DataReader, v3DataWriter, column, indexType);
}
- } else {
- if (indexType != StandardIndexes.vector()) {
+ continue;
+ }
+ // Vector index mirrors text: skip the standard copy when a legacy IVF sidecar (or
+ // HNSW directory) is present ({@code copyVectorIndexIfExists} handles those as
+ // sibling copies). Combined .vector.ivfflat.combined.index / .ivfpq.combined.index
+ // files fall through to the standard path and get packed into columns.psf.
+ // Mixed state: if both legacy and combined exist (e.g. operator toggled the flag
+ // and rebuilt without cleaning the old file), combined wins — pack it and let
+ // copyVectorIndexIfExists drop the legacy sibling so the operator gets exactly one
+ // copy of the bytes in the V3 segment.
+ if (indexType == StandardIndexes.vector()) {
+ boolean hasCombined = VectorIndexUtils.hasCombinedFormVectorIndex(v2Directory, column);
+ if (hasCombined || !VectorIndexUtils.hasVectorIndex(v2Directory, column)) {
copyIndexIfExists(v2DataReader, v3DataWriter, column, indexType);
}
+ continue;
}
Review Comment:
When `hasCombinedFormVectorIndex(...)` is true, this path calls `copyIndexIfExists(...)` for `StandardIndexes.vector()`. In V1/V2 segments, the underlying `SegmentDirectory.Reader` is file-based and may resolve the vector index to a *directory* (legacy HNSW) rather than the combined packed file, which then fails when mapping as a regular file.
To avoid converter failures in mixed/crash states (legacy HNSW dir + combined file both present), consider only taking the “standard copy into columns.psf” path when *no* legacy vector index exists (`!VectorIndexUtils.hasVectorIndex(...)`). That still packs combined-only artifacts, while deferring mixed states to the sibling-copy path.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/VectorIndexHandler.java:
##########
@@ -90,6 +122,290 @@ public boolean needUpdateIndices(SegmentDirectory.Reader segmentReader) {
return false;
}
+ /**
+ * Absorbs a vector index sidecar into {@code columns.psf} as a typed entry, then deletes the
+ * sibling. For IVF backends the bytes are copied verbatim; the on-disk IVF header is the
+ * contract and the reader handles version dispatch. For HNSW, if a combined-form file already
+ * exists (written by a creator run with the flag on), it is absorbed directly; if only the
+ * Lucene directory remains (operator just flipped the flag on an existing segment), it is first
+ * packed into a transient combined file, which is then absorbed and removed.
+ *
+ * <p><b>Crash recovery:</b> if a prior absorb crashed between {@code newIndexFor} (which
+ * committed bytes into {@code columns.psf} and added a {@code _columnEntries} entry) and the
+ * sidecar file deletion, the next load will see both forms. We detect that state upfront via
+ * {@code hasIndexFor}, verify the typed entry's size matches the sidecar's length (so we do not
+ * delete a sidecar that happens to coexist with an unrelated typed entry), and clean up the
+ * orphan sidecar instead of re-running the absorb.</p>
+ */
+ private void absorbCombinedIntoColumnsPsf(SegmentDirectory.Writer segmentWriter, String column,
+ VectorBackendType backendType, File indexDir)
+ throws Exception {
+ File v3Dir = SegmentDirectoryPaths.segmentDirectoryFor(indexDir,
+ _segmentDirectory.getSegmentMetadata().getVersion());
+ String segmentName = _segmentDirectory.getSegmentMetadata().getName();
+
+ if (backendType == VectorBackendType.HNSW) {
+ absorbHnswIntoColumnsPsf(segmentWriter, column, v3Dir, segmentName);
+ return;
+ }
+
+ // IVF path: prefer the combined-form file (freshly-written by a creator run with flag=on);
+ // fall back to the legacy sidecar (segments built with flag=off that the operator now wants
+ // consolidated).
+ File combinedFile = new File(v3Dir,
+ column + VectorIndexUtils.getIndexFileExtension(backendType, /* combined */ true));
+ File legacyFile = new File(v3Dir,
+ column + VectorIndexUtils.getIndexFileExtension(backendType, /* combined */ false));
+ File picked = combinedFile.exists() ? combinedFile : legacyFile;
+ if (!picked.exists()) {
+ LOGGER.warn("Expected vector index file {} not found during vector consolidation; skipping", picked);
+ return;
+ }
+ // Crash-recovery: if a prior absorb committed bytes to columns.psf but crashed before
+ // deleting the sidecar, the typed entry is already present. Detect that directly via
+ // hasIndexFor + size check, rather than catching the duplicate-key exception by message.
+ if (segmentWriter.hasIndexFor(column, StandardIndexes.vector())) {
+ long existingSize = segmentWriter.getIndexFor(column, StandardIndexes.vector()).size();
+ if (existingSize == picked.length()) {
+ LOGGER.warn("Vector index already present in columns.psf for segment: {}, column: {}; "
+ + "deleting orphan sidecar file from a previously-crashed absorb run", segmentName, column);
+ FileUtils.deleteQuietly(picked);
+ return;
+ }
+ // Size mismatch — the typed entry is from a different build than the sidecar. Refuse to
+ // proceed; an operator must reconcile manually rather than have us guess.
+ throw new IOException("Vector index already present in columns.psf for column: " + column
+ + " (size=" + existingSize + ") but sidecar file " + picked.getName()
+ + " has different size " + picked.length() + ". Refusing to overwrite — please remove "
+ + "the conflicting sidecar or rebuild the segment.");
+ }
+ LOGGER.info("Absorbing vector sidecar file into columns.psf for segment: {}, column: {} (backend={})",
+ segmentName, column, backendType);
+ LoaderUtils.writeIndexToV3Format(segmentWriter, column, picked, StandardIndexes.vector());
+ }
+
+ /**
+ * HNSW-specific absorb: packs the Lucene directory (or uses an existing combined-form file) into
+ * a single combined file and absorbs it into {@code columns.psf}.
+ *
+ * <p>If the combined-form file already exists (creator ran with the flag on and produced it),
+ * it is absorbed directly. If only the Lucene directory exists (operator just flipped the flag
+ * on an existing segment), the directory is first packed into a transient combined file, which is
+ * then absorbed and cleaned up.</p>
+ */
+ private void absorbHnswIntoColumnsPsf(SegmentDirectory.Writer segmentWriter, String column, File v3Dir,
+ String segmentName)
+ throws Exception {
+ File combinedFile = new File(v3Dir,
+ column + VectorIndexUtils.getIndexFileExtension(VectorBackendType.HNSW, /* combined */ true));
+ File hnswDir = new File(v3Dir,
+ column + VectorIndexUtils.getIndexFileExtension(VectorBackendType.HNSW, /* combined */ false));
+ if (!combinedFile.exists()) {
+ if (!hnswDir.exists() || !hnswDir.isDirectory()) {
+ LOGGER.warn("Expected HNSW directory or combined file for column {} not found; skipping absorb", column);
+ return;
+ }
+ // Pack the Lucene directory into a transient combined file for absorption. The directory
+ // and the combined file are both cleaned up below after a successful absorb (or by the
+ // crash-recovery branch above on a subsequent retry).
+ HnswVectorIndexCombined.combineHnswIndexFiles(hnswDir, combinedFile.getAbsolutePath(), v3Dir.getParentFile(),
+ column);
+ }
+
+ // Crash-recovery: check whether the typed entry already exists.
+ if (segmentWriter.hasIndexFor(column, StandardIndexes.vector())) {
+ long existingSize = segmentWriter.getIndexFor(column, StandardIndexes.vector()).size();
+ if (existingSize == combinedFile.length()) {
+ LOGGER.warn("HNSW vector index already present in columns.psf for segment: {}, column: {}; "
+ + "deleting orphan files from a previously-crashed absorb run", segmentName, column);
+ FileUtils.deleteQuietly(combinedFile);
+ // Clean up the Lucene directory regardless of createdTransient: if we packed the directory
+ // into a transient combined file in this very run, the directory still exists and must be
+ // removed to prevent hasCombinedFile from re-triggering on the next segment load.
+ if (hnswDir.exists()) {
+ FileUtils.deleteDirectory(hnswDir);
+ }
+ return;
+ }
+ throw new IOException("HNSW vector index already present in columns.psf for column: " + column
+ + " (size=" + existingSize + ") but combined file " + combinedFile.getName()
+ + " has different size " + combinedFile.length() + ". Refusing to overwrite — please remove "
+ + "the conflicting sidecar or rebuild the segment.");
+ }
Review Comment:
Same issue as the IVF absorb path: `absorbHnswIntoColumnsPsf` uses `segmentWriter.hasIndexFor(column, StandardIndexes.vector())` for crash-recovery, but that can be true solely because the legacy HNSW Lucene directory exists on disk. That makes `getIndexFor(...).size()` throw and prevents sidecar→columns.psf migration for existing segments when `storeInSegmentFile` is toggled on.
Use the same “try getIndexFor; if present run size-check recovery” approach here.
##########
pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/creator/VectorIndexConfig.java:
##########
@@ -139,6 +149,31 @@ public VectorIndexConfig setProperties(Map<String, String> properties) {
return this;
}
+ /**
+ * Whether the IVF vector index payload is consolidated into the segment's combined index file
+ * ({@code columns.psf}) rather than written as a combined file. Default {@code false}. Only
+ * applies to V3 segments and IVF backends; HNSW is unaffected.
+ *
Review Comment:
The Javadoc for `isStoreInSegmentFile()` says this flag “only applies … to IVF backends; HNSW is unaffected”, but this PR wires the same flag through the HNSW creator/reader/conversion path (combined packed file + `columns.psf` typed entry). This is misleading for config consumers and reviewers.
Update the Javadoc to reflect that the flag applies to vector indexes in V3 segments (IVF and HNSW), with backend-specific details in the subsequent paragraphs.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/vector/lucene99/HnswVectorIndexCombined.java:
##########
@@ -0,0 +1,351 @@
+/**
+ * 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.creator.impl.vector.lucene99;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.channels.FileChannel;
+import java.nio.file.StandardOpenOption;
+import java.util.Map;
+import java.util.TreeMap;
+import javax.annotation.Nullable;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.segment.local.segment.creator.impl.text.LuceneCombinedTextIndexConstants;
+import org.apache.pinot.segment.spi.V1Constants;
+import org.apache.pinot.segment.spi.store.SegmentDirectoryPaths;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Utility class to pack a Lucene HNSW index directory (and its optional docId mapping file) into
+ * a single combined file using the {@link LuceneCombinedTextIndexConstants#MAGIC_NUMBER LUCENE_V2}
+ * layout. Mirrors {@code LuceneTextIndexCombined} — reuses the same format constants to keep a
+ * single on-disk format shared across text and HNSW vector indexes.
+ *
+ * <p>Layout (identical to the text-index LUCENE_V2 format):</p>
+ * <pre>
+ * [Header]
+ * Magic "LUCENE_V2" 9 bytes
+ * Version 4 bytes (little-endian int)
+ * Total buffer size 8 bytes (little-endian long)
+ * File count 4 bytes (little-endian int)
+ * Reserved 4 bytes
+ *
+ * [File metadata, one entry per file]
+ * Name length 2 bytes (little-endian short)
+ * Name variable
+ * File offset 8 bytes (little-endian long)
+ * File size 8 bytes (little-endian long)
+ *
+ * [File data]
+ * Raw bytes of each file concatenated in metadata order
+ * </pre>
+ */
+public final class HnswVectorIndexCombined {
+ private static final Logger LOGGER = LoggerFactory.getLogger(HnswVectorIndexCombined.class);
+
+ private HnswVectorIndexCombined() {
+ }
+
+ /**
+ * Packs all files in {@code hnswIndexDir} (plus the optional docId mapping file) into a single
+ * combined file at {@code outputFilePath}.
+ *
+ * @param hnswIndexDir the Lucene HNSW index directory to pack
+ * @param outputFilePath destination path for the combined file
+ * @param segmentIndexDir when non-null, the segment's top-level index directory; used to
+ * locate the docId mapping file for inclusion in the packed output
+ * @param column column name; used to locate the docId mapping file when
+ * {@code segmentIndexDir} is provided
+ * @throws IOException if any file operations fail
+ */
+ public static void combineHnswIndexFiles(File hnswIndexDir, String outputFilePath,
+ @Nullable File segmentIndexDir, @Nullable String column)
+ throws IOException {
+ if (!hnswIndexDir.exists() || !hnswIndexDir.isDirectory()) {
+ throw new IllegalArgumentException(
+ "HNSW index directory does not exist or is not a directory: " + hnswIndexDir);
+ }
+
+ LOGGER.info("Combining HNSW index files from directory: {}", hnswIndexDir.getAbsolutePath());
+
+ Map<String, FileInfo> fileInfoMap = collectFiles(hnswIndexDir, segmentIndexDir, column);
+ int fileCount = fileInfoMap.size();
+
+ if (fileCount == 0) {
+ throw new IOException("No files found in HNSW index directory: " + hnswIndexDir);
+ }
+
+ long totalSize = calculateTotalBufferSize(fileInfoMap);
+ if (totalSize > Integer.MAX_VALUE) {
+ throw new IOException("Combined HNSW index size too large: " + totalSize + " bytes");
+ }
+
+ File outputFile = new File(outputFilePath);
+ try (FileChannel outputChannel = FileChannel.open(outputFile.toPath(), StandardOpenOption.CREATE,
+ StandardOpenOption.WRITE)) {
+ writeHeader(outputChannel, fileCount, (int) totalSize);
Review Comment:
`combineHnswIndexFiles` opens the output file with `CREATE, WRITE` but without `TRUNCATE_EXISTING`. If the combined file already exists from a prior failed attempt and the new combined payload is smaller, stale trailing bytes will remain on disk. That can break later size-based checks (e.g., crash-recovery comparing `combinedFile.length()` to a typed-entry size) and wastes space.
Open the channel with `TRUNCATE_EXISTING` (or explicitly delete the file first).
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Consolidate IVF and HNSW vector indexes into segment columns.psf (opt-in via storeInSegmentFile) [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18852:
URL: https://github.com/apache/pinot/pull/18852#discussion_r3483406960
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/store/SingleFileIndexDirectory.java:
##########
@@ -160,8 +160,11 @@ public boolean hasIndexFor(String column, IndexType<?, ?, ?> type) {
if (type == StandardIndexes.text() && TextIndexUtils.hasTextIndex(_segmentDirectory, column)) {
return true;
}
- if (type == StandardIndexes.vector()) {
- return VectorIndexUtils.hasVectorIndex(_segmentDirectory, column);
+ // Vector index may live either as a combined file (legacy / storeInSegmentFile=false) or as
+ // a typed entry inside columns.psf (storeInSegmentFile=true). Check both — mirror the text
+ // pattern of "combined OR _columnEntries".
+ if (type == StandardIndexes.vector() && VectorIndexUtils.hasVectorIndex(_segmentDirectory, column)) {
+ return true;
}
IndexKey key = new IndexKey(column, type);
return _columnEntries.containsKey(key);
Review Comment:
Kept `hasIndexFor(col, vector())` as "legacy sidecar OR typed entry" on purpose: `PhysicalColumnIndexContainer` gates *all* reader creation on `hasIndexFor`, so a legacy-sidecar segment (payload only on disk, not in `_columnEntries`) must report `true` or its vector reader would never be built. Limiting it to `_columnEntries` would stop legacy vector indexes from loading (and would diverge from the existing text-index pattern, where a Lucene sidecar directory also makes `hasIndexFor` true).
Instead, the callers that specifically need "is the payload already packed into columns.psf" now probe the typed entry directly via a new `VectorIndexUtils.getConsolidatedVectorEntry()`, which returns `null` (rather than throwing) when no typed slot exists. That fixes the mis-detection in `VectorIndexHandler` (threads below) without breaking the load path. Fixed in c04730d.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/VectorIndexHandler.java:
##########
@@ -90,6 +122,290 @@ public boolean needUpdateIndices(SegmentDirectory.Reader segmentReader) {
return false;
}
+ /**
+ * Absorbs a vector index sidecar into {@code columns.psf} as a typed entry, then deletes the
+ * sibling. For IVF backends the bytes are copied verbatim; the on-disk IVF header is the
+ * contract and the reader handles version dispatch. For HNSW, if a combined-form file already
+ * exists (written by a creator run with the flag on), it is absorbed directly; if only the
+ * Lucene directory remains (operator just flipped the flag on an existing segment), it is first
+ * packed into a transient combined file, which is then absorbed and removed.
+ *
+ * <p><b>Crash recovery:</b> if a prior absorb crashed between {@code newIndexFor} (which
+ * committed bytes into {@code columns.psf} and added a {@code _columnEntries} entry) and the
+ * sidecar file deletion, the next load will see both forms. We detect that state upfront via
+ * {@code hasIndexFor}, verify the typed entry's size matches the sidecar's length (so we do not
+ * delete a sidecar that happens to coexist with an unrelated typed entry), and clean up the
+ * orphan sidecar instead of re-running the absorb.</p>
+ */
+ private void absorbCombinedIntoColumnsPsf(SegmentDirectory.Writer segmentWriter, String column,
+ VectorBackendType backendType, File indexDir)
+ throws Exception {
+ File v3Dir = SegmentDirectoryPaths.segmentDirectoryFor(indexDir,
+ _segmentDirectory.getSegmentMetadata().getVersion());
+ String segmentName = _segmentDirectory.getSegmentMetadata().getName();
+
+ if (backendType == VectorBackendType.HNSW) {
+ absorbHnswIntoColumnsPsf(segmentWriter, column, v3Dir, segmentName);
+ return;
+ }
+
+ // IVF path: prefer the combined-form file (freshly-written by a creator run with flag=on);
+ // fall back to the legacy sidecar (segments built with flag=off that the operator now wants
+ // consolidated).
+ File combinedFile = new File(v3Dir,
+ column + VectorIndexUtils.getIndexFileExtension(backendType, /* combined */ true));
+ File legacyFile = new File(v3Dir,
+ column + VectorIndexUtils.getIndexFileExtension(backendType, /* combined */ false));
+ File picked = combinedFile.exists() ? combinedFile : legacyFile;
+ if (!picked.exists()) {
+ LOGGER.warn("Expected vector index file {} not found during vector consolidation; skipping", picked);
+ return;
+ }
+ // Crash-recovery: if a prior absorb committed bytes to columns.psf but crashed before
+ // deleting the sidecar, the typed entry is already present. Detect that directly via
+ // hasIndexFor + size check, rather than catching the duplicate-key exception by message.
+ if (segmentWriter.hasIndexFor(column, StandardIndexes.vector())) {
+ long existingSize = segmentWriter.getIndexFor(column, StandardIndexes.vector()).size();
+ if (existingSize == picked.length()) {
+ LOGGER.warn("Vector index already present in columns.psf for segment: {}, column: {}; "
+ + "deleting orphan sidecar file from a previously-crashed absorb run", segmentName, column);
+ FileUtils.deleteQuietly(picked);
+ return;
+ }
+ // Size mismatch — the typed entry is from a different build than the sidecar. Refuse to
+ // proceed; an operator must reconcile manually rather than have us guess.
+ throw new IOException("Vector index already present in columns.psf for column: " + column
+ + " (size=" + existingSize + ") but sidecar file " + picked.getName()
+ + " has different size " + picked.length() + ". Refusing to overwrite — please remove "
+ + "the conflicting sidecar or rebuild the segment.");
+ }
Review Comment:
Fixed in c04730d. The IVF absorb crash-recovery no longer trusts `hasIndexFor`; it probes the typed entry via `getConsolidatedVectorEntry()` and treats a missing slot as the normal first-absorb path (only running the size-check recovery when a typed entry actually exists). Added a regression test `testUpdateIndicesAbsorbsLegacySidecarWhenNoTypedEntryYet` that mocks the real behavior (`hasIndexFor` true because the sidecar is on disk, `getIndexFor` throws because no typed entry) and asserts the absorb proceeds and deletes the sidecar.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/VectorIndexHandler.java:
##########
@@ -90,6 +122,290 @@ public boolean needUpdateIndices(SegmentDirectory.Reader segmentReader) {
return false;
}
+ /**
+ * Absorbs a vector index sidecar into {@code columns.psf} as a typed entry, then deletes the
+ * sibling. For IVF backends the bytes are copied verbatim; the on-disk IVF header is the
+ * contract and the reader handles version dispatch. For HNSW, if a combined-form file already
+ * exists (written by a creator run with the flag on), it is absorbed directly; if only the
+ * Lucene directory remains (operator just flipped the flag on an existing segment), it is first
+ * packed into a transient combined file, which is then absorbed and removed.
+ *
+ * <p><b>Crash recovery:</b> if a prior absorb crashed between {@code newIndexFor} (which
+ * committed bytes into {@code columns.psf} and added a {@code _columnEntries} entry) and the
+ * sidecar file deletion, the next load will see both forms. We detect that state upfront via
+ * {@code hasIndexFor}, verify the typed entry's size matches the sidecar's length (so we do not
+ * delete a sidecar that happens to coexist with an unrelated typed entry), and clean up the
+ * orphan sidecar instead of re-running the absorb.</p>
+ */
+ private void absorbCombinedIntoColumnsPsf(SegmentDirectory.Writer segmentWriter, String column,
+ VectorBackendType backendType, File indexDir)
+ throws Exception {
+ File v3Dir = SegmentDirectoryPaths.segmentDirectoryFor(indexDir,
+ _segmentDirectory.getSegmentMetadata().getVersion());
+ String segmentName = _segmentDirectory.getSegmentMetadata().getName();
+
+ if (backendType == VectorBackendType.HNSW) {
+ absorbHnswIntoColumnsPsf(segmentWriter, column, v3Dir, segmentName);
+ return;
+ }
+
+ // IVF path: prefer the combined-form file (freshly-written by a creator run with flag=on);
+ // fall back to the legacy sidecar (segments built with flag=off that the operator now wants
+ // consolidated).
+ File combinedFile = new File(v3Dir,
+ column + VectorIndexUtils.getIndexFileExtension(backendType, /* combined */ true));
+ File legacyFile = new File(v3Dir,
+ column + VectorIndexUtils.getIndexFileExtension(backendType, /* combined */ false));
+ File picked = combinedFile.exists() ? combinedFile : legacyFile;
+ if (!picked.exists()) {
+ LOGGER.warn("Expected vector index file {} not found during vector consolidation; skipping", picked);
+ return;
+ }
+ // Crash-recovery: if a prior absorb committed bytes to columns.psf but crashed before
+ // deleting the sidecar, the typed entry is already present. Detect that directly via
+ // hasIndexFor + size check, rather than catching the duplicate-key exception by message.
+ if (segmentWriter.hasIndexFor(column, StandardIndexes.vector())) {
+ long existingSize = segmentWriter.getIndexFor(column, StandardIndexes.vector()).size();
+ if (existingSize == picked.length()) {
+ LOGGER.warn("Vector index already present in columns.psf for segment: {}, column: {}; "
+ + "deleting orphan sidecar file from a previously-crashed absorb run", segmentName, column);
+ FileUtils.deleteQuietly(picked);
+ return;
+ }
+ // Size mismatch — the typed entry is from a different build than the sidecar. Refuse to
+ // proceed; an operator must reconcile manually rather than have us guess.
+ throw new IOException("Vector index already present in columns.psf for column: " + column
+ + " (size=" + existingSize + ") but sidecar file " + picked.getName()
+ + " has different size " + picked.length() + ". Refusing to overwrite — please remove "
+ + "the conflicting sidecar or rebuild the segment.");
+ }
+ LOGGER.info("Absorbing vector sidecar file into columns.psf for segment: {}, column: {} (backend={})",
+ segmentName, column, backendType);
+ LoaderUtils.writeIndexToV3Format(segmentWriter, column, picked, StandardIndexes.vector());
+ }
+
+ /**
+ * HNSW-specific absorb: packs the Lucene directory (or uses an existing combined-form file) into
+ * a single combined file and absorbs it into {@code columns.psf}.
+ *
+ * <p>If the combined-form file already exists (creator ran with the flag on and produced it),
+ * it is absorbed directly. If only the Lucene directory exists (operator just flipped the flag
+ * on an existing segment), the directory is first packed into a transient combined file, which is
+ * then absorbed and cleaned up.</p>
+ */
+ private void absorbHnswIntoColumnsPsf(SegmentDirectory.Writer segmentWriter, String column, File v3Dir,
+ String segmentName)
+ throws Exception {
+ File combinedFile = new File(v3Dir,
+ column + VectorIndexUtils.getIndexFileExtension(VectorBackendType.HNSW, /* combined */ true));
+ File hnswDir = new File(v3Dir,
+ column + VectorIndexUtils.getIndexFileExtension(VectorBackendType.HNSW, /* combined */ false));
+ if (!combinedFile.exists()) {
+ if (!hnswDir.exists() || !hnswDir.isDirectory()) {
+ LOGGER.warn("Expected HNSW directory or combined file for column {} not found; skipping absorb", column);
+ return;
+ }
+ // Pack the Lucene directory into a transient combined file for absorption. The directory
+ // and the combined file are both cleaned up below after a successful absorb (or by the
+ // crash-recovery branch above on a subsequent retry).
+ HnswVectorIndexCombined.combineHnswIndexFiles(hnswDir, combinedFile.getAbsolutePath(), v3Dir.getParentFile(),
+ column);
+ }
+
+ // Crash-recovery: check whether the typed entry already exists.
+ if (segmentWriter.hasIndexFor(column, StandardIndexes.vector())) {
+ long existingSize = segmentWriter.getIndexFor(column, StandardIndexes.vector()).size();
+ if (existingSize == combinedFile.length()) {
+ LOGGER.warn("HNSW vector index already present in columns.psf for segment: {}, column: {}; "
+ + "deleting orphan files from a previously-crashed absorb run", segmentName, column);
+ FileUtils.deleteQuietly(combinedFile);
+ // Clean up the Lucene directory regardless of createdTransient: if we packed the directory
+ // into a transient combined file in this very run, the directory still exists and must be
+ // removed to prevent hasCombinedFile from re-triggering on the next segment load.
+ if (hnswDir.exists()) {
+ FileUtils.deleteDirectory(hnswDir);
+ }
+ return;
+ }
+ throw new IOException("HNSW vector index already present in columns.psf for column: " + column
+ + " (size=" + existingSize + ") but combined file " + combinedFile.getName()
+ + " has different size " + combinedFile.length() + ". Refusing to overwrite — please remove "
+ + "the conflicting sidecar or rebuild the segment.");
+ }
Review Comment:
Same fix as the IVF path, applied to `absorbHnswIntoColumnsPsf` in c04730d — it probes `getConsolidatedVectorEntry()` instead of `hasIndexFor` (which would be true here just because the still-present legacy Lucene directory is on disk), so a normal first absorb is no longer mis-read as already-absorbed.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/converter/SegmentV1V2ToV3FormatConverter.java:
##########
@@ -147,16 +148,31 @@ private void copyIndexData(File v2Directory, SegmentMetadataImpl v2Metadata, Fil
SegmentDirectory.Writer v3DataWriter = v3Segment.createWriter()) {
for (String column : v2Metadata.getAllColumns()) {
for (IndexType<?, ?, ?> indexType : IndexService.getInstance().getAllIndexes()) {
- //If Text index files are combined merge into columns.psf else no-op
+ // Text index: skip the standard copy when a legacy Lucene directory is present
+ // ({@code copyLuceneTextIndexIfExists} below handles that as a sibling copy).
+ // Combined .text.index files fall through to the standard path and get packed.
if (indexType == StandardIndexes.text()) {
if (!TextIndexUtils.hasTextIndex(v2Directory, column)) {
copyIndexIfExists(v2DataReader, v3DataWriter, column, indexType);
}
- } else {
- if (indexType != StandardIndexes.vector()) {
+ continue;
+ }
+ // Vector index mirrors text: skip the standard copy when a legacy IVF sidecar (or
+ // HNSW directory) is present ({@code copyVectorIndexIfExists} handles those as
+ // sibling copies). Combined .vector.ivfflat.combined.index / .ivfpq.combined.index
+ // files fall through to the standard path and get packed into columns.psf.
+ // Mixed state: if both legacy and combined exist (e.g. operator toggled the flag
+ // and rebuilt without cleaning the old file), combined wins — pack it and let
+ // copyVectorIndexIfExists drop the legacy sibling so the operator gets exactly one
+ // copy of the bytes in the V3 segment.
+ if (indexType == StandardIndexes.vector()) {
+ boolean hasCombined = VectorIndexUtils.hasCombinedFormVectorIndex(v2Directory, column);
+ if (hasCombined || !VectorIndexUtils.hasVectorIndex(v2Directory, column)) {
copyIndexIfExists(v2DataReader, v3DataWriter, column, indexType);
}
+ continue;
}
Review Comment:
Fixed in c04730d, with one adjustment: switching purely to `!hasVectorIndex(...)` would silently lose the index in a mixed **IVF** state — `copyVectorIndexIfExists` skips a legacy IVF sidecar when a combined sibling exists, and the combined form would no longer be packed either, leaving the column with no vector index in V3.
So the gate now packs the combined form into columns.psf only when the resolved on-disk artifact is a regular file (`!resolved.isDirectory()`). A legacy HNSW Lucene **directory** (which `findHnswVectorIndexFile` resolves before the combined file) is deferred to the sibling copy — avoiding the directory-as-buffer mapping you flagged — while a legacy IVF **file** still packs, so the index is never lost. `VectorIndexHandler` re-absorbs the deferred HNSW directory later if `storeInSegmentFile` stays on.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/vector/lucene99/HnswVectorIndexCombined.java:
##########
@@ -0,0 +1,351 @@
+/**
+ * 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.creator.impl.vector.lucene99;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.channels.FileChannel;
+import java.nio.file.StandardOpenOption;
+import java.util.Map;
+import java.util.TreeMap;
+import javax.annotation.Nullable;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.segment.local.segment.creator.impl.text.LuceneCombinedTextIndexConstants;
+import org.apache.pinot.segment.spi.V1Constants;
+import org.apache.pinot.segment.spi.store.SegmentDirectoryPaths;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Utility class to pack a Lucene HNSW index directory (and its optional docId mapping file) into
+ * a single combined file using the {@link LuceneCombinedTextIndexConstants#MAGIC_NUMBER LUCENE_V2}
+ * layout. Mirrors {@code LuceneTextIndexCombined} — reuses the same format constants to keep a
+ * single on-disk format shared across text and HNSW vector indexes.
+ *
+ * <p>Layout (identical to the text-index LUCENE_V2 format):</p>
+ * <pre>
+ * [Header]
+ * Magic "LUCENE_V2" 9 bytes
+ * Version 4 bytes (little-endian int)
+ * Total buffer size 8 bytes (little-endian long)
+ * File count 4 bytes (little-endian int)
+ * Reserved 4 bytes
+ *
+ * [File metadata, one entry per file]
+ * Name length 2 bytes (little-endian short)
+ * Name variable
+ * File offset 8 bytes (little-endian long)
+ * File size 8 bytes (little-endian long)
+ *
+ * [File data]
+ * Raw bytes of each file concatenated in metadata order
+ * </pre>
+ */
+public final class HnswVectorIndexCombined {
+ private static final Logger LOGGER = LoggerFactory.getLogger(HnswVectorIndexCombined.class);
+
+ private HnswVectorIndexCombined() {
+ }
+
+ /**
+ * Packs all files in {@code hnswIndexDir} (plus the optional docId mapping file) into a single
+ * combined file at {@code outputFilePath}.
+ *
+ * @param hnswIndexDir the Lucene HNSW index directory to pack
+ * @param outputFilePath destination path for the combined file
+ * @param segmentIndexDir when non-null, the segment's top-level index directory; used to
+ * locate the docId mapping file for inclusion in the packed output
+ * @param column column name; used to locate the docId mapping file when
+ * {@code segmentIndexDir} is provided
+ * @throws IOException if any file operations fail
+ */
+ public static void combineHnswIndexFiles(File hnswIndexDir, String outputFilePath,
+ @Nullable File segmentIndexDir, @Nullable String column)
+ throws IOException {
+ if (!hnswIndexDir.exists() || !hnswIndexDir.isDirectory()) {
+ throw new IllegalArgumentException(
+ "HNSW index directory does not exist or is not a directory: " + hnswIndexDir);
+ }
+
+ LOGGER.info("Combining HNSW index files from directory: {}", hnswIndexDir.getAbsolutePath());
+
+ Map<String, FileInfo> fileInfoMap = collectFiles(hnswIndexDir, segmentIndexDir, column);
+ int fileCount = fileInfoMap.size();
+
+ if (fileCount == 0) {
+ throw new IOException("No files found in HNSW index directory: " + hnswIndexDir);
+ }
+
+ long totalSize = calculateTotalBufferSize(fileInfoMap);
+ if (totalSize > Integer.MAX_VALUE) {
+ throw new IOException("Combined HNSW index size too large: " + totalSize + " bytes");
+ }
+
+ File outputFile = new File(outputFilePath);
+ try (FileChannel outputChannel = FileChannel.open(outputFile.toPath(), StandardOpenOption.CREATE,
+ StandardOpenOption.WRITE)) {
+ writeHeader(outputChannel, fileCount, (int) totalSize);
Review Comment:
Addressed — `combineHnswIndexFiles` opens the output channel with `StandardOpenOption.TRUNCATE_EXISTING` so a leftover larger file from a previously-crashed pack cannot leave stale trailing bytes that would inflate the length and break the size-based crash-recovery check.
##########
pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/creator/VectorIndexConfig.java:
##########
@@ -139,6 +149,31 @@ public VectorIndexConfig setProperties(Map<String, String> properties) {
return this;
}
+ /**
+ * Whether the IVF vector index payload is consolidated into the segment's combined index file
+ * ({@code columns.psf}) rather than written as a combined file. Default {@code false}. Only
+ * applies to V3 segments and IVF backends; HNSW is unaffected.
+ *
Review Comment:
Updated — the `isStoreInSegmentFile()` Javadoc now states the flag applies to V3 segments for both IVF and HNSW backends, with the backend-specific build/read details in the following paragraphs.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/readers/vector/HnswVectorIndexBufferReader.java:
##########
@@ -0,0 +1,119 @@
+/**
+ * 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 java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import javax.annotation.Nullable;
+import org.apache.lucene.store.Directory;
+import org.apache.pinot.segment.local.segment.index.readers.text.LuceneTextIndexBufferReader;
+import org.apache.pinot.segment.local.segment.index.readers.text.LuceneTextIndexHeader;
+import org.apache.pinot.segment.local.segment.index.readers.text.PinotBufferLuceneDirectory;
+import org.apache.pinot.segment.spi.V1Constants;
+import org.apache.pinot.segment.spi.memory.PinotDataBuffer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Utility for reading an HNSW vector index from a combined buffer ({@link PinotDataBuffer}).
+ *
+ * <p>The combined buffer uses the same {@code LUCENE_V2} layout as the text index (written by
+ * {@code HnswVectorIndexCombined}). This class delegates header and metadata parsing to
+ * {@link LuceneTextIndexBufferReader}, then constructs a {@link PinotBufferLuceneDirectory} that
+ * Lucene's {@code DirectoryReader} can open without any filesystem access.</p>
+ *
+ * <p>Two entry-points are provided:</p>
+ * <ul>
+ * <li>{@link #createLuceneDirectory} — a read-only {@link Directory} over the Lucene index
+ * files embedded in the buffer.</li>
+ * <li>{@link #extractDocIdMappingBuffer} — a sub-buffer view covering just the docId mapping
+ * bytes, or {@code null} if the mapping was not packed.</li>
+ * </ul>
+ */
+public final class HnswVectorIndexBufferReader {
+ private static final Logger LOGGER = LoggerFactory.getLogger(HnswVectorIndexBufferReader.class);
+
+ private HnswVectorIndexBufferReader() {
+ }
+
+ /**
+ * Creates a Lucene {@link Directory} that reads all Lucene index files from a combined HNSW
+ * buffer. The docId mapping file (a Pinot-only artifact with the
+ * {@link V1Constants.Indexes#VECTOR_HNSW_INDEX_DOCID_MAPPING_FILE_EXTENSION} suffix) is
+ * excluded from the directory view so that {@code DirectoryReader} does not see an unknown file.
+ *
+ * <p>The directory does <em>not</em> own the buffer — closing it is a no-op. The caller must
+ * keep the buffer alive for as long as the directory (and any reader opened on it) is in use.</p>
+ *
+ * @param indexBuffer combined buffer in LUCENE_V2 format
+ * @param column column name; used to identify and exclude the mapping file
+ * @return a read-only {@link Directory} backed by the buffer
+ * @throws IOException if the buffer is malformed or the magic/version check fails
+ */
+ public static Directory createLuceneDirectory(PinotDataBuffer indexBuffer, String column)
+ throws IOException {
+ List<String> fileNames = LuceneTextIndexBufferReader.listFiles(indexBuffer);
+ Map<String, LuceneTextIndexHeader.FileInfo> fileMap = new HashMap<>(fileNames.size());
+
+ String mappingFileName = column + V1Constants.Indexes.VECTOR_HNSW_INDEX_DOCID_MAPPING_FILE_EXTENSION;
+ for (String name : fileNames) {
+ if (name.equals(mappingFileName)) {
+ // Skip: not a Lucene index file; presenting it to DirectoryReader triggers an error.
+ continue;
+ }
+ LuceneTextIndexHeader.FileInfo info = LuceneTextIndexBufferReader.getFileInfo(indexBuffer, name);
+ if (info != null) {
+ fileMap.put(name, info);
+ }
+ }
+
+ LOGGER.debug("Creating buffer-backed Lucene directory for HNSW column '{}' with {} file(s)", column,
+ fileMap.size());
+ return new PinotBufferLuceneDirectory(indexBuffer, fileMap, column);
+ }
+
+ /**
+ * Extracts a sub-buffer covering the docId mapping bytes packed inside the combined buffer.
+ *
+ * <p>Returns {@code null} when the mapping was not included (e.g. the index was built from an
+ * in-memory segment that had not yet materialised a mapping file). The caller must build the
+ * mapping from the Lucene index in that case.</p>
+ *
+ * <p>The returned sub-buffer is a {@link PinotDataBuffer#view view} of the original buffer and
+ * shares its lifetime. The caller must not close it independently.</p>
+ *
+ * @param indexBuffer combined buffer in LUCENE_V2 format
+ * @param column column name; used to identify the mapping file entry
+ * @return sub-buffer for the mapping bytes, or {@code null} if not present
+ * @throws IOException if header parsing fails
+ */
+ @Nullable
+ public static PinotDataBuffer extractDocIdMappingBuffer(PinotDataBuffer indexBuffer, String column)
+ throws IOException {
+ String mappingFileName = column + V1Constants.Indexes.VECTOR_HNSW_INDEX_DOCID_MAPPING_FILE_EXTENSION;
+ LuceneTextIndexHeader.FileInfo fileInfo = LuceneTextIndexBufferReader.getFileInfo(indexBuffer, mappingFileName);
+ if (fileInfo == null) {
+ return null;
+ }
+ return indexBuffer.view(fileInfo.getOffset(), fileInfo.getOffset() + fileInfo.getSize());
Review Comment:
Good catch — fixed in df94a35. `extractDocIdMappingBuffer` now returns a `LITTLE_ENDIAN` view, so `DocIdTranslator.getInt()` reads the Lucene→Pinot doc ids unswapped regardless of the big-endian `columns.psf` order the view would otherwise inherit. Added a regression test (`testExtractDocIdMappingBufferIsLittleEndian`) that packs a known mapping and reads it back through a big-endian columns.psf-style buffer, which fails without the little-endian view.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Consolidate IVF and HNSW vector indexes into segment columns.psf (opt-in via storeInSegmentFile) [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18852:
URL: https://github.com/apache/pinot/pull/18852#discussion_r3498628717
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/vector/VectorIndexType.java:
##########
@@ -210,25 +211,85 @@ public VectorIndexReader createIndexReader(SegmentDirectory.Reader segmentReader
return null;
}
VectorBackendType backendType = indexConfig.resolveBackendType();
- File configuredIndexFile =
- SegmentDirectoryPaths.findVectorIndexIndexFile(segmentDir, metadata.getColumnName(), indexConfig);
- if (configuredIndexFile == null || !configuredIndexFile.exists()) {
- LOGGER.warn("Skipping vector index reader for column: {} because configured backend {} does not have a "
- + "matching on-disk artifact in segment: {}",
- metadata.getColumnName(), backendType, segmentDir);
- return null;
+ String column = metadata.getColumnName();
+
+ if (backendType == VectorBackendType.HNSW) {
+ // Combined form: load the HNSW index from the typed entry inside columns.psf when one
+ // actually exists. getConsolidatedVectorEntry (unlike hasIndexFor) does not mistake the
+ // legacy Lucene directory for a packed entry, so a segment whose handler has not yet
+ // migrated falls through to the legacy path below instead of failing the whole load. The
+ // buffer is owned by the segment directory — this reader must not close it.
+ if (indexConfig.isStoreInSegmentFile()) {
+ PinotDataBuffer buffer;
+ try {
+ buffer = VectorIndexUtils.getConsolidatedVectorEntry(segmentReader, column);
Review Comment:
This consolidated-entry probe is not safe on `FilePerIndexDirectory` segments that still have the legacy HNSW directory on disk. In that case `getIndexFor(StandardIndexes.vector())` resolves the existing directory candidate, `mapForReads()` rejects it because it is not a regular file, and the method throws before the fallback at lines 236-237 can run. That makes `storeInSegmentFile=true` non-rolling-upgrade-safe for existing V1/V2 HNSW segments. Please gate this probe to the packed-file/V3 case or catch the directory case and fall back to the legacy reader.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/VectorIndexHandler.java:
##########
@@ -78,6 +85,31 @@ public boolean needUpdateIndices(SegmentDirectory.Reader segmentReader) {
segmentName, column, existingBackend, desiredBackend);
return true;
}
+ // Migration: sidecar -> columns.psf. Fires either when the offline writer just wrote a
+ // sidecar (build with flag on) or when the operator just flipped the flag from off to on
+ // for an existing legacy segment. Same handler step covers both.
+ // For HNSW the "sidecar" is the combined packed file (.vector.hnsw.combined.index);
+ // for IVF it is the legacy combined extension.
+ if (desiredConfig.isStoreInSegmentFile()
+ && _segmentDirectory.getSegmentMetadata().getVersion() == SegmentVersion.v3
+ && existingBackend != null
+ && hasCombinedFile(indexDir, column, desiredBackend)) {
+ LOGGER.info("Need to consolidate Vector sidecar file into columns.psf for segment: {}, column: {}",
+ segmentName, column);
+ return true;
+ }
+ // Migration: columns.psf -> sidecar. Fires when the operator flipped the flag from on to
+ // off for a segment whose vector payload was previously absorbed. We detect this from the
+ // SegmentDirectory: the column has a vector index ({@code existingColumns} contains it via
+ // {@code SingleFileIndexDirectory._columnEntries}) but no sidecar file is on disk.
+ if (!desiredConfig.isStoreInSegmentFile()
+ && _segmentDirectory.getSegmentMetadata().getVersion() == SegmentVersion.v3
+ && existingBackend == null
Review Comment:
Once `existingColumns` can include a vector index that lives only in `columns.psf`, `existingBackend == null` no longer means "there is no existing backend". If a V3 segment already holds, say, a consolidated HNSW payload and the table config flips to `IVF_PQ`, this branch treats it as a pure `columns.psf -> sidecar` migration, extracts the old HNSW bytes under the IVF_PQ extension, and removes the typed entry instead of rebuilding. The next load then instantiates the new backend reader against the old backend payload. This backend-drift check needs to infer the existing backend from the consolidated entry too, not just from sidecar files.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Consolidate IVF and HNSW vector indexes into segment columns.psf (opt-in via storeInSegmentFile) [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on PR #18852:
URL: https://github.com/apache/pinot/pull/18852#issuecomment-4895048893
Docs follow-up PR: https://github.com/pinot-contrib/pinot-docs/pull/898
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org