Support hybrid_policy for filtered FT.SEARCH and FT.HYBRID queries

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

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",

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,21 +519,33 @@ 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(
     client: "async_redis.Redis",
     *,
     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,
+    )

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.