This is an automated email from the ASF dual-hosted git repository. xiangfu0 pushed a commit to branch xiangfu0/codex1/lucene-index-rollback-recovery in repository https://gitbox.apache.org/repos/asf/pinot.git
commit a51e726816669965664d2577acc48a1da7586cba Author: Xiang Fu <[email protected]> AuthorDate: Thu Oct 15 16:12:25 2026 -0700 Rebuild incompatible Lucene indexes during segment loading Keep Lucene 9.12 and rebuild newer text/HNSW indexes from retained forward values, restarting interrupted builds safely. Reproduce by loading a Lucene 10-generated segment on 9.12; loading fails before this change. --- README.md | 15 + .../segment/index/loader/SegmentPreProcessor.java | 28 +- .../invertedindex/MultiColumnTextIndexHandler.java | 83 +++- .../loader/invertedindex/TextIndexHandler.java | 183 ++++++--- .../loader/invertedindex/VectorIndexHandler.java | 97 +++++ .../local/segment/store/TextIndexUtils.java | 20 + .../local/utils/LuceneIndexCompatibilityUtils.java | 117 ++++++ .../invertedindex/HnswIndexRollbackTest.java | 318 ++++++++++++++++ .../TextIndexRollbackRecoveryTest.java | 417 +++++++++++++++++++++ .../invertedindex/VectorIndexHandlerTest.java | 13 +- .../utils/LuceneIndexCompatibilityUtilsTest.java | 107 ++++++ .../src/test/resources/data/lucene10-rollback.zip | Bin 0 -> 9060 bytes 12 files changed, 1333 insertions(+), 65 deletions(-) diff --git a/README.md b/README.md index aa3e5713f20..fc0fd979c4e 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, and corruption or ordinary I/O errors are not treated as format incompatibility. + ## 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 1bac733774a..1bc541f7459 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 @@ -39,6 +39,8 @@ 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; @@ -154,6 +156,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; @@ -308,7 +320,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; } @@ -358,10 +370,9 @@ 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, @@ -378,7 +389,7 @@ public class SegmentPreProcessor implements AutoCloseable { if (newConfig == null) { remove = true; } else { - if (MultiColumnTextIndexHandler.shouldModifyMultiColTextIndex(newConfig, oldConfig)) { + if (needProcessMultiColumnTextIndex(segmentWriter)) { LOGGER.info("Change detected in multi-column text index for segment: {}", segmentName); } else { create = false; @@ -398,13 +409,14 @@ public class SegmentPreProcessor implements AutoCloseable { LOGGER.info("Removing multi-column text index from segment: {}", segmentName); removeMultiColumnTextIndex(indexDir); } else if (create) { + MultiColumnTextIndexHandler handler = + new MultiColumnTextIndexHandler(_segmentDirectory, _indexLoadingConfig, newConfig); + handler.validateRecoverySources(segmentWriter); 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); } 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..fb1f6b93937 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,66 @@ public class MultiColumnTextIndexHandler extends BaseIndexHandler { } } + /// Verify all original source columns before removing an incompatible shared text index. + public void validateRecoverySources(SegmentDirectory.Reader segmentReader) + throws IOException { + if (_textIndexConfig != null && (isBuildInProgress() || isIncompatible())) { + for (String column : _textIndexConfig.getColumns()) { + 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() { + if (_textIndexConfig == null) { + return false; + } + 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 _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,6 +220,7 @@ public class MultiColumnTextIndexHandler extends BaseIndexHandler { @Override public void updateIndices(SegmentDirectory.Writer segmentWriter) throws Exception { + validateRecoverySources(segmentWriter); File indexDir = _segmentDirectory.getSegmentMetadata().getIndexDir(); File segmentDirectory = SegmentDirectoryPaths.segmentDirectoryFor(indexDir, _segmentDirectory.getSegmentMetadata().getVersion()); @@ -166,6 +234,18 @@ public class MultiColumnTextIndexHandler extends BaseIndexHandler { columnsSV.add(metadata.isSingleValue()); } + File marker = getBuildMarker(); + FileUtils.touch(marker); + 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); + } + try (MultiColumnLuceneTextIndexCreator creator = new MultiColumnLuceneTextIndexCreator(_textIndexConfig.getColumns(), columnsSV, segmentDirectory, true, false, null, null, _textIndexConfig)) { @@ -178,6 +258,7 @@ public class MultiColumnTextIndexHandler extends BaseIndexHandler { _textIndexConfig.getColumns(), _textIndexConfig.getProperties(), _textIndexConfig.getPerColumnProperties()); SegmentMetadataUtils.savePropertiesConfiguration(metadataProperties, indexDir); } + FileUtils.forceDelete(marker); } private void createMultiColumnTextIndex(SegmentDirectory.Writer segmentWriter, 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..be507b2c91b 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,84 @@ 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 { + Set<String> existingColumns = segmentReader.toSegmentDirectory().getColumnsWithIndex(StandardIndexes.vector()); + for (Map.Entry<String, VectorIndexConfig> entry : _vectorConfigs.entrySet()) { + String column = entry.getKey(); + if (entry.getValue().resolveBackendType() == VectorBackendType.HNSW + && (hasHnswRecoveryMarker(column) || existingColumns.contains(column) + && isIncompatibleHnswIndex(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 +579,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 +685,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 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..78f2f33c04e --- /dev/null +++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/HnswIndexRollbackTest.java @@ -0,0 +1,318 @@ +/** + * 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.io.InputStream; +import java.nio.file.Files; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.zip.ZipEntry; +import java.util.zip.ZipInputStream; +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.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 fixture = new File(tempDir, "lucene10-hnsw"); + FileUtils.forceMkdir(fixture); + try (InputStream input = HnswIndexRollbackTest.class.getResourceAsStream("/data/lucene10-rollback.zip")) { + assertNotNull(input); + try (ZipInputStream zip = new ZipInputStream(input)) { + ZipEntry entry; + while ((entry = zip.getNextEntry()) != null) { + if (!entry.isDirectory() && entry.getName().startsWith("hnsw/")) { + String fileName = entry.getName().substring("hnsw/".length()); + Files.copy(zip, new File(fixture, fileName).toPath()); + } + } + } + } + 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..c4efb7b265a --- /dev/null +++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/TextIndexRollbackRecoveryTest.java @@ -0,0 +1,417 @@ +/** + * 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.InputStream; +import java.nio.file.Files; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.zip.ZipEntry; +import java.util.zip.ZipInputStream; +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.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.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 = new File(_tempDir, "fixtures"); + try (InputStream resource = getClass().getClassLoader().getResourceAsStream("data/lucene10-rollback.zip"); + ZipInputStream zip = new ZipInputStream(resource)) { + ZipEntry entry; + while ((entry = zip.getNextEntry()) != null) { + File output = new File(_fixtures, entry.getName()); + assertTrue(output.toPath().normalize().startsWith(_fixtures.toPath())); + if (!entry.isDirectory()) { + FileUtils.forceMkdirParent(output); + Files.copy(zip, output.toPath()); + } + } + } + } + + @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", null).toArray(), new int[]{0}); + assertEquals(reader.getDocIds("other", "beta", null).toArray(), new int[]{1}); + } 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", null).toArray(), new int[]{0}); + assertEquals(reader.getDocIds("text", "other", null).toArray(), new int[]{1}); + assertTrue(reader.getDocIds("text", "partial", 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..7862c85c8ba --- /dev/null +++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/LuceneIndexCompatibilityUtilsTest.java @@ -0,0 +1,107 @@ +/** + * 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 java.util.zip.ZipEntry; +import java.util.zip.ZipInputStream; +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 = Files.createTempDirectory("lucene-compatibility").toFile(); + // 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 (ZipInputStream zip = new ZipInputStream(getClass().getResourceAsStream("/data/lucene10-rollback.zip"))) { + ZipEntry entry; + while ((entry = zip.getNextEntry()) != null) { + File output = new File(_fixtures, entry.getName()); + FileUtils.forceMkdirParent(output); + Files.copy(zip, output.toPath()); + } + } + } + + @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/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]
