This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 50b0f70452 [core] Add bucket-level primary-key vector index
maintenance (#8562)
50b0f70452 is described below
commit 50b0f7045265cb1968a89b0d3b8400b0f6706b61
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Jul 11 23:27:45 2026 +0800
[core] Add bucket-level primary-key vector index maintenance (#8562)
Builds the bucket-local maintenance layer for primary-key vector indexes
on top of #8549. The maintainer restores active ANN state, reads
complete vectors from compact data files, and produces index-file
increments for compaction transitions without activating the write or
query paths yet.
---
.../org/apache/paimon/index/IndexFileHandler.java | 5 +
.../pkvector/BucketedVectorIndexMaintainer.java | 353 +++++++++++++++++++++
.../index/pkvector/PkVectorBucketIndexState.java | 103 ++++++
.../index/pkvector/PkVectorDataFileReader.java | 132 ++++++++
.../index/pkvector/PkVectorSourcePolicy.java | 38 +++
.../paimon/io/KeyValueFileReaderFactory.java | 6 +
.../paimon/manifest/IndexManifestFileHandler.java | 1 +
.../BucketedVectorIndexMaintainerTest.java | 233 ++++++++++++++
.../pkvector/PkVectorBucketIndexStateTest.java | 95 ++++++
.../index/pkvector/PkVectorDataFileReaderTest.java | 160 ++++++++++
.../manifest/IndexManifestFileHandlerTest.java | 56 ++++
11 files changed, 1182 insertions(+)
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 e848ed205e..92909a618d 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.DeletionVector;
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 class IndexFileHandler {
fileIO, pathFactories.get(partition, bucket),
dvTargetFileSize, dvBitmap64);
}
+ public PkVectorAnnSegmentFile pkVectorAnnSegment(BinaryRow partition, int
bucket) {
+ return new PkVectorAnnSegmentFile(fileIO, pathFactories.get(partition,
bucket));
+ }
+
public Optional<IndexFileMeta> scanHashIndex(
Snapshot snapshot, BinaryRow partition, int bucket) {
List<IndexFileMeta> 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 0000000000..56ba5f9cfc
--- /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<IndexFileMeta> annSegments;
+ private final Map<String, DataFileMeta> activeSourceFiles;
+
+ BucketedVectorIndexMaintainer(
+ int vectorFieldId,
+ PkVectorAnnSegmentFile annSegmentFile,
+ DataField vectorField,
+ Options indexOptions,
+ String metric,
+ String algorithm,
+ PkVectorDataFileReader.Factory vectorReaderFactory,
+ List<DataFileMeta> restoredDataFiles,
+ List<IndexFileMeta> 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<String, DataFileMeta> nextSources = new
LinkedHashMap<>(activeSourceFiles);
+ Set<String> 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<IndexFileMeta> retained = new ArrayList<>();
+ List<IndexFileMeta> removed = new ArrayList<>();
+ for (IndexFileMeta ann : annSegments) {
+ if (referencesAny(ann, deletedFiles)) {
+ removed.add(ann);
+ } else {
+ retained.add(ann);
+ }
+ }
+
+ Set<String> covered = coveredSources(retained);
+ List<DataFileMeta> uncovered = new ArrayList<>();
+ for (DataFileMeta file : nextSources.values()) {
+ if (!covered.contains(file.fileName())) {
+ uncovered.add(file);
+ }
+ }
+ uncovered.sort(Comparator.comparing(DataFileMeta::fileName));
+
+ List<IndexFileMeta> created = new ArrayList<>();
+ try {
+ if (!uncovered.isEmpty()) {
+ created.add(buildAnnSegment(uncovered));
+ }
+ List<IndexFileMeta> nextAnn = new ArrayList<>(retained);
+ nextAnn.addAll(created);
+ validateCoverage(nextAnn, nextSources);
+
+ activeSourceFiles.clear();
+ activeSourceFiles.putAll(nextSources);
+ annSegments.clear();
+ annSegments.addAll(nextAnn);
+ Optional<VectorIndexIncrement> 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<DataFileMeta> files) {
+ try {
+ List<PkVectorAnnSegmentFile.Source> 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<IndexFileMeta> candidateAnn, Map<String, DataFileMeta>
sourceFiles) {
+ PkVectorBucketIndexState state =
+ new PkVectorBucketIndexState(vectorFieldId, algorithm,
candidateAnn);
+ for (Map.Entry<String, IndexFileMeta> 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<IndexFileMeta> segments() {
+ return Collections.unmodifiableList(new ArrayList<>(annSegments));
+ }
+
+ public synchronized PkVectorBucketIndexState state() {
+ return new PkVectorBucketIndexState(vectorFieldId, algorithm,
annSegments);
+ }
+
+ private void deleteAnnSegments(List<IndexFileMeta> segments) {
+ for (IndexFileMeta segment : segments) {
+ annSegmentFile.delete(segment);
+ }
+ }
+
+ private static List<DataFileMeta> eligibleFiles(List<DataFileMeta> files) {
+ List<DataFileMeta> eligible = new ArrayList<>();
+ for (DataFileMeta file : files) {
+ if (PkVectorSourcePolicy.shouldRead(file)) {
+ eligible.add(file);
+ }
+ }
+ return eligible;
+ }
+
+ private static boolean containsFile(List<DataFileMeta> files, String
fileName) {
+ for (DataFileMeta file : files) {
+ if (file.fileName().equals(fileName)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static boolean referencesAny(IndexFileMeta ann, Set<String>
dataFiles) {
+ for (PkVectorSourceFile source : sourceMeta(ann).sourceFiles()) {
+ if (dataFiles.contains(source.fileName())) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static Set<String> coveredSources(List<IndexFileMeta> segments) {
+ Set<String> 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<VectorIndexIncrement> appendIncrement;
+ private final Optional<VectorIndexIncrement> compactIncrement;
+
+ private VectorIndexCommit(
+ Optional<VectorIndexIncrement> appendIncrement,
+ Optional<VectorIndexIncrement> compactIncrement) {
+ this.appendIncrement = appendIncrement;
+ this.compactIncrement = compactIncrement;
+ }
+
+ public Optional<VectorIndexIncrement> appendIncrement() {
+ return appendIncrement;
+ }
+
+ public Optional<VectorIndexIncrement> compactIncrement() {
+ return compactIncrement;
+ }
+ }
+
+ /** Index-file additions and deletions emitted by one bucket state update.
*/
+ public static class VectorIndexIncrement {
+
+ private final List<IndexFileMeta> newIndexFiles;
+ private final List<IndexFileMeta> deletedIndexFiles;
+
+ private VectorIndexIncrement(
+ List<IndexFileMeta> newIndexFiles, List<IndexFileMeta>
deletedIndexFiles) {
+ this.newIndexFiles = Collections.unmodifiableList(new
ArrayList<>(newIndexFiles));
+ this.deletedIndexFiles =
+ Collections.unmodifiableList(new
ArrayList<>(deletedIndexFiles));
+ }
+
+ public List<IndexFileMeta> newIndexFiles() {
+ return newIndexFiles;
+ }
+
+ public List<IndexFileMeta> 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<DataFileMeta> restoredDataFiles,
+ @Nullable List<IndexFileMeta> restoredPayloads) {
+ checkArgument(indexOptions != null, "ANN index options are not
configured.");
+ List<DataFileMeta> dataFiles =
+ restoredDataFiles == null ? Collections.emptyList() :
restoredDataFiles;
+ List<IndexFileMeta> 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 0000000000..12eb770b0e
--- /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<IndexFileMeta> annSegments;
+ private final Map<String, IndexFileMeta> sourceFileToAnnSegment;
+
+ public static PkVectorBucketIndexState fromActivePayloads(
+ int vectorFieldId, String indexType, List<IndexFileMeta>
activePayloads) {
+ return new PkVectorBucketIndexState(vectorFieldId, indexType,
activePayloads);
+ }
+
+ public PkVectorBucketIndexState(
+ int vectorFieldId, String indexType, List<IndexFileMeta>
activePayloads) {
+ this.vectorFieldId = vectorFieldId;
+ this.indexType = indexType;
+
+ Map<String, IndexFileMeta> payloadsByName = new LinkedHashMap<>();
+ Map<String, IndexFileMeta> 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<IndexFileMeta> annSegments() {
+ return annSegments;
+ }
+
+ public Map<String, IndexFileMeta> 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 0000000000..e136fe7ef1
--- /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<KeyValue> 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 0000000000..9db854e11a
--- /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 17aa99866f..c594e4e082 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 class KeyValueFileReaderFactory implements
FileReaderFactory<KeyValue> {
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 f992780855..486f39fef4 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
@@ -241,6 +241,7 @@ public class IndexManifestFileHandler {
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/index/pkvector/BucketedVectorIndexMaintainerTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/pkvector/BucketedVectorIndexMaintainerTest.java
new file mode 100644
index 0000000000..31b4437151
--- /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 0000000000..6fd35521fe
--- /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 0000000000..420e0d5ed5
--- /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<DataField> 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<KeyValue, DataFileMeta> 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<DataField> keyFields(TableSchema schema) {
+ return keyType.getFields();
+ }
+
+ @Override
+ public List<DataField> 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 2eae202b30..cfcc896525 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,48 @@ public class IndexManifestFileHandlerTest {
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);
+ }
+
+ @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(),
@@ -226,4 +268,18 @@ public class IndexManifestFileHandlerTest {
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));
+ }
}
