feat(hybrid-search): support HYBRID_SEARCH SQL syntax for OceanBase 4…

README.md

@@ -399,16 +399,16 @@ client.insert(
             "source_id": "3b767712b57211f09c170242ac130008",
             "enabled": 1,
             "vector": [1, 1, 1],
-            "title": "企业版和社区版的功能差异",
-            "content": "OceanBase 数据库提供企业版和社区版两种形态。",
+            "title": "Differences between enterprise and community editions",
+            "content": "OceanBase database provides both enterprise and community editions.",
         },
         {
             "id": 2,
             "vector": [1, 2, 3],
             "enabled": 1,
             "source_id": "3b791472b57211f09c170242ac130008",
-            "title": "快速体验 OceanBase 社区版",
-            "content": "本文根据使用场景详细介绍如何快速部署 OceanBase 数据库。",
+            "title": "Quick start with OceanBase community edition",
+            "content": "This article introduces how to quickly deploy the OceanBase database in different scenarios.",
         },
         # ... more data
     ]
@@ -426,7 +426,7 @@ query = {
                 "query_string": {
                     "fields": ["title^10", "content"],  # field weights
                     "type": "best_fields",
-                    "query": "oceanbase 数据 迁移",
+                    "query": "oceanbase database migration",
                     "minimum_should_match": "30%",
                     "boost": 1
                 }
@@ -501,3 +501,103 @@ You can also get the actual SQL that will be executed:
 sql = client.get_sql(index=test_table_name, body=body)
 print(sql)  # prints the SQL query
 ```
+
+#### SQL-level Hybrid Search (OceanBase >= 4.6.0)
+
+Since OceanBase 4.6.0, hybrid search can be performed with the SQL-level `HYBRID_SEARCH` table function:
+
+```sql
+SELECT column, expr, ... FROM HYBRID_SEARCH(TABLE table_name, DSL_STRING);
+```
+
+Unlike the `DBMS_HYBRID_SEARCH` package interface (which composes a union SQL internally), the SQL-level syntax builds a logical fusion plan at plan stage, providing better hybrid search performance. `pyobvector` exposes it through `HybridSearch.sql_search`:
+
+```python
+from pyobvector.client.hybrid_search import HybridSearch
+
+client = HybridSearch(uri="127.0.0.1:2881", user="test@test")
+
+rows = client.sql_search(
+    table_name=test_table_name,
+    dsl={
+        # full-text route
+        "query": {"match": {"content": {"query": "oceanbase database", "boost": 0.3}}},
+        # vector route
+        "knn": {
+            "field": "vector",
+            "k": 5,
+            "query_vector": "[1, 2, 3]",
+            "boost": 0.7,
+        },
+        # fusion algorithm
+        "rank": {"rrf": {"rank_constant": 60, "rank_window_size": 10}},
+        "size": 10,
+    },
+)
+# rows is a list of dict, the relevance score of each row is in the `__score` field
+for row in rows:
+    print(row["id"], row["__score"])
+```
+
+**Note**: `sql_search` requires OceanBase version >= 4.6.0.0. The table must be a heap table (`ORGANIZATION = HEAP`, partitioned tables are supported). The `__score` relevance column is always included in the returned rows, even when only a subset of columns is requested via `columns`.
+
+##### DSL Reference
+
+The DSL string is a JSON document whose syntax is mostly compatible with Elasticsearch:
+
+- **Top-level keys**:
+  - `query`: full-text/scalar/json/array query route (scored)
+  - `knn`: vector search route, a single object or an array of objects (multi-path vector search)
+  - `rank`: fusion algorithm, `weighted_sum` (default) or `rrf`
+  - `min_score`: filter results whose final `__score` is below the threshold
+  - `from` / `size`: pagination (`from + size` must be in `[0, 10000]`, default `size` is 10)
+- **Full-text queries** (require full-text index on the searched columns):
+  - `match`: single field, multiple keywords. Supports `operator` (`OR`/`AND`), `minimum_should_match`, `boost`
+  - `match_phrase`: phrase search. Supports `slop`, `boost`
+  - `multi_match`: multiple fields. Supports `fields` (with weights like `title^0.3`), `type` (`best_fields`/`most_fields`), `operator`, `minimum_should_match`, `boost`
+  - `query_string`: like `multi_match`, plus keyword weights (e.g. `"query": "gatsby^0.2 dream"`) and `default_operator`
+- **Scalar queries** (non-scoring, cannot appear in scoring `must`/`should` of `bool`):
+  - `term` / `terms`: exact match(es)
+  - `range`: range conditions with `gt`/`gte`/`lt`/`lte`
+- **JSON queries**: `json_contains`, `json_overlaps`, `json_member_of` (with `candidate` and optional `path`); a dotted field name such as `doc_json.name` in scalar queries works like `json_extract`
+- **Array queries**: `array_contains`, `array_contains_all`, `array_overlaps`
+- **`bool` query**: combine sub-queries with `must` (scored), `should` (scored), `filter` (non-scored), `must_not`, plus `minimum_should_match` and `boost`. At least one positive clause (`must`/`should`/`filter`) is required
+- **`knn` parameters**:
+  - `field` (required): vector column name
+  - `k` (required, `[1, 16384]`): return top-K results
+  - `query_vector` (required): string form like `"[0.1, 0.2, 0.3]"` is recommended
+  - `similarity` (optional, `[0, 1]`): similarity threshold of this route (not supported for inner product)
+  - `boost` (optional): weight of this route in fusion
+  - `filter` (optional): per-route filter conditions, same syntax as `query` (non-scoring)
+  - `search_options` (optional): vector search tuning - `ef_search` (`[1, 1000]`), `refine_k` (`[1.0, 1000.0]`), `filter_mode` (`pre`, `pre-knn`, `pre-brute`, `post`, `post-index-merge`)
+- **`rank` fusion**:
+  - `weighted_sum` (default): sum of per-route scores weighted by outer `boost`, with optional `normalizer: "minmax"` to normalize per-route scores into `[0, 1]` first, and `rank_window_size`
+  - `rrf`: Reciprocal Rank Fusion, score `1 / (rank + rank_constant)` per route (`rank_constant` defaults to 60, `rank_window_size` must be >= `size`); setting per-route `boost` makes it weighted RRF
+
+Each `query`/`knn` route is an independent query: filters are not shared between routes, results are unioned and re-ranked by the fusion algorithm, and `size` rows are returned. Field names in the DSL are case-insensitive.
+
+##### Restrictions
+
+- Only heap tables are supported; partitioned tables are supported
+- Vector search requires a vector index (currently HNSW series only); full-text search requires a full-text index (a multi-column full-text index is not effective for hybrid search)
+- Scalar/JSON/ARRAY filter conditions work with or without indexes (indexes recommended)
+- `WHERE` / `ORDER BY` / `LIMIT` are not allowed at the same level as `HYBRID_SEARCH`. Filter or sort on the result with the `where` / `order_by` arguments of `sql_search` (the query is wrapped in a subquery automatically). **Security**: `where` and `order_by` are interpolated into the generated SQL verbatim, so they must be trusted SQL fragments and must never contain untrusted user input (SQL injection risk); prefer the DSL `filter` clauses for user-provided values:
+
+```python
+rows = client.sql_search(
+    table_name=test_table_name,
+    dsl={
+        "knn": {
+            "field": "vector",
+            "k": 10,
+            "query_vector": "[1, 2, 3]",
+            "filter": [{"range": {"id": {"gte": 5}}}],  # per-route filter
+        }
+    },
+    columns=["id", "title"],
+    where="enabled = 1",  # extra filter on the hybrid search result
+    order_by="id DESC",  # extra sorting on the hybrid search result
+)
+```
+
+- Multi-path vector search does not support sparse vectors; generated columns cannot be used in the DSL

pyobvector/client/hybrid_search.py

@@ -14,6 +14,11 @@
 logger.setLevel(logging.DEBUG)
 
 
+def _quote_identifier(identifier: str) -> str:
+    """Quote a MySQL/OceanBase identifier with backticks."""
+    return "`" + identifier.replace("`", "``") + "`"
+
+
 class HybridSearch(Client):
     """The OceanBase Hybrid Search Client"""
 
@@ -95,3 +100,110 @@ def get_sql(
                 if res[0] is None:
                     return ""
                 return res[0]
+
+    def _check_sql_search_version(self):
+        min_required_version = ObVersion.from_db_version_nums(4, 6, 0, 0)
+        if self.ob_version < min_required_version and not self._is_seekdb():
+            raise ClusterVersionException(
+                code=ErrorCode.NOT_SUPPORTED,
+                message=ExceptionsMessage.ClusterVersionIsLow
+                % ("Hybrid Search SQL syntax (HYBRID_SEARCH)", "4.6.0.0"),
+            )
+
+    def sql_search(
+        self,
+        table_name: str,
+        dsl: dict[str, Any] | str,
+        columns: list[str] | None = None,
+        where: str | None = None,
+        order_by: str | None = None,
+    ) -> list[dict[str, Any]]:
+        """Execute hybrid search with the SQL-level `HYBRID_SEARCH` syntax.
+
+        This uses the `HYBRID_SEARCH` table function introduced in OceanBase 4.6.0:
+        `SELECT ... FROM HYBRID_SEARCH(TABLE table_name, DSL_STRING)`.
+
+        Compared with `search` (based on the `DBMS_HYBRID_SEARCH` package), the
+        SQL-level syntax builds a logical fusion plan at plan stage and provides
+        better hybrid search performance. The DSL string is a JSON document whose
+        syntax is mostly compatible with Elasticsearch, e.g.::
+
+            {
+                "query": {"match": {"content": "python javascript"}},
+                "knn": {
+                    "field": "vector_col",
+                    "k": 5,
+                    "query_vector": "[0.1, 0.2, 0.3, 0.4]",
+                    "boost": 0.7,
+                },
+                "rank": {"rrf": {"rank_constant": 60, "rank_window_size": 10}},
+                "size": 10,
+            }
+
+        Args:
+            table_name: The name of the table to search. Only heap tables are
+                supported (partitioned tables included).
+            dsl: The hybrid search DSL, either a dict (serialized to JSON) or a
+                JSON string.
+            columns: Plain column names to select. If None, all columns of the
+                table plus the `__score` relevance column are returned. The
+                `__score` column is always included in the result, even when
+                it is not listed in `columns`.
+            where: Extra filter condition applied on the hybrid search result.
+                OceanBase does not allow WHERE/ORDER BY/LIMIT at the same level
+                as `HYBRID_SEARCH`, so the query is wrapped in a subquery
+                automatically when `where` or `order_by` is set. This fragment
+                is interpolated into the SQL statement verbatim: it must be a
+                trusted SQL fragment and must never contain untrusted user
+                input, otherwise it becomes a SQL injection risk. Prefer the
+                DSL `filter` clauses for user-provided values.
+            order_by: Sort expression applied on the hybrid search result, e.g.
+                `id DESC` (wrapped in a subquery automatically). Like `where`,
+                this fragment is interpolated verbatim and must be trusted SQL
+                that never contains untrusted user input.
+
+        Returns:
+            A list of rows (dict). The relevance score of each row is in the
+            `__score` field.
+
+        Raises:
+            ClusterVersionException: If the OceanBase cluster version is below
+                4.6.0.0.
+        """
+        self._check_sql_search_version()
+
+        if isinstance(dsl, dict):
+            dsl_str = json.dumps(dsl, ensure_ascii=False)
+        else:
+            dsl_str = dsl
+
+        if columns is None:
+            col_expr = "*"
+        else:
+            # `HYBRID_SEARCH` always produces a `__score` relevance column.
+            # Keep it in the projection so that every returned row carries
+            # its score even when a column subset is requested.
+            selected = [_quote_identifier(c) for c in columns]
+            if all(c.lower() != "__score" for c in columns):
+                selected.append(_quote_identifier("__score"))
+            col_expr = ", ".join(selected)
+        hybrid_search_from = (
+            f"HYBRID_SEARCH(TABLE {_quote_identifier(table_name)}, :dsl)"
+        )
+
+        if where is None and order_by is None:
+            stmt = f"SELECT {col_expr} FROM {hybrid_search_from}"
+        else:
+            stmt = (
+                f"SELECT {col_expr} FROM "
+                f"(SELECT * FROM {hybrid_search_from}) AS __hybrid_search_result"
+            )
+            if where is not None:
+                stmt += f" WHERE {where}"
+            if order_by is not None:
+                stmt += f" ORDER BY {order_by}"
+
+        with self.engine.connect() as conn:
+            with conn.begin():
+                res = conn.execute(text(stmt), {"dsl": dsl_str})
+                return [dict(row) for row in res.mappings().fetchall()]

pyobvector/json_table/virtual_data_type.py

@@ -78,7 +78,9 @@ def check_float(v):
         else:
             integer_part, decimal_part = decimal_str, ""
 
-        integer_count = len(integer_part.lstrip("-"))  # 去掉负号的长度
+        integer_count = len(
+            integer_part.lstrip("-")
+        )  # length without the negative sign
         decimal_count = len(decimal_part)
 
         if integer_count + min(decimal_count, y) > x:

source/pyobvector.client.rst

@@ -28,6 +28,14 @@ pyobvector.client.exceptions module
    :undoc-members:
    :show-inheritance:
 
+pyobvector.client.hybrid\_search module
+---------------------------------------
+
+.. automodule:: pyobvector.client.hybrid_search
+   :members:
+   :undoc-members:
+   :show-inheritance:
+
 pyobvector.client.index\_param module
 -------------------------------------
 

tests/test_fts_index.py

@@ -165,8 +165,8 @@ def test_fts_parser_types(self):
                 # Insert test data
                 test_data = [
                     {"id": 1, "doc": "OceanBase is a distributed database"},
-                    {"id": 2, "doc": "全文索引测试 Full text search test"},
-                    {"id": 3, "doc": "我喜欢编程 I like coding"},
+                    {"id": 2, "doc": "Full text search index test"},
+                    {"id": 3, "doc": "I like coding and programming"},
                 ]
                 self.client.insert(test_collection_name, data=test_data)
 
@@ -276,18 +276,18 @@ def test_fts_multi_fields(self):
         test_data = [
             {
                 "id": 1,
-                "title": "OceanBase 数据库",
+                "title": "OceanBase database",
                 "content": "OceanBase is a distributed database",
             },
-            {"id": 2, "title": "全文索引", "content": "Full text search index"},
+            {"id": 2, "title": "Full text search", "content": "Full text search index"},
         ]
         self.client.insert(test_collection_name, data=test_data)
 
         # Test searching title field
         res_title = self.client.get(
             test_collection_name,
             ids=None,
-            where_clause=[MatchAgainst("数据库", "title")],
+            where_clause=[MatchAgainst("database", "title")],
             output_column_name=["id", "title"],
             n_limits=10,
         )
@@ -315,19 +315,19 @@ def test_fts_multi_fields(self):
 
         self.client.drop_table_if_exist(test_collection_name)
 
-    def test_fts_chinese_search(self):
-        """Test Chinese full-text search"""
-        test_collection_name = "fts_chinese_test"
+    def test_fts_ik_parser_search(self):
+        """Test full-text search with the IK parser"""
+        test_collection_name = "fts_ik_parser_test"
         self.client.drop_table_if_exist(test_collection_name)
 
         cols = [
             Column("id", Integer, primary_key=True, autoincrement=False),
             Column("doc", TEXT),
         ]
 
-        # Use IK parser, suitable for Chinese
+        # Use IK parser
         fts_index_param = FtsIndexParam(
-            index_name="fts_idx_chinese",
+            index_name="fts_idx_ik",
             field_names=["doc"],
             parser_type=FtsParser.IK,
         )
@@ -339,27 +339,27 @@ def test_fts_chinese_search(self):
         )
 
         test_data = [
-            {"id": 1, "doc": "海洋数据库 OceanBase"},
-            {"id": 2, "doc": "全文索引功能测试"},
-            {"id": 3, "doc": "我喜欢使用 OceanBase 数据库"},
-            {"id": 4, "doc": "测试数据 test data"},
+            {"id": 1, "doc": "Ocean database OceanBase"},
+            {"id": 2, "doc": "Full text search feature test"},
+            {"id": 3, "doc": "I like using OceanBase database"},
+            {"id": 4, "doc": "sample test data"},
         ]
         self.client.insert(test_collection_name, data=test_data)
 
-        # Test Chinese search
+        # Test search with the IK parser
         res = self.client.get(
             test_collection_name,
             ids=None,
-            where_clause=[MatchAgainst("数据库", "doc")],
+            where_clause=[MatchAgainst("database", "doc")],
             output_column_name=["id", "doc"],
             n_limits=10,
         )
         rows = res.fetchall()
-        # Verify that Chinese full-text search works by checking at least one result
+        # Verify that full-text search works by checking at least one result
         self.assertGreater(
             len(rows),
             0,
-            "Chinese search should return at least one result for '数据库'",
+            "IK parser search should return at least one result for 'database'",
         )
 
         self.client.drop_table_if_exist(test_collection_name)