From 35604e28319a0d8cc319af10fc30bda0e76c7997 Mon Sep 17 00:00:00 2001 From: slorello89 Date: Wed, 21 Oct 2026 14:47:36 -0400 Subject: [PATCH] feat: support explicit policies for filtered vector SQL queries --- .../user_guide/how_to_guides/vector-search.md | 23 +++ sql_redis/__init__.py | 3 +- sql_redis/executor.py | 61 +++++- sql_redis/translator.py | 46 ++++- tests/test_hybrid_policy_unit.py | 182 ++++++++++++++++++ 5 files changed, 303 insertions(+), 12 deletions(-) create mode 100644 tests/test_hybrid_policy_unit.py diff --git a/docs/user_guide/how_to_guides/vector-search.md b/docs/user_guide/how_to_guides/vector-search.md index f941929..dd269a2 100644 --- a/docs/user_guide/how_to_guides/vector-search.md +++ b/docs/user_guide/how_to_guides/vector-search.md @@ -54,6 +54,29 @@ result = executor.execute( The filter narrows the candidate set; the KNN runs over what survives. +## Explicit filter execution policy + +Some Redis deployments require an explicit policy when vector queries have a +pre-filter. Set `hybrid_policy` when creating the executor: + +```python +from sql_redis import create_executor + +executor = create_executor(client, hybrid_policy="ADHOC_BF") +``` + +The same option is accepted by `create_async_executor`, `Executor`, +`AsyncExecutor`, and `Translator`. Values are case-insensitive; choose +`"ADHOC_BF"` or `"BATCHES"`. Invalid values raise `ValueError`. + +- Filtered `FT.SEARCH` KNN queries emit `HYBRID_POLICY ADHOC_BF` or + `HYBRID_POLICY BATCHES`. +- Filtered `FT.HYBRID` queries emit `FILTER ... POLICY ADHOC` or + `FILTER ... POLICY BATCHES`. The public `ADHOC_BF` value maps to `ADHOC` + in this command's syntax. +- With the default `None`, or when the query has no pre-filter, translation + is unchanged. + ## Hybrid fusion (FT.HYBRID) `hybrid_vector_search()` fuses a full-text query and a vector query into a single diff --git a/sql_redis/__init__.py b/sql_redis/__init__.py index 2070068..984b43a 100644 --- a/sql_redis/__init__.py +++ b/sql_redis/__init__.py @@ -9,12 +9,13 @@ create_executor, ) from sql_redis.schema import AsyncSchemaRegistry, SchemaRegistry -from sql_redis.translator import TranslatedQuery, Translator +from sql_redis.translator import HybridPolicy, TranslatedQuery, Translator from sql_redis.version import __version__ __all__ = [ "Translator", "TranslatedQuery", + "HybridPolicy", "SchemaRegistry", "AsyncSchemaRegistry", "Executor", diff --git a/sql_redis/executor.py b/sql_redis/executor.py index a9549ba..a391adc 100644 --- a/sql_redis/executor.py +++ b/sql_redis/executor.py @@ -9,7 +9,7 @@ import redis from sql_redis.schema import AsyncSchemaRegistry, SchemaRegistry -from sql_redis.translator import Translator +from sql_redis.translator import HybridPolicy, Translator, _validate_hybrid_policy if TYPE_CHECKING: import redis.asyncio as async_redis @@ -209,11 +209,27 @@ def _resolve_score_alias( class Executor(_ScoreParseMixin): """Executes SQL queries against Redis.""" - def __init__(self, client: redis.Redis, schema_registry: SchemaRegistry) -> None: - """Initialize executor with Redis client and schema registry.""" + def __init__( + self, + client: redis.Redis, + schema_registry: SchemaRegistry, + *, + hybrid_policy: str | None = None, + ) -> None: + """Initialize executor with Redis client and schema registry. + + Args: + client: A synchronous Redis client. + schema_registry: A SchemaRegistry instance. + hybrid_policy: Optional filtered-vector policy, ``"ADHOC_BF"`` + or ``"BATCHES"``. See :class:`Translator` for command mapping. + """ self._client = client self._schema_registry = schema_registry - self._translator = Translator(schema_registry) + self._translator = Translator( + schema_registry, + hybrid_policy=hybrid_policy, + ) def execute(self, sql: str, *, params: dict | None = None) -> QueryResult: """Execute a SQL query and return results.""" @@ -348,16 +364,23 @@ def __init__( self, client: "async_redis.Redis", schema_registry: AsyncSchemaRegistry, + *, + hybrid_policy: str | None = None, ) -> None: """Initialize async executor with Redis client and schema registry. Args: client: An async Redis client (redis.asyncio.Redis). schema_registry: An AsyncSchemaRegistry instance. + hybrid_policy: Optional filtered-vector policy, ``"ADHOC_BF"`` + or ``"BATCHES"``. See :class:`Translator` for command mapping. """ self._client = client self._schema_registry = schema_registry - self._translator = Translator(schema_registry) + self._translator = Translator( + schema_registry, + hybrid_policy=hybrid_policy, + ) async def execute(self, sql: str, *, params: dict | None = None) -> QueryResult: """Execute a SQL query asynchronously and return results.""" @@ -486,6 +509,7 @@ def create_executor( *, schema_registry: SchemaRegistry | None = None, schema_cache_strategy: SchemaCacheStrategy = "lazy", + hybrid_policy: str | None = None, ) -> Executor: """Create a sync SQL executor with the requested schema cache strategy. @@ -495,14 +519,25 @@ def create_executor( schema_cache_strategy: Schema loading strategy. ``"lazy"`` defers ``FT.INFO`` calls until a referenced index is needed. ``"load_all"`` preserves the historical eager behavior by preloading all schemas. + hybrid_policy: Optional filtered-vector execution policy, ``"BATCHES"`` + or ``"ADHOC_BF"``. Applies to FT.SEARCH and FT.HYBRID; the latter + emits ``POLICY ADHOC`` for ``"ADHOC_BF"``. Unfiltered queries and + queries with no explicit policy are unchanged. """ schema_cache_strategy = _validate_schema_cache_strategy(schema_cache_strategy) + validated_hybrid_policy: HybridPolicy | None = _validate_hybrid_policy( + hybrid_policy + ) registry = schema_registry or SchemaRegistry(client) if schema_cache_strategy == "load_all": registry.load_all() - return Executor(client, registry) + return Executor( + client, + registry, + hybrid_policy=validated_hybrid_policy, + ) async def create_async_executor( @@ -510,6 +545,7 @@ async def create_async_executor( *, schema_registry: AsyncSchemaRegistry | None = None, schema_cache_strategy: SchemaCacheStrategy = "lazy", + hybrid_policy: str | None = None, ) -> AsyncExecutor: """Create an async SQL executor with the requested schema cache strategy. @@ -519,11 +555,22 @@ async def create_async_executor( schema_cache_strategy: Schema loading strategy. ``"lazy"`` defers ``FT.INFO`` calls until a referenced index is needed. ``"load_all"`` preserves the historical eager behavior by preloading all schemas. + hybrid_policy: Optional filtered-vector execution policy, ``"BATCHES"`` + or ``"ADHOC_BF"``. Applies to FT.SEARCH and FT.HYBRID; the latter + emits ``POLICY ADHOC`` for ``"ADHOC_BF"``. Unfiltered queries and + queries with no explicit policy are unchanged. """ schema_cache_strategy = _validate_schema_cache_strategy(schema_cache_strategy) + validated_hybrid_policy: HybridPolicy | None = _validate_hybrid_policy( + hybrid_policy + ) registry = schema_registry or AsyncSchemaRegistry(client) if schema_cache_strategy == "load_all": await registry.load_all() - return AsyncExecutor(client, registry) + return AsyncExecutor( + client, + registry, + hybrid_policy=validated_hybrid_policy, + ) diff --git a/sql_redis/translator.py b/sql_redis/translator.py index 372ce02..28770f3 100644 --- a/sql_redis/translator.py +++ b/sql_redis/translator.py @@ -5,6 +5,7 @@ import re import warnings from dataclasses import dataclass, field +from typing import Literal, cast from sql_redis.analyzer import AnalyzedQuery, Analyzer from sql_redis.parser import ( @@ -20,6 +21,18 @@ from sql_redis.query_builder import QueryBuilder from sql_redis.schema import AsyncSchemaRegistry, SchemaRegistry +HybridPolicy = Literal["BATCHES", "ADHOC_BF"] + + +def _validate_hybrid_policy(hybrid_policy: str | None) -> HybridPolicy | None: + """Validate the optional vector pre-filter execution policy.""" + if hybrid_policy is None: + return None + normalized = hybrid_policy.upper() + if normalized not in {"BATCHES", "ADHOC_BF"}: + raise ValueError("hybrid_policy must be one of: 'BATCHES', 'ADHOC_BF'") + return cast(HybridPolicy, normalized) + def _fmt_num(value: float) -> str: """Format a numeric FT.HYBRID parameter without trailing float noise.""" @@ -62,17 +75,28 @@ def to_command_string(self) -> str: class Translator: """Translates SQL queries to Redis FT.SEARCH/FT.AGGREGATE commands.""" - def __init__(self, schema_registry: SchemaRegistry | AsyncSchemaRegistry) -> None: + def __init__( + self, + schema_registry: SchemaRegistry | AsyncSchemaRegistry, + *, + hybrid_policy: str | None = None, + ) -> None: """Initialize translator with schema registry. Args: schema_registry: Registry containing index schemas. Can be either sync (SchemaRegistry) or async (AsyncSchemaRegistry) - only the sync get_schema() method is used. + hybrid_policy: Optional execution policy for filtered vector queries. + ``"BATCHES"`` or ``"ADHOC_BF"`` emits ``HYBRID_POLICY`` for + FT.SEARCH and ``FILTER ... POLICY`` for FT.HYBRID. FT.HYBRID + maps ``"ADHOC_BF"`` to ``"ADHOC"``. Queries without a + pre-filter are unchanged; None preserves the server default. """ self._schema_registry = schema_registry self._parser = SQLParser() self._query_builder = QueryBuilder() + self._hybrid_policy = _validate_hybrid_policy(hybrid_policy) def parse(self, sql: str) -> ParsedQuery: """Parse a SQL SELECT into a ParsedQuery AST. @@ -233,8 +257,13 @@ def _build_query_string(self, analyzed: AnalyzedQuery) -> str: vs = analyzed.vector_search # Vector search uses KNN syntax if analyzed.has_prefilter and combined: - # Prefilter: (filter)=>[KNN k @field $vec] - return f"({combined})=>[KNN {vs.k} @{vs.field} $vector AS {vs.alias}]" + # Prefilter: (filter)=>[KNN k @field $vec [HYBRID_POLICY p]] + policy = ( + f" HYBRID_POLICY {self._hybrid_policy}" + if self._hybrid_policy is not None + else "" + ) + return f"({combined})=>[KNN {vs.k} @{vs.field} $vector{policy} AS {vs.alias}]" else: # Pure KNN: *=>[KNN k @field $vec] return f"*=>[KNN {vs.k} @{vs.field} $vector AS {vs.alias}]" @@ -576,7 +605,16 @@ def _build_hybrid(self, analyzed: AnalyzedQuery) -> TranslatedQuery: method_tokens.extend(["EF_RUNTIME", str(spec.ef_runtime)]) args.extend(["KNN", str(len(method_tokens)), *method_tokens]) if has_filter: - args.extend(["FILTER", "1", filter_expr]) + filter_tokens = [filter_expr] + if self._hybrid_policy is not None: + # FT.HYBRID names the brute-force policy ADHOC, unlike FT.SEARCH. + policy = ( + "ADHOC" + if self._hybrid_policy == "ADHOC_BF" + else self._hybrid_policy + ) + filter_tokens.extend(["POLICY", policy]) + args.extend(["FILTER", str(len(filter_tokens)), *filter_tokens]) # COMBINE: RRF (default) or LINEAR. The combined score is yielded under # the SELECT alias so it comes back as a column. diff --git a/tests/test_hybrid_policy_unit.py b/tests/test_hybrid_policy_unit.py new file mode 100644 index 0000000..243ccac --- /dev/null +++ b/tests/test_hybrid_policy_unit.py @@ -0,0 +1,182 @@ +"""Unit coverage for explicit filtered-vector execution policy translation.""" + +from typing import cast + +import pytest +import redis +import redis.asyncio as async_redis + +from sql_redis.executor import create_async_executor, create_executor +from sql_redis.schema import AsyncSchemaRegistry, SchemaRegistry +from sql_redis.translator import Translator + + +class _StaticSchemaRegistry: + def get_schema(self, index: str) -> dict[str, str]: + if index != "vectors": + return {} + return { + "title": "TEXT", + "category": "TAG", + "embedding": "VECTOR", + } + + +FILTERED_SQL = ( + "SELECT title, vector_distance(embedding, :vec) AS score " + "FROM vectors WHERE category = 'books' LIMIT 5" +) +UNFILTERED_SQL = ( + "SELECT title, vector_distance(embedding, :vec) AS score " "FROM vectors LIMIT 5" +) + + +@pytest.mark.parametrize("policy", ["BATCHES", "ADHOC_BF"]) +def test_filtered_knn_emits_hybrid_policy(policy: str): + translator = Translator(_StaticSchemaRegistry(), hybrid_policy=policy) # type: ignore[arg-type] + + result = translator.translate(FILTERED_SQL) + + assert f"HYBRID_POLICY {policy}" in result.query_string + + +def test_hybrid_policy_is_normalized_to_uppercase(): + translator = Translator( # type: ignore[arg-type] + _StaticSchemaRegistry(), hybrid_policy="batches" + ) + + assert "HYBRID_POLICY BATCHES" in translator.translate(FILTERED_SQL).query_string + + +def test_hybrid_policy_does_not_change_unfiltered_knn(): + translator = Translator( # type: ignore[arg-type] + _StaticSchemaRegistry(), hybrid_policy="BATCHES" + ) + + assert "HYBRID_POLICY" not in translator.translate(UNFILTERED_SQL).query_string + + +def test_invalid_hybrid_policy_is_rejected(): + with pytest.raises(ValueError, match="hybrid_policy must be one of"): + Translator(_StaticSchemaRegistry(), hybrid_policy="invalid") # type: ignore[arg-type] + + +def test_sync_executor_factory_passes_hybrid_policy_to_translator(): + registry = cast(SchemaRegistry, _StaticSchemaRegistry()) + executor = create_executor( + cast(redis.Redis, object()), + schema_registry=registry, + hybrid_policy="ADHOC_BF", + ) + + assert ( + "HYBRID_POLICY ADHOC_BF" + in executor._translator.translate(FILTERED_SQL).query_string + ) + + +@pytest.mark.asyncio +async def test_async_executor_factory_passes_hybrid_policy_to_translator(): + registry = cast(AsyncSchemaRegistry, _StaticSchemaRegistry()) + executor = await create_async_executor( + cast(async_redis.Redis, object()), + schema_registry=registry, + hybrid_policy="BATCHES", + ) + + assert ( + "HYBRID_POLICY BATCHES" + in executor._translator.translate(FILTERED_SQL).query_string + ) + + +HYBRID_SQL = ( + "SELECT title, hybrid_vector_search(" + "cosine_distance(embedding, :vec), fulltext(title, 'redis'), rrf()" + ") AS score FROM vectors WHERE category = 'books' LIMIT 5" +) + + +@pytest.mark.parametrize( + "policy, expected", [("ADHOC_BF", "ADHOC"), ("batches", "BATCHES")] +) +@pytest.mark.parametrize("fusion", ["rrf()", "linear(alpha => 0.3)"]) +@pytest.mark.parametrize( + "vector", + [ + "cosine_distance(embedding, :vec)", + "vector_range(embedding, :vec, radius => 0.2)", + ], +) +def test_filtered_hybrid_policy_preserves_query_semantics( + policy, expected, fusion, vector +): + sql = HYBRID_SQL.replace("rrf()", fusion).replace( + "cosine_distance(embedding, :vec)", vector + ) + baseline = Translator(_StaticSchemaRegistry()).translate(sql).to_command_list() + command = ( + Translator(_StaticSchemaRegistry(), hybrid_policy=policy) + .translate(sql) + .to_command_list() + ) + position = command.index("FILTER") + assert command[position : position + 5] == [ + "FILTER", + "3", + "@category:{books}", + "POLICY", + expected, + ] + # Only the policy and FILTER argument count may change. + command[position + 1] = "1" + del command[position + 3 : position + 5] + assert command == baseline + + +@pytest.mark.parametrize("policy", ["ADHOC_BF", "BATCHES"]) +def test_unfiltered_hybrid_unchanged_with_policy(policy): + sql = HYBRID_SQL.replace(" WHERE category = 'books'", "") + baseline = Translator(_StaticSchemaRegistry()).translate(sql).to_command_list() + command = ( + Translator(_StaticSchemaRegistry(), hybrid_policy=policy) + .translate(sql) + .to_command_list() + ) + assert command == baseline + assert "FILTER" not in command + + +def test_filtered_hybrid_without_policy_preserves_default(): + command = ( + Translator(_StaticSchemaRegistry()).translate(HYBRID_SQL).to_command_list() + ) + position = command.index("FILTER") + assert command[position : position + 3] == ["FILTER", "1", "@category:{books}"] + assert "POLICY" not in command + + +@pytest.mark.parametrize( + "policy, expected", [("ADHOC_BF", "ADHOC"), ("BATCHES", "BATCHES")] +) +@pytest.mark.asyncio +async def test_executor_factories_forward_hybrid_filter_policy(policy, expected): + registry = _StaticSchemaRegistry() + sync_executor = create_executor( + cast(redis.Redis, object()), schema_registry=registry, hybrid_policy=policy + ) + async_executor = await create_async_executor( + cast(async_redis.Redis, object()), + schema_registry=registry, + hybrid_policy=policy, + ) + for executor in [sync_executor, async_executor]: + command = executor._translator.translate(HYBRID_SQL).to_command_list() + position = command.index("FILTER") + assert command[position : position + 5] == [ + "FILTER", + "3", + "@category:{books}", + "POLICY", + expected, + ]