From 830f8003210e91655910415efe1476b666eab904 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Sat, 11 Jul 2026 22:17:31 +0800 Subject: [PATCH 1/2] [core] Add bucket-level primary-key vector index maintenance --- .../apache/paimon/index/IndexFileHandler.java | 5 + .../BucketedVectorIndexMaintainer.java | 353 ++++++++++++++++++ .../pkvector/PkVectorBucketIndexState.java | 103 +++++ .../pkvector/PkVectorDataFileReader.java | 132 +++++++ .../index/pkvector/PkVectorSourcePolicy.java | 38 ++ .../paimon/io/KeyValueFileReaderFactory.java | 6 + .../manifest/IndexManifestFileHandler.java | 40 +- .../BucketedVectorIndexMaintainerTest.java | 233 ++++++++++++ .../PkVectorBucketIndexStateTest.java | 95 +++++ .../pkvector/PkVectorDataFileReaderTest.java | 160 ++++++++ .../IndexManifestFileHandlerTest.java | 37 ++ 11 files changed, 1195 insertions(+), 7 deletions(-) create mode 100644 paimon-core/src/main/java/org/apache/paimon/index/pkvector/BucketedVectorIndexMaintainer.java create mode 100644 paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorBucketIndexState.java create mode 100644 paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorDataFileReader.java create mode 100644 paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorSourcePolicy.java create mode 100644 paimon-core/src/test/java/org/apache/paimon/index/pkvector/BucketedVectorIndexMaintainerTest.java create mode 100644 paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorBucketIndexStateTest.java create mode 100644 paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorDataFileReaderTest.java diff --git a/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java b/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java index e848ed205e14..92909a618da4 100644 --- a/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java +++ b/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java @@ -25,6 +25,7 @@ import org.apache.paimon.deletionvectors.DeletionVectorsIndexFile; import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.Path; +import org.apache.paimon.index.pkvector.PkVectorAnnSegmentFile; import org.apache.paimon.manifest.IndexManifestEntry; import org.apache.paimon.manifest.IndexManifestEntrySerializer; import org.apache.paimon.manifest.IndexManifestFile; @@ -84,6 +85,10 @@ public DeletionVectorsIndexFile dvIndex(BinaryRow partition, int bucket) { fileIO, pathFactories.get(partition, bucket), dvTargetFileSize, dvBitmap64); } + public PkVectorAnnSegmentFile pkVectorAnnSegment(BinaryRow partition, int bucket) { + return new PkVectorAnnSegmentFile(fileIO, pathFactories.get(partition, bucket)); + } + public Optional scanHashIndex( Snapshot snapshot, BinaryRow partition, int bucket) { List result = scan(snapshot, HASH_INDEX, partition, bucket); diff --git a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/BucketedVectorIndexMaintainer.java b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/BucketedVectorIndexMaintainer.java new file mode 100644 index 000000000000..56ba5f9cfc00 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/BucketedVectorIndexMaintainer.java @@ -0,0 +1,353 @@ +/* + * 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.paimon.index.pkvector; + +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.index.IndexFileHandler; +import org.apache.paimon.index.IndexFileMeta; +import org.apache.paimon.io.CompactIncrement; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.io.DataIncrement; +import org.apache.paimon.io.KeyValueFileReaderFactory; +import org.apache.paimon.options.Options; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.VectorType; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** Maintains bucket-local ANN payloads from complete compact-output data files. */ +public class BucketedVectorIndexMaintainer { + + private final int vectorFieldId; + private final PkVectorAnnSegmentFile annSegmentFile; + private final DataField vectorField; + private final Options indexOptions; + private final String metric; + private final String algorithm; + private final PkVectorDataFileReader.Factory vectorReaderFactory; + private final List annSegments; + private final Map activeSourceFiles; + + BucketedVectorIndexMaintainer( + int vectorFieldId, + PkVectorAnnSegmentFile annSegmentFile, + DataField vectorField, + Options indexOptions, + String metric, + String algorithm, + PkVectorDataFileReader.Factory vectorReaderFactory, + List restoredDataFiles, + List restoredPayloads) { + this.vectorFieldId = vectorFieldId; + this.annSegmentFile = annSegmentFile; + this.vectorField = vectorField; + this.indexOptions = indexOptions; + this.metric = metric; + this.algorithm = algorithm; + this.vectorReaderFactory = vectorReaderFactory; + PkVectorBucketIndexState restoredState = + new PkVectorBucketIndexState(vectorFieldId, algorithm, restoredPayloads); + this.annSegments = new ArrayList<>(restoredState.annSegments()); + this.activeSourceFiles = new LinkedHashMap<>(); + for (DataFileMeta file : restoredDataFiles) { + if (PkVectorSourcePolicy.shouldRead(file)) { + activeSourceFiles.put(file.fileName(), file); + } + } + validateCoverage(annSegments, activeSourceFiles); + } + + /** Produces ANN changes in the same compact increment as their source data-file transition. */ + public synchronized VectorIndexCommit prepareCommit( + DataIncrement appendIncrement, CompactIncrement compactIncrement) { + checkArgument( + eligibleFiles(appendIncrement.newFiles()).isEmpty(), + "Append files must not be primary-key vector index sources."); + + Map nextSources = new LinkedHashMap<>(activeSourceFiles); + Set deletedFiles = new HashSet<>(); + for (DataFileMeta file : compactIncrement.compactBefore()) { + if (!containsFile(compactIncrement.compactAfter(), file.fileName())) { + deletedFiles.add(file.fileName()); + nextSources.remove(file.fileName()); + } + } + for (DataFileMeta file : compactIncrement.compactAfter()) { + if (PkVectorSourcePolicy.shouldRead(file)) { + nextSources.put(file.fileName(), file); + } + } + + List retained = new ArrayList<>(); + List removed = new ArrayList<>(); + for (IndexFileMeta ann : annSegments) { + if (referencesAny(ann, deletedFiles)) { + removed.add(ann); + } else { + retained.add(ann); + } + } + + Set covered = coveredSources(retained); + List uncovered = new ArrayList<>(); + for (DataFileMeta file : nextSources.values()) { + if (!covered.contains(file.fileName())) { + uncovered.add(file); + } + } + uncovered.sort(Comparator.comparing(DataFileMeta::fileName)); + + List created = new ArrayList<>(); + try { + if (!uncovered.isEmpty()) { + created.add(buildAnnSegment(uncovered)); + } + List nextAnn = new ArrayList<>(retained); + nextAnn.addAll(created); + validateCoverage(nextAnn, nextSources); + + activeSourceFiles.clear(); + activeSourceFiles.putAll(nextSources); + annSegments.clear(); + annSegments.addAll(nextAnn); + Optional compactChange = + created.isEmpty() && removed.isEmpty() + ? Optional.empty() + : Optional.of(new VectorIndexIncrement(created, removed)); + return new VectorIndexCommit(Optional.empty(), compactChange); + } catch (RuntimeException e) { + deleteAnnSegments(created); + throw e; + } + } + + private IndexFileMeta buildAnnSegment(List files) { + try { + List sources = new ArrayList<>(files.size()); + for (DataFileMeta file : files) { + PkVectorSourceFile sourceFile = + new PkVectorSourceFile(file.fileName(), file.rowCount()); + sources.add( + PkVectorAnnSegmentFile.Source.lazy( + sourceFile, () -> vectorReaderFactory.create(file))); + } + return annSegmentFile.build(sources, vectorField, indexOptions, metric, algorithm); + } catch (IOException e) { + throw new UncheckedIOException("Failed to build an ANN vector segment.", e); + } + } + + private void validateCoverage( + List candidateAnn, Map sourceFiles) { + PkVectorBucketIndexState state = + new PkVectorBucketIndexState(vectorFieldId, algorithm, candidateAnn); + for (Map.Entry entry : state.sourceFileToAnnSegment().entrySet()) { + DataFileMeta file = sourceFiles.get(entry.getKey()); + checkArgument( + file != null, + "ANN segment %s references inactive data file %s.", + entry.getValue().fileName(), + entry.getKey()); + PkVectorSourceMeta sourceMeta = sourceMeta(entry.getValue()); + for (PkVectorSourceFile source : sourceMeta.sourceFiles()) { + if (source.fileName().equals(entry.getKey())) { + checkArgument( + source.rowCount() == file.rowCount(), + "ANN source %s row count does not match its active data file.", + source.fileName()); + } + } + } + } + + public synchronized List segments() { + return Collections.unmodifiableList(new ArrayList<>(annSegments)); + } + + public synchronized PkVectorBucketIndexState state() { + return new PkVectorBucketIndexState(vectorFieldId, algorithm, annSegments); + } + + private void deleteAnnSegments(List segments) { + for (IndexFileMeta segment : segments) { + annSegmentFile.delete(segment); + } + } + + private static List eligibleFiles(List files) { + List eligible = new ArrayList<>(); + for (DataFileMeta file : files) { + if (PkVectorSourcePolicy.shouldRead(file)) { + eligible.add(file); + } + } + return eligible; + } + + private static boolean containsFile(List files, String fileName) { + for (DataFileMeta file : files) { + if (file.fileName().equals(fileName)) { + return true; + } + } + return false; + } + + private static boolean referencesAny(IndexFileMeta ann, Set dataFiles) { + for (PkVectorSourceFile source : sourceMeta(ann).sourceFiles()) { + if (dataFiles.contains(source.fileName())) { + return true; + } + } + return false; + } + + private static Set coveredSources(List segments) { + Set sources = new HashSet<>(); + for (IndexFileMeta ann : segments) { + for (PkVectorSourceFile source : sourceMeta(ann).sourceFiles()) { + sources.add(source.fileName()); + } + } + return sources; + } + + private static PkVectorSourceMeta sourceMeta(IndexFileMeta ann) { + return PkVectorSourceMeta.fromIndexFile(ann); + } + + /** Vector index changes for Paimon's append snapshot followed by its compact snapshot. */ + public static class VectorIndexCommit { + + private final Optional appendIncrement; + private final Optional compactIncrement; + + private VectorIndexCommit( + Optional appendIncrement, + Optional compactIncrement) { + this.appendIncrement = appendIncrement; + this.compactIncrement = compactIncrement; + } + + public Optional appendIncrement() { + return appendIncrement; + } + + public Optional compactIncrement() { + return compactIncrement; + } + } + + /** Index-file additions and deletions emitted by one bucket state update. */ + public static class VectorIndexIncrement { + + private final List newIndexFiles; + private final List deletedIndexFiles; + + private VectorIndexIncrement( + List newIndexFiles, List deletedIndexFiles) { + this.newIndexFiles = Collections.unmodifiableList(new ArrayList<>(newIndexFiles)); + this.deletedIndexFiles = + Collections.unmodifiableList(new ArrayList<>(deletedIndexFiles)); + } + + public List newIndexFiles() { + return newIndexFiles; + } + + public List deletedIndexFiles() { + return deletedIndexFiles; + } + } + + /** Factory to restore a bucket maintainer from data files and active ANN metadata. */ + public static class Factory { + + private final IndexFileHandler handler; + private final KeyValueFileReaderFactory.Builder readerFactoryBuilder; + private final DataField vectorField; + private final int vectorDimension; + private final String metric; + private final String algorithm; + private Options indexOptions; + + public Factory( + IndexFileHandler handler, + KeyValueFileReaderFactory.Builder readerFactoryBuilder, + DataField vectorField, + String metric, + String algorithm) { + this.handler = handler; + this.readerFactoryBuilder = readerFactoryBuilder; + this.vectorField = vectorField; + checkArgument( + vectorField.type() instanceof VectorType, + "Primary-key vector field must have VECTOR type."); + this.vectorDimension = ((VectorType) vectorField.type()).getLength(); + this.metric = metric; + this.algorithm = algorithm; + } + + public Factory withIndexOptions(Options indexOptions) { + this.indexOptions = new Options(indexOptions.toMap()); + return this; + } + + public IndexFileHandler indexFileHandler() { + return handler; + } + + public BucketedVectorIndexMaintainer create( + BinaryRow partition, + int bucket, + @Nullable List restoredDataFiles, + @Nullable List restoredPayloads) { + checkArgument(indexOptions != null, "ANN index options are not configured."); + List dataFiles = + restoredDataFiles == null ? Collections.emptyList() : restoredDataFiles; + List payloads = + restoredPayloads == null ? Collections.emptyList() : restoredPayloads; + return new BucketedVectorIndexMaintainer( + vectorField.id(), + handler.pkVectorAnnSegment(partition, bucket), + vectorField, + indexOptions, + metric, + algorithm, + new PkVectorDataFileReader.Factory( + readerFactoryBuilder, partition, bucket, vectorField, vectorDimension), + dataFiles, + payloads); + } + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorBucketIndexState.java b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorBucketIndexState.java new file mode 100644 index 000000000000..12eb770b0ebf --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorBucketIndexState.java @@ -0,0 +1,103 @@ +/* + * 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.paimon.index.pkvector; + +import org.apache.paimon.index.IndexFileMeta; + +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** Immutable primary-key vector-index state derived from one bucket's active payload metadata. */ +public final class PkVectorBucketIndexState { + + private final int vectorFieldId; + private final String indexType; + private final List annSegments; + private final Map sourceFileToAnnSegment; + + public static PkVectorBucketIndexState fromActivePayloads( + int vectorFieldId, String indexType, List activePayloads) { + return new PkVectorBucketIndexState(vectorFieldId, indexType, activePayloads); + } + + public PkVectorBucketIndexState( + int vectorFieldId, String indexType, List activePayloads) { + this.vectorFieldId = vectorFieldId; + this.indexType = indexType; + + Map payloadsByName = new LinkedHashMap<>(); + Map annBySource = new LinkedHashMap<>(); + for (IndexFileMeta payload : activePayloads) { + checkArgument( + payloadsByName.put(payload.fileName(), payload) == null, + "Active vector payload %s appears more than once in the index manifest.", + payload.fileName()); + PkVectorSourceMeta sourceMeta = validatedSourceMeta(payload); + for (PkVectorSourceFile sourceFile : sourceMeta.sourceFiles()) { + IndexFileMeta previous = annBySource.put(sourceFile.fileName(), payload); + checkArgument( + previous == null, + "Source data file %s is covered by both ANN segments %s and %s.", + sourceFile.fileName(), + previous == null ? "" : previous.fileName(), + payload.fileName()); + } + } + + this.annSegments = Collections.unmodifiableList(new java.util.ArrayList<>(activePayloads)); + this.sourceFileToAnnSegment = Collections.unmodifiableMap(annBySource); + } + + public int vectorFieldId() { + return vectorFieldId; + } + + public String indexType() { + return indexType; + } + + public List annSegments() { + return annSegments; + } + + public Map sourceFileToAnnSegment() { + return sourceFileToAnnSegment; + } + + private PkVectorSourceMeta validatedSourceMeta(IndexFileMeta payload) { + validateIdentity(payload); + return PkVectorSourceMeta.fromIndexFile(payload); + } + + private void validateIdentity(IndexFileMeta payload) { + checkArgument( + indexType.equals(payload.indexType()), + "Vector payload %s has a different index type.", + payload.fileName()); + checkArgument( + payload.globalIndexMeta() != null + && vectorFieldId == payload.globalIndexMeta().indexFieldId(), + "Vector payload %s has a different vector field.", + payload.fileName()); + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorDataFileReader.java b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorDataFileReader.java new file mode 100644 index 000000000000..e136fe7ef10f --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorDataFileReader.java @@ -0,0 +1,132 @@ +/* + * 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.paimon.index.pkvector; + +import org.apache.paimon.KeyValue; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.InternalArray; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.io.KeyValueFileReaderFactory; +import org.apache.paimon.reader.RecordReaderIterator; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.RowType; + +import java.io.IOException; + +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** Reads one projected vector value from every physical row of a data file. */ +public class PkVectorDataFileReader implements PkVectorReader { + + private final DataFileMeta dataFile; + private final int dimension; + private final RecordReaderIterator iterator; + private long rowsRead; + + public PkVectorDataFileReader( + KeyValueFileReaderFactory readerFactory, DataFileMeta dataFile, int dimension) + throws IOException { + checkArgument(dimension > 0, "Vector dimension must be positive."); + this.dataFile = dataFile; + this.dimension = dimension; + this.iterator = new RecordReaderIterator<>(readerFactory.createRecordReader(dataFile)); + } + + @Override + public int dimension() { + return dimension; + } + + @Override + public long rowCount() { + return dataFile.rowCount(); + } + + @Override + public boolean readNextVector(float[] reuse) { + checkArgument(reuse.length == dimension, "Vector buffer dimension does not match reader."); + checkArgument( + rowsRead < rowCount(), "Read past data file %s row count.", dataFile.fileName()); + checkArgument( + iterator.hasNext(), + "Data file %s ended before its declared row count.", + dataFile.fileName()); + KeyValue keyValue = iterator.next(); + rowsRead++; + if (rowsRead == rowCount()) { + checkArgument( + !iterator.hasNext(), + "Data file %s contains more rows than declared.", + dataFile.fileName()); + } + InternalRow value = keyValue.value(); + if (value.isNullAt(0)) { + return false; + } + InternalArray vector = value.getVector(0); + checkArgument( + vector.size() == dimension, + "Vector in data file %s has dimension %s instead of %s.", + dataFile.fileName(), + vector.size(), + dimension); + for (int i = 0; i < dimension; i++) { + reuse[i] = vector.getFloat(i); + } + return true; + } + + @Override + public void close() throws IOException { + try { + iterator.close(); + } catch (IOException e) { + throw e; + } catch (Exception e) { + throw new IOException("Failed to close vector reader for " + dataFile.fileName(), e); + } + } + + /** Bucket-scoped factory with vector-only value projection and no deletion-vector filtering. */ + public static class Factory { + + private final KeyValueFileReaderFactory readerFactory; + private final int dimension; + + public Factory( + KeyValueFileReaderFactory.Builder readerFactoryBuilder, + BinaryRow partition, + int bucket, + DataField vectorField, + int dimension) { + this.readerFactory = + readerFactoryBuilder + .copyWithoutProjection() + .withReadKeyType(RowType.of()) + .withReadValueType(RowType.of(vectorField)) + .buildWithoutDeletionVector(partition, bucket); + this.dimension = dimension; + } + + public PkVectorDataFileReader create(DataFileMeta dataFile) throws IOException { + return new PkVectorDataFileReader(readerFactory, dataFile, dimension); + } + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorSourcePolicy.java b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorSourcePolicy.java new file mode 100644 index 000000000000..9db854e11af3 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorSourcePolicy.java @@ -0,0 +1,38 @@ +/* + * 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.paimon.index.pkvector; + +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.manifest.FileSource; + +/** Selects data files whose complete rows can be used to build vector index payloads. */ +public final class PkVectorSourcePolicy { + + private PkVectorSourcePolicy() {} + + public static boolean shouldWrite(FileSource fileSource, int level) { + return fileSource == FileSource.COMPACT && level > 0; + } + + public static boolean shouldRead(DataFileMeta file) { + return file.fileSource() + .map(fileSource -> shouldWrite(fileSource, file.level())) + .orElse(false); + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileReaderFactory.java b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileReaderFactory.java index 17aa99866fa7..c594e4e0821d 100644 --- a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileReaderFactory.java +++ b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileReaderFactory.java @@ -309,6 +309,12 @@ public KeyValueFileReaderFactory build( return build(partition, bucket, dvFactory, true, Collections.emptyList()); } + /** Builds a reader which preserves every physical row position. */ + public KeyValueFileReaderFactory buildWithoutDeletionVector( + BinaryRow partition, int bucket) { + return build(partition, bucket, ignored -> Optional.empty()); + } + public KeyValueFileReaderFactory build( BinaryRow partition, int bucket, diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFileHandler.java b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFileHandler.java index f99278085550..c94bd3b3599e 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFileHandler.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFileHandler.java @@ -73,11 +73,13 @@ String write(@Nullable String previousIndexManifest, List ne indexes.addAll(previous.keySet()); indexes.addAll(current.keySet()); for (String indexName : indexes) { + List previousEntries = + previous.getOrDefault(indexName, Collections.emptyList()); + List currentEntries = + current.getOrDefault(indexName, Collections.emptyList()); indexEntries.addAll( - getIndexManifestFileCombine(indexName) - .combine( - previous.getOrDefault(indexName, Collections.emptyList()), - current.getOrDefault(indexName, Collections.emptyList()))); + getIndexManifestFileCombine(indexName, previousEntries, currentEntries) + .combine(previousEntries, currentEntries)); } return indexManifestFile.writeWithoutRolling(indexEntries); @@ -94,9 +96,15 @@ private Map> separateIndexEntries( return result; } - private IndexManifestFileCombiner getIndexManifestFileCombine(String indexType) { + private IndexManifestFileCombiner getIndexManifestFileCombine( + String indexType, + List previousEntries, + List currentEntries) { + if (hasSourceMeta(previousEntries) || hasSourceMeta(currentEntries)) { + return new GlobalIndexCombiner(false); + } if (!DELETION_VECTORS_INDEX.equals(indexType) && !HASH_INDEX.equals(indexType)) { - return new GlobalIndexCombiner(); + return new GlobalIndexCombiner(true); } if (DELETION_VECTORS_INDEX.equals(indexType) && BucketMode.BUCKET_UNAWARE == bucketMode) { @@ -106,6 +114,16 @@ private IndexManifestFileCombiner getIndexManifestFileCombine(String indexType) } } + private static boolean hasSourceMeta(List entries) { + for (IndexManifestEntry entry : entries) { + GlobalIndexMeta globalIndexMeta = entry.indexFile().globalIndexMeta(); + if (globalIndexMeta != null && globalIndexMeta.sourceMeta() != null) { + return true; + } + } + return false; + } + interface IndexManifestFileCombiner { List combine( List prevIndexFiles, List newIndexFiles); @@ -202,6 +220,12 @@ public List combine( /** We combine the previous and new index files by file name. */ static class GlobalIndexCombiner implements IndexManifestFileCombiner { + private final boolean validateRowRangeOverlap; + + private GlobalIndexCombiner(boolean validateRowRangeOverlap) { + this.validateRowRangeOverlap = validateRowRangeOverlap; + } + @Override public List combine( List prevIndexFiles, List newIndexFiles) { @@ -222,7 +246,9 @@ public List combine( for (IndexManifestEntry entry : removed) { indexEntries.remove(entry.indexFile().fileName()); } - validateRetainedIndexFiles(indexEntries.values(), added); + if (validateRowRangeOverlap) { + validateRetainedIndexFiles(indexEntries.values(), added); + } for (IndexManifestEntry entry : added) { indexEntries.put(entry.indexFile().fileName(), entry); } diff --git a/paimon-core/src/test/java/org/apache/paimon/index/pkvector/BucketedVectorIndexMaintainerTest.java b/paimon-core/src/test/java/org/apache/paimon/index/pkvector/BucketedVectorIndexMaintainerTest.java new file mode 100644 index 000000000000..31b44371516b --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/index/pkvector/BucketedVectorIndexMaintainerTest.java @@ -0,0 +1,233 @@ +/* + * 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.paimon.index.pkvector; + +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.index.IndexFileMeta; +import org.apache.paimon.index.IndexPathFactory; +import org.apache.paimon.io.CompactIncrement; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.io.DataIncrement; +import org.apache.paimon.manifest.FileSource; +import org.apache.paimon.options.Options; +import org.apache.paimon.stats.SimpleStats; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataTypes; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.IOException; +import java.util.Arrays; +import java.util.Collections; +import java.util.UUID; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** Tests ANN maintenance over compact data-file sources. */ +class BucketedVectorIndexMaintainerTest { + + @TempDir java.nio.file.Path tempPath; + + @Test + void testPartialCompactionRebuildsMultiSourceAnnFromDataFiles() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + PkVectorAnnSegmentFile annFile = new PkVectorAnnSegmentFile(fileIO, pathFactory()); + DataField vectorField = + new DataField(7, "embedding", DataTypes.VECTOR(2, DataTypes.FLOAT())); + Options options = indexOptions(); + DataFileMeta data1 = dataFile("data-1"); + DataFileMeta data2 = dataFile("data-2"); + DataFileMeta data3 = dataFile("data-3"); + IndexFileMeta initialAnn = + annFile.build( + Arrays.asList( + new PkVectorAnnSegmentFile.Source( + data1, new ArrayReader(new float[][] {{1, 0}})), + new PkVectorAnnSegmentFile.Source( + data2, new ArrayReader(new float[][] {{2, 0}}))), + vectorField, + options, + "l2", + "test-vector-ann"); + PkVectorDataFileReader.Factory readerFactory = mock(PkVectorDataFileReader.Factory.class); + PkVectorDataFileReader reader2 = reader(new float[][] {{2, 0}}); + PkVectorDataFileReader reader3 = reader(new float[][] {{3, 0}}); + when(readerFactory.create(data2)).thenReturn(reader2); + when(readerFactory.create(data3)).thenReturn(reader3); + BucketedVectorIndexMaintainer maintainer = + new BucketedVectorIndexMaintainer( + 7, + annFile, + vectorField, + options, + "l2", + "test-vector-ann", + readerFactory, + Arrays.asList(data1, data2), + Collections.singletonList(initialAnn)); + + BucketedVectorIndexMaintainer.VectorIndexCommit commit = + maintainer.prepareCommit( + DataIncrement.emptyIncrement(), + new CompactIncrement( + Collections.singletonList(data1), + Collections.singletonList(data3), + Collections.emptyList())); + + BucketedVectorIndexMaintainer.VectorIndexIncrement increment = + commit.compactIncrement().get(); + assertThat(increment.deletedIndexFiles()).containsExactly(initialAnn); + assertThat(increment.newIndexFiles()).hasSize(1); + IndexFileMeta repaired = increment.newIndexFiles().get(0); + assertThat(repaired.indexType()).isEqualTo("test-vector-ann"); + assertThat(PkVectorSourceMeta.fromIndexFile(repaired).sourceFiles()) + .extracting(PkVectorSourceFile::fileName) + .containsExactly("data-2", "data-3"); + } + + @Test + void testBuildsAnnForSingleCompactSource() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + PkVectorAnnSegmentFile annFile = new PkVectorAnnSegmentFile(fileIO, pathFactory()); + DataField vectorField = + new DataField(7, "embedding", DataTypes.VECTOR(2, DataTypes.FLOAT())); + DataFileMeta data = dataFile("data"); + PkVectorDataFileReader.Factory readerFactory = mock(PkVectorDataFileReader.Factory.class); + PkVectorDataFileReader dataReader = reader(new float[][] {{1, 0}}); + when(readerFactory.create(data)).thenReturn(dataReader); + BucketedVectorIndexMaintainer maintainer = + new BucketedVectorIndexMaintainer( + 7, + annFile, + vectorField, + indexOptions(), + "l2", + "test-vector-ann", + readerFactory, + Collections.emptyList(), + Collections.emptyList()); + + BucketedVectorIndexMaintainer.VectorIndexCommit commit = + maintainer.prepareCommit( + DataIncrement.emptyIncrement(), + new CompactIncrement( + Collections.emptyList(), + Collections.singletonList(data), + Collections.emptyList())); + + assertThat(commit.compactIncrement()).isPresent(); + assertThat(commit.compactIncrement().get().newIndexFiles()).hasSize(1); + } + + private static Options indexOptions() { + Options options = new Options(); + options.setString("test.vector.dimension", "2"); + options.setString("test.vector.metric", "l2"); + return options; + } + + private static PkVectorDataFileReader reader(float[][] vectors) throws IOException { + PkVectorDataFileReader reader = mock(PkVectorDataFileReader.class); + when(reader.dimension()).thenReturn(2); + when(reader.rowCount()).thenReturn((long) vectors.length); + final int[] position = {0}; + when(reader.readNextVector(org.mockito.ArgumentMatchers.any(float[].class))) + .thenAnswer( + invocation -> { + float[] vector = vectors[position[0]++]; + System.arraycopy( + vector, 0, invocation.getArgument(0), 0, vector.length); + return true; + }); + return reader; + } + + private static DataFileMeta dataFile(String fileName) { + return DataFileMeta.forAppend( + fileName, + 100, + 1, + SimpleStats.EMPTY_STATS, + 0, + 0, + 1, + Collections.emptyList(), + null, + FileSource.COMPACT, + null, + null, + null, + null) + .upgrade(1); + } + + private IndexPathFactory pathFactory() { + Path directory = new Path(tempPath.toUri()); + return new IndexPathFactory() { + @Override + public Path toPath(String fileName) { + return new Path(directory, fileName); + } + + @Override + public Path newPath() { + return new Path(directory, UUID.randomUUID().toString()); + } + + @Override + public boolean isExternalPath() { + return false; + } + }; + } + + private static class ArrayReader implements PkVectorReader { + + private final float[][] vectors; + private int position; + + private ArrayReader(float[][] vectors) { + this.vectors = vectors; + } + + @Override + public int dimension() { + return 2; + } + + @Override + public long rowCount() { + return vectors.length; + } + + @Override + public boolean readNextVector(float[] reuse) { + float[] vector = vectors[position++]; + System.arraycopy(vector, 0, reuse, 0, vector.length); + return true; + } + + @Override + public void close() throws IOException {} + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorBucketIndexStateTest.java b/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorBucketIndexStateTest.java new file mode 100644 index 000000000000..6fd35521fec7 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorBucketIndexStateTest.java @@ -0,0 +1,95 @@ +/* + * 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.paimon.index.pkvector; + +import org.apache.paimon.index.GlobalIndexMeta; +import org.apache.paimon.index.IndexFileMeta; + +import org.junit.jupiter.api.Test; + +import java.util.Collections; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link PkVectorBucketIndexState}. */ +class PkVectorBucketIndexStateTest { + + @Test + void testDerivesAnnCoverage() { + IndexFileMeta ann = segment("ann", "data-1", "test-vector-ann"); + + PkVectorBucketIndexState state = + PkVectorBucketIndexState.fromActivePayloads( + 7, "test-vector-ann", Collections.singletonList(ann)); + + assertThat(state.vectorFieldId()).isEqualTo(7); + assertThat(state.annSegments()).extracting(IndexFileMeta::fileName).containsExactly("ann"); + assertThat(state.sourceFileToAnnSegment()).containsOnlyKeys("data-1"); + } + + @Test + void testRejectsDuplicateAnnSource() { + assertThatThrownBy( + () -> + PkVectorBucketIndexState.fromActivePayloads( + 7, + "test-vector-ann", + java.util.Arrays.asList( + segment("ann-1", "data", "test-vector-ann"), + segment("ann-2", "data", "test-vector-ann")))) + .hasMessageContaining("both ANN segments"); + } + + @Test + void testRejectsDifferentIndexType() { + assertThatThrownBy( + () -> + PkVectorBucketIndexState.fromActivePayloads( + 7, + "test-vector-ann", + Collections.singletonList( + segment("ann", "data", "other-vector-ann")))) + .hasMessageContaining("different index type"); + } + + @Test + void testEmptyPayloadsProduceEmptyState() { + PkVectorBucketIndexState state = + PkVectorBucketIndexState.fromActivePayloads( + 7, "test-vector-ann", Collections.emptyList()); + + assertThat(state.vectorFieldId()).isEqualTo(7); + assertThat(state.annSegments()).isEmpty(); + } + + private static IndexFileMeta segment( + String segmentFileName, String sourceFileName, String indexType) { + PkVectorSourceFile sourceFile = new PkVectorSourceFile(sourceFileName, 10); + byte[] sourceMeta = + new PkVectorSourceMeta(Collections.singletonList(sourceFile)).serialize(); + return new IndexFileMeta( + indexType, + segmentFileName, + 100, + 10, + new GlobalIndexMeta(0, 10, 7, null, new byte[] {1}, sourceMeta), + null); + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorDataFileReaderTest.java b/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorDataFileReaderTest.java new file mode 100644 index 000000000000..420e0d5ed5c0 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorDataFileReaderTest.java @@ -0,0 +1,160 @@ +/* + * 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.paimon.index.pkvector; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.KeyValue; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.BinaryVector; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.format.FlushingFileFormat; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.io.KeyValueFileReaderFactory; +import org.apache.paimon.io.KeyValueFileWriterFactory; +import org.apache.paimon.io.RollingFileWriter; +import org.apache.paimon.manifest.FileSource; +import org.apache.paimon.options.Options; +import org.apache.paimon.schema.KeyValueFieldsExtractor; +import org.apache.paimon.schema.SchemaManager; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowKind; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.FileStorePathFactory; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import static org.apache.paimon.options.MemorySize.VALUE_128_MB; +import static org.apache.paimon.utils.FileStorePathFactoryTest.createNonPartFactory; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests reading physical vector positions from compact data files. */ +class PkVectorDataFileReaderTest { + + @TempDir java.nio.file.Path tempDir; + + @Test + void testReadsProjectedVectorsAndPreservesNullPositions() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + DataField keyField = new DataField(0, "id", DataTypes.INT()); + DataField vectorField = + new DataField(1, "embedding", DataTypes.VECTOR(2, DataTypes.FLOAT())); + DataField payloadField = new DataField(2, "payload", DataTypes.STRING()); + RowType keyType = RowType.of(keyField); + RowType valueType = RowType.of(vectorField, payloadField); + List fields = Arrays.asList(keyField, vectorField, payloadField); + TableSchema schema = + new TableSchema( + 0, + fields, + 2, + Collections.emptyList(), + Collections.singletonList("id"), + Collections.emptyMap(), + ""); + SchemaManager schemaManager = new SchemaManager(fileIO, tablePath); + assertThat(schemaManager.commit(schema)).isTrue(); + FileStorePathFactory pathFactory = createNonPartFactory(tablePath); + FlushingFileFormat format = new FlushingFileFormat("avro"); + CoreOptions options = new CoreOptions(new Options()); + + KeyValueFileWriterFactory writerFactory = + KeyValueFileWriterFactory.builder( + fileIO, + schema.id(), + keyType, + valueType, + format, + ignored -> pathFactory, + VALUE_128_MB.getBytes()) + .build(BinaryRow.EMPTY_ROW, 0, options); + RollingFileWriter writer = + writerFactory.createRollingMergeTreeFileWriter(1, FileSource.COMPACT); + try { + writer.write(keyValue(1, BinaryVector.fromPrimitiveArray(new float[] {1, 2}), "first")); + writer.write(keyValue(2, null, "second")); + writer.write(keyValue(3, BinaryVector.fromPrimitiveArray(new float[] {3, 4}), "third")); + } finally { + writer.close(); + } + DataFileMeta dataFile = writer.result().get(0); + + KeyValueFileReaderFactory.Builder readerFactoryBuilder = + KeyValueFileReaderFactory.builder( + fileIO, + schemaManager, + schema, + keyType, + valueType, + ignored -> format, + pathFactory, + fieldsExtractor(keyType, valueType), + options); + PkVectorDataFileReader reader = + new PkVectorDataFileReader.Factory( + readerFactoryBuilder, BinaryRow.EMPTY_ROW, 0, vectorField, 2) + .create(dataFile); + float[] reuse = new float[2]; + try { + assertThat(reader.rowCount()).isEqualTo(3); + assertThat(reader.readNextVector(reuse)).isTrue(); + assertThat(reuse).containsExactly(1, 2); + assertThat(reader.readNextVector(reuse)).isFalse(); + assertThat(reader.readNextVector(reuse)).isTrue(); + assertThat(reuse).containsExactly(3, 4); + assertThatThrownBy(() -> reader.readNextVector(reuse)) + .hasMessageContaining("Read past data file"); + } finally { + reader.close(); + } + } + + private static KeyValue keyValue(int key, BinaryVector vector, String payload) { + return new KeyValue() + .replace( + GenericRow.of(key), + RowKind.INSERT, + GenericRow.of(vector, BinaryString.fromString(payload))); + } + + private static KeyValueFieldsExtractor fieldsExtractor(RowType keyType, RowType valueType) { + return new KeyValueFieldsExtractor() { + @Override + public List keyFields(TableSchema schema) { + return keyType.getFields(); + } + + @Override + public List valueFields(TableSchema schema) { + return valueType.getFields(); + } + }; + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestFileHandlerTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestFileHandlerTest.java index 2eae202b306c..9456dd771b8a 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestFileHandlerTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestFileHandlerTest.java @@ -202,6 +202,29 @@ public void testGlobalIndexNonOverlappingRangeAllowedForSameFieldId() throws Exc assertThat(entries).containsExactlyInAnyOrder(previous, added); } + @Test + public void testPrimaryKeyVectorSegmentsAllowFileLocalOverlappingRanges() throws Exception { + TestAppendFileStore fileStore = + TestAppendFileStore.createAppendStore(tempDir, new HashMap<>()); + IndexManifestFile indexManifestFile = createIndexManifestFile(fileStore); + IndexManifestFileHandler handler = + new IndexManifestFileHandler(indexManifestFile, BucketMode.HASH_FIXED); + IndexManifestEntry first = pkVectorEntry("test-vector-ann", "ann-1"); + String firstManifest = handler.write(null, Arrays.asList(first)); + IndexManifestEntry second = pkVectorEntry("test-vector-ann", "ann-2"); + + String secondManifest = handler.write(firstManifest, Arrays.asList(second)); + + assertThat(indexManifestFile.read(secondManifest)).containsExactlyInAnyOrder(first, second); + + IndexManifestEntry ann = pkVectorEntry("test-vector-ann", "ann-3"); + String annManifest = + handler.write( + secondManifest, + Arrays.asList(first.toDeleteEntry(), second.toDeleteEntry(), ann)); + assertThat(indexManifestFile.read(annManifest)).containsExactly(ann); + } + private IndexManifestFile createIndexManifestFile(TestAppendFileStore fileStore) { return new IndexManifestFile.Factory( fileStore.fileIO(), @@ -226,4 +249,18 @@ private IndexManifestEntry globalIndexEntry( new GlobalIndexMeta(rowRangeStart, rowRangeEnd, indexFieldId, null, null), null)); } + + private IndexManifestEntry pkVectorEntry(String indexType, String fileName) { + return new IndexManifestEntry( + FileKind.ADD, + BinaryRow.EMPTY_ROW, + 0, + new IndexFileMeta( + indexType, + fileName, + 1L, + 1L, + new GlobalIndexMeta(0, 1, 1, null, null, new byte[] {1}), + null)); + } } From e946f8e9a68c9a6e5aef938cd415f927959981d0 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Sat, 11 Jul 2026 23:01:09 +0800 Subject: [PATCH 2/2] [core] Preserve global index range validation --- .../manifest/IndexManifestFileHandler.java | 41 ++++--------------- .../IndexManifestFileHandlerTest.java | 19 +++++++++ 2 files changed, 27 insertions(+), 33 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFileHandler.java b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFileHandler.java index c94bd3b3599e..486f39fef4da 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFileHandler.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFileHandler.java @@ -73,13 +73,11 @@ String write(@Nullable String previousIndexManifest, List ne indexes.addAll(previous.keySet()); indexes.addAll(current.keySet()); for (String indexName : indexes) { - List previousEntries = - previous.getOrDefault(indexName, Collections.emptyList()); - List currentEntries = - current.getOrDefault(indexName, Collections.emptyList()); indexEntries.addAll( - getIndexManifestFileCombine(indexName, previousEntries, currentEntries) - .combine(previousEntries, currentEntries)); + getIndexManifestFileCombine(indexName) + .combine( + previous.getOrDefault(indexName, Collections.emptyList()), + current.getOrDefault(indexName, Collections.emptyList()))); } return indexManifestFile.writeWithoutRolling(indexEntries); @@ -96,15 +94,9 @@ private Map> separateIndexEntries( return result; } - private IndexManifestFileCombiner getIndexManifestFileCombine( - String indexType, - List previousEntries, - List currentEntries) { - if (hasSourceMeta(previousEntries) || hasSourceMeta(currentEntries)) { - return new GlobalIndexCombiner(false); - } + private IndexManifestFileCombiner getIndexManifestFileCombine(String indexType) { if (!DELETION_VECTORS_INDEX.equals(indexType) && !HASH_INDEX.equals(indexType)) { - return new GlobalIndexCombiner(true); + return new GlobalIndexCombiner(); } if (DELETION_VECTORS_INDEX.equals(indexType) && BucketMode.BUCKET_UNAWARE == bucketMode) { @@ -114,16 +106,6 @@ private IndexManifestFileCombiner getIndexManifestFileCombine( } } - private static boolean hasSourceMeta(List entries) { - for (IndexManifestEntry entry : entries) { - GlobalIndexMeta globalIndexMeta = entry.indexFile().globalIndexMeta(); - if (globalIndexMeta != null && globalIndexMeta.sourceMeta() != null) { - return true; - } - } - return false; - } - interface IndexManifestFileCombiner { List combine( List prevIndexFiles, List newIndexFiles); @@ -220,12 +202,6 @@ public List combine( /** We combine the previous and new index files by file name. */ static class GlobalIndexCombiner implements IndexManifestFileCombiner { - private final boolean validateRowRangeOverlap; - - private GlobalIndexCombiner(boolean validateRowRangeOverlap) { - this.validateRowRangeOverlap = validateRowRangeOverlap; - } - @Override public List combine( List prevIndexFiles, List newIndexFiles) { @@ -246,9 +222,7 @@ public List combine( for (IndexManifestEntry entry : removed) { indexEntries.remove(entry.indexFile().fileName()); } - if (validateRowRangeOverlap) { - validateRetainedIndexFiles(indexEntries.values(), added); - } + validateRetainedIndexFiles(indexEntries.values(), added); for (IndexManifestEntry entry : added) { indexEntries.put(entry.indexFile().fileName(), entry); } @@ -267,6 +241,7 @@ private void validateRetainedIndexFiles( for (IndexManifestEntry added : addedIndexFiles) { GlobalIndexMeta addedMeta = added.indexFile().globalIndexMeta(); if (addedMeta == null + || (retainedMeta.sourceMeta() != null && addedMeta.sourceMeta() != null) || retainedMeta.indexFieldId() != addedMeta.indexFieldId() || (Arrays.equals( retainedMeta.extraFieldIds(), addedMeta.extraFieldIds()) diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestFileHandlerTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestFileHandlerTest.java index 9456dd771b8a..cfcc896525f3 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestFileHandlerTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestFileHandlerTest.java @@ -225,6 +225,25 @@ public void testPrimaryKeyVectorSegmentsAllowFileLocalOverlappingRanges() throws assertThat(indexManifestFile.read(annManifest)).containsExactly(ann); } + @Test + public void testSourceMetaDoesNotDisableRangeValidationForOtherFields() throws Exception { + TestAppendFileStore fileStore = + TestAppendFileStore.createAppendStore(tempDir, new HashMap<>()); + IndexManifestFile indexManifestFile = createIndexManifestFile(fileStore); + IndexManifestFileHandler handler = + new IndexManifestFileHandler(indexManifestFile, BucketMode.HASH_FIXED); + IndexManifestEntry sourceBacked = pkVectorEntry("btree", "source-backed"); + IndexManifestEntry previousRange = globalIndexEntry("previous-range", 0, 99, 2); + String manifest = handler.write(null, Arrays.asList(sourceBacked, previousRange)); + IndexManifestEntry overlappingRange = globalIndexEntry("overlapping-range", 50, 149, 2); + + assertThatThrownBy(() -> handler.write(manifest, Arrays.asList(overlappingRange))) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("previous-range") + .hasMessageContaining("overlapping-range") + .hasMessageContaining("overlapping row range"); + } + private IndexManifestFile createIndexManifestFile(TestAppendFileStore fileStore) { return new IndexManifestFile.Factory( fileStore.fileIO(),