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 21d8e0ca59 [core] Add primary-key vector index foundation (#8549)
21d8e0ca59 is described below
commit 21d8e0ca59ed29e3551729d4c3dbb74f06ae756a
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Jul 11 22:39:29 2026 +0800
[core] Add primary-key vector index foundation (#8549)
Introduce the foundation for bucket-local vector indexes on primary-key
tables. This PR defines the table-option and validation contract, ANN
payload metadata, and ANN build/search primitives used by later write,
compaction, manifest, and query integrations.
---
docs/generated/core_configuration.html | 6 +
.../main/java/org/apache/paimon/CoreOptions.java | 116 +++++++++
.../testvector/TestVectorGlobalIndexer.java | 19 +-
.../testvector/TestVectorGlobalIndexerTest.java | 41 ++++
.../org/apache/paimon/index/GlobalIndexMeta.java | 22 +-
.../paimon/index/IndexFileMetaSerializer.java | 16 +-
.../index/pkvector/PkVectorAnnSegmentFile.java | 269 ++++++++++++++++++++
.../index/pkvector/PkVectorAnnSegmentSearcher.java | 272 +++++++++++++++++++++
.../paimon/index/pkvector/PkVectorReader.java | 33 +++
.../paimon/index/pkvector/PkVectorSourceFile.java | 61 +++++
.../paimon/index/pkvector/PkVectorSourceMeta.java | 91 +++++++
.../manifest/IndexManifestEntrySerializer.java | 16 +-
.../org/apache/paimon/schema/SchemaManager.java | 14 ++
.../org/apache/paimon/schema/SchemaValidation.java | 71 ++++++
.../paimon/index/IndexFileMetaSerializerTest.java | 23 ++
.../index/pkvector/PkVectorAnnSegmentFileTest.java | 237 ++++++++++++++++++
.../index/pkvector/PkVectorSourceMetaTest.java | 56 +++++
.../pkvector/PrimaryKeyVectorIndexOptionsTest.java | 157 ++++++++++++
.../manifest/IndexManifestEntrySerializerTest.java | 38 +++
.../PrimaryKeyVectorIndexValidationTest.java | 236 ++++++++++++++++++
.../apache/paimon/schema/SchemaManagerTest.java | 29 +++
21 files changed, 1812 insertions(+), 11 deletions(-)
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index 401d451b86..d22fb24cdf 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1205,6 +1205,12 @@ This config option does not affect the default
filesystem metastore.</td>
<td>Boolean</td>
<td>Enables clustering by non-primary key fields. When set to
true, the physical sort order of data files is determined by the configured
'clustering.columns' instead of the primary key, optimizing query performance
for non-PK columns.</td>
</tr>
+ <tr>
+ <td><h5>pk-vector.index.columns</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Comma-separated VECTOR columns indexed by primary-key vector
indexes. Each column owns one index and must define
fields.<column>.pk-vector.index.type. Index options and distance metric
are also field-scoped. The first release supports exactly one column.</td>
+ </tr>
<tr>
<td><h5>postpone.batch-write-fixed-bucket</h5></td>
<td style="word-wrap: break-word;">true</td>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index b353b56327..0dd64d8d25 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -33,6 +33,7 @@ import org.apache.paimon.options.Options;
import org.apache.paimon.options.description.DescribedEnum;
import org.apache.paimon.options.description.Description;
import org.apache.paimon.options.description.InlineElement;
+import org.apache.paimon.utils.JsonSerdeUtil;
import org.apache.paimon.utils.MathUtils;
import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.StringUtils;
@@ -48,11 +49,13 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.TreeMap;
import java.util.UUID;
import java.util.stream.Collectors;
@@ -2726,6 +2729,17 @@ public class CoreOptions implements Serializable {
"The batch size for lateral vector search. Each
batch executes vector "
+ "topK search and table lookup for
multiple query vectors.");
+ public static final ConfigOption<String> PK_VECTOR_INDEX_COLUMNS =
+ key("pk-vector.index.columns")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Comma-separated VECTOR columns indexed by
primary-key vector indexes. "
+ + "Each column owns one index and must
define "
+ + "fields.<column>.pk-vector.index.type.
Index options and distance "
+ + "metric are also field-scoped. The first
release supports exactly "
+ + "one column.");
+
@Immutable
public static final ConfigOption<Boolean> PK_CLUSTERING_OVERRIDE =
key("pk-clustering-override")
@@ -4254,6 +4268,108 @@ public class CoreOptions implements Serializable {
return options.get(VECTOR_SEARCH_LATERAL_JOIN_BATCH_SIZE);
}
+ public boolean primaryKeyVectorIndexEnabled() {
+ return options.getOptional(PK_VECTOR_INDEX_COLUMNS).isPresent();
+ }
+
+ public List<String> primaryKeyVectorIndexColumns() {
+ String columns = options.get(PK_VECTOR_INDEX_COLUMNS);
+ if (columns == null) {
+ return Collections.emptyList();
+ }
+ return Arrays.stream(columns.split(",",
-1)).map(String::trim).collect(Collectors.toList());
+ }
+
+ public String primaryKeyVectorIndexColumn() {
+ List<String> columns = primaryKeyVectorIndexColumns();
+ checkArgument(
+ columns.size() == 1,
+ "pk-vector.index.columns must contain exactly one column in
the first release, but is %s.",
+ columns);
+ return columns.get(0);
+ }
+
+ @Nullable
+ public String primaryKeyVectorIndexType(String column) {
+ return options.get("fields." + column + ".pk-vector.index.type");
+ }
+
+ @Nullable
+ private String primaryKeyVectorIndexOptionsJson(String column) {
+ return options.get("fields." + column + ".pk-vector.index.options");
+ }
+
+ public Options primaryKeyVectorIndexOptions(String column) {
+ Options resolved = new Options(toConfiguration().toMap());
+ for (Map.Entry<String, String> option :
+ primaryKeyVectorAlgorithmOptions(column).entrySet()) {
+ resolved.setString(option.getKey(), option.getValue());
+ }
+ return resolved;
+ }
+
+ private Map<String, String> primaryKeyVectorAlgorithmOptions(String
column) {
+ String indexTypeKey = "fields." + column + ".pk-vector.index.type";
+ String indexOptionsKey = "fields." + column +
".pk-vector.index.options";
+ String algorithm = primaryKeyVectorIndexType(column);
+ checkArgument(
+ algorithm != null && !algorithm.trim().isEmpty(),
+ "%s must be configured before resolving index options.",
+ indexTypeKey);
+ TreeMap<String, String> algorithmOptions = new TreeMap<>();
+ String algorithmPrefix = algorithm + ".";
+ String fieldPrefix = "fields." + column + ".";
+ for (Map.Entry<String, String> entry :
toConfiguration().toMap().entrySet()) {
+ if (entry.getKey().startsWith(algorithmPrefix)
+ || (entry.getKey().startsWith(fieldPrefix)
+ && !entry.getKey().startsWith(fieldPrefix +
"pk-vector."))) {
+ algorithmOptions.put(entry.getKey(), entry.getValue());
+ }
+ }
+ String serialized = primaryKeyVectorIndexOptionsJson(column);
+ if (serialized != null && !serialized.trim().isEmpty()) {
+ LinkedHashMap<String, String> parsed;
+ try {
+ parsed = JsonSerdeUtil.parseJsonMap(serialized, String.class);
+ } catch (RuntimeException e) {
+ throw new IllegalArgumentException(
+ indexOptionsKey + " must be a JSON object of option
key-value pairs.", e);
+ }
+ for (Map.Entry<String, String> entry : parsed.entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+ checkArgument(
+ key != null && !key.trim().isEmpty(),
+ "%s contains an empty option key.",
+ indexOptionsKey);
+ checkArgument(
+ value != null,
+ "%s value for key %s must not be null.",
+ indexOptionsKey,
+ key);
+ String qualifiedKey =
+ key.startsWith(algorithmPrefix) ||
key.startsWith("fields.")
+ ? key
+ : algorithmPrefix + key;
+ String previous = algorithmOptions.put(qualifiedKey, value);
+ checkArgument(
+ previous == null || previous.equals(value),
+ "%s defines conflicting values for %s.",
+ indexOptionsKey,
+ qualifiedKey);
+ }
+ }
+ algorithmOptions.put(algorithmPrefix + "metric",
primaryKeyVectorDistanceMetric(column));
+ return algorithmOptions;
+ }
+
+ public String primaryKeyVectorDistanceMetric(String column) {
+ String metric = options.get("fields." + column +
".pk-vector.distance.metric");
+ return (metric == null ? "inner_product" : metric)
+ .toLowerCase(Locale.ROOT)
+ .replace('-', '_');
+ }
+
/** Specifies the merge engine for table with primary key. */
public enum MergeEngine implements DescribedEnum {
DEDUPLICATE("deduplicate", "De-duplicate and keep the last row."),
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestVectorGlobalIndexer.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestVectorGlobalIndexer.java
index 0b6d8d9fd6..eb0365427a 100644
---
a/paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestVectorGlobalIndexer.java
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestVectorGlobalIndexer.java
@@ -28,6 +28,7 @@ import org.apache.paimon.options.Options;
import org.apache.paimon.types.ArrayType;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.FloatType;
+import org.apache.paimon.types.VectorType;
import java.io.IOException;
import java.util.List;
@@ -74,17 +75,27 @@ public class TestVectorGlobalIndexer implements
VectorGlobalIndexer {
public TestVectorGlobalIndexer(DataType fieldType, Options options) {
checkArgument(
- fieldType instanceof ArrayType
- && ((ArrayType) fieldType).getElementType() instanceof
FloatType,
- "TestVectorGlobalIndexer only supports ARRAY<FLOAT>, but got:
" + fieldType);
+ isFloatVector(fieldType),
+ "TestVectorGlobalIndexer only supports VECTOR<FLOAT> or
ARRAY<FLOAT>, but got: "
+ + fieldType);
this.fieldType = fieldType;
- this.dimension = options.getInteger(OPT_DIMENSION, 0);
+ this.dimension =
+ fieldType instanceof VectorType
+ ? ((VectorType) fieldType).getLength()
+ : options.getInteger(OPT_DIMENSION, 0);
this.metric = options.getString(OPT_METRIC, "l2");
this.reverseScore = options.getBoolean(OPT_REVERSE_SCORE, false);
this.requiredOptionKey = options.getString(OPT_REQUIRED_OPTION_KEY,
null);
this.requiredOptionValue =
options.getString(OPT_REQUIRED_OPTION_VALUE, null);
}
+ private static boolean isFloatVector(DataType fieldType) {
+ return (fieldType instanceof VectorType
+ && ((VectorType) fieldType).getElementType()
instanceof FloatType)
+ || (fieldType instanceof ArrayType
+ && ((ArrayType) fieldType).getElementType() instanceof
FloatType);
+ }
+
@Override
public GlobalIndexWriter createWriter(GlobalIndexFileWriter fileWriter)
throws IOException {
return new TestVectorGlobalIndexWriter(fileWriter, dimension);
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestVectorGlobalIndexerTest.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestVectorGlobalIndexerTest.java
new file mode 100644
index 0000000000..7ab5c45cef
--- /dev/null
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestVectorGlobalIndexerTest.java
@@ -0,0 +1,41 @@
+/*
+ * 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.globalindex.testvector;
+
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link TestVectorGlobalIndexer}. */
+class TestVectorGlobalIndexerTest {
+
+ @Test
+ void testUsesDimensionFromVectorType() {
+ Options options = new Options();
+ options.setString(TestVectorGlobalIndexer.OPT_DIMENSION, "99");
+
+ TestVectorGlobalIndexer indexer =
+ new TestVectorGlobalIndexer(DataTypes.VECTOR(2,
DataTypes.FLOAT()), options);
+
+ assertThat(indexer.dimension()).isEqualTo(2);
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/GlobalIndexMeta.java
b/paimon-core/src/main/java/org/apache/paimon/index/GlobalIndexMeta.java
index 83a6224f3e..727a6536e3 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/GlobalIndexMeta.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/GlobalIndexMeta.java
@@ -43,13 +43,15 @@ public class GlobalIndexMeta {
new DataField(2, "_INDEX_FIELD_ID", new
IntType(false)),
new DataField(
3, "_EXTRA_FIELD_IDS", DataTypes.ARRAY(new
IntType(false))),
- new DataField(4, "_INDEX_META",
DataTypes.BYTES())));
+ new DataField(4, "_INDEX_META", DataTypes.BYTES()),
+ new DataField(5, "_SOURCE_META",
DataTypes.BYTES())));
private final long rowRangeStart;
private final long rowRangeEnd;
private final int indexFieldId;
@Nullable private final int[] extraFieldIds;
@Nullable private final byte[] indexMeta;
+ @Nullable private final byte[] sourceMeta;
public GlobalIndexMeta(
long rowRangeStart,
@@ -57,11 +59,22 @@ public class GlobalIndexMeta {
int indexFieldId,
@Nullable int[] extraFieldIds,
@Nullable byte[] indexMeta) {
+ this(rowRangeStart, rowRangeEnd, indexFieldId, extraFieldIds,
indexMeta, null);
+ }
+
+ public GlobalIndexMeta(
+ long rowRangeStart,
+ long rowRangeEnd,
+ int indexFieldId,
+ @Nullable int[] extraFieldIds,
+ @Nullable byte[] indexMeta,
+ @Nullable byte[] sourceMeta) {
this.rowRangeStart = rowRangeStart;
this.rowRangeEnd = rowRangeEnd;
this.indexFieldId = indexFieldId;
this.extraFieldIds = extraFieldIds;
this.indexMeta = indexMeta;
+ this.sourceMeta = sourceMeta;
}
public long rowRangeStart() {
@@ -85,11 +98,18 @@ public class GlobalIndexMeta {
return extraFieldIds;
}
+ /** Metadata produced and consumed by the global-index implementation. */
@Nullable
public byte[] indexMeta() {
return indexMeta;
}
+ /** Metadata describing how index row ids map to their source data. */
+ @Nullable
+ public byte[] sourceMeta() {
+ return sourceMeta;
+ }
+
/** All indexed field ids in order: the primary {@link #indexFieldId}
followed by the rest. */
public List<Integer> getIndexedFieldIds() {
List<Integer> ids = new ArrayList<>();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/IndexFileMetaSerializer.java
b/paimon-core/src/main/java/org/apache/paimon/index/IndexFileMetaSerializer.java
index 6d98e61248..a2acfd0ffc 100644
---
a/paimon-core/src/main/java/org/apache/paimon/index/IndexFileMetaSerializer.java
+++
b/paimon-core/src/main/java/org/apache/paimon/index/IndexFileMetaSerializer.java
@@ -50,7 +50,8 @@ public class IndexFileMetaSerializer extends
ObjectSerializer<IndexFileMeta> {
globalIndexMeta.extraFieldIds() == null
? null
: new
GenericArray(globalIndexMeta.extraFieldIds()),
- globalIndexMeta.indexMeta());
+ globalIndexMeta.indexMeta(),
+ globalIndexMeta.sourceMeta());
return GenericRow.of(
fromString(record.indexType()),
fromString(record.fileName()),
@@ -65,16 +66,25 @@ public class IndexFileMetaSerializer extends
ObjectSerializer<IndexFileMeta> {
public IndexFileMeta fromRow(InternalRow row) {
GlobalIndexMeta globalIndexMeta = null;
if (!row.isNullAt(6)) {
- InternalRow globalIndexRow = row.getRow(6, 5);
+ InternalRow globalIndexRow = row.getRow(6, 6);
Long rowRangeStart = globalIndexRow.getLong(0);
Long rowRangeEnd = globalIndexRow.getLong(1);
Integer indexFieldId = globalIndexRow.getInt(2);
int[] extralFields =
globalIndexRow.isNullAt(3) ? null :
globalIndexRow.getArray(3).toIntArray();
byte[] indexMeta = globalIndexRow.isNullAt(4) ? null :
globalIndexRow.getBinary(4);
+ byte[] sourceMeta =
+ globalIndexRow.getFieldCount() <= 5 ||
globalIndexRow.isNullAt(5)
+ ? null
+ : globalIndexRow.getBinary(5);
globalIndexMeta =
new GlobalIndexMeta(
- rowRangeStart, rowRangeEnd, indexFieldId,
extralFields, indexMeta);
+ rowRangeStart,
+ rowRangeEnd,
+ indexFieldId,
+ extralFields,
+ indexMeta,
+ sourceMeta);
}
return new IndexFileMeta(
row.getString(0).toString(),
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentFile.java
b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentFile.java
new file mode 100644
index 0000000000..9a3306734e
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentFile.java
@@ -0,0 +1,269 @@
+/*
+ * 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.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
+import org.apache.paimon.globalindex.GlobalIndexWriter;
+import org.apache.paimon.globalindex.GlobalIndexer;
+import org.apache.paimon.globalindex.ResultEntry;
+import org.apache.paimon.globalindex.VectorGlobalIndexer;
+import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFile;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.IndexPathFactory;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.VectorType;
+import org.apache.paimon.utils.IOUtils;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.function.LongPredicate;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Builds immutable ANN payloads whose index ids are source data-file row
positions. */
+public class PkVectorAnnSegmentFile extends IndexFile {
+
+ public PkVectorAnnSegmentFile(FileIO fileIO, IndexPathFactory pathFactory)
{
+ super(fileIO, pathFactory);
+ }
+
+ public IndexFileMeta build(
+ List<Source> sources,
+ DataField vectorField,
+ Options indexOptions,
+ String metric,
+ String indexType)
+ throws IOException {
+ checkArgument(!sources.isEmpty(), "An ANN segment must reference
source files.");
+ long totalRowCount = 0;
+ List<PkVectorSourceFile> sourceFiles = new ArrayList<>(sources.size());
+ for (Source source : sources) {
+ totalRowCount = Math.addExact(totalRowCount,
source.sourceFile.rowCount());
+ sourceFiles.add(source.sourceFile);
+ }
+ checkArgument(totalRowCount > 0, "An ANN segment must reference at
least one source row.");
+
+ GlobalIndexer indexer = GlobalIndexer.create(indexType, vectorField,
indexOptions);
+ checkArgument(
+ indexer instanceof VectorGlobalIndexer,
+ "Index algorithm %s does not implement VectorGlobalIndexer.",
+ indexType);
+ String indexerMetric = normalizeMetric(((VectorGlobalIndexer)
indexer).metric());
+ checkArgument(
+ normalizeMetric(metric).equals(indexerMetric),
+ "Configured metric %s does not match index algorithm metric
%s.",
+ metric,
+ indexerMetric);
+
+ SegmentFileWriter fileWriter = new SegmentFileWriter();
+ GlobalIndexWriter writer = null;
+ boolean success = false;
+ try {
+ writer = indexer.createWriter(fileWriter);
+ checkArgument(
+ writer instanceof GlobalIndexSingleColumnWriter,
+ "Index algorithm %s does not create a single-column
writer.",
+ indexType);
+ GlobalIndexSingleColumnWriter singleColumnWriter =
+ (GlobalIndexSingleColumnWriter) writer;
+ long liveRowCount = 0;
+ long fileOffset = 0;
+ int dimension = -1;
+ for (Source source : sources) {
+ PkVectorReader vectors = source.openReader();
+ try {
+ checkArgument(
+ vectors.rowCount() == source.sourceFile.rowCount(),
+ "Vector row count %s does not match source file %s
row count %s.",
+ vectors.rowCount(),
+ source.sourceFile.fileName(),
+ source.sourceFile.rowCount());
+ if (dimension < 0) {
+ dimension = vectors.dimension();
+ }
+ checkArgument(
+ vectors.dimension() == dimension,
+ "Vector source %s dimension %s does not match
dimension %s.",
+ source.sourceFile.fileName(),
+ vectors.dimension(),
+ dimension);
+ if (vectorField.type() instanceof VectorType) {
+ checkArgument(
+ ((VectorType) vectorField.type()).getLength()
== dimension,
+ "Vector field dimension %s does not match
source vector dimension %s.",
+ ((VectorType) vectorField.type()).getLength(),
+ dimension);
+ }
+
+ float[] vector = new float[dimension];
+ for (long rowPosition = 0; rowPosition <
vectors.rowCount(); rowPosition++) {
+ boolean present = vectors.readNextVector(vector);
+ if (!present ||
source.excludedPosition.test(rowPosition)) {
+ continue;
+ }
+ singleColumnWriter.write(vector, fileOffset +
rowPosition);
+ liveRowCount++;
+ }
+ } finally {
+ source.closeReader(vectors);
+ }
+ fileOffset += source.sourceFile.rowCount();
+ }
+
+ List<ResultEntry> results = writer.finish();
+ checkArgument(
+ results.size() == 1,
+ "ANN segment build must produce exactly one payload file,
but produced %s.",
+ results.size());
+ ResultEntry result = results.get(0);
+ Path payloadPath = fileWriter.path(result.fileName());
+ byte[] payloadMetadata = result.meta() == null ? new byte[0] :
result.meta();
+ IndexFileMeta segment =
+ new IndexFileMeta(
+ indexType,
+ result.fileName(),
+ fileIO.getFileSize(payloadPath),
+ liveRowCount,
+ new GlobalIndexMeta(
+ 0,
+ totalRowCount - 1,
+ vectorField.id(),
+ null,
+ payloadMetadata,
+ new
PkVectorSourceMeta(sourceFiles).serialize()),
+ pathFactory.isExternalPath() ?
payloadPath.toString() : null);
+ success = true;
+ return segment;
+ } finally {
+ if (writer instanceof AutoCloseable) {
+ IOUtils.closeQuietly((AutoCloseable) writer);
+ }
+ if (!success) {
+ fileWriter.deleteCreatedFiles();
+ }
+ }
+ }
+
+ private static PkVectorSourceFile sourceMetadata(DataFileMeta sourceFile) {
+ return new PkVectorSourceFile(sourceFile.fileName(),
sourceFile.rowCount());
+ }
+
+ private static String normalizeMetric(String metric) {
+ return metric.toLowerCase(Locale.ROOT).replace('-', '_');
+ }
+
+ private class SegmentFileWriter implements GlobalIndexFileWriter {
+
+ private final Map<String, Path> createdFiles = new HashMap<>();
+
+ @Override
+ public String newFileName(String prefix) {
+ Path path = pathFactory.newPath();
+ createdFiles.put(path.getName(), path);
+ return path.getName();
+ }
+
+ @Override
+ public PositionOutputStream newOutputStream(String fileName) throws
IOException {
+ return fileIO.newOutputStream(path(fileName), false);
+ }
+
+ private Path path(String fileName) {
+ Path path = createdFiles.get(fileName);
+ checkArgument(path != null, "ANN payload file %s was not
allocated.", fileName);
+ return path;
+ }
+
+ private void deleteCreatedFiles() {
+ for (Path path : createdFiles.values()) {
+ fileIO.deleteQuietly(path);
+ }
+ }
+ }
+
+ /** One vector source used while building an ANN segment. */
+ public static class Source {
+
+ private final PkVectorSourceFile sourceFile;
+ @Nullable private final PkVectorReader vectors;
+ @Nullable private final ReaderFactory readerFactory;
+ private final LongPredicate excludedPosition;
+
+ public Source(DataFileMeta sourceFile, PkVectorReader vectors) {
+ this(sourceFile, vectors, position -> false);
+ }
+
+ public Source(
+ DataFileMeta sourceFile, PkVectorReader vectors, LongPredicate
excludedPosition) {
+ this(sourceMetadata(sourceFile), vectors, excludedPosition);
+ }
+
+ Source(
+ PkVectorSourceFile sourceFile,
+ PkVectorReader vectors,
+ LongPredicate excludedPosition) {
+ this.sourceFile = sourceFile;
+ this.vectors = vectors;
+ this.readerFactory = null;
+ this.excludedPosition = excludedPosition;
+ }
+
+ private Source(
+ PkVectorSourceFile sourceFile,
+ ReaderFactory readerFactory,
+ LongPredicate excludedPosition) {
+ this.sourceFile = sourceFile;
+ this.vectors = null;
+ this.readerFactory = readerFactory;
+ this.excludedPosition = excludedPosition;
+ }
+
+ static Source lazy(PkVectorSourceFile sourceFile, ReaderFactory
readerFactory) {
+ return new Source(sourceFile, readerFactory, position -> false);
+ }
+
+ private PkVectorReader openReader() throws IOException {
+ return vectors != null ? vectors : readerFactory.open();
+ }
+
+ private void closeReader(PkVectorReader reader) throws IOException {
+ if (vectors == null) {
+ reader.close();
+ }
+ }
+
+ @FunctionalInterface
+ interface ReaderFactory {
+ PkVectorReader open() throws IOException;
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentSearcher.java
b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentSearcher.java
new file mode 100644
index 0000000000..e3b2fbba7f
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentSearcher.java
@@ -0,0 +1,272 @@
+/*
+ * 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.deletionvectors.DeletionVector;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.globalindex.GlobalIndexIOMeta;
+import org.apache.paimon.globalindex.GlobalIndexReader;
+import org.apache.paimon.globalindex.GlobalIndexer;
+import org.apache.paimon.globalindex.ScoredGlobalIndexResult;
+import org.apache.paimon.globalindex.VectorGlobalIndexer;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.VectorSearch;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.utils.IOUtils;
+import org.apache.paimon.utils.Range;
+import org.apache.paimon.utils.RoaringNavigableMap64;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Searches one ANN payload and maps its segment-local ids back to source row
positions. */
+public class PkVectorAnnSegmentSearcher {
+
+ private static final Comparator<Candidate> BEST_FIRST =
+ Comparator.comparingDouble((Candidate candidate) ->
candidate.distance)
+ .thenComparing(candidate -> candidate.dataFileName)
+ .thenComparingLong(candidate -> candidate.rowPosition);
+
+ private final FileIO fileIO;
+ private final PkVectorAnnSegmentFile annSegmentFile;
+ private final DataField vectorField;
+ private final Options indexOptions;
+ private final String metric;
+ private final ExecutorService executor;
+
+ public PkVectorAnnSegmentSearcher(
+ FileIO fileIO,
+ PkVectorAnnSegmentFile annSegmentFile,
+ DataField vectorField,
+ Options indexOptions,
+ String metric,
+ ExecutorService executor) {
+ this.fileIO = fileIO;
+ this.annSegmentFile = annSegmentFile;
+ this.vectorField = vectorField;
+ this.indexOptions = indexOptions;
+ this.metric = normalizeMetric(metric);
+ this.executor = executor;
+ }
+
+ public List<Candidate> search(
+ IndexFileMeta segment,
+ PkVectorSourceMeta sourceMeta,
+ float[] query,
+ int limit,
+ @Nullable DeletionVector deletionVector,
+ Map<String, String> searchOptions) {
+ Map<String, DeletionVector> deletionVectors = new HashMap<>();
+ if (deletionVector != null) {
+ checkArgument(
+ sourceMeta.sourceFiles().size() == 1,
+ "A single deletion vector can only search a single-source
ANN segment.");
+ deletionVectors.put(sourceMeta.sourceFiles().get(0).fileName(),
deletionVector);
+ }
+ return search(segment, sourceMeta, query, limit, deletionVectors,
searchOptions);
+ }
+
+ public List<Candidate> search(
+ IndexFileMeta segment,
+ PkVectorSourceMeta sourceMeta,
+ float[] query,
+ int limit,
+ Map<String, DeletionVector> deletionVectors,
+ Map<String, String> searchOptions) {
+ checkArgument(limit > 0, "Vector search limit must be positive: %s.",
limit);
+ GlobalIndexMeta globalIndexMeta = segment.globalIndexMeta();
+ checkArgument(
+ globalIndexMeta != null && globalIndexMeta.sourceMeta() !=
null,
+ "Vector segment %s has no source metadata.",
+ segment.fileName());
+ GlobalIndexer indexer =
+ GlobalIndexer.create(segment.indexType(), vectorField,
indexOptions);
+ checkArgument(
+ indexer instanceof VectorGlobalIndexer,
+ "Index algorithm %s does not implement VectorGlobalIndexer.",
+ segment.indexType());
+ String readerMetric = normalizeMetric(((VectorGlobalIndexer)
indexer).metric());
+ checkArgument(
+ metric.equals(readerMetric),
+ "ANN segment metric %s does not match index reader metric %s.",
+ metric,
+ readerMetric);
+
+ GlobalIndexIOMeta ioMeta =
+ new GlobalIndexIOMeta(
+ annSegmentFile.path(segment),
+ segment.fileSize(),
+ globalIndexMeta.indexMeta());
+ GlobalIndexReader reader =
+ indexer.createReader(
+ meta -> fileIO.newInputStream(meta.filePath()),
+ Collections.singletonList(ioMeta),
+ executor);
+ try {
+ VectorSearch search = new VectorSearch(query, limit,
vectorField.name(), searchOptions);
+ RoaringNavigableMap64 liveRows =
+ liveRowPositions(sourceMeta.sourceFiles(),
deletionVectors);
+ if (liveRows != null) {
+ search.withIncludeRowIds(liveRows);
+ }
+ Optional<ScoredGlobalIndexResult> result =
reader.visitVectorSearch(search).join();
+ if (!result.isPresent()) {
+ return Collections.emptyList();
+ }
+
+ long sourceRowCount = totalRowCount(sourceMeta.sourceFiles());
+ List<Candidate> candidates = new ArrayList<>();
+ ScoredGlobalIndexResult scored = result.get();
+ for (long ordinal : scored.results()) {
+ checkArgument(
+ ordinal >= 0 && ordinal < sourceRowCount,
+ "ANN segment %s returned ordinal %s outside [0, %s).",
+ segment.fileName(),
+ ordinal,
+ sourceRowCount);
+ FilePosition filePosition =
filePosition(sourceMeta.sourceFiles(), ordinal);
+ DeletionVector deletionVector =
deletionVectors.get(filePosition.dataFileName);
+ checkArgument(
+ deletionVector == null
+ ||
!deletionVector.isDeleted(filePosition.rowPosition),
+ "ANN segment %s returned snapshot-deleted row position
%s.",
+ segment.fileName(),
+ filePosition.rowPosition);
+ candidates.add(
+ new Candidate(
+ filePosition.dataFileName,
+ filePosition.rowPosition,
+
scoreToDistance(scored.scoreGetter().score(ordinal), metric)));
+ }
+ Collections.sort(candidates, BEST_FIRST);
+ return Collections.unmodifiableList(candidates);
+ } finally {
+ IOUtils.closeQuietly(reader);
+ }
+ }
+
+ @Nullable
+ private static RoaringNavigableMap64 liveRowPositions(
+ List<PkVectorSourceFile> sourceFiles, Map<String, DeletionVector>
deletionVectors) {
+ if (deletionVectors.isEmpty()) {
+ return null;
+ }
+ RoaringNavigableMap64 live = new RoaringNavigableMap64();
+ RoaringNavigableMap64 deleted = new RoaringNavigableMap64();
+ long fileOffset = 0;
+ for (PkVectorSourceFile sourceFile : sourceFiles) {
+ if (sourceFile.rowCount() > 0) {
+ live.addRange(new Range(fileOffset, fileOffset +
sourceFile.rowCount() - 1));
+ }
+ DeletionVector deletionVector =
deletionVectors.get(sourceFile.fileName());
+ if (deletionVector != null) {
+ final long offset = fileOffset;
+ deletionVector.forEachDeletedPosition(position ->
deleted.add(offset + position));
+ }
+ fileOffset += sourceFile.rowCount();
+ }
+ live.andNot(deleted);
+ return live;
+ }
+
+ private static long totalRowCount(List<PkVectorSourceFile> sourceFiles) {
+ long total = 0;
+ for (PkVectorSourceFile sourceFile : sourceFiles) {
+ total = Math.addExact(total, sourceFile.rowCount());
+ }
+ return total;
+ }
+
+ private static FilePosition filePosition(List<PkVectorSourceFile>
sourceFiles, long ordinal) {
+ long fileOffset = 0;
+ for (PkVectorSourceFile sourceFile : sourceFiles) {
+ long nextOffset = fileOffset + sourceFile.rowCount();
+ if (ordinal < nextOffset) {
+ return new FilePosition(sourceFile.fileName(), ordinal -
fileOffset);
+ }
+ fileOffset = nextOffset;
+ }
+ throw new IllegalArgumentException("ANN ordinal is outside source
files: " + ordinal);
+ }
+
+ private static float scoreToDistance(float score, String metric) {
+ if ("l2".equals(metric)) {
+ return 1F / score - 1F;
+ } else if ("cosine".equals(metric)) {
+ return 1F - score;
+ } else if ("inner_product".equals(metric)) {
+ return -score;
+ }
+ throw new IllegalArgumentException("Unsupported ANN vector metric: " +
metric);
+ }
+
+ private static String normalizeMetric(String metric) {
+ return metric.toLowerCase(Locale.ROOT).replace('-', '_');
+ }
+
+ /** One ANN candidate addressed by source-file row position. */
+ public static class Candidate {
+
+ private final long rowPosition;
+ private final float distance;
+ private final String dataFileName;
+
+ private Candidate(String dataFileName, long rowPosition, float
distance) {
+ this.dataFileName = dataFileName;
+ this.rowPosition = rowPosition;
+ this.distance = distance;
+ }
+
+ public String dataFileName() {
+ return dataFileName;
+ }
+
+ public long rowPosition() {
+ return rowPosition;
+ }
+
+ public float distance() {
+ return distance;
+ }
+ }
+
+ private static class FilePosition {
+
+ private final String dataFileName;
+ private final long rowPosition;
+
+ private FilePosition(String dataFileName, long rowPosition) {
+ this.dataFileName = dataFileName;
+ this.rowPosition = rowPosition;
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorReader.java
b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorReader.java
new file mode 100644
index 0000000000..b173607ce0
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorReader.java
@@ -0,0 +1,33 @@
+/*
+ * 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 java.io.Closeable;
+import java.io.IOException;
+
+/** Sequential vectors in physical data-file row order. */
+public interface PkVectorReader extends Closeable {
+
+ int dimension();
+
+ long rowCount();
+
+ /** Reads the next physical row and returns whether it contains a non-null
vector. */
+ boolean readNextVector(float[] reuse) throws IOException;
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorSourceFile.java
b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorSourceFile.java
new file mode 100644
index 0000000000..87429d5694
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorSourceFile.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.index.pkvector;
+
+import java.util.Objects;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Immutable source data-file identity captured when a vector segment is
built. */
+public final class PkVectorSourceFile {
+
+ private final String fileName;
+ private final long rowCount;
+
+ public PkVectorSourceFile(String fileName, long rowCount) {
+ this.fileName = Objects.requireNonNull(fileName);
+ this.rowCount = rowCount;
+ checkArgument(rowCount >= 0, "Source file row count must not be
negative.");
+ }
+
+ public String fileName() {
+ return fileName;
+ }
+
+ public long rowCount() {
+ return rowCount;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ PkVectorSourceFile that = (PkVectorSourceFile) o;
+ return rowCount == that.rowCount && Objects.equals(fileName,
that.fileName);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(fileName, rowCount);
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorSourceMeta.java
b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorSourceMeta.java
new file mode 100644
index 0000000000..d0a6a8171c
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorSourceMeta.java
@@ -0,0 +1,91 @@
+/*
+ * 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.apache.paimon.io.DataInputDeserializer;
+import org.apache.paimon.io.DataOutputSerializer;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Ordered source data files for a primary-key vector index payload. */
+public final class PkVectorSourceMeta {
+
+ private static final int VERSION = 1;
+
+ private final List<PkVectorSourceFile> sourceFiles;
+
+ public PkVectorSourceMeta(List<PkVectorSourceFile> sourceFiles) {
+ this.sourceFiles = Collections.unmodifiableList(new
ArrayList<>(sourceFiles));
+ checkArgument(!this.sourceFiles.isEmpty(), "A vector index must
reference source files.");
+ }
+
+ public List<PkVectorSourceFile> sourceFiles() {
+ return sourceFiles;
+ }
+
+ public static PkVectorSourceMeta fromIndexFile(IndexFileMeta indexFile) {
+ GlobalIndexMeta globalIndexMeta = indexFile.globalIndexMeta();
+ checkArgument(
+ globalIndexMeta != null && globalIndexMeta.sourceMeta() !=
null,
+ "Vector index file %s has no source metadata.",
+ indexFile.fileName());
+ return deserialize(globalIndexMeta.sourceMeta());
+ }
+
+ public byte[] serialize() {
+ try {
+ DataOutputSerializer output = new DataOutputSerializer(128);
+ output.writeInt(VERSION);
+ output.writeInt(sourceFiles.size());
+ for (PkVectorSourceFile sourceFile : sourceFiles) {
+ output.writeUTF(sourceFile.fileName());
+ output.writeLong(sourceFile.rowCount());
+ }
+ return output.getCopyOfBuffer();
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to serialize vector source
metadata.", e);
+ }
+ }
+
+ public static PkVectorSourceMeta deserialize(byte[] bytes) {
+ try {
+ DataInputDeserializer input = new DataInputDeserializer(bytes);
+ int version = input.readInt();
+ checkArgument(version == VERSION, "Unsupported vector source
version: %s.", version);
+ int sourceFileCount = input.readInt();
+ checkArgument(sourceFileCount > 0, "A vector index must reference
source files.");
+ List<PkVectorSourceFile> sourceFiles = new
ArrayList<>(sourceFileCount);
+ for (int i = 0; i < sourceFileCount; i++) {
+ sourceFiles.add(new PkVectorSourceFile(input.readUTF(),
input.readLong()));
+ }
+ checkArgument(
+ input.available() == 0, "Unexpected trailing bytes in
vector source metadata.");
+ return new PkVectorSourceMeta(sourceFiles);
+ } catch (IOException e) {
+ throw new IllegalArgumentException("Failed to deserialize vector
source metadata.", e);
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntrySerializer.java
b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntrySerializer.java
index 60113adff1..ef0b68d142 100644
---
a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntrySerializer.java
+++
b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntrySerializer.java
@@ -60,7 +60,8 @@ public class IndexManifestEntrySerializer extends
VersionedObjectSerializer<Inde
globalIndexMeta.extraFieldIds() == null
? null
: new
GenericArray(globalIndexMeta.extraFieldIds()),
- globalIndexMeta.indexMeta());
+ globalIndexMeta.indexMeta(),
+ globalIndexMeta.sourceMeta());
return GenericRow.of(
record.kind().toByteValue(),
serializeBinaryRow(record.partition()),
@@ -82,16 +83,25 @@ public class IndexManifestEntrySerializer extends
VersionedObjectSerializer<Inde
GlobalIndexMeta globalIndexMeta = null;
if (!row.isNullAt(9)) {
- InternalRow globalIndexRow = row.getRow(9, 5);
+ InternalRow globalIndexRow = row.getRow(9, 6);
long rowRangeStart = globalIndexRow.getLong(0);
long rowRangeEnd = globalIndexRow.getLong(1);
int indexFieldId = globalIndexRow.getInt(2);
int[] extralFields =
globalIndexRow.isNullAt(3) ? null :
globalIndexRow.getArray(3).toIntArray();
byte[] indexMeta = globalIndexRow.isNullAt(4) ? null :
globalIndexRow.getBinary(4);
+ byte[] sourceMeta =
+ globalIndexRow.getFieldCount() <= 5 ||
globalIndexRow.isNullAt(5)
+ ? null
+ : globalIndexRow.getBinary(5);
globalIndexMeta =
new GlobalIndexMeta(
- rowRangeStart, rowRangeEnd, indexFieldId,
extralFields, indexMeta);
+ rowRangeStart,
+ rowRangeEnd,
+ indexFieldId,
+ extralFields,
+ indexMeta,
+ sourceMeta);
}
return new IndexManifestEntry(
diff --git
a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
index 0791c2539e..06f6a1e077 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
@@ -431,6 +431,7 @@ public class SchemaManager implements Serializable {
} else if (change instanceof RenameColumn) {
RenameColumn rename = (RenameColumn) change;
assertNotUpdatingPartitionKeys(oldTableSchema,
rename.fieldNames(), "rename");
+ assertNotRenamingPrimaryKeyVectorIndexColumn(oldTableSchema,
rename.fieldNames());
assertNotRenamingBlobColumn(newFields, rename.fieldNames());
new NestedColumnModifier(rename.fieldNames(), lazyIdentifier) {
@Override
@@ -975,6 +976,19 @@ public class SchemaManager implements Serializable {
}
}
+ private static void assertNotRenamingPrimaryKeyVectorIndexColumn(
+ TableSchema schema, String[] fieldNames) {
+ if (fieldNames.length > 1) {
+ return;
+ }
+ String fieldName = fieldNames[0];
+ if (new
CoreOptions(schema.options()).primaryKeyVectorIndexColumns().contains(fieldName))
{
+ throw new UnsupportedOperationException(
+ String.format(
+ "Cannot rename primary-key vector index column:
[%s]", fieldName));
+ }
+ }
+
private static void assertNotChangingBlobColumnType(
List<DataField> fields, String[] fieldNames, DataType newType) {
if (fieldNames.length > 1) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index 56b4676dca..2122c04ed5 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -350,6 +350,8 @@ public class SchemaValidation {
fieldNamesSpecifiedAsVector.isEmpty(),
"Some of the columns specified as vector-field are unknown.");
+ validatePrimaryKeyVectorIndex(schema, options);
+
validateMergeFunctionFactory(schema);
validateMapStorageLayout(schema, options);
@@ -883,6 +885,75 @@ public class SchemaValidation {
}
}
+ private static void validatePrimaryKeyVectorIndex(TableSchema schema,
CoreOptions options) {
+ if (!options.primaryKeyVectorIndexEnabled()) {
+ return;
+ }
+
+ List<String> indexColumns = options.primaryKeyVectorIndexColumns();
+ checkArgument(
+ new HashSet<>(indexColumns).size() == indexColumns.size(),
+ "pk-vector.index.columns must not contain duplicate columns,
but is %s.",
+ indexColumns);
+ checkArgument(
+ indexColumns.size() == 1,
+ "pk-vector.index.columns must contain exactly one column in
the first release, but is %s.",
+ indexColumns);
+ String indexColumn = indexColumns.get(0);
+ String indexType = options.primaryKeyVectorIndexType(indexColumn);
+ checkArgument(
+ !StringUtils.isNullOrWhitespaceOnly(indexColumn),
+ "pk-vector.index.columns must contain a non-empty column.");
+ checkArgument(
+ !StringUtils.isNullOrWhitespaceOnly(indexType),
+ "fields.%s.pk-vector.index.type must be configured when a
primary-key vector index is defined.",
+ indexColumn);
+ checkArgument(
+ !schema.primaryKeys().isEmpty(),
+ "Primary-key vector index requires a primary-key table.");
+ checkArgument(
+ options.deletionVectorsEnabled(),
+ "Primary-key vector index requires deletion-vectors.enabled =
true.");
+ checkArgument(
+ options.mergeEngine() == MergeEngine.DEDUPLICATE
+ || options.mergeEngine() == MergeEngine.PARTIAL_UPDATE,
+ "Primary-key vector index only supports merge-engine =
deduplicate or partial-update, but is %s.",
+ options.mergeEngine());
+ checkArgument(
+ !options.deletionVectorsMergeOnRead(),
+ "Primary-key vector index with merge-engine = %s requires
deletion-vectors.merge-on-read = false.",
+ options.mergeEngine());
+ checkArgument(
+ options.bucket() > 0,
+ "Primary-key vector index requires fixed bucket mode (bucket >
0), but bucket is %s.",
+ options.bucket());
+ checkArgument(
+ !options.pkClusteringOverride(),
+ "Primary-key vector index does not support
pk-clustering-override.");
+ options.primaryKeyVectorIndexOptions(indexColumn);
+
+ DataField vectorField =
+ schema.fields().stream()
+ .filter(field -> field.name().equals(indexColumn))
+ .findFirst()
+ .orElse(null);
+ checkArgument(
+ vectorField != null && vectorField.type().getTypeRoot() ==
VECTOR,
+ "pk-vector.index.columns entry '%s' must reference a VECTOR
column.",
+ indexColumn);
+ checkArgument(
+ ((VectorType)
vectorField.type()).getElementType().getTypeRoot()
+ == DataTypeRoot.FLOAT,
+ "pk-vector.index.columns entry '%s' must use FLOAT elements.",
+ indexColumn);
+ checkArgument(
+ Arrays.asList("l2", "cosine", "inner_product")
+
.contains(options.primaryKeyVectorDistanceMetric(indexColumn)),
+ "fields.%s.pk-vector.distance.metric must be one of l2,
cosine, inner_product, but is %s.",
+ indexColumn,
+ options.primaryKeyVectorDistanceMetric(indexColumn));
+ }
+
private static void validateSequenceField(TableSchema schema, CoreOptions
options) {
List<String> sequenceField = options.sequenceField();
if (!sequenceField.isEmpty()) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/index/IndexFileMetaSerializerTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/IndexFileMetaSerializerTest.java
index 7e4fe92c8d..00d97d5435 100644
---
a/paimon-core/src/test/java/org/apache/paimon/index/IndexFileMetaSerializerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/index/IndexFileMetaSerializerTest.java
@@ -22,12 +22,35 @@ import
org.apache.paimon.deletionvectors.DeletionVectorsIndexFile;
import org.apache.paimon.utils.ObjectSerializer;
import org.apache.paimon.utils.ObjectSerializerTestBase;
+import org.junit.jupiter.api.Test;
+
import java.util.LinkedHashMap;
import java.util.Random;
+import static org.assertj.core.api.Assertions.assertThat;
+
/** Test for {@link org.apache.paimon.index.IndexFileMetaSerializer}. */
public class IndexFileMetaSerializerTest extends
ObjectSerializerTestBase<IndexFileMeta> {
+ @Test
+ void testGlobalIndexSourceMetaRoundTrip() {
+ IndexFileMetaSerializer serializer = new IndexFileMetaSerializer();
+ IndexFileMeta indexFile =
+ new IndexFileMeta(
+ "ivf-pq",
+ "index-file",
+ 100,
+ 10,
+ new GlobalIndexMeta(0, 9, 7, null, new byte[] {3, 4},
new byte[] {1, 2}),
+ null);
+
+ GlobalIndexMeta restored =
+
serializer.fromRow(serializer.toRow(indexFile)).globalIndexMeta();
+
+ assertThat(restored.sourceMeta()).containsExactly(1, 2);
+ assertThat(restored.indexMeta()).containsExactly(3, 4);
+ }
+
@Override
protected ObjectSerializer<IndexFileMeta> serializer() {
return new IndexFileMetaSerializer();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentFileTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentFileTest.java
new file mode 100644
index 0000000000..1a30e207f4
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentFileTest.java
@@ -0,0 +1,237 @@
+/*
+ * 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.deletionvectors.BitmapDeletionVector;
+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.DataFileMeta;
+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.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests ANN construction from generic vector readers. */
+class PkVectorAnnSegmentFileTest {
+
+ @TempDir java.nio.file.Path tempPath;
+
+ @Test
+ void testBuildSkipsNullAndExcludedPhysicalRows() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ IndexFileMeta segment =
+ annFile(fileIO)
+ .build(
+ Collections.singletonList(
+ new PkVectorAnnSegmentFile.Source(
+ dataFile("data-1", 3),
+ new ArrayReader(
+ new float[][] {{0, 0},
null, {2, 0}}),
+ position -> position == 0)),
+ vectorField(),
+ indexOptions(),
+ "l2",
+ "test-vector-ann");
+
+ assertThat(segment.indexType()).isEqualTo("test-vector-ann");
+ assertThat(segment.rowCount()).isEqualTo(1);
+ PkVectorSourceMeta sourceMeta =
PkVectorSourceMeta.fromIndexFile(segment);
+ assertThat(sourceMeta.sourceFiles())
+ .extracting(PkVectorSourceFile::fileName)
+ .containsExactly("data-1");
+ }
+
+ @Test
+ void testBuildsAndSearchesMultiSourceSegment() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ PkVectorAnnSegmentFile annFile = annFile(fileIO);
+ IndexFileMeta segment =
+ annFile.build(
+ Arrays.asList(
+ new PkVectorAnnSegmentFile.Source(
+ dataFile("data-1", 2),
+ new ArrayReader(new float[][] {{5, 0},
{10, 0}})),
+ new PkVectorAnnSegmentFile.Source(
+ dataFile("data-2", 2),
+ new ArrayReader(new float[][] {{0, 0},
{2, 0}}))),
+ vectorField(),
+ indexOptions(),
+ "l2",
+ "test-vector-ann");
+ assertThat(segment.globalIndexMeta().rowRangeStart()).isZero();
+ assertThat(segment.globalIndexMeta().rowRangeEnd()).isEqualTo(3);
+ PkVectorSourceMeta sourceMeta =
PkVectorSourceMeta.fromIndexFile(segment);
+ BitmapDeletionVector data2Deletes = new BitmapDeletionVector();
+ data2Deletes.delete(0);
+ Map<String, org.apache.paimon.deletionvectors.DeletionVector>
deletionVectors =
+ new HashMap<>();
+ deletionVectors.put("data-2", data2Deletes);
+
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ List<PkVectorAnnSegmentSearcher.Candidate> candidates;
+ try {
+ candidates =
+ new PkVectorAnnSegmentSearcher(
+ fileIO, annFile, vectorField(),
indexOptions(), "l2", executor)
+ .search(
+ segment,
+ sourceMeta,
+ new float[] {0, 0},
+ 3,
+ deletionVectors,
+ Collections.emptyMap());
+ } finally {
+ executor.shutdownNow();
+ }
+
+ assertThat(candidates)
+ .extracting(
+ PkVectorAnnSegmentSearcher.Candidate::dataFileName,
+ PkVectorAnnSegmentSearcher.Candidate::rowPosition)
+ .containsExactly(
+ org.assertj.core.groups.Tuple.tuple("data-2", 1L),
+ org.assertj.core.groups.Tuple.tuple("data-1", 0L),
+ org.assertj.core.groups.Tuple.tuple("data-1", 1L));
+ }
+
+ @Test
+ void testRejectsSourcesWithoutRows() {
+ LocalFileIO fileIO = LocalFileIO.create();
+
+ assertThatThrownBy(
+ () ->
+ annFile(fileIO)
+ .build(
+ Collections.singletonList(
+ new
PkVectorAnnSegmentFile.Source(
+
dataFile("empty", 0),
+ new
ArrayReader(new float[0][]))),
+ vectorField(),
+ indexOptions(),
+ "l2",
+ "test-vector-ann"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("at least one source row");
+ }
+
+ private PkVectorAnnSegmentFile annFile(LocalFileIO fileIO) {
+ return new PkVectorAnnSegmentFile(fileIO, pathFactory());
+ }
+
+ private static DataField vectorField() {
+ return new DataField(7, "embedding", DataTypes.VECTOR(2,
DataTypes.FLOAT()));
+ }
+
+ private static Options indexOptions() {
+ Options options = new Options();
+ options.setString("test.vector.dimension", "2");
+ options.setString("test.vector.metric", "l2");
+ return options;
+ }
+
+ private static DataFileMeta dataFile(String fileName, long rowCount) {
+ return DataFileMeta.forAppend(
+ fileName,
+ 100,
+ rowCount,
+ SimpleStats.EMPTY_STATS,
+ 0,
+ 0,
+ 1,
+ Collections.emptyList(),
+ null,
+ FileSource.COMPACT,
+ null,
+ null,
+ null,
+ null);
+ }
+
+ 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++];
+ if (vector == null) {
+ return false;
+ }
+ System.arraycopy(vector, 0, reuse, 0, reuse.length);
+ return true;
+ }
+
+ @Override
+ public void close() throws IOException {}
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorSourceMetaTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorSourceMetaTest.java
new file mode 100644
index 0000000000..9d8e967267
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorSourceMetaTest.java
@@ -0,0 +1,56 @@
+/*
+ * 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.DataOutputSerializer;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link PkVectorSourceMeta}. */
+class PkVectorSourceMetaTest {
+
+ @Test
+ void testRoundTrip() {
+ PkVectorSourceMeta metadata =
+ new PkVectorSourceMeta(
+ Arrays.asList(
+ new PkVectorSourceFile("data-1", 10),
+ new PkVectorSourceFile("data-2", 20)));
+
+ PkVectorSourceMeta restored =
PkVectorSourceMeta.deserialize(metadata.serialize());
+
+ assertThat(restored.sourceFiles()).isEqualTo(metadata.sourceFiles());
+ }
+
+ @Test
+ void testRejectsTruncatedSourceMetadata() throws Exception {
+ DataOutputSerializer output = new DataOutputSerializer(128);
+ output.writeInt(1);
+ output.writeInt(1);
+ output.writeUTF("data-1");
+
+ assertThatThrownBy(() ->
PkVectorSourceMeta.deserialize(output.getCopyOfBuffer()))
+ .hasMessageContaining("Failed to deserialize vector source
metadata");
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PrimaryKeyVectorIndexOptionsTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PrimaryKeyVectorIndexOptionsTest.java
new file mode 100644
index 0000000000..449d640aa3
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PrimaryKeyVectorIndexOptionsTest.java
@@ -0,0 +1,157 @@
+/*
+ * 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.options.Options;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for primary-key vector index options in {@link CoreOptions}. */
+class PrimaryKeyVectorIndexOptionsTest {
+
+ @Test
+ void testPluralFieldRegistryEnablesIndex() {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.PK_VECTOR_INDEX_COLUMNS.key(), "embedding");
+
+ assertThat(new
CoreOptions(options).primaryKeyVectorIndexEnabled()).isTrue();
+ }
+
+ @Test
+ void testResolvesSingleIndexColumn() {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.PK_VECTOR_INDEX_COLUMNS.key(), "embedding");
+
+ assertThat(new
CoreOptions(options).primaryKeyVectorIndexColumn()).isEqualTo("embedding");
+ }
+
+ @Test
+ void testFieldRegistryIsTheOnlyEnableSwitch() {
+ Map<String, String> options = new HashMap<>();
+ options.put("pk-vector.index.column", "embedding");
+ options.put("pk-vector.index.type", "ivf-pq");
+
+ assertThat(new
CoreOptions(options).primaryKeyVectorIndexEnabled()).isFalse();
+ }
+
+ @Test
+ void testIndexTypeMustBeFieldScoped() {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.PK_VECTOR_INDEX_COLUMNS.key(), "embedding");
+ options.put("pk-vector.index.type", "ivf-pq");
+
+ assertThat(new
CoreOptions(options).primaryKeyVectorIndexType("embedding")).isNull();
+ }
+
+ @Test
+ void testIndexOptionsMustBeFieldScoped() {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.PK_VECTOR_INDEX_COLUMNS.key(), "embedding");
+ options.put("fields.embedding.pk-vector.index.type", "ivf-pq");
+ options.put("pk-vector.index.options", "{\"nlist\":64}");
+
+ assertThat(
+ new CoreOptions(options)
+ .primaryKeyVectorIndexOptions("embedding")
+ .get("ivf-pq.nlist"))
+ .isNull();
+ }
+
+ @Test
+ void testDistanceMetricMustBeFieldScoped() {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.PK_VECTOR_INDEX_COLUMNS.key(), "embedding");
+ options.put("pk-vector.distance.metric", "l2");
+
+ assertThat(new
CoreOptions(options).primaryKeyVectorDistanceMetric("embedding"))
+ .isEqualTo("inner_product");
+ }
+
+ @Test
+ void testFieldScopedDistanceMetricOverridesTableDefault() {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.PK_VECTOR_INDEX_COLUMNS.key(), "embedding");
+ options.put("pk-vector.distance.metric", "l2");
+ options.put("fields.embedding.pk-vector.distance.metric", "cosine");
+
+ assertThat(new
CoreOptions(options).primaryKeyVectorDistanceMetric("embedding"))
+ .isEqualTo("cosine");
+ }
+
+ @Test
+ void testFieldScopedJsonOptionsOverrideTableDefault() {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.PK_VECTOR_INDEX_COLUMNS.key(), "embedding");
+ options.put("fields.embedding.pk-vector.index.type", "ivf-pq");
+ options.put("pk-vector.index.options", "{\"nlist\":64}");
+ options.put("fields.embedding.pk-vector.index.options",
"{\"nlist\":128}");
+
+ Options resolved = new
CoreOptions(options).primaryKeyVectorIndexOptions("embedding");
+
+ assertThat(resolved.get("ivf-pq.nlist")).isEqualTo("128");
+ }
+
+ @Test
+ void testResolvesShortAndQualifiedAlgorithmOptions() {
+ CoreOptions coreOptions =
+ coreOptions(
+ "{\"nlist\":64,\"ivf-pq.pq.m\":\"8\"," +
"\"fields.embedding.hnsw.m\":16}");
+
+ Options resolved =
coreOptions.primaryKeyVectorIndexOptions("embedding");
+
+ assertThat(resolved.get("ivf-pq.nlist")).isEqualTo("64");
+ assertThat(resolved.get("ivf-pq.pq.m")).isEqualTo("8");
+ assertThat(resolved.get("fields.embedding.hnsw.m")).isEqualTo("16");
+ assertThat(resolved.get("ivf-pq.metric")).isEqualTo("l2");
+ }
+
+ @Test
+ void testRejectsNonObjectOptions() {
+ assertThatThrownBy(() ->
coreOptions("[1,2]").primaryKeyVectorIndexOptions("embedding"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("pk-vector.index.options")
+ .hasMessageContaining("JSON object");
+ }
+
+ private static CoreOptions coreOptions(String indexOptions) {
+ return coreOptions(indexOptions, null, null);
+ }
+
+ private static CoreOptions coreOptions(
+ String indexOptions, String additionalKey, String additionalValue)
{
+ Map<String, String> options = new HashMap<>();
+ options.put("fields.embedding.pk-vector.index.type", "ivf-pq");
+ options.put(CoreOptions.PK_VECTOR_INDEX_COLUMNS.key(), "embedding");
+ options.put("fields.embedding.pk-vector.distance.metric", "l2");
+ if (indexOptions != null) {
+ options.put("fields.embedding.pk-vector.index.options",
indexOptions);
+ }
+ if (additionalKey != null) {
+ options.put(additionalKey, additionalValue);
+ }
+ return new CoreOptions(options);
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestEntrySerializerTest.java
b/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestEntrySerializerTest.java
index 0429b8dae3..0678ed787a 100644
---
a/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestEntrySerializerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestEntrySerializerTest.java
@@ -18,17 +18,55 @@
package org.apache.paimon.manifest;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.utils.ObjectSerializer;
import org.apache.paimon.utils.ObjectSerializerTestBase;
+import org.junit.jupiter.api.Test;
+
import java.util.Random;
import static
org.apache.paimon.index.IndexFileMetaSerializerTest.randomIndexFile;
import static org.apache.paimon.io.DataFileTestUtils.row;
+import static org.assertj.core.api.Assertions.assertThat;
/** Test for {@link IndexManifestEntrySerializer}. */
public class IndexManifestEntrySerializerTest extends
ObjectSerializerTestBase<IndexManifestEntry> {
+ @Test
+ void testReadsGlobalIndexWithoutSourceMeta() {
+ IndexManifestEntrySerializer serializer = new
IndexManifestEntrySerializer();
+ IndexManifestEntry entry =
+ new IndexManifestEntry(
+ FileKind.ADD,
+ BinaryRow.EMPTY_ROW,
+ 0,
+ new IndexFileMeta(
+ "btree",
+ "index-file",
+ 100,
+ 10,
+ new GlobalIndexMeta(0, 9, 7, null, new byte[]
{1}),
+ null));
+ GenericRow serialized = (GenericRow) serializer.convertTo(entry);
+ serialized.setField(9, GenericRow.of(0L, 9L, 7, null, new byte[] {1}));
+
+ InternalRow globalIndexRow = serialized.getRow(9, 5);
+ assertThat(globalIndexRow.getFieldCount()).isEqualTo(5);
+ GlobalIndexMeta restored =
+ serializer
+ .convertFrom(serializer.getVersion(), serialized)
+ .indexFile()
+ .globalIndexMeta();
+
+ assertThat(restored.indexMeta()).containsExactly(1);
+ assertThat(restored.sourceMeta()).isNull();
+ }
+
@Override
protected ObjectSerializer<IndexManifestEntry> serializer() {
return new IndexManifestEntrySerializer();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/schema/PrimaryKeyVectorIndexValidationTest.java
b/paimon-core/src/test/java/org/apache/paimon/schema/PrimaryKeyVectorIndexValidationTest.java
new file mode 100644
index 0000000000..0daa7bc5c4
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/schema/PrimaryKeyVectorIndexValidationTest.java
@@ -0,0 +1,236 @@
+/*
+ * 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.schema;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.paimon.schema.SchemaValidation.validateTableSchema;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for primary-key vector index option validation. */
+class PrimaryKeyVectorIndexValidationTest {
+
+ @Test
+ void testValidPluralPrimaryKeyVectorIndexConfiguration() {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), "1");
+ options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
+ options.put(CoreOptions.PK_VECTOR_INDEX_COLUMNS.key(), " embedding ");
+ options.put("fields.embedding.pk-vector.index.type", "ivf-pq");
+
+ assertThatCode(() ->
validateTableSchema(schema(options))).doesNotThrowAnyException();
+ }
+
+ @Test
+ void testRejectsMultiplePrimaryKeyVectorIndexColumnsForFirstRelease() {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), "1");
+ options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
+ options.put(CoreOptions.PK_VECTOR_INDEX_COLUMNS.key(),
"embedding,other_embedding");
+ options.put("fields.embedding.pk-vector.index.type", "ivf-pq");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining("exactly one column")
+ .hasMessageContaining("pk-vector.index.columns");
+ }
+
+ @Test
+ void testRejectsDuplicatePrimaryKeyVectorIndexColumns() {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), "1");
+ options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
+ options.put(CoreOptions.PK_VECTOR_INDEX_COLUMNS.key(),
"embedding,embedding");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining("pk-vector.index.columns")
+ .hasMessageContaining("duplicate");
+ }
+
+ @Test
+ void testValidPrimaryKeyVectorIndex() {
+ assertThatCode(() -> validateTableSchema(schema(enabledOptions())))
+ .doesNotThrowAnyException();
+ }
+
+ @Test
+ void testRequiresFieldScopedIndexType() {
+ Map<String, String> options = enabledOptions();
+ options.remove("fields.embedding.pk-vector.index.type");
+ options.put("pk-vector.index.type", "ivf-pq");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining("fields.embedding.pk-vector.index.type");
+ }
+
+ @Test
+ void testRequiresPrimaryKeyTable() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.BUCKET_KEY.key(), "id");
+ assertThatThrownBy(
+ () ->
+ validateTableSchema(
+ new TableSchema(
+ 0,
+ fields(),
+ 0,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ options,
+ "")))
+ .hasMessageContaining("requires a primary-key table");
+ }
+
+ @Test
+ void testRequiresDeletionVectors() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "false");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining("requires deletion-vectors.enabled =
true");
+ }
+
+ @Test
+ void testSupportsPartialUpdateMergeEngine() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.MERGE_ENGINE.key(), "partial-update");
+
+ assertThatCode(() ->
validateTableSchema(schema(options))).doesNotThrowAnyException();
+ }
+
+ @Test
+ void testPartialUpdateRejectsDeletionVectorMergeOnRead() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.MERGE_ENGINE.key(), "partial-update");
+ options.put(CoreOptions.DELETION_VECTORS_MERGE_ON_READ.key(), "true");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining(
+ "partial-update requires
deletion-vectors.merge-on-read = false");
+ }
+
+ @Test
+ void testDeduplicateRejectsDeletionVectorMergeOnRead() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.DELETION_VECTORS_MERGE_ON_READ.key(), "true");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining("requires deletion-vectors.merge-on-read
= false");
+ }
+
+ @Test
+ void testRequiresFixedBucket() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.BUCKET.key(), "-1");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining("requires fixed bucket mode");
+ }
+
+ @Test
+ void testRejectsPkClusteringOverride() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.PK_CLUSTERING_OVERRIDE.key(), "true");
+ options.put(CoreOptions.CLUSTERING_COLUMNS.key(), "payload");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining("does not support
pk-clustering-override");
+ }
+
+ @Test
+ void testRequiresVectorColumn() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.PK_VECTOR_INDEX_COLUMNS.key(), "payload");
+ options.put("fields.payload.pk-vector.index.type", "ivf-pq");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining("must reference a VECTOR column");
+ }
+
+ @Test
+ void testIndexNameIsNotRequired() {
+ Map<String, String> options = enabledOptions();
+ assertThatCode(() ->
validateTableSchema(schema(options))).doesNotThrowAnyException();
+ }
+
+ @Test
+ void testRequiresFloatVectorElements() {
+ TableSchema schema =
+ new TableSchema(
+ 0,
+ Arrays.asList(
+ new DataField(0, "id",
DataTypes.INT().notNull()),
+ new DataField(
+ 1, "embedding", DataTypes.VECTOR(8,
DataTypes.DOUBLE()))),
+ 1,
+ Collections.emptyList(),
+ Collections.singletonList("id"),
+ enabledOptions(),
+ "");
+
+ assertThatThrownBy(() -> validateTableSchema(schema))
+ .hasMessageContaining("must use FLOAT elements");
+ }
+
+ @Test
+ void testRejectsUnsupportedDistanceMetric() {
+ Map<String, String> options = enabledOptions();
+ options.put("fields.embedding.pk-vector.distance.metric", "manhattan");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining("pk-vector.distance.metric")
+ .hasMessageContaining("l2, cosine, inner_product");
+ }
+
+ private static Map<String, String> enabledOptions() {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), "1");
+ options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
+ options.put(CoreOptions.PK_VECTOR_INDEX_COLUMNS.key(), "embedding");
+ options.put("fields.embedding.pk-vector.index.type", "ivf-pq");
+ return options;
+ }
+
+ private static TableSchema schema(Map<String, String> options) {
+ return new TableSchema(
+ 0,
+ fields(),
+ 0,
+ Collections.emptyList(),
+ Collections.singletonList("id"),
+ options,
+ "");
+ }
+
+ private static java.util.List<DataField> fields() {
+ return Arrays.asList(
+ new DataField(0, "id", DataTypes.INT().notNull()),
+ new DataField(1, "embedding", DataTypes.VECTOR(8,
DataTypes.FLOAT())),
+ new DataField(2, "payload", DataTypes.STRING()));
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java
b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java
index 59a4f90f61..b982ff1df1 100644
--- a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java
@@ -170,6 +170,35 @@ public class SchemaManagerTest {
assertThat(latest.get().options()).containsEntry("new_k", "new_v");
}
+ @Test
+ public void testRejectRenamePrimaryKeyVectorIndexColumn() throws Exception
{
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), "1");
+ options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
+ options.put(CoreOptions.PK_VECTOR_INDEX_COLUMNS.key(), "embedding");
+ options.put("fields.embedding.pk-vector.index.type", "ivf-pq");
+ Schema schema =
+ new Schema(
+ Arrays.asList(
+ new DataField(0, "id",
DataTypes.INT().notNull()),
+ new DataField(
+ 1, "embedding", DataTypes.VECTOR(8,
DataTypes.FLOAT()))),
+ Collections.emptyList(),
+ Collections.singletonList("id"),
+ options,
+ "");
+ SchemaManager manager = new SchemaManager(LocalFileIO.create(), path);
+ manager.createTable(schema);
+
+ assertThatThrownBy(
+ () ->
+ manager.commitChanges(
+ SchemaChange.renameColumn(
+ new String[] {"embedding"},
"renamed_embedding")))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessage("Cannot rename primary-key vector index column:
[embedding]");
+ }
+
@Test
public void testResetSequenceGroupForAggregateFunction() throws Exception {
Map<String, String> options = new HashMap<>();
