feat(vector_store): add OceanBase as a VectorStore

pyproject.toml

@@ -49,6 +49,9 @@ dependencies = [
     "openai>=2.8.1",
     "pandas>=2.3.3",
     "pydantic>=2.12.4",
+    "pyobvector>=0.1.20",
+    # pyobvector imports Expression from sqlglot; removed from sqlglot 30+ top-level API
+    "sqlglot>=25,<30",
     "qdrant-client>=1.16.0",
     "tavily-python>=0.7.13",
     "tiktoken>=0.12.0",

reme/core/vector_store/__init__.py

@@ -4,6 +4,7 @@ from .base_vector_store import BaseVectorStore
 from .chroma_vector_store import ChromaVectorStore
 from .es_vector_store import ESVectorStore
 from .local_vector_store import LocalVectorStore
+from .obvec_vector_store import ObVecVectorStore
 from .pgvector_store import PGVectorStore
 from .qdrant_vector_store import QdrantVectorStore
 from ..registry_factory import R
@@ -13,6 +14,7 @@ __all__ = [
     "ChromaVectorStore",
     "ESVectorStore",
     "LocalVectorStore",
+    "ObVecVectorStore",
     "PGVectorStore",
     "QdrantVectorStore",
 ]
@@ -20,5 +22,6 @@ __all__ = [
 R.vector_stores.register("chroma")(ChromaVectorStore)
 R.vector_stores.register("es")(ESVectorStore)
 R.vector_stores.register("local")(LocalVectorStore)
+R.vector_stores.register("obvec")(ObVecVectorStore)
 R.vector_stores.register("pgvector")(PGVectorStore)
 R.vector_stores.register("qdrant")(QdrantVectorStore)

reme/core/vector_store/obvec_vector_store.py

@@ -0,0 +1,609 @@
+"""OceanBase vector store implementation for ReMe.
+
+This module provides an OceanBase-based vector store that implements the BaseVectorStore
+interface for high-performance dense vector storage and retrieval using pyobvector.
+"""
+
+import json
+from pathlib import Path
+from typing import Any
+
+from loguru import logger
+
+from .base_vector_store import BaseVectorStore
+from ..embedding import BaseEmbeddingModel
+from ..schema import VectorNode
+
+_OBVECTOR_IMPORT_ERROR: Exception | None = None
+
+try:
+    from pyobvector import IndexParams, ObVecClient, VecIndexType, VECTOR
+except Exception as e:
+    _OBVECTOR_IMPORT_ERROR = e
+    IndexParams = None
+    ObVecClient = None
+    VecIndexType = None
+    VECTOR = None
+
+
+class ObVecVectorStore(BaseVectorStore):
+    """OceanBase-based vector store for dense vector storage and kNN search."""
+
+    @staticmethod
+    def _coerce_db_vector(raw: Any) -> list[float] | None:
+        if raw is None:
+            return None
+        if isinstance(raw, list):
+            return [float(x) for x in raw]
+        if isinstance(raw, str):
+            try:
+                parsed = json.loads(raw)
+                if isinstance(parsed, list):
+                    return [float(x) for x in parsed]
+            except (json.JSONDecodeError, TypeError, ValueError):
+                pass
+        return None
+
+    @staticmethod
+    def _coerce_db_metadata(raw: Any) -> dict[str, Any]:
+        if raw is None:
+            return {}
+        if isinstance(raw, dict):
+            return raw
+        if isinstance(raw, str):
+            try:
+                parsed = json.loads(raw)
+                if isinstance(parsed, dict):
+                    return parsed
+            except (json.JSONDecodeError, TypeError):
+                pass
+        return {}
+
+    def __init__(
+        self,
+        collection_name: str,
+        db_path: str | Path,
+        embedding_model: BaseEmbeddingModel,
+        uri: str = "127.0.0.1:2881",
+        user: str = "root",
+        password: str = "",
+        database: str = "test",
+        index_type: str = "HNSW",
+        index_metric: str = "cosine",
+        index_ef_search: int = 100,
+        **kwargs,
+    ):
+        """Initialize the OceanBase vector store with connection parameters.
+
+        Args:
+            collection_name: Name of the collection (table).
+            db_path: Database path (not used for remote OceanBase, kept for API consistency).
+            embedding_model: Model instance used to generate vector embeddings.
+            uri: Connection URI for OceanBase server.
+            user: Username for authentication.
+            password: Password for authentication.
+            database: Database name to use.
+            index_type: Type of vector index (HNSW).
+            index_metric: Distance metric (cosine, l2, ip).
+            index_ef_search: HNSW ef_search parameter.
+            **kwargs: Additional configuration passed to the base class.
+        """
+        if _OBVECTOR_IMPORT_ERROR is not None:
+            raise ImportError(
+                "ObVecVectorStore requires extra dependencies. Install with `pip install pyobvector`",
+            ) from _OBVECTOR_IMPORT_ERROR
+
+        super().__init__(
+            collection_name=collection_name,
+            db_path=db_path,
+            embedding_model=embedding_model,
+            **kwargs,
+        )
+
+        self.uri = uri
+        self.user = user
+        self.password = password
+        self.database = database
+        self.index_type = index_type.upper()
+        self.index_metric = index_metric.lower()
+        self.index_ef_search = index_ef_search
+
+        self.client: ObVecClient | None = None
+        self.embedding_model_dims = embedding_model.dimensions
+
+    async def list_collections(self) -> list[str]:
+        """List all available table names in the current database."""
+        # OceanBase doesn't have a direct list collections API
+        # We'll return the current collection name if it exists
+        if self.client is None:
+            return []
+        try:
+            result = self.client.perform_raw_text_sql(
+                f"SHOW TABLES FROM `{self.database}`",
+            )
+            rows = result.fetchall()
+            return [row[0] for row in rows if row]
+        except Exception as e:
+            logger.warning(f"Failed to list collections: {e}")
+            return []
+
+    async def create_collection(self, collection_name: str, **kwargs):
+        """Create a new table with vector support and appropriate indexing."""
+        dimensions = kwargs.get("dimensions", self.embedding_model_dims)
+
+        if self.client is None:
+            logger.warning("Client not initialized, skipping collection creation")
+            return
+
+        # Check if table already exists
+        if self.client.check_table_exists(collection_name):
+            logger.info(f"Collection {collection_name} already exists")
+            return
+
+        from sqlalchemy import Column, String, Text, JSON, create_engine
+        from sqlalchemy.dialects.mysql import LONGTEXT
+
+        # Create columns
+        columns = [
+            Column("id", String(255), primary_key=True),
+            Column("content", LONGTEXT),
+            Column("vector", VECTOR(dimensions)),
+            Column("metadata", JSON),
+        ]
+
+        vidxs: IndexParams | None = None
+        if self.index_type == "HNSW":
+            metric_map = {
+                "cosine": "cosine",
+                "l2": "l2_distance",
+                "ip": "inner_product",
+            }
+            metric = metric_map.get(self.index_metric, "cosine")
+            vidxs = IndexParams()
+            vidxs.add_index(
+                "vector",
+                VecIndexType.HNSW,
+                f"{collection_name}_vidx",
+                metric_type=metric,
+                params={"efSearch": self.index_ef_search},
+            )
+
+        try:
+            # Create table with index
+            self.client.create_table_with_index_params(
+                table_name=collection_name,
+                columns=columns,
+                vidxs=vidxs,
+            )
+            logger.info(f"Created collection {collection_name} with dimensions={dimensions}")
+        except Exception as e:
+            logger.error(f"Failed to create collection {collection_name}: {e}")
+            raise
+
+    async def delete_collection(self, collection_name: str, **kwargs):
+        """Permanently delete a collection table from the database."""
+        if self.client is None:
+            logger.warning("Client not initialized, skipping collection deletion")
+            return
+
+        try:
+            self.client.drop_table_if_exist(collection_name)
+            logger.info(f"Deleted collection {collection_name}")
+        except Exception as e:
+            logger.error(f"Failed to delete collection {collection_name}: {e}")
+            raise
+
+    async def copy_collection(self, collection_name: str, **kwargs):
+        """Duplicate the current collection to a new one with the given name."""
+        if self.client is None:
+            logger.warning("Client not initialized, skipping collection copy")
+            return
+
+        # Check if source collection exists
+        if not self.client.check_table_exists(self.collection_name):
+            raise ValueError(f"Source collection {self.collection_name} does not exist")
+
+        # Create new collection with same structure
+        await self.create_collection(collection_name)
+
+        # Copy data from source to target
+        try:
+            # Get all data from source
+            source_data = await self.list(limit=None)
+
+            # Insert into new collection
+            if source_data:
+                await self.insert(source_data, collection_name=collection_name)
+
+            logger.info(f"Copied collection {self.collection_name} to {collection_name}")
+        except Exception as e:
+            # Clean up the created collection if copy fails
+            try:
+                self.client.drop_table_if_exist(collection_name)
+            except:
+                pass
+            raise
+
+    async def insert(self, nodes: VectorNode | list[VectorNode], **kwargs):
+        """Insert or upsert vector nodes into the OceanBase collection."""
+        if isinstance(nodes, VectorNode):
+            nodes = [nodes]
+
+        if not nodes:
+            return
+
+        nodes_without_vectors = [node for node in nodes if node.vector is None]
+        if nodes_without_vectors:
+            nodes_with_vectors = await self.get_node_embeddings(nodes_without_vectors)
+            vector_map = {n.vector_id: n for n in nodes_with_vectors}
+            nodes_to_insert = [vector_map.get(n.vector_id, n) if n.vector is None else n for n in nodes]
+        else:
+            nodes_to_insert = nodes
+
+        # Prepare data for insertion
+        data = []
+        for node in nodes_to_insert:
+            # Convert vector to list if it's not already
+            vector_list = node.vector if node.vector is not None else []
+
+            data.append({
+                "id": node.vector_id,
+                "content": node.content,
+                "vector": vector_list,
+                "metadata": node.metadata if node.metadata else {},
+            })
+
+        # Determine which collection to use
+        target_collection = kwargs.get("collection_name", self.collection_name)
+
+        try:
+            self.client.insert(
+                table_name=target_collection,
+                data=data,
+            )
+            logger.info(f"Inserted {len(nodes_to_insert)} documents into {target_collection}")
+        except Exception as e:
+            logger.error(f"Failed to insert documents: {e}")
+            raise
+
+    def _build_filter_clause(self, filters: dict | None) -> str:
+        """Generate a WHERE clause from filter dictionary.
+
+        Supports two filter formats:
+        1. Range query: {"field": [start_value, end_value]} - filters for field >= start_value AND field <= end_value
+        2. Exact match: {"field": value} - filters for field == value
+        """
+        if not filters:
+            return ""
+
+        conditions = []
+        for key, value in filters.items():
+            # Sanitize key to prevent SQL injection
+            if not key.replace("_", "").replace(".", "").isalnum():
+                continue
+
+            # New syntax: [start, end] represents a range query
+            if isinstance(value, list) and len(value) == 2:
+                lo, hi = value[0], value[1]
+                if isinstance(lo, (int, float)) and isinstance(hi, (int, float)):
+                    conditions.append(
+                        f"(JSON_EXTRACT(metadata, '$.{key}') >= {lo} AND "
+                        f"JSON_EXTRACT(metadata, '$.{key}') <= {hi})",
+                    )
+                else:
+                    conditions.append(
+                        f"(JSON_EXTRACT(metadata, '$.{key}') >= '{lo}' AND "
+                        f"JSON_EXTRACT(metadata, '$.{key}') <= '{hi}')",
+                    )
+            else:
+                if isinstance(value, (int, float)):
+                    conditions.append(f"JSON_EXTRACT(metadata, '$.{key}') = {value}")
+                else:
+                    conditions.append(f"JSON_EXTRACT(metadata, '$.{key}') = '{value}'")
+
+        return " AND ".join(conditions)
+
+    async def search(
+        self,
+        query: str,
+        limit: int = 5,
+        filters: dict | None = None,
+        **kwargs,
+    ) -> list[VectorNode]:
+        """Perform a kNN similarity search based on a text query."""
+        query_vector = await self.get_embedding(query)
+        if hasattr(query_vector, "tolist"):
+            query_vector = query_vector.tolist()
+        else:
+            query_vector = [float(x) for x in query_vector]
+
+        # Determine distance function based on metric
+        if self.index_metric == "cosine":
+            from pyobvector import cosine_distance
+            distance_func = cosine_distance
+        elif self.index_metric == "l2":
+            from pyobvector import l2_distance
+            distance_func = l2_distance
+        else:  # ip (inner product)
+            from pyobvector import inner_product
+            distance_func = inner_product
+
+        # ann_search passes where_clause to SQLAlchemy where(*...); use text(), not raw str.
+        from sqlalchemy import text as sa_text
+
+        filter_sql = self._build_filter_clause(filters)
+        where_parts = [sa_text(filter_sql)] if filter_sql else None
+
+        try:
+            # pyobvector expects vec_data as a flat list of floats, not [vector].
+            results = self.client.ann_search(
+                table_name=self.collection_name,
+                vec_data=query_vector,
+                vec_column_name="vector",
+                distance_func=distance_func,
+                with_dist=True,
+                topk=limit,
+                output_column_names=["id", "content", "metadata"],
+                where_clause=where_parts,
+            )
+
+            search_results = []
+            score_threshold = kwargs.get("score_threshold")
+
+            for row in results:
+                # row format: (id, content, metadata, distance)
+                if len(row) >= 4:
+                    vector_id = row[0]
+                    content = row[1]
+                    metadata_str = row[2]
+                    distance = row[3]
+
+                    # Convert distance to similarity score
+                    if self.index_metric == "cosine":
+                        score = max(0.0, 1.0 - distance / 2.0)
+                    elif self.index_metric == "l2":
+                        # For L2, we need to normalize - this is a rough approximation
+                        score = max(0.0, 1.0 - distance / 2.0)
+                    else:  # inner product
+                        # For IP, higher is better, no conversion needed
+                        score = max(0.0, distance)
+
+                    if score_threshold is not None and score < score_threshold:
+                        continue
+
+                    # Parse metadata
+                    metadata = {}
+                    if metadata_str:
+                        metadata = self._coerce_db_metadata(metadata_str)
+
+                    metadata["score"] = score
+                    metadata["_distance"] = distance
+
+                    search_results.append(
+                        VectorNode(
+                            vector_id=vector_id,
+                            content=content or "",
+                            vector=None,  # Don't return vector in search results
+                            metadata=metadata,
+                        ),
+                    )
+
+            return search_results
+        except Exception as e:
+            logger.error(f"Search failed: {e}")
+            return []
+
+    async def delete(self, vector_ids: str | list[str], **kwargs):
+        """Remove specific vector records from the collection by their IDs."""
+        if isinstance(vector_ids, str):
+            vector_ids = [vector_ids]
+
+        if not vector_ids:
+            return
+
+        try:
+            self.client.delete(self.collection_name, ids=vector_ids)
+            logger.info(f"Deleted {len(vector_ids)} documents from {self.collection_name}")
+        except Exception as e:
+            logger.error(f"Failed to delete documents: {e}")
+            raise
+
+    async def delete_all(self, **kwargs):
+        """Remove all vectors from the collection."""
+        try:
+            self.client.delete(self.collection_name)
+            logger.info(f"Deleted all documents from {self.collection_name}")
+        except Exception as e:
+            logger.error(f"Failed to delete all documents: {e}")
+            raise
+
+    async def update(self, nodes: VectorNode | list[VectorNode], **kwargs):
+        """Update existing vector nodes with new content, embeddings, or metadata."""
+        if isinstance(nodes, VectorNode):
+            nodes = [nodes]
+
+        if not nodes:
+            return
+
+        nodes_without_vectors = [node for node in nodes if node.vector is None and node.content]
+        if nodes_without_vectors:
+            nodes_with_vectors = await self.get_node_embeddings(nodes_without_vectors)
+            vector_map = {n.vector_id: n for n in nodes_with_vectors}
+            nodes_to_update = [vector_map.get(n.vector_id, n) if n.vector is None and n.content else n for n in nodes]
+        else:
+            nodes_to_update = nodes
+
+        try:
+            from sqlalchemy import text
+
+            for node in nodes_to_update:
+                updates: list[str] = []
+                params: dict[str, Any] = {}
+
+                if node.content is not None:
+                    updates.append("content = :content")
+                    params["content"] = node.content
+
+                if node.vector is not None:
+                    updates.append("vector = :vector")
+                    # VECTOR column expects a single literal, not a Python list bound as multi-column
+                    params["vector"] = "[" + ",".join(str(float(v)) for v in node.vector) + "]"
+
+                if node.metadata is not None:
+                    updates.append("metadata = :metadata")
+                    params["metadata"] = json.dumps(node.metadata)
+
+                if updates:
+                    params["vid"] = node.vector_id
+                    update_sql = (
+                        f"UPDATE `{self.collection_name}` SET {', '.join(updates)} WHERE id = :vid"
+                    )
+                    with self.client.engine.connect() as conn:
+                        with conn.begin():
+                            conn.execute(text(update_sql), params)
+
+            logger.info(f"Updated {len(nodes_to_update)} documents in {self.collection_name}")
+        except Exception as e:
+            logger.error(f"Failed to update documents: {e}")
+            raise
+
+    async def get(self, vector_ids: str | list[str]) -> VectorNode | list[VectorNode] | None:
+        """Retrieve vector nodes by their unique identifiers."""
+        single_result = isinstance(vector_ids, str)
+        if single_result:
+            vector_ids = [vector_ids]
+
+        if not vector_ids:
+            return [] if not single_result else None
+
+        try:
+            ids_str = "', '".join(vector_ids)
+            select_sql = (
+                f"SELECT id, content, vector, metadata FROM `{self.collection_name}` "
+                f"WHERE id IN ('{ids_str}')"
+            )
+
+            result = self.client.perform_raw_text_sql(select_sql)
+            rows = result.fetchall()
+
+            results = []
+            for row in rows:
+                if row:
+                    results.append(
+                        VectorNode(
+                            vector_id=row[0],
+                            content=row[1] or "",
+                            vector=self._coerce_db_vector(row[2]),
+                            metadata=self._coerce_db_metadata(row[3]),
+                        ),
+                    )
+
+            if single_result:
+                return results[0] if results else None
+            return results
+        except Exception as e:
+            logger.error(f"Failed to get documents: {e}")
+            return [] if not single_result else None
+
+    async def list(
+        self,
+        filters: dict | None = None,
+        limit: int | None = None,
+        sort_key: str | None = None,
+        reverse: bool = False,
+    ) -> list[VectorNode]:
+        """Retrieve a list of vector nodes matching the provided filters and limit.
+
+        Args:
+            filters: Dictionary of filter conditions to match vectors
+            limit: Maximum number of vectors to return
+            sort_key: Key to sort the results by (e.g., field name in metadata). None for no sorting
+            reverse: If True, sort in descending order; if False, sort in ascending order
+        """
+        try:
+            # Build select SQL
+            select_sql = f"SELECT id, content, vector, metadata FROM `{self.collection_name}`"
+
+            where_clause = self._build_filter_clause(filters)
+            if where_clause:
+                select_sql += f" WHERE {where_clause}"
+
+            # Add sorting
+            if sort_key:
+                order = "DESC" if reverse else "ASC"
+                select_sql += f" ORDER BY JSON_EXTRACT(metadata, '$.{sort_key}') {order}"
+
+            # Add limit
+            if limit:
+                select_sql += f" LIMIT {limit}"
+
+            result = self.client.perform_raw_text_sql(select_sql)
+            rows = result.fetchall()
+
+            results = []
+            for row in rows:
+                if row:
+                    results.append(
+                        VectorNode(
+                            vector_id=row[0],
+                            content=row[1] or "",
+                            vector=self._coerce_db_vector(row[2]),
+                            metadata=self._coerce_db_metadata(row[3]),
+                        ),
+                    )
+
+            return results
+        except Exception as e:
+            logger.error(f"Failed to list documents: {e}")
+            return []
+
+    async def collection_info(self) -> dict[str, Any]:
+        """Fetch metadata including record count for the collection."""
+        try:
+            count_sql = f"SELECT COUNT(*) FROM `{self.collection_name}`"
+            result = self.client.perform_raw_text_sql(count_sql)
+            row = result.fetchone()
+            count = row[0] if row else 0
+
+            return {
+                "name": self.collection_name,
+                "count": count,
+            }
+        except Exception as e:
+            logger.error(f"Failed to get collection info: {e}")
+            return {"name": self.collection_name, "count": 0}
+
+    async def reset(self):
+        """Purge all data by dropping and recreating the collection table."""
+        logger.warning(f"Resetting collection {self.collection_name}...")
+        await self.delete_collection(self.collection_name)
+        await self.create_collection(self.collection_name)
+
+    async def reset_collection(self, collection_name: str):
+        """Reset collection with the given name."""
+        self.collection_name = collection_name
+        await self.create_collection(collection_name)
+        logger.info(f"Collection reset to {collection_name}")
+
+    async def start(self) -> None:
+        """Initialize the OceanBase client and ensure the collection exists.
+
+        Creates the collection table if it doesn't exist.
+        """
+        # Initialize the client
+        self.client = ObVecClient(
+            uri=self.uri,
+            user=self.user,
+            password=self.password,
+            db_name=self.database,
+        )
+
+        await super().start()
+        logger.info(f"OceanBase collection {self.collection_name} initialized")
+
+    async def close(self):
+        """Terminate the OceanBase client connection and release resources."""
+        # Note: ObVecClient may not have a close method
+        # We'll just log the closure
+        self.client = None
+        logger.info("OceanBase client connection closed")

tests/test_vector_store.py

@@ -11,12 +11,14 @@ Usage:
     python test_vector_store.py --pgvector   # Test PGVectorStore only
     python test_vector_store.py --qdrant     # Test QdrantVectorStore only
     python test_vector_store.py --chroma     # Test ChromaVectorStore only
+    python test_vector_store.py --obvec      # Test ObVecVectorStore only
     python test_vector_store.py --all        # Test all vector stores
 
 """
 
 import argparse
 import asyncio
+import os
 import shutil
 import tempfile
 from pathlib import Path
@@ -32,6 +34,7 @@ from reme.core.vector_store import (
     ChromaVectorStore,
     LocalVectorStore,
     ESVectorStore,
+    ObVecVectorStore,
     PGVectorStore,
     QdrantVectorStore,
 )
@@ -73,6 +76,13 @@ class TestConfig:
     CHROMA_TENANT = None  # Set for ChromaDB Cloud tenant
     CHROMA_DATABASE = None  # Set for ChromaDB Cloud database
 
+    # ObVecVectorStore: SeekDB uses user `root`; OceanBase multi-tenant often uses `root@test`.
+    # Defaults match docker-compose.obvec.yml (ROOT_PASSWORD=root).
+    OBVEC_URI = os.environ.get("OBVEC_URI", "127.0.0.1:2881")
+    OBVEC_USER = os.environ.get("OBVEC_USER", "root")
+    OBVEC_PASSWORD = os.environ.get("OBVEC_PASSWORD", "root")
+    OBVEC_DATABASE = os.environ.get("OBVEC_DATABASE", "test")
+
     # Embedding model settings
     EMBEDDING_MODEL_NAME = "text-embedding-v4"
     EMBEDDING_DIMENSIONS = 64
@@ -182,7 +192,7 @@ def get_store_type(store: BaseVectorStore) -> str:
         store: Vector store instance
 
     Returns:
-        str: Type identifier ("local", "es", "pgvector", "qdrant", or "chroma")
+        str: Type identifier ("local", "es", "pgvector", "qdrant", "chroma", or "obvec")
     """
     if isinstance(store, LocalVectorStore):
         return "local"
@@ -194,6 +204,8 @@ def get_store_type(store: BaseVectorStore) -> str:
         return "pgvector"
     elif isinstance(store, ChromaVectorStore):
         return "chroma"
+    elif isinstance(store, ObVecVectorStore):
+        return "obvec"
     else:
         raise ValueError(f"Unknown vector store type: {type(store)}")
 
@@ -202,7 +214,7 @@ def create_vector_store(store_type: str, collection_name: str) -> BaseVectorStor
     """Create a vector store instance based on type.
 
     Args:
-        store_type: Type of vector store ("local", "es", "pgvector", "qdrant", or "chroma")
+        store_type: Type of vector store ("local", "es", "pgvector", "qdrant", "chroma", or "obvec")
         collection_name: Name of the collection
 
     Returns:
@@ -264,6 +276,18 @@ def create_vector_store(store_type: str, collection_name: str) -> BaseVectorStor
             tenant=config.CHROMA_TENANT,
             database=config.CHROMA_DATABASE,
         )
+    elif store_type == "obvec":
+        return ObVecVectorStore(
+            collection_name=collection_name,
+            embedding_model=embedding_model,
+            db_path=tempfile.mkdtemp(prefix="test_obvec_"),
+            uri=config.OBVEC_URI,
+            user=config.OBVEC_USER,
+            password=config.OBVEC_PASSWORD,
+            database=config.OBVEC_DATABASE,
+            index_metric="cosine",
+            index_ef_search=100,
+        )
     else:
         raise ValueError(f"Unknown store type: {store_type}")
 
@@ -581,9 +605,9 @@ async def test_copy_collection(store: BaseVectorStore, store_name: str):
     config = TestConfig()
     copy_collection_name = f"{config.TEST_COLLECTION_PREFIX}_{store_name}_copy"
 
-    # Elasticsearch and PostgreSQL require lowercase table/index names
+    # Elasticsearch, PostgreSQL and OceanBase require lowercase table/index names
     store_type = get_store_type(store)
-    if store_type in ("es", "pgvector"):
+    if store_type in ("es", "pgvector", "obvec"):
         copy_collection_name = copy_collection_name.lower()
 
     # Clean up if exists
@@ -1800,6 +1824,11 @@ Examples:
         action="store_true",
         help="Test ChromaVectorStore",
     )
+    parser.add_argument(
+        "--obvec",
+        action="store_true",
+        help="Test ObVecVectorStore",
+    )
     parser.add_argument(
         "--all",
         action="store_true",
@@ -1818,6 +1847,7 @@ Examples:
             ("pgvector", "PGVectorStore"),
             ("qdrant", "QdrantVectorStore"),
             ("chroma", "ChromaVectorStore"),
+            ("obvec", "ObVecVectorStore"),
         ]
     else:
         # Build list based on individual flags
@@ -1831,6 +1861,8 @@ Examples:
             stores_to_test.append(("qdrant", "QdrantVectorStore"))
         if args.chroma:
             stores_to_test.append(("chroma", "ChromaVectorStore"))
+        if args.obvec:
+            stores_to_test.append(("obvec", "ObVecVectorStore"))
 
         if not stores_to_test:
             # Default to all vector stores if no argument provided
@@ -1840,10 +1872,11 @@ Examples:
                 ("pgvector", "PGVectorStore"),
                 ("qdrant", "QdrantVectorStore"),
                 ("chroma", "ChromaVectorStore"),
+                ("obvec", "ObVecVectorStore"),
             ]
             print("No vector store specified, defaulting to test all vector stores")
             print(
-                "Use --local/--es/--pgvector/--qdrant/--chroma to test specific ones\n",
+                "Use --local/--es/--pgvector/--qdrant/--chroma/--obvec to test specific ones\n",
             )
 
     # Run tests for each vector store