diff --git a/docs/api/query.rst b/docs/api/query.rst index 48e375c3..598e0ced 100644 --- a/docs/api/query.rst +++ b/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. diff --git a/redisvl/query/sql.py b/redisvl/query/sql.py index 0788452e..a3900168 100644 --- a/redisvl/query/sql.py +++ b/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. diff --git a/tests/unit/test_sql_hybrid_policy.py b/tests/unit/test_sql_hybrid_policy.py new file mode 100644 index 00000000..4bc0d512 --- /dev/null +++ b/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})