You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by xi...@apache.org on 2026/10/20 20:16:18 UTC
(pinot) branch release-1.6.0-rc updated: [1.6.0] Backport Lucene index rollback recovery (#19976)
This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch release-1.6.0-rc
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/release-1.6.0-rc by this push:
new a1c9200b95f [1.6.0] Backport Lucene index rollback recovery (#19976)
a1c9200b95f is described below
commit a1c9200b95f7f11c0405873b8033d3fadd24f6dd
Author: Xiang Fu <xi...@apache.org>
AuthorDate: Tue Oct 20 13:16:11 2026 -0700
[1.6.0] Backport Lucene index rollback recovery (#19976)
Backport #19900 so Pinot 1.6.0 can serve as the rollback target for a later Lucene upgrade. Rebuild incompatible text and HNSW indexes from retained original forward values; keep Lucene at 9.12.0.
(cherry picked from commit a45ebf01379257591c27c900c3525612a3c49b92)
---
README.md | 15 +
.../segment/index/loader/SegmentPreProcessor.java | 96 ++---
.../invertedindex/MultiColumnTextIndexHandler.java | 136 +++++-
.../loader/invertedindex/TextIndexHandler.java | 183 +++++---
.../loader/invertedindex/VectorIndexHandler.java | 97 ++++-
.../local/segment/store/TextIndexUtils.java | 20 +
.../local/utils/LuceneIndexCompatibilityUtils.java | 117 ++++++
.../invertedindex/HnswIndexRollbackTest.java | 304 ++++++++++++++
.../TextIndexRollbackRecoveryTest.java | 460 +++++++++++++++++++++
.../invertedindex/VectorIndexHandlerTest.java | 13 +-
.../utils/LuceneIndexCompatibilityUtilsTest.java | 96 +++++
.../segment/local/utils/LuceneIndexTestUtils.java | 61 +++
.../src/test/resources/data/lucene10-rollback.zip | Bin 0 -> 9060 bytes
13 files changed, 1452 insertions(+), 146 deletions(-)
diff --git a/README.md b/README.md
index aa3e5713f20..c27335478b2 100644
--- a/README.md
+++ b/README.md
@@ -175,6 +175,21 @@ To verify which profile is active, run:
If you encounter issues, check that the `protoc-gen-grpc-java` binary is present in one of the expected locations and is executable.
+### Lucene index recovery during loading
+
+Normal segment preprocessing automatically rebuilds text and HNSW indexes whose Lucene format cannot be read by
+the running version. Single-column text, multi-column text, and directory/packed HNSW indexes are rebuilt using
+the segment's original forward-index values and existing table configuration; no recovery flag is needed.
+Compatible indexes are reused, and subsequent loads do not rebuild the recovered indexes again. Interrupted builds
+restart from the retained source values before the segment is loaded.
+
+Install this recovery release on the intended rollback target before upgrading Lucene. After a downgrade, loading
+segments containing newer Lucene indexes rebuilds those indexes with the older writer. Retain the original forward
+indexes: dictionary/inverted-index reconstruction can lose multi-value ordering and duplicate values. When original
+values are unavailable, recovery fails before replacing the incompatible index; restore a compatible segment or
+refresh it from retained source data. Rebuilding adds CPU/I/O and segment-loading time. Existing preprocessing
+opt-outs remain in effect. Corruption and ordinary I/O errors fail loading instead of triggering format recovery.
+
## Deploying Pinot to Kubernetes
Please refer to [Running Pinot on Kubernetes](https://docs.pinot.apache.org/basics/getting-started/kubernetes-quickstart) in our project documentation. Pinot also provides Kubernetes integrations with the interactive query engine, [Trino](https://docs.pinot.apache.org/integrations/trino), and the data visualization tool, [Apache Superset](helm/superset.yaml).
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/SegmentPreProcessor.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/SegmentPreProcessor.java
index f8dbc0dddce..d7f0d280ebc 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/SegmentPreProcessor.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/SegmentPreProcessor.java
@@ -20,15 +20,13 @@ package org.apache.pinot.segment.local.segment.index.loader;
import com.google.common.base.Preconditions;
import java.io.File;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.LinkedHashSet;
import java.util.List;
+import java.util.Map;
import java.util.ServiceLoader;
import java.util.Set;
import javax.annotation.Nullable;
-import org.apache.commons.configuration2.PropertiesConfiguration;
-import org.apache.commons.configuration2.ex.ConfigurationException;
import org.apache.commons.io.FileUtils;
import org.apache.pinot.common.metrics.ServerMeter;
import org.apache.pinot.common.metrics.ServerMetrics;
@@ -38,23 +36,21 @@ import org.apache.pinot.segment.local.segment.index.loader.defaultcolumn.Default
import org.apache.pinot.segment.local.segment.index.loader.defaultcolumn.DefaultColumnHandlerFactory;
import org.apache.pinot.segment.local.segment.index.loader.invertedindex.LegacyRawValueInvertedIndexCleanup;
import org.apache.pinot.segment.local.segment.index.loader.invertedindex.MultiColumnTextIndexHandler;
+import org.apache.pinot.segment.local.segment.index.loader.invertedindex.TextIndexHandler;
+import org.apache.pinot.segment.local.segment.index.loader.invertedindex.VectorIndexHandler;
import org.apache.pinot.segment.local.startree.StarTreeBuilderUtils;
import org.apache.pinot.segment.local.startree.v2.builder.MultipleTreesBuilder;
import org.apache.pinot.segment.local.startree.v2.builder.StarTreeV2BuilderConfig;
import org.apache.pinot.segment.local.utils.SegmentOperationsThrottlerSet;
import org.apache.pinot.segment.spi.V1Constants;
+import org.apache.pinot.segment.spi.index.FieldIndexConfigs;
import org.apache.pinot.segment.spi.index.IndexHandler;
import org.apache.pinot.segment.spi.index.IndexService;
import org.apache.pinot.segment.spi.index.IndexType;
import org.apache.pinot.segment.spi.index.StandardIndexes;
import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
-import org.apache.pinot.segment.spi.index.multicolumntext.MultiColumnTextIndexConstants;
-import org.apache.pinot.segment.spi.index.multicolumntext.MultiColumnTextMetadata;
import org.apache.pinot.segment.spi.index.startree.StarTreeV2Metadata;
import org.apache.pinot.segment.spi.store.SegmentDirectory;
-import org.apache.pinot.segment.spi.store.SegmentDirectoryPaths;
-import org.apache.pinot.segment.spi.utils.SegmentMetadataUtils;
-import org.apache.pinot.spi.config.table.MultiColumnTextIndexConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.plugin.PluginManager;
@@ -149,6 +145,16 @@ public class SegmentPreProcessor implements AutoCloseable {
removeInvertedIndexTempFiles(indexDir);
try (SegmentDirectory.Writer segmentWriter = _segmentDirectory.createWriter()) {
+ // Recovery must use retained forward values. Other handlers can reconstruct multi-value forward indexes from
+ // dictionary/inverted indexes, losing ordering and duplicates, or drop the original source altogether.
+ Map<String, FieldIndexConfigs> recoveryConfigs = _indexLoadingConfig.getFieldIndexConfigByColName();
+ new TextIndexHandler(_segmentDirectory, recoveryConfigs, _tableConfig, _schema)
+ .validateRecoverySources(segmentWriter);
+ new VectorIndexHandler(_segmentDirectory, recoveryConfigs, _tableConfig, _schema)
+ .validateRecoverySources(segmentWriter);
+ new MultiColumnTextIndexHandler(_segmentDirectory, _indexLoadingConfig,
+ _indexLoadingConfig.getMultiColTextIndexConfig()).validateRecoverySources(segmentWriter);
+
// Backward-compat shim: invalidate any legacy raw-value embedded-dictionary inverted indexes left over from
// PR #17060 (reverted by PR #18410) so the standard handlers can rebuild them in the dict-id format. Must
// run before any handler that may try to read the inverted-index buffer. Safe to delete after Pinot 1.7;
@@ -222,7 +228,7 @@ public class SegmentPreProcessor implements AutoCloseable {
segmentWriter.save();
}
// Create/modify/remove multi-col text index if required.
- if (processMultiColTextIndex(indexDir, segmentWriter, segmentOperationsThrottlerSet)) {
+ if (processMultiColTextIndex(segmentWriter, segmentOperationsThrottlerSet)) {
// NOTE: When adding new steps after this, un-comment the next line.
//_segmentDirectory.reloadMetadata();
segmentWriter.save();
@@ -267,7 +273,7 @@ public class SegmentPreProcessor implements AutoCloseable {
}
// Check if there is need to create/modify/remove multi-col text index
- if (needProcessMultiColumnTextIndex()) {
+ if (needProcessMultiColumnTextIndex(segmentReader)) {
LOGGER.info("Found multi-column text index needs updates in segment: {}", segmentName);
return true;
}
@@ -317,56 +323,25 @@ public class SegmentPreProcessor implements AutoCloseable {
return !starTreeBuilderConfigs.isEmpty();
}
- private boolean needProcessMultiColumnTextIndex() {
- MultiColumnTextIndexConfig newConfig = _indexLoadingConfig.getMultiColTextIndexConfig();
- MultiColumnTextMetadata oldConfig = _segmentDirectory.getSegmentMetadata().getMultiColumnTextMetadata();
- return MultiColumnTextIndexHandler.shouldModifyMultiColTextIndex(newConfig, oldConfig);
+ private boolean needProcessMultiColumnTextIndex(SegmentDirectory.Reader segmentReader) {
+ return new MultiColumnTextIndexHandler(_segmentDirectory, _indexLoadingConfig,
+ _indexLoadingConfig.getMultiColTextIndexConfig()).needUpdateIndices(segmentReader);
}
- private boolean processMultiColTextIndex(File indexDir, SegmentDirectory.Writer segmentWriter,
+ private boolean processMultiColTextIndex(SegmentDirectory.Writer segmentWriter,
@Nullable SegmentOperationsThrottlerSet segmentOperationsThrottlerSet)
throws Exception {
- SegmentMetadataImpl segmentMetadata = _segmentDirectory.getSegmentMetadata();
- String segmentName = segmentMetadata.getName();
- MultiColumnTextMetadata oldConfig = segmentMetadata.getMultiColumnTextMetadata();
- MultiColumnTextIndexConfig newConfig = _indexLoadingConfig.getMultiColTextIndexConfig();
- boolean remove = false;
- boolean create = newConfig != null;
-
- if (oldConfig != null) {
- if (newConfig == null) {
- remove = true;
- } else {
- if (MultiColumnTextIndexHandler.shouldModifyMultiColTextIndex(newConfig, oldConfig)) {
- LOGGER.info("Change detected in multi-column text index for segment: {}", segmentName);
- } else {
- create = false;
- }
- }
- }
- if (!remove && !create) {
- LOGGER.info("No change detected in multi-column text index for segment: {}", segmentName);
+ MultiColumnTextIndexHandler handler = new MultiColumnTextIndexHandler(_segmentDirectory, _indexLoadingConfig,
+ _indexLoadingConfig.getMultiColTextIndexConfig());
+ if (!handler.needUpdateIndices(segmentWriter)) {
return false;
}
-
if (segmentOperationsThrottlerSet != null) {
segmentOperationsThrottlerSet.getSegmentMultiColTextIndexPreprocessThrottler().acquire();
}
try {
- if (remove) {
- LOGGER.info("Removing multi-column text index from segment: {}", segmentName);
- removeMultiColumnTextIndex(indexDir);
- } else if (create) {
- if (oldConfig != null) {
- // Drop existing multi-column text index before creating a new one
- // TODO: check if it's possible to only add/remove select columns
- removeMultiColumnTextIndex(indexDir);
- }
- MultiColumnTextIndexHandler handler =
- new MultiColumnTextIndexHandler(_segmentDirectory, _indexLoadingConfig, newConfig);
- handler.updateIndices(segmentWriter);
- handler.postUpdateIndicesCleanup(segmentWriter);
- }
+ handler.updateIndices(segmentWriter);
+ handler.postUpdateIndicesCleanup(segmentWriter);
} finally {
if (segmentOperationsThrottlerSet != null) {
segmentOperationsThrottlerSet.getSegmentMultiColTextIndexPreprocessThrottler().release();
@@ -375,27 +350,6 @@ public class SegmentPreProcessor implements AutoCloseable {
return true;
}
- private void removeMultiColumnTextIndex(File indexDir)
- throws ConfigurationException, IOException {
- // Remove the multi-col text index metadata
- PropertiesConfiguration metadataProperties = SegmentMetadataUtils.getPropertiesConfiguration(indexDir);
- metadataProperties.subset(MultiColumnTextIndexConstants.MetadataKey.ROOT_SUBSET).clear();
- SegmentMetadataUtils.savePropertiesConfiguration(metadataProperties, indexDir);
-
- // Remove the index file and index map file
- File segmentDirectory = SegmentDirectoryPaths.findSegmentDirectory(indexDir);
- File textIdxDir =
- SegmentDirectoryPaths.findTextIndexIndexFile(segmentDirectory, MultiColumnTextIndexConstants.INDEX_DIR_NAME);
-
- if (textIdxDir != null && textIdxDir.exists()) {
- FileUtils.forceDelete(textIdxDir);
- }
- File mappingFile = new File(segmentDirectory, MultiColumnTextIndexConstants.DOCID_MAPPING_FILE_NAME);
- if (mappingFile.exists()) {
- FileUtils.forceDelete(mappingFile);
- }
- }
-
private boolean processStarTrees(File indexDir,
@Nullable SegmentOperationsThrottlerSet segmentOperationsThrottlerSet)
throws Exception {
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/MultiColumnTextIndexHandler.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/MultiColumnTextIndexHandler.java
index 273ceb5974e..968dc26a878 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/MultiColumnTextIndexHandler.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/MultiColumnTextIndexHandler.java
@@ -26,10 +26,12 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import org.apache.commons.configuration2.PropertiesConfiguration;
+import org.apache.commons.io.FileUtils;
import org.apache.pinot.segment.local.segment.creator.impl.text.MultiColumnLuceneTextIndexCreator;
import org.apache.pinot.segment.local.segment.index.dictionary.DictionaryIndexType;
import org.apache.pinot.segment.local.segment.index.loader.BaseIndexHandler;
import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
+import org.apache.pinot.segment.local.utils.LuceneIndexCompatibilityUtils;
import org.apache.pinot.segment.local.utils.TableConfigUtils;
import org.apache.pinot.segment.spi.ColumnMetadata;
import org.apache.pinot.segment.spi.index.IndexReaderConstraintException;
@@ -37,6 +39,7 @@ import org.apache.pinot.segment.spi.index.IndexReaderFactory;
import org.apache.pinot.segment.spi.index.StandardIndexes;
import org.apache.pinot.segment.spi.index.TextIndexConfig;
import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
+import org.apache.pinot.segment.spi.index.multicolumntext.MultiColumnTextIndexConstants;
import org.apache.pinot.segment.spi.index.multicolumntext.MultiColumnTextMetadata;
import org.apache.pinot.segment.spi.index.reader.Dictionary;
import org.apache.pinot.segment.spi.index.reader.ForwardIndexReader;
@@ -92,8 +95,12 @@ public class MultiColumnTextIndexHandler extends BaseIndexHandler {
String segmentName = segmentMetadata.getName();
MultiColumnTextMetadata oldConfig = segmentMetadata.getMultiColumnTextMetadata();
- boolean needUpdate = shouldModifyMultiColTextIndex(_textIndexConfig, oldConfig);
+ boolean needUpdate = shouldModifyMultiColTextIndex(_textIndexConfig, oldConfig)
+ || isBuildInProgress() || isIncompatible();
if (needUpdate) {
+ if (_textIndexConfig == null) {
+ return true;
+ }
List<String> newColumns = _textIndexConfig.getColumns();
TableConfigUtils.checkForDuplicates(newColumns);
for (String column : newColumns) {
@@ -113,6 +120,73 @@ public class MultiColumnTextIndexHandler extends BaseIndexHandler {
}
}
+ /// Verify existing source columns before other handlers can replace their original forward values.
+ /// Newly added default columns are checked with all other sources immediately before recovery.
+ public void validateRecoverySources(SegmentDirectory.Reader segmentReader)
+ throws IOException {
+ validateRecoverySources(segmentReader, true);
+ }
+
+ private void validateRecoverySources(SegmentDirectory.Reader segmentReader, boolean existingColumnsOnly)
+ throws IOException {
+ if (_textIndexConfig != null && (isBuildInProgress() || isIncompatible())) {
+ for (String column : _textIndexConfig.getColumns()) {
+ // DefaultColumnHandler creates newly added columns after the early source validation.
+ if (existingColumnsOnly && _segmentDirectory.getSegmentMetadata().getColumnMetadataFor(column) == null) {
+ continue;
+ }
+ if (!segmentReader.hasIndexFor(column, StandardIndexes.forward())
+ || _fieldIndexConfigs.get(column).getConfig(StandardIndexes.forward()).isDisabled()) {
+ throw new IllegalStateException("Cannot rebuild multi-column Lucene text index for column: " + column
+ + ". Original forward index must be retained; restore a compatible segment or refresh it from source"
+ + " data");
+ }
+ }
+ }
+ }
+
+ private File getBuildMarker() {
+ File indexDir = _segmentDirectory.getSegmentMetadata().getIndexDir();
+ return indexDir == null
+ ? null
+ : new File(indexDir, MultiColumnTextIndexConstants.INDEX_DIR_NAME + ".text.inprogress");
+ }
+
+ private boolean isBuildInProgress() {
+ File marker = getBuildMarker();
+ if (marker == null) {
+ return false;
+ }
+ // Older preprocessing removed metadata before building and wrote no marker. A leftover shared directory
+ // without its metadata is also an unfinished build; it must never be appended to on retry.
+ if (marker.exists()) {
+ return true;
+ }
+ return _textIndexConfig != null && _segmentDirectory.getSegmentMetadata().getMultiColumnTextMetadata() == null
+ && SegmentDirectoryPaths.findTextIndexIndexFile(_segmentDirectory.getSegmentMetadata().getIndexDir(),
+ MultiColumnTextIndexConstants.INDEX_DIR_NAME) != null;
+ }
+
+ private boolean isIncompatible() {
+ if (_segmentDirectory.getSegmentMetadata().getMultiColumnTextMetadata() == null || _textIndexConfig == null) {
+ return false;
+ }
+ File indexDir = _segmentDirectory.getSegmentMetadata().getIndexDir();
+ if (indexDir == null) {
+ return false;
+ }
+ File luceneDirectory =
+ SegmentDirectoryPaths.findTextIndexIndexFile(indexDir, MultiColumnTextIndexConstants.INDEX_DIR_NAME);
+ if (luceneDirectory == null) {
+ return false;
+ }
+ try {
+ return LuceneIndexCompatibilityUtils.isIncompatible(luceneDirectory);
+ } catch (IOException e) {
+ throw new IllegalStateException("Cannot inspect multi-column Lucene text index in: " + luceneDirectory, e);
+ }
+ }
+
private void validate(DataType columnMeta, String column) {
if (columnMeta != DataType.STRING) {
throw new UnsupportedOperationException(
@@ -153,30 +227,56 @@ public class MultiColumnTextIndexHandler extends BaseIndexHandler {
@Override
public void updateIndices(SegmentDirectory.Writer segmentWriter)
throws Exception {
+ validateRecoverySources(segmentWriter, false);
File indexDir = _segmentDirectory.getSegmentMetadata().getIndexDir();
File segmentDirectory = SegmentDirectoryPaths.segmentDirectoryFor(indexDir,
_segmentDirectory.getSegmentMetadata().getVersion());
- TableConfigUtils.checkForDuplicates(_textIndexConfig.getColumns());
+ BooleanList columnsSV = new BooleanArrayList();
+ if (_textIndexConfig != null) {
+ TableConfigUtils.checkForDuplicates(_textIndexConfig.getColumns());
+ for (String column : _textIndexConfig.getColumns()) {
+ ColumnMetadata metadata = _segmentDirectory.getSegmentMetadata().getColumnMetadataFor(column);
+ validate(metadata.getFieldSpec().getDataType(), column);
+ columnsSV.add(metadata.isSingleValue());
+ }
+ }
+
+ File marker = getBuildMarker();
+ FileUtils.touch(marker);
+ removeMultiColumnTextIndex(indexDir, segmentDirectory);
- BooleanList columnsSV = new BooleanArrayList(_textIndexConfig.getColumns().size());
- for (String column : _textIndexConfig.getColumns()) {
- ColumnMetadata metadata = _segmentDirectory.getSegmentMetadata().getColumnMetadataFor(column);
- validate(metadata.getFieldSpec().getDataType(), column);
- columnsSV.add(metadata.isSingleValue());
+ if (_textIndexConfig != null) {
+ try (MultiColumnLuceneTextIndexCreator creator =
+ new MultiColumnLuceneTextIndexCreator(_textIndexConfig.getColumns(), columnsSV, segmentDirectory, true, false,
+ null, null, _textIndexConfig)) {
+ createMultiColumnTextIndex(segmentWriter, creator);
+ creator.seal();
+
+ // Write the metadata for multi-column text index
+ PropertiesConfiguration metadataProperties = SegmentMetadataUtils.getPropertiesConfiguration(indexDir);
+ MultiColumnTextMetadata.writeMetadata(metadataProperties, MultiColumnTextMetadata.VERSION_1,
+ _textIndexConfig.getColumns(), _textIndexConfig.getProperties(), _textIndexConfig.getPerColumnProperties());
+ SegmentMetadataUtils.savePropertiesConfiguration(metadataProperties, indexDir);
+ }
}
+ FileUtils.forceDelete(marker);
+ }
+
+ private void removeMultiColumnTextIndex(File indexDir, File segmentDirectory)
+ throws Exception {
+ PropertiesConfiguration metadataProperties = SegmentMetadataUtils.getPropertiesConfiguration(indexDir);
+ metadataProperties.subset(MultiColumnTextIndexConstants.MetadataKey.ROOT_SUBSET).clear();
+ SegmentMetadataUtils.savePropertiesConfiguration(metadataProperties, indexDir);
- try (MultiColumnLuceneTextIndexCreator creator =
- new MultiColumnLuceneTextIndexCreator(_textIndexConfig.getColumns(), columnsSV, segmentDirectory, true, false,
- null, null, _textIndexConfig)) {
- createMultiColumnTextIndex(segmentWriter, creator);
- creator.seal();
-
- // Write the metadata for multi-column text index
- PropertiesConfiguration metadataProperties = SegmentMetadataUtils.getPropertiesConfiguration(indexDir);
- MultiColumnTextMetadata.writeMetadata(metadataProperties, MultiColumnTextMetadata.VERSION_1,
- _textIndexConfig.getColumns(), _textIndexConfig.getProperties(), _textIndexConfig.getPerColumnProperties());
- SegmentMetadataUtils.savePropertiesConfiguration(metadataProperties, indexDir);
+ File existing =
+ SegmentDirectoryPaths.findTextIndexIndexFile(indexDir, MultiColumnTextIndexConstants.INDEX_DIR_NAME);
+ if (existing != null) {
+ FileUtils.forceDelete(existing);
+ }
+ File mapping = new File(segmentDirectory, MultiColumnTextIndexConstants.DOCID_MAPPING_FILE_NAME);
+ if (mapping.exists()) {
+ FileUtils.forceDelete(mapping);
}
}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/TextIndexHandler.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/TextIndexHandler.java
index 4273dfbe424..dad2a02059a 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/TextIndexHandler.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/TextIndexHandler.java
@@ -20,6 +20,7 @@ package org.apache.pinot.segment.local.segment.index.loader.invertedindex;
import java.io.File;
import java.io.IOException;
+import java.nio.ByteOrder;
import java.util.Collection;
import java.util.HashSet;
import java.util.Map;
@@ -29,6 +30,7 @@ import org.apache.pinot.segment.local.segment.index.dictionary.DictionaryIndexTy
import org.apache.pinot.segment.local.segment.index.loader.BaseIndexHandler;
import org.apache.pinot.segment.local.segment.index.loader.LoaderUtils;
import org.apache.pinot.segment.local.segment.store.TextIndexUtils;
+import org.apache.pinot.segment.local.utils.LuceneIndexCompatibilityUtils;
import org.apache.pinot.segment.spi.ColumnMetadata;
import org.apache.pinot.segment.spi.V1Constants;
import org.apache.pinot.segment.spi.creator.IndexCreationContext;
@@ -42,6 +44,7 @@ import org.apache.pinot.segment.spi.index.creator.TextIndexCreator;
import org.apache.pinot.segment.spi.index.reader.Dictionary;
import org.apache.pinot.segment.spi.index.reader.ForwardIndexReader;
import org.apache.pinot.segment.spi.index.reader.ForwardIndexReaderContext;
+import org.apache.pinot.segment.spi.memory.PinotDataBuffer;
import org.apache.pinot.segment.spi.store.SegmentDirectory;
import org.apache.pinot.segment.spi.store.SegmentDirectoryPaths;
import org.apache.pinot.spi.config.table.MultiColumnTextIndexConfig;
@@ -91,13 +94,15 @@ public class TextIndexHandler extends BaseIndexHandler {
legacyNativeColumns);
return true;
}
- Set<String> existingColumns = segmentReader.toSegmentDirectory().getColumnsWithIndex(StandardIndexes.text());
+ Set<String> existingColumns = getExistingColumns(segmentReader);
// Check if any existing index need to be removed.
// Check if existing indexes need configuration updates based on storeInSegmentFile
for (String column : existingColumns) {
ColumnMetadata columnMetadata = _segmentDirectory.getSegmentMetadata().getColumnMetadataFor(column);
try {
- if (columnMetadata != null && hasTextIndexConfigurationChanged(column, segmentReader)) {
+ if (columnsToAddIdx.contains(column) && columnMetadata != null
+ && (isBuildInProgress(column) || hasTextIndexConfigurationChanged(column, segmentReader)
+ || isIncompatible(column, segmentReader))) {
LOGGER.info("Need to update text index configuration for segment: {}, column: {}", segmentName, column);
return true;
}
@@ -131,28 +136,31 @@ public class TextIndexHandler extends BaseIndexHandler {
Set<String> columnsToAddIdx = new HashSet<>(_columnsToAddIdx);
for (String column : getColumnsWithLegacyNativeTextIndex()) {
LOGGER.info("Removing legacy native text index file from segment: {}, column: {}", segmentName, column);
- segmentWriter.removeIndex(column, StandardIndexes.text());
+ removeTextIndex(segmentWriter, column);
}
- Set<String> existingColumns = segmentWriter.toSegmentDirectory().getColumnsWithIndex(StandardIndexes.text());
- // Handle configuration changes for existing indexes
+ Set<String> existingColumns = getExistingColumns(segmentWriter);
for (String column : existingColumns) {
- if (!columnsToAddIdx.contains(column)) {
+ if (!columnsToAddIdx.remove(column)) {
+ LOGGER.info("Removing existing text index from segment: {}, column: {}", segmentName, column);
+ removeTextIndex(segmentWriter, column);
+ FileUtils.deleteQuietly(getBuildMarker(column));
continue;
}
ColumnMetadata columnMetadata = _segmentDirectory.getSegmentMetadata().getColumnMetadataFor(column);
- if (columnMetadata != null && hasTextIndexConfigurationChanged(column, segmentWriter)) {
- LOGGER.info("Updating text index configuration for segment: {}, column: {}", segmentName, column);
- // Remove existing index and recreate with new configuration
- segmentWriter.removeIndex(column, StandardIndexes.text());
- createTextIndexForColumn(segmentWriter, columnMetadata);
- columnsToAddIdx.remove(column);
- }
- }
- for (String column : existingColumns) {
- if (!columnsToAddIdx.remove(column)) {
- LOGGER.info("Removing existing text index from segment: {}, column: {}", segmentName, column);
- segmentWriter.removeIndex(column, StandardIndexes.text());
- LOGGER.info("Removed existing text index from segment: {}, column: {}", segmentName, column);
+ if (columnMetadata != null) {
+ // An interrupted build can contain a readable partial commit, or no commit at all. Its marker takes
+ // precedence over compatibility probes and requires a full restart from the original values.
+ boolean recovery = isBuildInProgress(column) || isIncompatible(column, segmentWriter);
+ if (recovery) {
+ validateRecoverySource(segmentWriter, column);
+ }
+ if (recovery || hasTextIndexConfigurationChanged(column, segmentWriter)) {
+ LOGGER.info("Rebuilding text index for segment: {}, column: {}, recovering Lucene index: {}", segmentName,
+ column, recovery);
+ FileUtils.touch(getBuildMarker(column));
+ removeTextIndex(segmentWriter, column);
+ createTextIndexForColumn(segmentWriter, columnMetadata);
+ }
}
}
for (String column : columnsToAddIdx) {
@@ -163,6 +171,108 @@ public class TextIndexHandler extends BaseIndexHandler {
}
}
+ /// Verify original source values before any preprocessing handler can remove or regenerate them.
+ public void validateRecoverySources(SegmentDirectory.Reader segmentReader)
+ throws IOException {
+ for (String column : getExistingColumns(segmentReader)) {
+ if (_columnsToAddIdx.contains(column)
+ && (isBuildInProgress(column) || isIncompatible(column, segmentReader))) {
+ validateRecoverySource(segmentReader, column);
+ }
+ }
+ }
+
+ private Set<String> getExistingColumns(SegmentDirectory.Reader segmentReader) {
+ Set<String> columns =
+ new HashSet<>(segmentReader.toSegmentDirectory().getColumnsWithIndex(StandardIndexes.text()));
+ File indexDir = _segmentDirectory.getSegmentMetadata().getIndexDir();
+ if (indexDir != null) {
+ File segmentDirectory =
+ SegmentDirectoryPaths.segmentDirectoryFor(indexDir, _segmentDirectory.getSegmentMetadata().getVersion());
+ for (String column : _segmentDirectory.getSegmentMetadata().getAllColumns()) {
+ File combined = new File(segmentDirectory,
+ column + V1Constants.Indexes.LUCENE_COMBINE_TEXT_INDEX_FILE_EXTENSION);
+ if (combined.exists() || isBuildInProgress(column)) {
+ columns.add(column);
+ }
+ }
+ }
+ return columns;
+ }
+
+ private void validateRecoverySource(SegmentDirectory.Reader segmentReader, String column) {
+ if (!segmentReader.hasIndexFor(column, StandardIndexes.forward())) {
+ throw new IllegalStateException("Cannot rebuild Lucene text index for column: " + column
+ + ". Original forward index must be retained; restore a compatible segment or refresh it from source data");
+ }
+ }
+
+ private File getBuildMarker(String column) {
+ File indexDir = _segmentDirectory.getSegmentMetadata().getIndexDir();
+ return indexDir == null ? null : new File(indexDir, column + ".text.inprogress");
+ }
+
+ private boolean isBuildInProgress(String column) {
+ File marker = getBuildMarker(column);
+ return marker != null && marker.exists();
+ }
+
+ private boolean isIncompatible(String column, SegmentDirectory.Reader segmentReader)
+ throws IOException {
+ boolean hasLocalRepresentation = false;
+ File indexDir = _segmentDirectory.getSegmentMetadata().getIndexDir();
+ if (indexDir != null) {
+ File luceneDirectory = SegmentDirectoryPaths.findTextIndexIndexFile(indexDir, column);
+ if (luceneDirectory != null) {
+ hasLocalRepresentation = true;
+ if (LuceneIndexCompatibilityUtils.isIncompatible(luceneDirectory)) {
+ return true;
+ }
+ }
+ File segmentDirectory =
+ SegmentDirectoryPaths.segmentDirectoryFor(indexDir, _segmentDirectory.getSegmentMetadata().getVersion());
+ File combined = new File(segmentDirectory, column + V1Constants.Indexes.LUCENE_COMBINE_TEXT_INDEX_FILE_EXTENSION);
+ if (combined.exists()) {
+ hasLocalRepresentation = true;
+ try (PinotDataBuffer buffer = PinotDataBuffer.mapFile(combined, true, 0, combined.length(),
+ ByteOrder.BIG_ENDIAN, "Lucene text compatibility check")) {
+ if (LuceneIndexCompatibilityUtils.isIncompatible(buffer, column)) {
+ return true;
+ }
+ }
+ }
+ }
+ if (hasLocalRepresentation && _segmentDirectory.getSegmentMetadata().getVersion() != SegmentVersion.v3) {
+ return false;
+ }
+ // A v3 packed entry can coexist with a directory or sidecar after an interrupted conversion. Check it too:
+ // the reader selected by the table config may use the packed entry rather than the compatible local directory.
+ PinotDataBuffer packed = _segmentDirectory.getSegmentMetadata().getVersion() == SegmentVersion.v3
+ ? TextIndexUtils.getConsolidatedTextEntry(segmentReader, column)
+ : segmentReader.getIndexFor(column, StandardIndexes.text());
+ if (packed == null) {
+ if (hasLocalRepresentation) {
+ return false;
+ }
+ throw new IOException("Existing text index has no readable representation for column: " + column);
+ }
+ return LuceneIndexCompatibilityUtils.isIncompatible(packed, column);
+ }
+
+ private void removeTextIndex(SegmentDirectory.Writer segmentWriter, String column)
+ throws IOException {
+ segmentWriter.removeIndex(column, StandardIndexes.text());
+ File indexDir = _segmentDirectory.getSegmentMetadata().getIndexDir();
+ if (indexDir != null) {
+ File segmentDirectory =
+ SegmentDirectoryPaths.segmentDirectoryFor(indexDir, _segmentDirectory.getSegmentMetadata().getVersion());
+ File combined = new File(segmentDirectory, column + V1Constants.Indexes.LUCENE_COMBINE_TEXT_INDEX_FILE_EXTENSION);
+ if (combined.exists()) {
+ FileUtils.forceDelete(combined);
+ }
+ }
+ }
+
private boolean shouldCreateTextIndex(ColumnMetadata columnMetadata) {
if (columnMetadata != null) {
// Fail fast upon unsupported operations.
@@ -208,17 +318,9 @@ public class TextIndexHandler extends BaseIndexHandler {
LOGGER.info("Creating new text index for column: {} in segment: {}, hasDictionary: {}, storeInSegmentFile: {}",
columnName, segmentName, hasDictionary, storeInSegmentFile);
- // Create in-progress marker file for recovery
- File inProgress = new File(indexDir, columnName + ".text.inprogress");
- if (!inProgress.exists()) {
- // Marker file does not exist, which means last run ended normally.
- // Create a marker file.
- FileUtils.touch(inProgress);
- } else {
- // Marker file exists, which means last run was interrupted.
- // Clean up any existing text index files
- cleanupExistingTextIndexFiles(indexDir, columnName);
- }
+ // Preserve the marker until the rebuilt index, including its packed representation, is complete.
+ File inProgress = getBuildMarker(columnName);
+ FileUtils.touch(inProgress);
// The handlers are always invoked by the preprocessor. Before this ImmutableSegmentLoader would have already
// up-converted the segment from v1/v2 -> v3 (if needed). So based on the segmentVersion, whatever segment
@@ -251,7 +353,7 @@ public class TextIndexHandler extends BaseIndexHandler {
}
// Delete the marker file.
- FileUtils.deleteQuietly(inProgress);
+ FileUtils.forceDelete(inProgress);
LOGGER.info("Created text index for column: {} in segment: {}", columnName, segmentName);
}
@@ -336,25 +438,6 @@ public class TextIndexHandler extends BaseIndexHandler {
return expectedFormat != currentFormat;
}
- /// Clean up existing text index files for recovery purposes.
- ///
- /// @param indexDir the index directory
- /// @param columnName the column name
- private void cleanupExistingTextIndexFiles(File indexDir, String columnName) {
- // Remove any existing text index files for this column
- File[] textIndexFiles = indexDir.listFiles((dir, name) -> {
- // Look for text index files for this column
- return name.startsWith(columnName) && (name.endsWith(V1Constants.Indexes.LUCENE_V912_TEXT_INDEX_FILE_EXTENSION)
- || name.endsWith(".text.inprogress"));
- });
-
- if (textIndexFiles != null) {
- for (File file : textIndexFiles) {
- FileUtils.deleteQuietly(file);
- }
- }
- }
-
/// Convert text index to V3 combined format by merging into columns.psf.
///
/// @param segmentWriter the segment writer
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/VectorIndexHandler.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/VectorIndexHandler.java
index 0e8aa0a2f58..7ba3054556e 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/VectorIndexHandler.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/VectorIndexHandler.java
@@ -23,6 +23,7 @@ import com.google.common.base.Preconditions;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
+import java.nio.ByteOrder;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
@@ -32,6 +33,7 @@ import org.apache.pinot.segment.local.segment.creator.impl.vector.lucene99.HnswV
import org.apache.pinot.segment.local.segment.index.loader.BaseIndexHandler;
import org.apache.pinot.segment.local.segment.index.loader.LoaderUtils;
import org.apache.pinot.segment.local.segment.store.VectorIndexUtils;
+import org.apache.pinot.segment.local.utils.LuceneIndexCompatibilityUtils;
import org.apache.pinot.segment.spi.ColumnMetadata;
import org.apache.pinot.segment.spi.creator.IndexCreationContext;
import org.apache.pinot.segment.spi.creator.SegmentVersion;
@@ -81,6 +83,11 @@ public class VectorIndexHandler extends BaseIndexHandler {
}
VectorIndexConfig desiredConfig = _vectorConfigs.get(column);
VectorBackendType desiredBackend = desiredConfig.resolveBackendType();
+ if (desiredBackend == VectorBackendType.HNSW && needsHnswRecovery(segmentReader, column)) {
+ LOGGER.info("Need to rebuild incompatible or interrupted HNSW index for segment: {}, column: {}", segmentName,
+ column);
+ return true;
+ }
VectorBackendType existingBackend = VectorIndexUtils.detectVectorIndexBackend(indexDir, column);
// Both detectVectorIndexBackend and the consolidated sniff report the STORAGE format
// (IVF_ON_DISK payloads/files register as IVF_FLAT), so compare against the desired
@@ -136,6 +143,82 @@ public class VectorIndexHandler extends BaseIndexHandler {
return false;
}
+ /// Checks recovery sources before other handlers can reconstruct a missing forward index. A multi-value forward
+ /// index regenerated from dictionary/inverted indexes does not retain vector coordinate order or duplicates.
+ public void validateRecoverySources(SegmentDirectory.Reader segmentReader)
+ throws IOException {
+ for (Map.Entry<String, VectorIndexConfig> entry : _vectorConfigs.entrySet()) {
+ String column = entry.getKey();
+ if (entry.getValue().resolveBackendType() == VectorBackendType.HNSW
+ && needsHnswRecovery(segmentReader, column)) {
+ requireVectorRecoverySource(segmentReader, column);
+ }
+ }
+ }
+
+ private boolean needsHnswRecovery(SegmentDirectory.Reader segmentReader, String column)
+ throws IOException {
+ // An interrupted creator can leave a readable but incomplete Lucene commit, or no commit at all.
+ return hasHnswRecoveryMarker(column) || isIncompatibleHnswIndex(segmentReader, column);
+ }
+
+ private boolean hasHnswRecoveryMarker(String column) {
+ File segmentDirectory = SegmentDirectoryPaths.segmentDirectoryFor(
+ _segmentDirectory.getSegmentMetadata().getIndexDir(), _segmentDirectory.getSegmentMetadata().getVersion());
+ return new File(segmentDirectory,
+ column + VectorIndexUtils.getIndexFileExtension(VectorBackendType.HNSW, false) + ".inprogress").exists()
+ || new File(segmentDirectory,
+ column + VectorIndexUtils.getIndexFileExtension(VectorBackendType.HNSW, true) + ".inprogress").exists();
+ }
+
+ private static void requireVectorRecoverySource(SegmentDirectory.Reader segmentReader, String column)
+ throws IOException {
+ if (!segmentReader.hasIndexFor(column, StandardIndexes.forward())) {
+ throw new IOException("Cannot rebuild incompatible or interrupted HNSW index for column: " + column
+ + " without the original forward index. Vector coordinate order cannot be recovered from dictionary and "
+ + "inverted indexes; refresh the segment from source data.");
+ }
+ }
+
+ private boolean isIncompatibleHnswIndex(SegmentDirectory.Reader segmentReader, String column)
+ throws IOException {
+ File indexDir = _segmentDirectory.getSegmentMetadata().getIndexDir();
+ VectorBackendType existingBackend = VectorIndexUtils.detectVectorIndexBackend(indexDir, column);
+ if (existingBackend != null && existingBackend != VectorBackendType.HNSW) {
+ return false;
+ }
+ File indexFile = SegmentDirectoryPaths.findVectorIndexIndexFile(indexDir, column, VectorBackendType.HNSW);
+ if (indexFile != null) {
+ if (indexFile.isDirectory()) {
+ if (LuceneIndexCompatibilityUtils.isIncompatible(indexFile)) {
+ return true;
+ }
+ // Absorption prefers a combined sibling over the legacy directory when both survived a previous load.
+ File combinedFile = new File(indexFile.getParentFile(),
+ column + VectorIndexUtils.getIndexFileExtension(VectorBackendType.HNSW, true));
+ if (combinedFile.exists() && isIncompatibleCombinedHnsw(combinedFile, column)) {
+ return true;
+ }
+ } else if (isIncompatibleCombinedHnsw(indexFile, column)) {
+ return true;
+ }
+ }
+ if (_segmentDirectory.getSegmentMetadata().getVersion() == SegmentVersion.v3
+ && VectorIndexUtils.detectConsolidatedVectorBackend(segmentReader, column) == VectorBackendType.HNSW) {
+ PinotDataBuffer buffer = VectorIndexUtils.getConsolidatedVectorEntry(segmentReader, column);
+ return buffer != null && LuceneIndexCompatibilityUtils.isIncompatible(buffer, column);
+ }
+ return false;
+ }
+
+ private static boolean isIncompatibleCombinedHnsw(File indexFile, String column)
+ throws IOException {
+ try (PinotDataBuffer buffer = PinotDataBuffer.mapFile(indexFile, true, 0, indexFile.length(),
+ ByteOrder.LITTLE_ENDIAN, "HNSW compatibility: " + column)) {
+ return LuceneIndexCompatibilityUtils.isIncompatible(buffer, column);
+ }
+ }
+
/// Returns the consolidated payload's backend when it drifts from the desired backend's storage
/// format, or `null` when the column has an on-disk sidecar (the sidecar drift check owns
/// that case), the segment is not V3 (consolidated entries only exist in `columns.psf`),
@@ -494,6 +577,15 @@ public class VectorIndexHandler extends BaseIndexHandler {
}
VectorIndexConfig desiredConfig = _vectorConfigs.get(column);
VectorBackendType desiredBackend = desiredConfig.resolveBackendType();
+ // Check before the storage migrations below: neither absorption nor extraction makes Lucene formats readable.
+ if (desiredBackend == VectorBackendType.HNSW && needsHnswRecovery(segmentWriter, column)) {
+ requireVectorRecoverySource(segmentWriter, column);
+ LOGGER.info("Rebuilding incompatible or interrupted HNSW index for segment: {}, column: {}", segmentName,
+ column);
+ segmentWriter.removeIndex(column, StandardIndexes.vector());
+ columnsToAddIdx.add(column);
+ continue;
+ }
VectorBackendType existingBackend = VectorIndexUtils.detectVectorIndexBackend(indexDir, column);
// Storage-format comparison — see the twin check in needUpdateIndices.
if (existingBackend != null && existingBackend != VectorIndexUtils.storageFormatOf(desiredBackend)) {
@@ -591,6 +683,9 @@ public class VectorIndexHandler extends BaseIndexHandler {
String columnName = columnMetadata.getColumnName();
VectorIndexConfig config = _fieldIndexConfigs.get(columnName).getConfig(StandardIndexes.vector());
VectorBackendType backendType = config.resolveBackendType();
+ if (backendType == VectorBackendType.HNSW && hasHnswRecoveryMarker(columnName)) {
+ requireVectorRecoverySource(segmentWriter, columnName);
+ }
// Resolve the on-disk extension (or directory name) that the creator writes to. When
// storeInSegmentFile=true the creator produces the combined-form directly; when false it writes
// the legacy form (a Lucene directory for HNSW, a flat file for IVF). The .inprogress marker
@@ -637,7 +732,7 @@ public class VectorIndexHandler extends BaseIndexHandler {
}
// Delete the marker file.
- FileUtils.deleteQuietly(inProgress);
+ FileUtils.forceDelete(inProgress);
LOGGER.info("Created Vector index for segment: {}, column: {}", segmentName, columnName);
}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/store/TextIndexUtils.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/store/TextIndexUtils.java
index 2ecf5110316..3f97bc15618 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/store/TextIndexUtils.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/store/TextIndexUtils.java
@@ -19,6 +19,7 @@
package org.apache.pinot.segment.local.segment.store;
import java.io.File;
+import java.io.IOException;
import java.lang.reflect.Constructor;
import java.util.ArrayList;
import java.util.Arrays;
@@ -43,7 +44,10 @@ import org.apache.pinot.segment.local.segment.index.text.CaseAwareStandardAnalyz
import org.apache.pinot.segment.local.segment.index.text.TextIndexConfigBuilder;
import org.apache.pinot.segment.spi.V1Constants;
import org.apache.pinot.segment.spi.V1Constants.Indexes;
+import org.apache.pinot.segment.spi.index.StandardIndexes;
import org.apache.pinot.segment.spi.index.TextIndexConfig;
+import org.apache.pinot.segment.spi.memory.PinotDataBuffer;
+import org.apache.pinot.segment.spi.store.SegmentDirectory;
import org.apache.pinot.spi.config.provider.PinotClusterConfigChangeListener;
import org.apache.pinot.spi.config.table.FieldConfig;
import org.apache.pinot.spi.env.CommonsConfigurationUtils;
@@ -114,6 +118,22 @@ public class TextIndexUtils {
//@formatter:on
}
+ /// Returns a v3 packed text entry, or null only when the store reports that the entry is absent.
+ /// Other store failures propagate. The returned buffer remains owned by the segment directory.
+ @Nullable
+ public static PinotDataBuffer getConsolidatedTextEntry(SegmentDirectory.Reader reader, String column)
+ throws IOException {
+ try {
+ return reader.getIndexFor(column, StandardIndexes.text());
+ } catch (RuntimeException e) {
+ String message = e.getMessage();
+ if (message != null && message.startsWith(SingleFileIndexDirectory.INDEX_NOT_FOUND_MESSAGE_PREFIX)) {
+ return null;
+ }
+ throw e;
+ }
+ }
+
/// Bulk form of [#hasTextIndex]: which of `columns` have a Lucene text index in `segDir`.
///
/// Prefer this over calling [#hasTextIndex] in a loop over a segment's columns. It lists the
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/LuceneIndexCompatibilityUtils.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/LuceneIndexCompatibilityUtils.java
new file mode 100644
index 00000000000..928f43b6489
--- /dev/null
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/LuceneIndexCompatibilityUtils.java
@@ -0,0 +1,117 @@
+/**
+ * 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.utils;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import org.apache.lucene.codecs.CodecUtil;
+import org.apache.lucene.index.CorruptIndexException;
+import org.apache.lucene.index.DirectoryReader;
+import org.apache.lucene.index.IndexFormatTooNewException;
+import org.apache.lucene.index.IndexFormatTooOldException;
+import org.apache.lucene.index.IndexNotFoundException;
+import org.apache.lucene.index.SegmentInfos;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.store.FSDirectory;
+import org.apache.lucene.store.IOContext;
+import org.apache.lucene.store.IndexInput;
+import org.apache.lucene.util.StringHelper;
+import org.apache.lucene.util.Version;
+import org.apache.pinot.segment.local.segment.index.readers.text.LuceneTextIndexBufferReader;
+import org.apache.pinot.segment.spi.memory.PinotDataBuffer;
+
+
+/// Detects indexes written in a newer, unreadable Lucene format before segment preprocessing replaces them.
+/// Current/older writers only require a commit-header read, avoiding a second full index open on normal loads.
+/// Corruption and ordinary I/O errors propagate instead of being treated as format incompatibility.
+/// Stateless and thread-safe; callers retain ownership of supplied directories and buffers.
+public final class LuceneIndexCompatibilityUtils {
+ private LuceneIndexCompatibilityUtils() {
+ }
+
+ public static boolean isIncompatible(File indexDirectory)
+ throws IOException {
+ if (!indexDirectory.isDirectory()) {
+ throw new FileNotFoundException("Lucene index directory does not exist: " + indexDirectory);
+ }
+ try (Directory directory = FSDirectory.open(indexDirectory.toPath())) {
+ return isIncompatible(directory);
+ }
+ }
+
+ public static boolean isIncompatible(PinotDataBuffer buffer, String column)
+ throws IOException {
+ try (Directory directory = LuceneTextIndexBufferReader.createLuceneDirectory(buffer, column)) {
+ return isIncompatible(directory);
+ }
+ }
+
+ public static boolean isIncompatible(Directory directory)
+ throws IOException {
+ String commit = SegmentInfos.getLastCommitSegmentsFileName(directory);
+ if (commit == null) {
+ throw new IndexNotFoundException("No Lucene commit found in " + directory);
+ }
+ Version writerVersion;
+ try (IndexInput input = directory.openInput(commit, IOContext.READONCE)) {
+ // Both Lucene 9.12 and 10.0 use this commit layout. Inspect its creation version before SegmentInfos,
+ // whose constructor rejects a future major with IllegalArgumentException rather than a format exception.
+ try {
+ CodecUtil.checkHeader(input, "segments", SegmentInfos.VERSION_74, SegmentInfos.VERSION_86);
+ } catch (IndexFormatTooNewException e) {
+ CodecUtil.checksumEntireFile(input);
+ return true;
+ } catch (IndexFormatTooOldException e) {
+ // Leave legacy-reader support (including the Lucene 8 expert-reader path) to the actual index reader.
+ return false;
+ }
+ input.skipBytes(StringHelper.ID_LENGTH);
+ CodecUtil.checkIndexHeaderSuffix(input,
+ Long.toString(SegmentInfos.generationFromSegmentsFileName(commit), Character.MAX_RADIX));
+ writerVersion = Version.fromBits(input.readVInt(), input.readVInt(), input.readVInt());
+ int createdMajor = input.readVInt();
+ if (createdMajor < 6 || createdMajor > writerVersion.major) {
+ throw new CorruptIndexException("Invalid Lucene index creation version: " + createdMajor, input);
+ }
+ if (Version.LATEST.onOrAfter(writerVersion)) {
+ return false;
+ }
+ CodecUtil.checksumEntireFile(input);
+ if (createdMajor > Version.LATEST.major) {
+ return true;
+ }
+ }
+ // A newer writer can append to an older index without changing its creation version. Only replace it if
+ // this runtime actually rejects the result; a metadata-only commit may leave every segment readable.
+ try (DirectoryReader ignored = DirectoryReader.open(directory)) {
+ return false;
+ } catch (IndexFormatTooNewException e) {
+ return true;
+ } catch (IllegalArgumentException e) {
+ // Lucene reports an unavailable built-in codec through this exception, not IndexFormatTooNewException.
+ // Restrict recovery to the future writer's codec family; missing older/custom codec plugins still fail.
+ String message = e.getMessage();
+ if (message != null && message.matches("Could not load codec 'Lucene" + writerVersion.major + "[0-9]*'\\..*")) {
+ return true;
+ }
+ throw e;
+ }
+ }
+}
diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/HnswIndexRollbackTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/HnswIndexRollbackTest.java
new file mode 100644
index 00000000000..0bd4c58eea8
--- /dev/null
+++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/HnswIndexRollbackTest.java
@@ -0,0 +1,304 @@
+/**
+ * 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.loader.invertedindex;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.segment.local.PinotBuffersAfterClassCheckRule;
+import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
+import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
+import org.apache.pinot.segment.local.segment.creator.impl.vector.HnswVectorIndexCreator;
+import org.apache.pinot.segment.local.segment.creator.impl.vector.lucene99.HnswVectorIndexCombined;
+import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
+import org.apache.pinot.segment.local.segment.index.loader.LoaderUtils;
+import org.apache.pinot.segment.local.segment.index.loader.SegmentPreProcessor;
+import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
+import org.apache.pinot.segment.local.segment.store.SegmentLocalFSDirectory;
+import org.apache.pinot.segment.local.utils.LuceneIndexCompatibilityUtils;
+import org.apache.pinot.segment.local.utils.LuceneIndexTestUtils;
+import org.apache.pinot.segment.spi.ImmutableSegment;
+import org.apache.pinot.segment.spi.V1Constants;
+import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
+import org.apache.pinot.segment.spi.creator.SegmentVersion;
+import org.apache.pinot.segment.spi.index.StandardIndexes;
+import org.apache.pinot.segment.spi.index.creator.VectorBackendType;
+import org.apache.pinot.segment.spi.index.creator.VectorIndexConfig;
+import org.apache.pinot.segment.spi.index.reader.VectorIndexReader;
+import org.apache.pinot.segment.spi.store.SegmentDirectory;
+import org.apache.pinot.segment.spi.store.SegmentDirectoryPaths;
+import org.apache.pinot.spi.config.table.FieldConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.apache.pinot.spi.utils.ReadMode;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
+
+
+/// Exercises server-load rollback from real Lucene 10 HNSW files to the running Lucene writer.
+public class HnswIndexRollbackTest implements PinotBuffersAfterClassCheckRule {
+ private static final String COLUMN = "embedding";
+ private static final int DIMENSIONS = 2048;
+ private static final Schema SCHEMA =
+ new Schema.SchemaBuilder().addMultiValueDimension(COLUMN, DataType.FLOAT).build();
+
+ @DataProvider(name = "layouts")
+ public Object[][] layouts() {
+ return new Object[][]{
+ {SegmentVersion.v1, "directory", false},
+ {SegmentVersion.v3, "directory", false},
+ {SegmentVersion.v3, "combined", true},
+ {SegmentVersion.v3, "consolidated", true},
+ {SegmentVersion.v3, "consolidated", false}
+ };
+ }
+
+ @Test(dataProvider = "layouts")
+ public void testLoadingRebuildsNewerHnswIndex(SegmentVersion version, String layout, boolean storeInSegmentFile)
+ throws Exception {
+ File tempDir = Files.createTempDirectory("hnsw-rollback").toFile();
+ try {
+ TableConfig tableConfig = tableConfig(storeInSegmentFile, false);
+ File segmentDir = buildSegment(tempDir, tableConfig, version);
+ installNewerIndex(tempDir, segmentDir, version, layout);
+ IndexLoadingConfig loadingConfig = new IndexLoadingConfig(tableConfig, SCHEMA);
+
+ try (SegmentDirectory directory = new SegmentLocalFSDirectory(segmentDir, ReadMode.mmap);
+ SegmentPreProcessor processor = new SegmentPreProcessor(directory, loadingConfig)) {
+ assertTrue(processor.needProcess(), "the newer HNSW format must trigger preprocessing");
+ }
+ assertLoadsAndQueries(segmentDir, loadingConfig);
+ try (SegmentDirectory directory = new SegmentLocalFSDirectory(segmentDir, ReadMode.mmap);
+ SegmentPreProcessor processor = new SegmentPreProcessor(directory, loadingConfig);
+ SegmentDirectory.Reader reader = directory.createReader()) {
+ assertFalse(processor.needProcess(), "a rebuilt index must be retained on the next load");
+ if (storeInSegmentFile) {
+ assertFalse(LuceneIndexCompatibilityUtils.isIncompatible(
+ reader.getIndexFor(COLUMN, StandardIndexes.vector()), COLUMN));
+ } else {
+ File hnswDir = SegmentDirectoryPaths.findVectorIndexIndexFile(segmentDir, COLUMN, VectorBackendType.HNSW);
+ assertNotNull(hnswDir);
+ assertFalse(LuceneIndexCompatibilityUtils.isIncompatible(hnswDir));
+ }
+ }
+ assertLoadsAndQueries(segmentDir, loadingConfig);
+ } finally {
+ FileUtils.deleteDirectory(tempDir);
+ }
+ }
+
+ @DataProvider(name = "interruptedLayouts")
+ public Object[][] interruptedLayouts() {
+ return new Object[][]{{"directory"}, {"combined"}, {"consolidated"}};
+ }
+
+ @Test(dataProvider = "interruptedLayouts")
+ public void testLoadingRestartsReadablePartialHnswIndex(String layout)
+ throws Exception {
+ File tempDir = Files.createTempDirectory("hnsw-rollback-interrupted").toFile();
+ try {
+ boolean storeInSegmentFile = !layout.equals("directory");
+ TableConfig tableConfig = tableConfig(storeInSegmentFile, false);
+ File segmentDir = buildSegment(tempDir, tableConfig, SegmentVersion.v3);
+ File storageDir = SegmentDirectoryPaths.segmentDirectoryFor(segmentDir, SegmentVersion.v3);
+ IndexLoadingConfig loadingConfig = new IndexLoadingConfig(tableConfig, SCHEMA);
+ try (SegmentDirectory directory = new SegmentLocalFSDirectory(segmentDir, ReadMode.mmap);
+ SegmentDirectory.Writer writer = directory.createWriter()) {
+ writer.removeIndex(COLUMN, StandardIndexes.vector());
+ writer.save();
+ }
+ VectorIndexConfig vectorConfig =
+ loadingConfig.getFieldIndexConfigByColName().get(COLUMN).getConfig(StandardIndexes.vector());
+ // A crash can leave a valid commit containing only the first input document. Compatibility alone cannot
+ // distinguish this from a completed rebuild; the creator's marker must force a restart from all source rows.
+ try (HnswVectorIndexCreator creator = new HnswVectorIndexCreator(COLUMN, storageDir, vectorConfig)) {
+ float[] firstVector = new float[DIMENSIONS];
+ firstVector[0] = 1;
+ creator.add(firstVector);
+ creator.seal();
+ }
+ File combined = new File(storageDir, COLUMN + V1Constants.Indexes.VECTOR_HNSW_COMBINED_INDEX_FILE_EXTENSION);
+ if (layout.equals("consolidated")) {
+ try (SegmentDirectory directory = new SegmentLocalFSDirectory(segmentDir, ReadMode.mmap);
+ SegmentDirectory.Writer writer = directory.createWriter()) {
+ LoaderUtils.writeIndexToV3Format(writer, COLUMN, combined, StandardIndexes.vector());
+ writer.save();
+ }
+ }
+ String extension = storeInSegmentFile
+ ? V1Constants.Indexes.VECTOR_HNSW_COMBINED_INDEX_FILE_EXTENSION
+ : V1Constants.Indexes.VECTOR_V912_HNSW_INDEX_FILE_EXTENSION;
+ File marker = new File(storageDir, COLUMN + extension + ".inprogress");
+ FileUtils.touch(marker);
+ try (SegmentDirectory directory = new SegmentLocalFSDirectory(segmentDir, ReadMode.mmap);
+ SegmentPreProcessor processor = new SegmentPreProcessor(directory, loadingConfig)) {
+ assertTrue(processor.needProcess(), "the owned marker must force recovery of a readable partial commit");
+ }
+ assertLoadsAndQueries(segmentDir, loadingConfig);
+ assertFalse(marker.exists(), "a successful rebuild must clear its marker");
+ try (SegmentDirectory directory = new SegmentLocalFSDirectory(segmentDir, ReadMode.mmap);
+ SegmentPreProcessor processor = new SegmentPreProcessor(directory, loadingConfig)) {
+ assertFalse(processor.needProcess(), "a completed retry must be retained");
+ }
+ } finally {
+ FileUtils.deleteDirectory(tempDir);
+ }
+ }
+
+ @Test
+ public void testMissingOriginalForwardIndexPreservesNewerHnswIndex()
+ throws Exception {
+ File tempDir = Files.createTempDirectory("hnsw-rollback-no-source").toFile();
+ try {
+ // Dictionary + inverted indexes cannot recover either repeated coordinates or their original order.
+ TableConfig tableConfig = tableConfig(false, true);
+ File segmentDir = buildSegment(tempDir, tableConfig, SegmentVersion.v3);
+ installNewerIndex(tempDir, segmentDir, SegmentVersion.v3, "directory");
+ File hnswDir = SegmentDirectoryPaths.findVectorIndexIndexFile(segmentDir, COLUMN, VectorBackendType.HNSW);
+ assertNotNull(hnswDir);
+ File commit = new File(hnswDir, "segments_1");
+ byte[] originalCommit = Files.readAllBytes(commit.toPath());
+ try (SegmentDirectory directory = new SegmentLocalFSDirectory(segmentDir, ReadMode.mmap);
+ SegmentDirectory.Writer writer = directory.createWriter()) {
+ writer.removeIndex(COLUMN, StandardIndexes.forward());
+ writer.save();
+ assertTrue(writer.hasIndexFor(COLUMN, StandardIndexes.dictionary()));
+ assertTrue(writer.hasIndexFor(COLUMN, StandardIndexes.inverted()));
+ }
+
+ // The unchanged table config still enables forward, so this also tests that recovery is checked before
+ // ForwardIndexHandler can regenerate an unordered/deduplicated replacement from the inverted index.
+ IndexLoadingConfig loadingConfig = new IndexLoadingConfig(tableConfig, SCHEMA);
+ try (SegmentDirectory directory = new SegmentLocalFSDirectory(segmentDir, ReadMode.mmap);
+ SegmentPreProcessor processor = new SegmentPreProcessor(directory, loadingConfig)) {
+ IOException error = expectThrows(IOException.class, processor::process);
+ assertTrue(error.getMessage().contains("original forward index"));
+ try (SegmentDirectory.Reader reader = directory.createReader()) {
+ assertFalse(reader.hasIndexFor(COLUMN, StandardIndexes.forward()));
+ }
+ }
+ assertEquals(Files.readAllBytes(commit.toPath()), originalCommit);
+ assertTrue(LuceneIndexCompatibilityUtils.isIncompatible(hnswDir));
+
+ // A crash before the creator opens its directory can leave only the marker. This must still be checked
+ // before any handler regenerates a forward index from the unordered inverted-index values.
+ FileUtils.deleteDirectory(hnswDir);
+ File marker = new File(hnswDir.getParentFile(), hnswDir.getName() + ".inprogress");
+ FileUtils.touch(marker);
+ try (SegmentDirectory directory = new SegmentLocalFSDirectory(segmentDir, ReadMode.mmap);
+ SegmentPreProcessor processor = new SegmentPreProcessor(directory, loadingConfig)) {
+ IOException error = expectThrows(IOException.class, processor::process);
+ assertTrue(error.getMessage().contains("original forward index"));
+ assertTrue(marker.exists());
+ }
+ } finally {
+ FileUtils.deleteDirectory(tempDir);
+ }
+ }
+
+ private static TableConfig tableConfig(boolean storeInSegmentFile, boolean dictionary) {
+ VectorIndexConfig vectorConfig = new VectorIndexConfig(false, "HNSW", DIMENSIONS, 1,
+ VectorIndexConfig.VectorDistanceFunction.EUCLIDEAN,
+ Map.of(VectorIndexConfig.STORE_IN_SEGMENT_FILE, Boolean.toString(storeInSegmentFile)));
+ TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("hnswRollback").build();
+ tableConfig.setFieldConfigList(List.of(new FieldConfig.Builder(COLUMN)
+ .withEncodingType(dictionary ? FieldConfig.EncodingType.DICTIONARY : FieldConfig.EncodingType.RAW)
+ .withIndexes(JsonUtils.objectToJsonNode(Map.of("vector", vectorConfig))).build()));
+ if (dictionary) {
+ tableConfig.getIndexingConfig().setInvertedIndexColumns(List.of(COLUMN));
+ }
+ return tableConfig;
+ }
+
+ private static File buildSegment(File tempDir, TableConfig tableConfig, SegmentVersion version)
+ throws Exception {
+ SegmentGeneratorConfig config = new SegmentGeneratorConfig(tableConfig, SCHEMA);
+ config.setOutDir(tempDir.getAbsolutePath());
+ config.setSegmentName("segment");
+ config.setSegmentVersion(version);
+ List<GenericRow> rows = new ArrayList<>();
+ for (int docId = 0; docId < 3; docId++) {
+ Float[] vector = new Float[DIMENSIONS];
+ Arrays.fill(vector, 0F);
+ vector[docId] = 1F;
+ GenericRow row = new GenericRow();
+ row.putValue(COLUMN, vector);
+ rows.add(row);
+ }
+ SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl();
+ driver.init(config, new GenericRowRecordReader(rows));
+ driver.build();
+ return new File(tempDir, "segment");
+ }
+
+ private static void installNewerIndex(File tempDir, File segmentDir, SegmentVersion version, String layout)
+ throws Exception {
+ File fixtures = LuceneIndexTestUtils.extractLucene10Fixtures(new File(tempDir, "lucene10-hnsw"));
+ File fixture = new File(fixtures, "hnsw");
+ assertTrue(LuceneIndexCompatibilityUtils.isIncompatible(fixture));
+ File storageDir = SegmentDirectoryPaths.segmentDirectoryFor(segmentDir, version);
+ try (SegmentDirectory directory = new SegmentLocalFSDirectory(segmentDir, ReadMode.mmap);
+ SegmentDirectory.Writer writer = directory.createWriter()) {
+ writer.removeIndex(COLUMN, StandardIndexes.vector());
+ if (layout.equals("directory")) {
+ FileUtils.copyDirectory(fixture,
+ new File(storageDir, COLUMN + V1Constants.Indexes.VECTOR_V912_HNSW_INDEX_FILE_EXTENSION));
+ } else {
+ File combined = new File(storageDir, COLUMN + V1Constants.Indexes.VECTOR_HNSW_COMBINED_INDEX_FILE_EXTENSION);
+ HnswVectorIndexCombined.combineHnswIndexFiles(fixture, combined.getAbsolutePath(), null, null);
+ if (layout.equals("consolidated")) {
+ LoaderUtils.writeIndexToV3Format(writer, COLUMN, combined, StandardIndexes.vector());
+ }
+ }
+ writer.save();
+ }
+ }
+
+ private static void assertLoadsAndQueries(File segmentDir, IndexLoadingConfig loadingConfig)
+ throws Exception {
+ ImmutableSegment segment = ImmutableSegmentLoader.load(segmentDir, loadingConfig, true);
+ try {
+ VectorIndexReader reader = segment.getDataSource(COLUMN).getVectorIndex();
+ assertNotNull(reader);
+ for (int docId = 0; docId < 3; docId++) {
+ float[] query = new float[DIMENSIONS];
+ query[docId] = 1;
+ assertEquals(reader.getDocIds(query, 1).toArray(), new int[]{docId});
+ }
+ } finally {
+ segment.destroy();
+ }
+ }
+}
diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/TextIndexRollbackRecoveryTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/TextIndexRollbackRecoveryTest.java
new file mode 100644
index 00000000000..453278715b5
--- /dev/null
+++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/TextIndexRollbackRecoveryTest.java
@@ -0,0 +1,460 @@
+/**
+ * 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.loader.invertedindex;
+
+import java.io.File;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import org.apache.commons.configuration2.PropertiesConfiguration;
+import org.apache.commons.io.FileUtils;
+import org.apache.lucene.analysis.standard.StandardAnalyzer;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.StoredField;
+import org.apache.lucene.document.TextField;
+import org.apache.lucene.index.DirectoryReader;
+import org.apache.lucene.index.IndexWriter;
+import org.apache.lucene.index.IndexWriterConfig;
+import org.apache.lucene.store.FSDirectory;
+import org.apache.pinot.segment.local.PinotBuffersAfterClassCheckRule;
+import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
+import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
+import org.apache.pinot.segment.local.segment.creator.impl.text.LuceneTextIndexCombined;
+import org.apache.pinot.segment.local.segment.creator.impl.text.LuceneTextIndexCreator;
+import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
+import org.apache.pinot.segment.local.segment.index.loader.LoaderUtils;
+import org.apache.pinot.segment.local.segment.index.readers.text.MultiColumnLuceneTextIndexReader;
+import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
+import org.apache.pinot.segment.local.segment.store.SegmentLocalFSDirectory;
+import org.apache.pinot.segment.local.utils.LuceneIndexCompatibilityUtils;
+import org.apache.pinot.segment.local.utils.LuceneIndexTestUtils;
+import org.apache.pinot.segment.spi.ImmutableSegment;
+import org.apache.pinot.segment.spi.V1Constants;
+import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
+import org.apache.pinot.segment.spi.creator.SegmentVersion;
+import org.apache.pinot.segment.spi.index.StandardIndexes;
+import org.apache.pinot.segment.spi.index.multicolumntext.MultiColumnTextIndexConstants;
+import org.apache.pinot.segment.spi.store.SegmentDirectory;
+import org.apache.pinot.segment.spi.store.SegmentDirectoryPaths;
+import org.apache.pinot.segment.spi.utils.SegmentMetadataUtils;
+import org.apache.pinot.spi.config.table.FieldConfig;
+import org.apache.pinot.spi.config.table.MultiColumnTextIndexConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.utils.ReadMode;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
+
+
+/// Exercises server loading with real Lucene 10 indexes and retained Pinot source values on Lucene 9.12.
+public class TextIndexRollbackRecoveryTest implements PinotBuffersAfterClassCheckRule {
+ private File _tempDir;
+ private File _fixtures;
+
+ @BeforeMethod
+ public void setUp()
+ throws Exception {
+ _tempDir = Files.createTempDirectory("text-rollback").toFile();
+ _fixtures = LuceneIndexTestUtils.extractLucene10Fixtures(new File(_tempDir, "fixtures"));
+ }
+
+ @AfterMethod
+ public void tearDown()
+ throws Exception {
+ FileUtils.deleteDirectory(_tempDir);
+ }
+
+ @DataProvider
+ public Object[][] textFormats() {
+ return new Object[][]{
+ {SegmentVersion.v1, false, false, false},
+ {SegmentVersion.v1, true, false, false},
+ {SegmentVersion.v3, false, false, true},
+ {SegmentVersion.v3, true, false, false},
+ {SegmentVersion.v3, false, true, false},
+ {SegmentVersion.v3, true, true, true}
+ };
+ }
+
+ @Test(dataProvider = "textFormats")
+ public void testSingleColumnRollback(SegmentVersion version, boolean combined, boolean multiValue, boolean raw)
+ throws Exception {
+ Schema schema = schema(multiValue);
+ TableConfig tableConfig = tableConfig(false, combined, raw);
+ File segmentDir = buildSegment(tableConfig, schema, version, multiValue);
+ installLucene10(segmentDir, version, false, combined);
+
+ IndexLoadingConfig loadingConfig = new IndexLoadingConfig(tableConfig, schema);
+ try (SegmentDirectory directory = new SegmentLocalFSDirectory(segmentDir, ReadMode.mmap)) {
+ assertTrue(ImmutableSegmentLoader.needPreprocess(directory, loadingConfig));
+ }
+ ImmutableSegment segment = ImmutableSegmentLoader.load(segmentDir, loadingConfig, true);
+ try {
+ assertEquals(segment.getDataSource("text").getTextIndex().getDocIds("hello", null).toArray(), new int[]{0});
+ assertEquals(segment.getDataSource("text").getTextIndex().getDocIds("other", null).toArray(), new int[]{1});
+ // For MV fields, original values contain duplicate terms and a phrase inside one value.
+ assertEquals(segment.getDataSource("text").getTextIndex().getDocIds("\"hello world\"", null).toArray(),
+ new int[]{0});
+ if (multiValue) {
+ assertEquals(
+ segment.getDataSource("text").getTextIndex().getDocIds("\"hello world hello world\"", null).toArray(),
+ new int[]{0});
+ assertEquals(segment.getDataSource("text").getTextIndex().getDocIds("\"value other\"", null).toArray(),
+ new int[]{1});
+ }
+ } finally {
+ segment.destroy();
+ }
+ assertStableReload(segmentDir, loadingConfig);
+ }
+
+ @DataProvider
+ public Object[][] multiColumnVersions() {
+ return new Object[][]{{SegmentVersion.v1}, {SegmentVersion.v3}};
+ }
+
+ @Test(dataProvider = "multiColumnVersions")
+ public void testMultiColumnRollback(SegmentVersion version)
+ throws Exception {
+ Schema schema = schema(false);
+ TableConfig tableConfig = tableConfig(true, false, false);
+ File segmentDir = buildSegment(tableConfig, schema, version, false);
+ installLucene10(segmentDir, version, true, false);
+ IndexLoadingConfig loadingConfig = new IndexLoadingConfig(tableConfig, schema);
+ try (SegmentDirectory directory = new SegmentLocalFSDirectory(segmentDir, ReadMode.mmap)) {
+ assertTrue(ImmutableSegmentLoader.needPreprocess(directory, loadingConfig));
+ }
+ ImmutableSegment segment = ImmutableSegmentLoader.load(segmentDir, loadingConfig, true);
+ try {
+ MultiColumnLuceneTextIndexReader reader = (MultiColumnLuceneTextIndexReader) segment.getMultiColumnTextIndex();
+ assertEquals(reader.getDocIds("text", "hello", (String) null).toArray(), new int[]{0});
+ assertEquals(reader.getDocIds("other", "beta", (String) null).toArray(), new int[]{1});
+ } finally {
+ segment.destroy();
+ }
+ assertStableReload(segmentDir, loadingConfig);
+ }
+
+ @Test(dataProvider = "multiColumnVersions")
+ public void testMultiColumnRollbackWithNewDefaultColumn(SegmentVersion version)
+ throws Exception {
+ Schema schema = schema(false);
+ TableConfig tableConfig = tableConfig(true, false, false);
+ File segmentDir = buildSegment(tableConfig, schema, version, false);
+ installLucene10(segmentDir, version, true, false);
+
+ Schema updatedSchema = new Schema.SchemaBuilder().setSchemaName("rollback")
+ .addSingleValueDimension("text", DataType.STRING).addSingleValueDimension("other", DataType.STRING)
+ .addSingleValueDimension("added", DataType.STRING, "defaultvalue").build();
+ tableConfig.getIndexingConfig().setMultiColumnTextIndexConfig(
+ new MultiColumnTextIndexConfig(List.of("text", "other", "added"), Map.of(), Map.of()));
+ IndexLoadingConfig loadingConfig = new IndexLoadingConfig(tableConfig, updatedSchema);
+ ImmutableSegment segment = ImmutableSegmentLoader.load(segmentDir, loadingConfig, true);
+ try {
+ MultiColumnLuceneTextIndexReader reader = (MultiColumnLuceneTextIndexReader) segment.getMultiColumnTextIndex();
+ assertEquals(reader.getDocIds("text", "hello", (String) null).toArray(), new int[]{0});
+ assertEquals(reader.getDocIds("other", "beta", (String) null).toArray(), new int[]{1});
+ assertEquals(reader.getDocIds("added", "defaultvalue", (String) null).toArray(), new int[]{0, 1});
+ } finally {
+ segment.destroy();
+ }
+ assertStableReload(segmentDir, loadingConfig);
+ }
+
+ @Test(dataProvider = "multiColumnVersions")
+ public void testInterruptedMultiColumnRemoval(SegmentVersion version)
+ throws Exception {
+ Schema schema = schema(false);
+ TableConfig tableConfig = tableConfig(true, false, false);
+ File segmentDir = buildSegment(tableConfig, schema, version, false);
+ File indexDir = SegmentDirectoryPaths.segmentDirectoryFor(segmentDir, version);
+ File luceneDir = SegmentDirectoryPaths.findTextIndexIndexFile(segmentDir,
+ MultiColumnTextIndexConstants.INDEX_DIR_NAME);
+ // A failed removal has cleared metadata, but the marker and index files remain.
+ PropertiesConfiguration properties = SegmentMetadataUtils.getPropertiesConfiguration(segmentDir);
+ properties.subset(MultiColumnTextIndexConstants.MetadataKey.ROOT_SUBSET).clear();
+ SegmentMetadataUtils.savePropertiesConfiguration(properties, segmentDir);
+ File marker = new File(segmentDir, MultiColumnTextIndexConstants.INDEX_DIR_NAME + ".text.inprogress");
+ FileUtils.touch(marker);
+ tableConfig.getIndexingConfig().setMultiColumnTextIndexConfig(null);
+
+ IndexLoadingConfig loadingConfig = new IndexLoadingConfig(tableConfig, schema);
+ ImmutableSegment segment = ImmutableSegmentLoader.load(segmentDir, loadingConfig, true);
+ try {
+ assertNull(segment.getMultiColumnTextIndex());
+ assertFalse(luceneDir.exists());
+ assertFalse(new File(indexDir, MultiColumnTextIndexConstants.DOCID_MAPPING_FILE_NAME).exists());
+ assertFalse(marker.exists());
+ } finally {
+ segment.destroy();
+ }
+ assertStableReload(segmentDir, loadingConfig);
+ }
+
+ @Test(dataProvider = "multiColumnVersions")
+ public void testMissingOriginalForwardIndexPreservesTextIndex(SegmentVersion version)
+ throws Exception {
+ // Dictionary/inverted reconstruction cannot retain MV order or duplicates. Fail before that handler runs.
+ for (boolean multiColumn : List.of(false, true)) {
+ Schema schema = schema(true);
+ TableConfig tableConfig = tableConfig(multiColumn, false, false);
+ tableConfig.getIndexingConfig().setInvertedIndexColumns(List.of("text"));
+ File segmentDir = buildSegment(tableConfig, schema, version, true);
+ installLucene10(segmentDir, version, multiColumn, false);
+ try (SegmentDirectory directory = new SegmentLocalFSDirectory(segmentDir, ReadMode.mmap);
+ SegmentDirectory.Writer writer = directory.createWriter()) {
+ writer.removeIndex("text", StandardIndexes.forward());
+ writer.save();
+ }
+ File luceneDir = SegmentDirectoryPaths.findTextIndexIndexFile(segmentDir,
+ multiColumn ? MultiColumnTextIndexConstants.INDEX_DIR_NAME : "text");
+ File commitFile = new File(luceneDir, "segments_1");
+ byte[] original = Files.readAllBytes(commitFile.toPath());
+ assertThrows(IllegalStateException.class,
+ () -> ImmutableSegmentLoader.load(segmentDir, new IndexLoadingConfig(tableConfig, schema), true));
+ assertEquals(Files.readAllBytes(commitFile.toPath()), original);
+ assertTrue(LuceneIndexCompatibilityUtils.isIncompatible(luceneDir));
+ FileUtils.deleteDirectory(segmentDir);
+ }
+ }
+
+ @DataProvider
+ public Object[][] textKinds() {
+ return new Object[][]{{false}, {true}};
+ }
+
+ @Test(dataProvider = "textKinds")
+ public void testInterruptedReadableIndexRestartsFromOriginalValues(boolean multiColumn)
+ throws Exception {
+ Schema schema = schema(false);
+ TableConfig tableConfig = tableConfig(multiColumn, false, false);
+ File segmentDir = buildSegment(tableConfig, schema, SegmentVersion.v3, false);
+ File partial = installInterruptedBuild(segmentDir, multiColumn);
+ assertFalse(LuceneIndexCompatibilityUtils.isIncompatible(partial));
+ IndexLoadingConfig loadingConfig = new IndexLoadingConfig(tableConfig, schema);
+ try (SegmentDirectory directory = new SegmentLocalFSDirectory(segmentDir, ReadMode.mmap)) {
+ assertTrue(ImmutableSegmentLoader.needPreprocess(directory, loadingConfig));
+ }
+ ImmutableSegment segment = ImmutableSegmentLoader.load(segmentDir, loadingConfig, true);
+ try {
+ if (multiColumn) {
+ MultiColumnLuceneTextIndexReader reader = (MultiColumnLuceneTextIndexReader) segment.getMultiColumnTextIndex();
+ assertEquals(reader.getDocIds("text", "hello", (String) null).toArray(), new int[]{0});
+ assertEquals(reader.getDocIds("text", "other", (String) null).toArray(), new int[]{1});
+ assertTrue(reader.getDocIds("text", "partial", (String) null).isEmpty());
+ } else {
+ assertEquals(segment.getDataSource("text").getTextIndex().getDocIds("hello", null).toArray(), new int[]{0});
+ assertEquals(segment.getDataSource("text").getTextIndex().getDocIds("other", null).toArray(), new int[]{1});
+ assertTrue(segment.getDataSource("text").getTextIndex().getDocIds("partial", null).isEmpty());
+ }
+ } finally {
+ segment.destroy();
+ }
+ try (FSDirectory directory = FSDirectory.open(partial.toPath());
+ DirectoryReader reader = DirectoryReader.open(directory)) {
+ assertEquals(reader.numDocs(), 2);
+ }
+ assertStableReload(segmentDir, loadingConfig);
+ }
+
+ @Test(dataProvider = "textKinds")
+ public void testInterruptedBuildRequiresOriginalForwardIndex(boolean multiColumn)
+ throws Exception {
+ Schema schema = schema(true);
+ TableConfig tableConfig = tableConfig(multiColumn, false, false);
+ tableConfig.getIndexingConfig().setInvertedIndexColumns(List.of("text"));
+ File segmentDir = buildSegment(tableConfig, schema, SegmentVersion.v3, true);
+ File partial = installInterruptedBuild(segmentDir, multiColumn);
+ removeForwardIndex(segmentDir);
+ assertThrows(IllegalStateException.class,
+ () -> ImmutableSegmentLoader.load(segmentDir, new IndexLoadingConfig(tableConfig, schema), true));
+ try (FSDirectory directory = FSDirectory.open(partial.toPath());
+ DirectoryReader reader = DirectoryReader.open(directory)) {
+ assertEquals(reader.numDocs(), 1);
+ }
+ try (SegmentDirectory directory = new SegmentLocalFSDirectory(segmentDir, ReadMode.mmap);
+ SegmentDirectory.Reader reader = directory.createReader()) {
+ assertFalse(reader.hasIndexFor("text", StandardIndexes.forward()));
+ }
+ }
+
+ @Test
+ public void testCompatibleDirectoryDoesNotHideIncompatiblePackedIndex()
+ throws Exception {
+ Schema schema = schema(true);
+ TableConfig tableConfig = tableConfig(false, true, false);
+ tableConfig.getIndexingConfig().setInvertedIndexColumns(List.of("text"));
+ File segmentDir = buildSegment(tableConfig, schema, SegmentVersion.v3, true);
+ installLucene10(segmentDir, SegmentVersion.v3, false, true);
+ File directory = new File(SegmentDirectoryPaths.segmentDirectoryFor(segmentDir, SegmentVersion.v3),
+ "text" + V1Constants.Indexes.LUCENE_V912_TEXT_INDEX_FILE_EXTENSION);
+ writePartialIndex(directory);
+ assertFalse(LuceneIndexCompatibilityUtils.isIncompatible(directory));
+ removeForwardIndex(segmentDir);
+ assertThrows(IllegalStateException.class,
+ () -> ImmutableSegmentLoader.load(segmentDir, new IndexLoadingConfig(tableConfig, schema), true));
+ try (SegmentDirectory segmentDirectory = new SegmentLocalFSDirectory(segmentDir, ReadMode.mmap);
+ SegmentDirectory.Reader reader = segmentDirectory.createReader()) {
+ assertTrue(LuceneIndexCompatibilityUtils.isIncompatible(
+ reader.getIndexFor("text", StandardIndexes.text()), "text"));
+ assertFalse(reader.hasIndexFor("text", StandardIndexes.forward()));
+ }
+ }
+
+ private File installInterruptedBuild(File segmentDir, boolean multiColumn)
+ throws Exception {
+ String column = multiColumn ? MultiColumnTextIndexConstants.INDEX_DIR_NAME : "text";
+ File partial = new File(SegmentDirectoryPaths.segmentDirectoryFor(segmentDir, SegmentVersion.v3),
+ column + V1Constants.Indexes.LUCENE_V912_TEXT_INDEX_FILE_EXTENSION);
+ writePartialIndex(partial);
+ if (multiColumn) {
+ // Reproduce the old loader's crash window: metadata was removed and the partial directory remains.
+ PropertiesConfiguration properties = SegmentMetadataUtils.getPropertiesConfiguration(segmentDir);
+ properties.subset(MultiColumnTextIndexConstants.MetadataKey.ROOT_SUBSET).clear();
+ SegmentMetadataUtils.savePropertiesConfiguration(properties, segmentDir);
+ } else {
+ FileUtils.touch(new File(segmentDir, "text.text.inprogress"));
+ }
+ return partial;
+ }
+
+ private void writePartialIndex(File directory)
+ throws Exception {
+ try (FSDirectory lucene = FSDirectory.open(directory.toPath());
+ StandardAnalyzer analyzer = new StandardAnalyzer();
+ IndexWriter writer = new IndexWriter(lucene,
+ new IndexWriterConfig(analyzer).setOpenMode(IndexWriterConfig.OpenMode.CREATE))) {
+ Document document = new Document();
+ document.add(new TextField("text", "partial", Field.Store.NO));
+ document.add(new StoredField(LuceneTextIndexCreator.LUCENE_INDEX_DOC_ID_COLUMN_NAME, 0));
+ writer.addDocument(document);
+ }
+ }
+
+ private void removeForwardIndex(File segmentDir)
+ throws Exception {
+ try (SegmentDirectory directory = new SegmentLocalFSDirectory(segmentDir, ReadMode.mmap);
+ SegmentDirectory.Writer writer = directory.createWriter()) {
+ writer.removeIndex("text", StandardIndexes.forward());
+ writer.save();
+ }
+ }
+
+ private void assertStableReload(File segmentDir, IndexLoadingConfig loadingConfig)
+ throws Exception {
+ try (SegmentDirectory directory = new SegmentLocalFSDirectory(segmentDir, ReadMode.mmap)) {
+ assertFalse(ImmutableSegmentLoader.needPreprocess(directory, loadingConfig));
+ }
+ ImmutableSegment segment = ImmutableSegmentLoader.load(segmentDir, loadingConfig, true);
+ segment.destroy();
+ }
+
+ private Schema schema(boolean multiValue) {
+ Schema.SchemaBuilder builder = new Schema.SchemaBuilder().setSchemaName("rollback");
+ if (multiValue) {
+ builder.addMultiValueDimension("text", DataType.STRING);
+ } else {
+ builder.addSingleValueDimension("text", DataType.STRING);
+ }
+ return builder.addSingleValueDimension("other", DataType.STRING).build();
+ }
+
+ private TableConfig tableConfig(boolean multiColumn, boolean combined, boolean raw) {
+ TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("rollback").build();
+ if (raw) {
+ tableConfig.getIndexingConfig().setNoDictionaryColumns(List.of("text"));
+ }
+ if (multiColumn) {
+ tableConfig.getIndexingConfig().setMultiColumnTextIndexConfig(
+ new MultiColumnTextIndexConfig(List.of("text", "other"), Map.of(), Map.of()));
+ } else {
+ tableConfig.setFieldConfigList(List.of(new FieldConfig("text",
+ raw ? FieldConfig.EncodingType.RAW : FieldConfig.EncodingType.DICTIONARY,
+ List.of(FieldConfig.IndexType.TEXT), null, Map.of("storeInSegmentFile", Boolean.toString(combined)))));
+ }
+ return tableConfig;
+ }
+
+ private File buildSegment(TableConfig tableConfig, Schema schema, SegmentVersion version, boolean multiValue)
+ throws Exception {
+ SegmentGeneratorConfig config = new SegmentGeneratorConfig(tableConfig, schema);
+ config.setOutDir(_tempDir.getAbsolutePath());
+ config.setSegmentName("segment");
+ config.setSegmentVersion(version);
+ List<GenericRow> rows = new ArrayList<>();
+ GenericRow first = new GenericRow();
+ first.putValue("text", multiValue ? new String[]{"hello world", "hello world"} : "hello world");
+ first.putValue("other", "alpha");
+ rows.add(first);
+ GenericRow second = new GenericRow();
+ second.putValue("text", multiValue ? new String[]{"other value", "other"} : "other value");
+ second.putValue("other", "beta");
+ rows.add(second);
+ SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl();
+ driver.init(config, new GenericRowRecordReader(rows));
+ driver.build();
+ return new File(_tempDir, "segment");
+ }
+
+ private void installLucene10(File segmentDir, SegmentVersion version, boolean multiColumn, boolean combined)
+ throws Exception {
+ String column = multiColumn ? MultiColumnTextIndexConstants.INDEX_DIR_NAME : "text";
+ File actualDirectory = SegmentDirectoryPaths.segmentDirectoryFor(segmentDir, version);
+ File luceneDir = new File(actualDirectory, column + V1Constants.Indexes.LUCENE_V912_TEXT_INDEX_FILE_EXTENSION);
+ if (!multiColumn) {
+ try (SegmentDirectory directory = new SegmentLocalFSDirectory(segmentDir, ReadMode.mmap);
+ SegmentDirectory.Writer writer = directory.createWriter()) {
+ writer.removeIndex(column, StandardIndexes.text());
+ writer.save();
+ }
+ }
+ FileUtils.deleteQuietly(luceneDir);
+ FileUtils.copyDirectory(new File(_fixtures, multiColumn ? "multi" : "text"), luceneDir);
+ try (FSDirectory directory = FSDirectory.open(luceneDir.toPath())) {
+ assertThrows(IllegalArgumentException.class, () -> DirectoryReader.open(directory));
+ }
+ if (combined) {
+ File combinedFile =
+ new File(actualDirectory, column + V1Constants.Indexes.LUCENE_COMBINE_TEXT_INDEX_FILE_EXTENSION);
+ LuceneTextIndexCombined.combineLuceneIndexFiles(luceneDir, combinedFile.getAbsolutePath());
+ FileUtils.deleteDirectory(luceneDir);
+ if (version == SegmentVersion.v3) {
+ try (SegmentDirectory directory = new SegmentLocalFSDirectory(segmentDir, ReadMode.mmap);
+ SegmentDirectory.Writer writer = directory.createWriter()) {
+ LoaderUtils.writeIndexToV3Format(writer, column, combinedFile, StandardIndexes.text());
+ writer.save();
+ }
+ }
+ }
+ }
+}
diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/VectorIndexHandlerTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/VectorIndexHandlerTest.java
index 9bd1e7a0460..c5409172fb4 100644
--- a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/VectorIndexHandlerTest.java
+++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/VectorIndexHandlerTest.java
@@ -31,6 +31,7 @@ import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;
import org.apache.commons.io.FileUtils;
+import org.apache.pinot.segment.local.segment.creator.impl.vector.HnswVectorIndexCreator;
import org.apache.pinot.segment.local.segment.creator.impl.vector.lucene99.HnswVectorIndexCombined;
import org.apache.pinot.segment.spi.V1Constants;
import org.apache.pinot.segment.spi.creator.SegmentVersion;
@@ -268,12 +269,12 @@ public class VectorIndexHandlerTest {
File indexDir = createEmptyV3SegmentDir();
try {
File v3Dir = new File(indexDir, SegmentDirectoryPaths.V3_SUBDIRECTORY_NAME);
- // Build a real combined HNSW payload from a fake source dir (the extract copies bytes; it does
- // not need a valid Lucene index for this test). Capture the bytes, then drop the scratch files.
- File srcDir = new File(v3Dir, "src-hnsw");
- assertTrue(srcDir.mkdirs());
- Files.write(new File(srcDir, "_0.cfe").toPath(), "lucene-file-one".getBytes(StandardCharsets.UTF_8));
- Files.write(new File(srcDir, "segments_1").toPath(), "lucene-file-two".getBytes(StandardCharsets.UTF_8));
+ // Use a readable HNSW index so the format check permits extraction without a rebuild.
+ File srcDir = new File(v3Dir, COLUMN + V1Constants.Indexes.VECTOR_V912_HNSW_INDEX_FILE_EXTENSION);
+ try (HnswVectorIndexCreator creator = new HnswVectorIndexCreator(COLUMN, v3Dir, vectorIndexConfig("HNSW"))) {
+ creator.add(new float[]{1, 0, 0, 0});
+ creator.seal();
+ }
File scratchCombined = new File(v3Dir, COLUMN + ".vector.hnsw.scratch");
HnswVectorIndexCombined.combineHnswIndexFiles(srcDir, scratchCombined.getAbsolutePath(), null, null);
byte[] combinedBytes = Files.readAllBytes(scratchCombined.toPath());
diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/LuceneIndexCompatibilityUtilsTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/LuceneIndexCompatibilityUtilsTest.java
new file mode 100644
index 00000000000..2b15fbf8745
--- /dev/null
+++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/LuceneIndexCompatibilityUtilsTest.java
@@ -0,0 +1,96 @@
+/**
+ * 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.utils;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import org.apache.commons.io.FileUtils;
+import org.apache.lucene.index.CorruptIndexException;
+import org.apache.lucene.index.IndexNotFoundException;
+import org.apache.lucene.index.IndexWriter;
+import org.apache.lucene.index.IndexWriterConfig;
+import org.apache.lucene.store.ByteBuffersDirectory;
+import org.apache.lucene.store.Directory;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
+
+
+/// Verifies that format recovery does not mask corrupt commits or ordinary I/O failures.
+public class LuceneIndexCompatibilityUtilsTest {
+ private File _fixtures;
+
+ @BeforeMethod
+ public void setUp()
+ throws IOException {
+ _fixtures = LuceneIndexTestUtils.extractLucene10Fixtures(
+ Files.createTempDirectory("lucene-compatibility").toFile());
+ }
+
+ @AfterMethod
+ public void tearDown()
+ throws IOException {
+ FileUtils.deleteDirectory(_fixtures);
+ }
+
+ @Test
+ public void testNewerFormatsIncludingOlderCreationVersion()
+ throws IOException {
+ for (String name : new String[]{"text", "multi", "hnsw", "created9", "created9-unknowncodec"}) {
+ assertTrue(LuceneIndexCompatibilityUtils.isIncompatible(new File(_fixtures, name)), name);
+ }
+ try (Directory directory = new ByteBuffersDirectory()) {
+ try (IndexWriter writer = new IndexWriter(directory, new IndexWriterConfig())) {
+ writer.commit();
+ }
+ assertFalse(LuceneIndexCompatibilityUtils.isIncompatible(directory));
+ }
+ }
+
+ @Test
+ public void testCorruptFutureCommitIsNotRecoverable()
+ throws IOException {
+ File directory = new File(_fixtures, "text");
+ File commit = new File(directory, "segments_1");
+ byte[] bytes = Files.readAllBytes(commit.toPath());
+ bytes[bytes.length - 1] ^= 1;
+ Files.write(commit.toPath(), bytes);
+ assertThrows(CorruptIndexException.class, () -> LuceneIndexCompatibilityUtils.isIncompatible(directory));
+ }
+
+ @Test
+ public void testMissingFilesPropagate()
+ throws IOException {
+ assertThrows(IOException.class,
+ () -> LuceneIndexCompatibilityUtils.isIncompatible(new File(_fixtures, "does-not-exist")));
+ try (Directory directory = new ByteBuffersDirectory()) {
+ assertThrows(IndexNotFoundException.class, () -> LuceneIndexCompatibilityUtils.isIncompatible(directory));
+ }
+ File directory = new File(_fixtures, "created9");
+ File[] compoundFiles = directory.listFiles((parent, name) -> name.endsWith(".cfs"));
+ assertTrue(compoundFiles != null && compoundFiles.length == 1);
+ Files.delete(compoundFiles[0].toPath());
+ assertThrows(IOException.class, () -> LuceneIndexCompatibilityUtils.isIncompatible(directory));
+ }
+}
diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/LuceneIndexTestUtils.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/LuceneIndexTestUtils.java
new file mode 100644
index 00000000000..f9301bec16c
--- /dev/null
+++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/LuceneIndexTestUtils.java
@@ -0,0 +1,61 @@
+/**
+ * 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.utils;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipInputStream;
+import org.apache.commons.io.FileUtils;
+
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+
+
+/// Extracts the shared Lucene 10 rollback fixtures into a caller-owned directory, without shared state.
+public final class LuceneIndexTestUtils {
+ private LuceneIndexTestUtils() {
+ }
+
+ public static File extractLucene10Fixtures(File directory)
+ throws IOException {
+ FileUtils.forceMkdir(directory);
+ // Generated with Lucene 10.0.0 (text/multi) and Pinot f9cd6821's HnswCodec (hnsw).
+ // created9 variants set the creation major to 9, with Lucene912 and Lucene100 codecs respectively.
+ try (InputStream input = LuceneIndexTestUtils.class.getResourceAsStream("/data/lucene10-rollback.zip")) {
+ assertNotNull(input);
+ try (ZipInputStream zip = new ZipInputStream(input)) {
+ ZipEntry entry;
+ while ((entry = zip.getNextEntry()) != null) {
+ File output = new File(directory, entry.getName());
+ assertTrue(output.toPath().normalize().startsWith(directory.toPath().normalize()));
+ if (entry.isDirectory()) {
+ FileUtils.forceMkdir(output);
+ } else {
+ FileUtils.forceMkdirParent(output);
+ Files.copy(zip, output.toPath());
+ }
+ }
+ }
+ }
+ return directory;
+ }
+}
diff --git a/pinot-segment-local/src/test/resources/data/lucene10-rollback.zip b/pinot-segment-local/src/test/resources/data/lucene10-rollback.zip
new file mode 100644
index 00000000000..61ece7c4eee
Binary files /dev/null and b/pinot-segment-local/src/test/resources/data/lucene10-rollback.zip differ
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org