[ISSUE-851] implemented VectorStore and VectorStoreMetadata interfaces with validation and persistence

geaflow-ai/src/main/java/org/apache/geaflow/ai/index/vectorstore/DistanceMetric.java

@@ -0,0 +1,24 @@
+/*
+ * 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.geaflow.ai.index.vectorstore;
+
+public enum DistanceMetric {
+    COSINE,
+    L2,
+    DOT_PRODUCT
+}

geaflow-ai/src/main/java/org/apache/geaflow/ai/index/vectorstore/DistanceUtils.java

@@ -0,0 +1,70 @@
+/*
+ * 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.geaflow.ai.index.vectorstore;
+
+public class DistanceUtils {
+
+    public static double compute(double[] a, double[] b, DistanceMetric metric) {
+        if (a.length != b.length) {
+            throw new VectorStoreException(VectorStoreException.ErrorCode.DIMENSION_MISMATCH,
+                "Vector lengths differ: " + a.length + " vs " + b.length);
+        }
+        switch (metric) {
+            case COSINE:
+                return cosineSimilarity(a, b);
+            case L2:
+                return 1.0 / (1.0 + euclideanDistance(a, b));
+            case DOT_PRODUCT:
+                return dotProduct(a, b);
+            default:
+                throw new IllegalArgumentException("Unknown metric: " + metric);
+        }
+    }
+
+    private static double cosineSimilarity(double[] a, double[] b) {
+        double dot = 0.0;
+        double normA = 0.0;
+        double normB = 0.0;
+        for (int i = 0; i < a.length; i++) {
+            dot += a[i] * b[i];
+            normA += a[i] * a[i];
+            normB += b[i] * b[i];
+        }
+        if (normA == 0.0 || normB == 0.0) {
+            return 0.0;
+        }
+        return dot / (Math.sqrt(normA) * Math.sqrt(normB));
+    }
+
+    private static double euclideanDistance(double[] a, double[] b) {
+        double sum = 0.0;
+        for (int i = 0; i < a.length; i++) {
+            double diff = a[i] - b[i];
+            sum += diff * diff;
+        }
+        return Math.sqrt(sum);
+    }
+
+    private static double dotProduct(double[] a, double[] b) {
+        double dot = 0.0;
+        for (int i = 0; i < a.length; i++) {
+            dot += a[i] * b[i];
+        }
+        return dot;
+    }
+}

geaflow-ai/src/main/java/org/apache/geaflow/ai/index/vectorstore/InMemoryVectorStore.java

@@ -0,0 +1,119 @@
+/*
+ * 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.geaflow.ai.index.vectorstore;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.PriorityQueue;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class InMemoryVectorStore implements VectorStore {
+    private final VectorStoreMetadata metadata;
+    private final ConcurrentHashMap<String, VectorRecord> store = new ConcurrentHashMap<>();
+    private final Set<String> deletedIds = ConcurrentHashMap.newKeySet();
+
+    public InMemoryVectorStore(VectorStoreMetadata metadata) {
+        this.metadata = Objects.requireNonNull(metadata);
+    }
+
+    @Override
+    public void upsert(VectorRecord record) {
+        if (record.getEmbedding().length != metadata.getDimension()) {
+            throw new VectorStoreException(VectorStoreException.ErrorCode.DIMENSION_MISMATCH,
+                    "Expected dimension " + metadata.getDimension() + ", got " + record.getEmbedding().length);
+        }
+        store.put(record.getVectorId(), record);
+        deletedIds.remove(record.getVectorId());
+    }
+
+    @Override
+    public void upsertBatch(List<VectorRecord> records) {
+        for (VectorRecord record : records) {
+            upsert(record);
+        }
+    }
+
+    @Override
+    public List<VectorHit> search(VectorQuery query) {
+        if (query.getFilterMetadata().containsKey("model_name")) {
+            String filterModel = query.getFilterMetadata().get("model_name");
+            if (!Objects.equals(filterModel, metadata.getModelName())) {
+                throw new VectorStoreException(VectorStoreException.ErrorCode.MODEL_MISMATCH,
+                        "Expected model " + metadata.getModelName() + ", got " + filterModel);
+            }
+        }
+
+        PriorityQueue<VectorHit> hits = new PriorityQueue<>(query.getTopK(), Comparator.comparingDouble(VectorHit::getScore));
+        for (VectorRecord record : store.values()) {
+            if (!deletedIds.contains(record.getVectorId())) {
+                boolean match = true;
+                for (Map.Entry<String, String> entry : query.getFilterMetadata().entrySet()) {
+                    if (entry.getKey().equals("model_name")) {
+                        continue;
+                    }
+                    if (!Objects.equals(record.getMetadata().get(entry.getKey()), entry.getValue()) && !Objects.equals(record.getSourceType(), entry.getValue())) {
+                        match = false;
+                        break;
+                    }
+                }
+                if (match) {
+                    double score = DistanceUtils.compute(query.getQueryVector(), record.getEmbedding(), metadata.getDistance());
+
+                    if (hits.size() < query.getTopK()) {
+                        hits.add(new VectorHit(record.getVectorId(), score, record));
+                    } else if (score > hits.peek().getScore()) {
+                        hits.poll();
+                        hits.add(new VectorHit(record.getVectorId(), score, record));
+                    }
+                }
+            }
+        }
+
+        List<VectorHit> topKHits = new ArrayList<>();
+        while (!hits.isEmpty()) {
+            topKHits.add(hits.poll());
+        }
+
+        Collections.reverse(topKHits);
+        return topKHits;
+    }
+
+    @Override
+    public void markDeleted(String vectorId) {
+        if (!store.containsKey(vectorId)) {
+            throw new VectorStoreException(VectorStoreException.ErrorCode.RECORD_NOT_FOUND,
+                    "Record not found: " + vectorId);
+        }
+        deletedIds.add(vectorId);
+    }
+
+    @Override
+    public VectorStoreMetadata getMetadata() {
+        return metadata;
+    }
+
+    @Override
+    public void close() {
+
+    }
+}