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 <[email protected]>
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: [email protected]
For additional commands, e-mail: [email protected]
