Expose hybrid_policy on SQLQuery

docs/api/query.rst

@@ -267,3 +267,19 @@ SQLQuery
    fusing a text and a vector query server-side. This is the SQL front-end to
    :class:`HybridQuery` and requires ``sql-redis >= 0.7.0``, Redis 8.4+, and
    redis-py >= 7.1.0.
+
+.. note::
+   For filtered vector queries, pass ``hybrid_policy="ADHOC_BF"`` or
+   ``hybrid_policy="BATCHES"`` to ``SQLQuery``. The value is forwarded to
+   ``sql-redis`` for translation and for synchronous or asynchronous execution.
+   For ``FT.SEARCH``, the translator emits ``HYBRID_POLICY``; for ``FT.HYBRID``,
+   it emits ``FILTER ... POLICY``, mapping ``ADHOC_BF`` to ``ADHOC``.
+   Unfiltered queries are unaffected.
+
+   An explicit ``hybrid_policy`` takes precedence over the same key in
+   ``sql_redis_options``. Omitting it preserves the options dictionary, which
+   is copied rather than modified.
+
+   This requires a ``sql-redis`` version implementing the ``hybrid_policy``
+   executor option; ``sql-redis 0.7.1`` does not support it. The RedisVL
+   argument alone does not add policy translation to older sql-redis builds.

redisvl/query/sql.py

@@ -52,6 +52,7 @@ def __init__(
         sql: str,
         params: dict[str, Any] | None = None,
         *,
+        hybrid_policy: str | None = None,
         sql_redis_options: dict[str, Any] | None = None,
     ):
         """Initialize a SQLQuery.
@@ -60,6 +61,15 @@ def __init__(
             sql: The SQL SELECT statement to execute.
             params: Optional dictionary of parameters for parameterized queries.
                    Useful for passing vector data for similarity searches.
+            hybrid_policy: Optional execution policy for filtered vector queries.
+                Use ``"BATCHES"`` or ``"ADHOC_BF"`` to emit ``HYBRID_POLICY``
+                for FT.SEARCH or ``FILTER ... POLICY`` for FT.HYBRID.
+                FT.HYBRID maps ``"ADHOC_BF"`` to ``"ADHOC"``. Requires a
+                sql-redis version supporting the ``hybrid_policy`` executor
+                option (not available in sql-redis 0.7.1). An explicit value
+                overrides ``sql_redis_options["hybrid_policy"]``. With None,
+                the options dictionary is used unchanged. Queries without a
+                pre-filter are unchanged by the policy.
             sql_redis_options: Optional passthrough options forwarded to
                 ``sql-redis`` executor creation. Use this to tune how SQL
                 query translation loads and caches index schema metadata.
@@ -78,6 +88,8 @@ def __init__(
         self.sql = sql
         self.params = params or {}
         self.sql_redis_options = dict(sql_redis_options or {})
+        if hybrid_policy is not None:
+            self.sql_redis_options["hybrid_policy"] = hybrid_policy
 
     def _substitute_params(self, sql: str, params: dict[str, Any]) -> str:
         """Substitute parameter placeholders in SQL with actual values.

tests/unit/test_sql_hybrid_policy.py

@@ -0,0 +1,114 @@
+"""Policy forwarding through SQL translation and sync/async index execution."""
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, call, patch
+
+import pytest
+
+from redisvl.index import AsyncSearchIndex, SearchIndex
+from redisvl.query import SQLQuery
+
+SCHEMA = {"index": {"name": "idx", "prefix": "idx:"}, "fields": []}
+SQL = "SELECT title FROM idx"
+
+
+@pytest.mark.parametrize("policy", ["ADHOC_BF", "BATCHES"])
+def test_policy_forwarded_for_translation(policy):
+    client = MagicMock()
+    executor = MagicMock()
+    executor._translator.translate.return_value.to_command_string.return_value = "cmd"
+    factory = MagicMock(return_value=executor)
+    query = SQLQuery(SQL, hybrid_policy=policy)
+
+    with patch.dict(
+        "sys.modules", {"sql_redis": SimpleNamespace(create_executor=factory)}
+    ):
+        assert query.redis_query_string(redis_client=client) == "cmd"
+
+    factory.assert_called_once_with(
+        client, schema_cache_strategy="lazy", hybrid_policy=policy
+    )
+    executor._translator.translate.assert_called_once_with(SQL)
+
+
+def test_explicit_policy_overrides_options_without_mutating_them():
+    options = {"hybrid_policy": "BATCHES", "schema_cache_strategy": "load_all"}
+    query = SQLQuery(SQL, hybrid_policy="ADHOC_BF", sql_redis_options=options)
+
+    assert query.sql_redis_options == {
+        "hybrid_policy": "ADHOC_BF",
+        "schema_cache_strategy": "load_all",
+    }
+    assert options == {"hybrid_policy": "BATCHES", "schema_cache_strategy": "load_all"}
+
+
+@pytest.mark.parametrize("options", [{}, {"hybrid_policy": "BATCHES"}])
+def test_none_preserves_existing_options(options):
+    query = SQLQuery(SQL, hybrid_policy=None, sql_redis_options=options)
+    assert query.sql_redis_options == options
+    assert query.sql_redis_options is not options
+
+
+def test_omitted_policy_not_forwarded_to_legacy_executor():
+    client = MagicMock()
+    executor = MagicMock()
+    factory = MagicMock(return_value=executor)
+    with patch.dict(
+        "sys.modules", {"sql_redis": SimpleNamespace(create_executor=factory)}
+    ):
+        SQLQuery(SQL).redis_query_string(redis_client=client)
+    factory.assert_called_once_with(client, schema_cache_strategy="lazy")
+
+
+def test_sync_query_keeps_policy_executor_caches_separate():
+    client = MagicMock()
+    index = SearchIndex.from_dict(SCHEMA, redis_client=client)
+    result = SimpleNamespace(rows=[{"title": "Redis"}])
+    executors = [MagicMock(), MagicMock(), MagicMock()]
+    for executor in executors:
+        executor.execute.return_value = result
+    factory = MagicMock(side_effect=executors)
+    with patch.dict(
+        "sys.modules", {"sql_redis": SimpleNamespace(create_executor=factory)}
+    ):
+        for policy in [None, "ADHOC_BF", "BATCHES", "ADHOC_BF", None]:
+            query = SQLQuery(SQL, params={"id": 1}, hybrid_policy=policy)
+            assert index.query(query) == result.rows
+
+    assert factory.call_args_list == [
+        call(client, schema_cache_strategy="lazy"),
+        call(client, schema_cache_strategy="lazy", hybrid_policy="ADHOC_BF"),
+        call(client, schema_cache_strategy="lazy", hybrid_policy="BATCHES"),
+    ]
+    assert [e.execute.call_count for e in executors] == [2, 2, 1]
+    for executor in executors:
+        executor.execute.assert_called_with(SQL, params={"id": 1})
+
+
+@pytest.mark.asyncio
+async def test_async_query_keeps_policy_executor_caches_separate():
+    client = MagicMock()
+    index = AsyncSearchIndex.from_dict(SCHEMA)
+    result = SimpleNamespace(rows=[{"title": "Redis"}])
+    executors = [
+        SimpleNamespace(execute=AsyncMock(return_value=result)) for _ in range(3)
+    ]
+    factory = AsyncMock(side_effect=executors)
+    with (
+        patch.dict(
+            "sys.modules", {"sql_redis": SimpleNamespace(create_async_executor=factory)}
+        ),
+        patch.object(index, "_get_client", new=AsyncMock(return_value=client)),
+    ):
+        for policy in [None, "ADHOC_BF", "BATCHES", "ADHOC_BF", None]:
+            query = SQLQuery(SQL, params={"id": 1}, hybrid_policy=policy)
+            assert await index.query(query) == result.rows
+
+    assert factory.await_args_list == [
+        call(client, schema_cache_strategy="lazy"),
+        call(client, schema_cache_strategy="lazy", hybrid_policy="ADHOC_BF"),
+        call(client, schema_cache_strategy="lazy", hybrid_policy="BATCHES"),
+    ]
+    assert [e.execute.await_count for e in executors] == [2, 2, 1]
+    for executor in executors:
+        executor.execute.assert_awaited_with(SQL, params={"id": 1})