[core] Add primary-key vector index foundation

docs/generated/core_configuration.html

@@ -1205,6 +1205,12 @@
  <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.&lt;column&gt;.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>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -33,6 +33,7 @@
 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.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 String toString() {
  "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 int vectorSearchLateralJoinBatchSize() {
  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."),

paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestVectorGlobalIndexer.java

@@ -28,6 +28,7 @@
 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);

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);
+ }
+}

paimon-core/src/main/java/org/apache/paimon/index/GlobalIndexMeta.java

@@ -43,25 +43,38 @@ 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,
  long rowRangeEnd,
  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 int[] extraFieldIds() {
  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<>();

paimon-core/src/main/java/org/apache/paimon/index/IndexFileMetaSerializer.java

@@ -50,7 +50,8 @@ public InternalRow toRow(IndexFileMeta record) {
  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 InternalRow toRow(IndexFileMeta record) {
  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(),