[core] Validate ANN segment metric against the built-in metric

paimon-common/src/main/java/org/apache/paimon/globalindex/VectorGlobalIndexer.java

@@ -23,4 +23,13 @@ public interface VectorGlobalIndexer extends GlobalIndexer {
 
     /** Returns the metric name used to convert vector distances to comparable scores. */
     String metric();
+
+    /**
+     * Returns the metric recorded in a segment's index metadata when it was built, or {@code null}
+     * when the metadata records none (legacy segments or indexers that do not persist it).
+     * Searchers use it to reject segments built with a different metric than the current one.
+     */
+    default String segmentMetric(byte[] indexMeta) {
+        return null;
+    }
 }

paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentSearcher.java

@@ -193,6 +193,7 @@ CompletableFuture<List<PkVectorSearchResult>> searchAsync(
                 "ANN segment metric %s does not match index reader metric %s.",
                 metric,
                 readerMetric);
+        checkSegmentMetric(indexer, metric, globalIndexMeta.indexMeta());
 
         GlobalIndexIOMeta ioMeta =
                 new GlobalIndexIOMeta(
@@ -291,6 +292,7 @@ CompletableFuture<List<List<PkVectorSearchResult>>> searchBatchAsync(
                 "ANN segment metric %s does not match index reader metric %s.",
                 metric,
                 readerMetric);
+        checkSegmentMetric(indexer, metric, globalIndexMeta.indexMeta());
 
         GlobalIndexIOMeta ioMeta =
                 new GlobalIndexIOMeta(
@@ -491,4 +493,25 @@ private FilePosition(String dataFileName, long rowPosition) {
             this.rowPosition = rowPosition;
         }
     }
+
+    /**
+     * The guard above compares two values from the current config; the segment metadata records
+     * what the index was actually built with, and a mismatch would mean silently wrong distances.
+     * Legacy segments record no metric and are not checked.
+     */
+    static void checkSegmentMetric(
+            GlobalIndexer indexer, String normalizedMetric, byte[] indexMeta) {
+        if (!(indexer instanceof VectorGlobalIndexer)) {
+            return;
+        }
+        String segmentMetric = ((VectorGlobalIndexer) indexer).segmentMetric(indexMeta);
+        if (segmentMetric != null) {
+            String normalized = VectorSearchMetric.normalize(segmentMetric);
+            checkArgument(
+                    normalizedMetric.equals(normalized),
+                    "ANN segment was built with metric %s but the current metric is %s.",
+                    normalized,
+                    normalizedMetric);
+        }
+    }
 }

paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentSearcherMetricTest.java

@@ -0,0 +1,94 @@
+/*
+ * 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.globalindex.GlobalIndexIOMeta;
+import org.apache.paimon.globalindex.GlobalIndexReader;
+import org.apache.paimon.globalindex.GlobalIndexWriter;
+import org.apache.paimon.globalindex.VectorGlobalIndexer;
+import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
+import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for the segment metric validation in {@link PkVectorAnnSegmentSearcher}. */
+class PkVectorAnnSegmentSearcherMetricTest {
+
+    private static VectorGlobalIndexer indexerWithMetric(String segmentMetric) {
+        return new VectorGlobalIndexer() {
+            @Override
+            public String metric() {
+                return "inner_product";
+            }
+
+            @Override
+            public String segmentMetric(byte[] indexMeta) {
+                return segmentMetric;
+            }
+
+            @Override
+            public GlobalIndexWriter createWriter(GlobalIndexFileWriter fileWriter)
+                    throws IOException {
+                throw new UnsupportedOperationException();
+            }
+
+            @Override
+            public GlobalIndexReader createReader(
+                    GlobalIndexFileReader fileReader,
+                    List<GlobalIndexIOMeta> files,
+                    long totalRowCount,
+                    List<org.apache.paimon.utils.Range> rowRanges,
+                    ExecutorService executor) {
+                throw new UnsupportedOperationException();
+            }
+        };
+    }
+
+    @Test
+    void testSegmentMetricMismatchIsRejected() {
+        assertThatThrownBy(
+                        () ->
+                                PkVectorAnnSegmentSearcher.checkSegmentMetric(
+                                        indexerWithMetric("cosine"), "l2", new byte[] {1}))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("built with metric cosine but the current metric is l2");
+    }
+
+    @Test
+    void testMatchingAndLegacySegmentMetricsPass() {
+        assertThatCode(
+                        () ->
+                                PkVectorAnnSegmentSearcher.checkSegmentMetric(
+                                        indexerWithMetric("cosine"), "cosine", new byte[] {1}))
+                .doesNotThrowAnyException();
+        // legacy segments record no metric and must not be rejected
+        assertThatCode(
+                        () ->
+                                PkVectorAnnSegmentSearcher.checkSegmentMetric(
+                                        indexerWithMetric(null), "l2", new byte[] {1}))
+                .doesNotThrowAnyException();
+    }
+}

paimon-vector/src/main/java/org/apache/paimon/vector/index/NativeVectorGlobalIndexWriter.java

@@ -290,7 +290,10 @@ private ResultEntry buildIndex() throws IOException {
                     identifier,
                     System.currentTimeMillis() - buildStart);
 
-            VectorIndexMeta meta = new VectorIndexMeta();
+            VectorIndexMeta meta =
+                    new VectorIndexMeta(
+                            nativeOptions.getOrDefault(
+                                    "metric", NativeVectorGlobalIndexer.DEFAULT_METRIC));
             return new ResultEntry(fileName, rowCount, meta.serialize());
         }
     }

paimon-vector/src/main/java/org/apache/paimon/vector/index/NativeVectorGlobalIndexer.java

@@ -27,6 +27,8 @@
 import org.apache.paimon.types.DataType;
 import org.apache.paimon.utils.Range;
 
+import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -86,6 +88,15 @@ public GlobalIndexReader createReader(
         return new NativeVectorGlobalIndexReader(fileReader, files, fieldType, executor);
     }
 
+    @Override
+    public String segmentMetric(byte[] indexMeta) {
+        try {
+            return VectorIndexMeta.deserialize(indexMeta).metric();
+        } catch (IOException e) {
+            throw new UncheckedIOException("Failed to read vector index metadata.", e);
+        }
+    }
+
     @Override
     public String metric() {
         return options.getOrDefault("metric", DEFAULT_METRIC);

paimon-vector/src/main/java/org/apache/paimon/vector/index/VectorIndexMeta.java

@@ -21,17 +21,20 @@
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.type.TypeReference;
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
 
+import javax.annotation.Nullable;
+
 import java.io.IOException;
 import java.io.Serializable;
-import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.Map;
 
 /**
  * Metadata for a vector index file.
  *
- * <p>Serialized as an empty JSON {@code Map<String, String>}. Search-time parameters are passed
- * through {@link org.apache.paimon.predicate.VectorSearch#options()}.
+ * <p>Serialized as a JSON {@code Map<String, String>}; it records the metric the index was built
+ * with so searches can reject segments built under a different metric. Legacy segments carry an
+ * empty map. Search-time parameters are passed through {@link
+ * org.apache.paimon.predicate.VectorSearch#options()}.
  */
 public class VectorIndexMeta implements Serializable {
 
@@ -42,14 +45,30 @@ public class VectorIndexMeta implements Serializable {
     private static final TypeReference<LinkedHashMap<String, String>> MAP_TYPE_REF =
             new TypeReference<LinkedHashMap<String, String>>() {};
 
-    VectorIndexMeta() {}
+    private static final String METRIC_KEY = "metric";
+
+    @Nullable private final String metric;
+
+    VectorIndexMeta(@Nullable String metric) {
+        this.metric = metric;
+    }
 
     public byte[] serialize() throws IOException {
-        return OBJECT_MAPPER.writeValueAsBytes(Collections.<String, String>emptyMap());
+        Map<String, String> data = new LinkedHashMap<>();
+        if (metric != null) {
+            data.put(METRIC_KEY, metric);
+        }
+        return OBJECT_MAPPER.writeValueAsBytes(data);
     }
 
     public static VectorIndexMeta deserialize(byte[] data) throws IOException {
-        Map<String, String> ignored = OBJECT_MAPPER.readValue(data, MAP_TYPE_REF);
-        return new VectorIndexMeta();
+        Map<String, String> map = OBJECT_MAPPER.readValue(data, MAP_TYPE_REF);
+        return new VectorIndexMeta(map.get(METRIC_KEY));
+    }
+
+    /** The metric this index was built with, or null for legacy segments that record none. */
+    @Nullable
+    public String metric() {
+        return metric;
     }
 }

paimon-vector/src/test/java/org/apache/paimon/vector/index/NativeVectorGlobalIndexTest.java

@@ -229,7 +229,7 @@ public void testVectorBatchSizeProtectsSingleJavaArrayAllocation() {
 
     @Test
     public void testMetaSerializationIsEmptyMap() throws IOException {
-        VectorIndexMeta meta = new VectorIndexMeta();
+        VectorIndexMeta meta = new VectorIndexMeta(null);
         byte[] serialized = meta.serialize();
         VectorIndexMeta deserialized = VectorIndexMeta.deserialize(serialized);
 

paimon-vector/src/test/java/org/apache/paimon/vector/index/VectorIndexMetaTest.java

@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.vector.index;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link VectorIndexMeta} metric persistence. */
+class VectorIndexMetaTest {
+
+    @Test
+    void testMetricRoundTrip() throws Exception {
+        byte[] data = new VectorIndexMeta("cosine").serialize();
+        assertThat(VectorIndexMeta.deserialize(data).metric()).isEqualTo("cosine");
+
+        // a null metric writes an empty map, matching legacy segments
+        byte[] legacy = new VectorIndexMeta(null).serialize();
+        assertThat(VectorIndexMeta.deserialize(legacy).metric()).isNull();
+        assertThat(VectorIndexMeta.deserialize("{}".getBytes()).metric()).isNull();
+    }
+}