Validate EF_RUNTIME query parameter (#319)

The EF_RUNTIME parameter is only valid in queries that target vector
fields using the HNSW algorithm. If you run the same query against a
'flat' vector field, Redis will return an error. Because we have all the
information to prevent you from running the bad query, we can introduce
validation that raises a helpful error instead of relying on Redis error
handling.

---------

Co-authored-by: Tyler Hutcherson <[email protected]>

Author: abrookins

.cursor/rules/redisvl.mdc

@@ -0,0 +1,11 @@
+---
+description: 
+globs: 
+alwaysApply: true
+---
+
+# Rules for working on RedisVL
+- Do not change this line of code unless explicitly asked. It's already correct:
+```
+token.strip().strip(",").replace("“", "").replace("”", "").lower()
+```

redisvl/exceptions.py

@@ -30,3 +30,9 @@ def __init__(self, message, index=None):
         if index is not None:
             message = f"Validation failed for object at index {index}: {message}"
         super().__init__(message)
+
+
+class QueryValidationError(RedisVLError):
+    """Error when validating a query."""
+
+    pass

redisvl/index/index.py

@@ -18,6 +18,7 @@
     Union,
 )
 
+from redisvl.query.query import VectorQuery
 from redisvl.redis.utils import convert_bytes, make_dict
 from redisvl.utils.utils import deprecated_argument, deprecated_function, sync_wrapper
 
@@ -34,6 +35,7 @@
 from redis.commands.search.indexDefinition import IndexDefinition
 
 from redisvl.exceptions import (
+    QueryValidationError,
     RedisModuleVersionError,
     RedisSearchError,
     RedisVLError,
@@ -46,16 +48,18 @@
     BaseVectorQuery,
     CountQuery,
     FilterQuery,
-    HybridQuery,
 )
 from redisvl.query.filter import FilterExpression
 from redisvl.redis.connection import (
     RedisConnectionFactory,
     convert_index_info_to_schema,
 )
-from redisvl.redis.utils import convert_bytes
 from redisvl.schema import IndexSchema, StorageType
-from redisvl.schema.fields import VECTOR_NORM_MAP, VectorDistanceMetric
+from redisvl.schema.fields import (
+    VECTOR_NORM_MAP,
+    VectorDistanceMetric,
+    VectorIndexAlgorithm,
+)
 from redisvl.utils.log import get_logger
 
 logger = get_logger(__name__)
@@ -194,6 +198,15 @@ def _storage(self) -> BaseStorage:
             index_schema=self.schema
         )
 
+    def _validate_query(self, query: BaseQuery) -> None:
+        """Validate a query."""
+        if isinstance(query, VectorQuery):
+            field = self.schema.fields[query._vector_field_name]
+            if query.ef_runtime and field.attrs.algorithm != VectorIndexAlgorithm.HNSW:  # type: ignore
+                raise QueryValidationError(
+                    "Vector field using 'flat' algorithm does not support EF_RUNTIME query parameter."
+                )
+
     @property
     def name(self) -> str:
         """The name of the Redis search index."""
@@ -592,6 +605,27 @@ def drop_keys(self, keys: Union[str, List[str]]) -> int:
         else:
             return self._redis_client.delete(keys)  # type: ignore
 
+    def drop_documents(self, ids: Union[str, List[str]]) -> int:
+        """Remove documents from the index by their document IDs.
+
+        This method converts document IDs to Redis keys automatically by applying
+        the index's key prefix and separator configuration.
+
+        Args:
+            ids (Union[str, List[str]]): The document ID or IDs to remove from the index.
+
+        Returns:
+            int: Count of documents deleted from Redis.
+        """
+        if isinstance(ids, list):
+            if not ids:
+                return 0
+            keys = [self.key(id) for id in ids]
+            return self._redis_client.delete(*keys)  # type: ignore
+        else:
+            key = self.key(ids)
+            return self._redis_client.delete(key)  # type: ignore
+
     def expire_keys(
         self, keys: Union[str, List[str]], ttl: int
     ) -> Union[int, List[int]]:
@@ -816,6 +850,10 @@ def batch_query(
 
     def _query(self, query: BaseQuery) -> List[Dict[str, Any]]:
         """Execute a query and process results."""
+        try:
+            self._validate_query(query)
+        except QueryValidationError as e:
+            raise QueryValidationError(f"Invalid query: {str(e)}") from e
         results = self.search(query.query, query_params=query.params)
         return process_results(results, query=query, schema=self.schema)
 
@@ -1236,6 +1274,28 @@ async def drop_keys(self, keys: Union[str, List[str]]) -> int:
         else:
             return await client.delete(keys)
 
+    async def drop_documents(self, ids: Union[str, List[str]]) -> int:
+        """Remove documents from the index by their document IDs.
+
+        This method converts document IDs to Redis keys automatically by applying
+        the index's key prefix and separator configuration.
+
+        Args:
+            ids (Union[str, List[str]]): The document ID or IDs to remove from the index.
+
+        Returns:
+            int: Count of documents deleted from Redis.
+        """
+        client = await self._get_client()
+        if isinstance(ids, list):
+            if not ids:
+                return 0
+            keys = [self.key(id) for id in ids]
+            return await client.delete(*keys)
+        else:
+            key = self.key(ids)
+            return await client.delete(key)
+
     async def expire_keys(
         self, keys: Union[str, List[str]], ttl: int
     ) -> Union[int, List[int]]:
@@ -1356,9 +1416,10 @@ async def fetch(self, id: str) -> Optional[Dict[str, Any]]:
     async def _aggregate(
         self, aggregation_query: AggregationQuery
     ) -> List[Dict[str, Any]]:
-        """Execute an aggretation query and processes the results."""
+        """Execute an aggregation query and processes the results."""
         results = await self.aggregate(
-            aggregation_query, query_params=aggregation_query.params  # type: ignore[attr-defined]
+            aggregation_query,
+            query_params=aggregation_query.params,  # type: ignore[attr-defined]
         )
         return process_aggregate_results(
             results,
@@ -1486,6 +1547,10 @@ async def batch_query(
 
     async def _query(self, query: BaseQuery) -> List[Dict[str, Any]]:
         """Asynchronously execute a query and process results."""
+        try:
+            self._validate_query(query)
+        except QueryValidationError as e:
+            raise QueryValidationError(f"Invalid query: {str(e)}") from e
         results = await self.search(query.query, query_params=query.params)
         return process_results(results, query=query, schema=self.schema)