Python: fix(vectors): compare in-memory filters in the field's own type

python/packages/core/agent_framework/_in_memory.py

@@ -5,11 +5,12 @@
 from __future__ import annotations
 
 import math
-from collections.abc import Mapping, Sequence
+import re
+from collections.abc import Callable, Mapping, Sequence
 from copy import deepcopy
 from dataclasses import dataclass
 from datetime import date, datetime, time, timedelta
-from decimal import Decimal
+from decimal import ROUND_HALF_EVEN, Decimal
 from typing import Any, ClassVar, Generic, cast
 from uuid import UUID, uuid4
 
@@ -18,8 +19,10 @@
 from ._feature_stage import ExperimentalFeature, experimental
 from ._telemetry import FeatureIndex, mark_feature_used
 from ._vector_filters import (
+    _TEXT_VALUE_OPERATORS,  # pyright: ignore[reportPrivateUsage]
     Filter,
     FilterExpression,
+    FilterGroup,
     validate_filter,
 )
 from ._vectors import (
@@ -152,6 +155,120 @@ def _calculate_score(left: Vector, right: Vector, distance_function: DistanceFun
     return score
 
 
+_DURATION_PATTERN = re.compile(
+    r"^(?P<sign>-)?P(?:(?P<days>[0-9]+(?:\.[0-9]+)?)D)?"
+    r"(?:T(?:(?P<hours>[0-9]+(?:\.[0-9]+)?)H)?(?:(?P<minutes>[0-9]+(?:\.[0-9]+)?)M)?"
+    r"(?:(?P<seconds>[0-9]+(?:\.[0-9]+)?)S)?)?$"
+)
+
+
+def _restore_datetime(value: str) -> datetime:
+    """Parse a timestamp the shared serializer wrote."""
+    if value.endswith("Z"):
+        value = f"{value[:-1]}+00:00"
+    return datetime.fromisoformat(value)
+
+
+def _restore_timedelta(value: str) -> timedelta:
+    """Parse the ISO 8601 duration the shared serializer wrote.
+
+    The components are summed as exact decimals of microseconds: the string
+    the serializer writes carries every microsecond, while a ``float`` of the
+    total seconds loses them on the largest durations (``timedelta.max`` even
+    rounds past the range the type supports).
+    """
+    match = _DURATION_PATTERN.match(value)
+    if match is None:
+        raise ValueError(f"Unsupported duration '{value}'.")
+    parts = match.groupdict()
+    microseconds = (
+        Decimal(parts["days"] or "0") * 86_400_000_000
+        + Decimal(parts["hours"] or "0") * 3_600_000_000
+        + Decimal(parts["minutes"] or "0") * 60_000_000
+        + Decimal(parts["seconds"] or "0") * 1_000_000
+    )
+    delta = timedelta(microseconds=int(microseconds.quantize(Decimal(1), rounding=ROUND_HALF_EVEN)))
+    return -delta if parts["sign"] else delta
+
+
+def _restore_time(value: str) -> time:
+    """Parse a time of day the shared serializer wrote."""
+    if value.endswith("Z"):
+        # The serializer writes a UTC time with a trailing "Z", which
+        # ``time.fromisoformat`` accepts only from Python 3.11.
+        value = f"{value[:-1]}+00:00"
+    return time.fromisoformat(value)
+
+
+_RESTORE_IN_MEMORY_VALUE_TYPES: dict[str, Callable[[str], Any]] = {
+    "datetime": _restore_datetime,
+    "date": date.fromisoformat,
+    "time": _restore_time,
+    "timedelta": _restore_timedelta,
+    "UUID": UUID,
+    "Decimal": Decimal,
+}
+_SERIALIZED_SCALAR_TYPE_NAMES = frozenset(_RESTORE_IN_MEMORY_VALUE_TYPES)
+_COMPARISON_OPERATORS = frozenset({"eq", "ne", "gt", "gte", "lt", "lte"})
+_SEQUENCE_OPERAND_OPERATORS = frozenset({"between", "in", "not_in"})
+
+
+def _restore_in_memory_value(value: Any, type_tag: str | None) -> Any:
+    """Return a stored value in the type its field declares.
+
+    In-memory records are stored through the shared serializer, so a native
+    scalar such as a datetime is held as its serialized value. A filter
+    operand arrives native, so the stored value is restored before the
+    comparison and the comparison keeps the field's semantics.
+    """
+    if not isinstance(value, str) or type_tag is None:
+        return value
+    restore = _RESTORE_IN_MEMORY_VALUE_TYPES.get(type_tag)
+    if restore is None:
+        return value
+    try:
+        return restore(value)
+    except (TypeError, ValueError):
+        return value
+
+
+def _evaluate_in_memory_filter(
+    expression: FilterExpression,
+    record: Mapping[str, Any],
+    definition: VectorStoreCollectionDefinition,
+) -> bool:
+    """Evaluate a filter against the record as it is stored, one leaf at a time.
+
+    A comparison operator restores its own leaf's field to the type the field
+    declares, so the native operand compares in the field's semantics. The
+    text operators stay on the stored string, which lets one group mix a text
+    operator and a native comparison over the same field.
+    """
+    if isinstance(expression, FilterGroup):
+        values = (_evaluate_in_memory_filter(item, record, definition) for item in expression.filters)
+        match expression.operator:
+            case "and":
+                return all(values)
+            case "or":
+                return any(values)
+            case "not":
+                return not next(values)
+            case _:
+                raise ValueError(f"Unknown filter group operator '{expression.operator}'.")
+    field = definition.try_get_field(expression.field_name)
+    if (
+        field is None
+        or field.type_ not in _SERIALIZED_SCALAR_TYPE_NAMES
+        or expression.operator in _TEXT_VALUE_OPERATORS
+    ):
+        return _evaluate_filter(expression, record, definition)
+    restored = dict(record)
+    storage_name = field.storage_name or field.name
+    if storage_name in restored:
+        restored[storage_name] = _restore_in_memory_value(restored[storage_name], field.type_)
+    return _evaluate_filter(expression, restored, definition)
+
+
 def _validate_in_memory_filter_value(value: Any) -> None:
     if value is None or isinstance(value, _SCALAR_FILTER_TYPES):
         if isinstance(value, float) and not math.isfinite(value):
@@ -307,7 +424,8 @@ async def _inner_get(
         records = list(self._state.records.values())
         if filter is not None:
             _validate_in_memory_filter(filter, self.definition)
-            records = [record for record in records if _evaluate_filter(filter, record, self.definition)]
+            _reject_incompatible_filter_operands(filter, self.definition)
+            records = [record for record in records if _evaluate_in_memory_filter(filter, record, self.definition)]
         if order_by:
             for field_name, ascending in reversed(tuple(order_by.items())):
                 if not isinstance(ascending, bool):
@@ -362,6 +480,7 @@ async def _inner_search(
         _validate_vector_dimensions(query_vector, vector_field)
         if filter is not None:
             _validate_in_memory_filter(filter, self.definition)
+            _reject_incompatible_filter_operands(filter, self.definition)
 
         distance_function = vector_field.distance_function or "DEFAULT"
         if distance_function == "DEFAULT":
@@ -375,7 +494,7 @@ async def _inner_search(
         results: list[dict[str, Any]] = []
         key_storage_name = self.definition.key_field_storage_name
         for record in self._state.records.values():
-            if filter is not None and not _evaluate_filter(filter, record, self.definition):
+            if filter is not None and not _evaluate_in_memory_filter(filter, record, self.definition):
                 continue
             stored_vector = record.get(storage_name)
             if stored_vector is None:
@@ -434,6 +553,57 @@ def _validate_in_memory_filter(
         _validate_in_memory_filter_value(item.value)
 
 
+def _filter_operand_values(item: Filter) -> tuple[Any, ...]:
+    """Return the operands a filter compares against the field value."""
+    if item.operator in _COMPARISON_OPERATORS:
+        return (item.value,)
+    if item.operator in _SEQUENCE_OPERAND_OPERATORS:
+        value = item.value
+        if isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
+            return tuple(cast(Sequence[Any], value))
+        return (value,)
+    return ()
+
+
+def _operand_matches_field_type(field_type: str, operand: Any) -> bool:
+    """Return True when the operand can be compared in the type the field declares."""
+    if operand is None:
+        return True
+    operand_type = type(operand).__name__
+    if field_type in _SERIALIZED_SCALAR_TYPE_NAMES:
+        return operand_type == field_type
+    return operand_type not in _SERIALIZED_SCALAR_TYPE_NAMES
+
+
+def _reject_incompatible_filter_operands(
+    expression: FilterExpression,
+    definition: VectorStoreCollectionDefinition,
+) -> None:
+    """Reject an operand that cannot be compared in the type the field declares.
+
+    A record is stored through the shared serializer, so the scalar types it
+    writes as text are restored before the comparison. An operand of another
+    type cannot be compared in the field's own semantics, so it is rejected
+    instead of quietly matching no record. The text operators are exempt:
+    they compare the stored string itself, and their operand is already
+    required to be a string by the shared filter validation.
+    """
+    for item in _walk_filters(expression):
+        field = definition.try_get_field(item.field_name)
+        if field is None or field.type_ is None:
+            continue
+        if item.operator in _TEXT_VALUE_OPERATORS:
+            continue
+        for operand in _filter_operand_values(item):
+            if _operand_matches_field_type(field.type_, operand):
+                continue
+            raise TypeError(
+                f"InMemoryCollection cannot compare field '{item.field_name}' of type '{field.type_}' "
+                f"with a filter operand of type '{type(operand).__name__}'. The in-memory connector "
+                "compares native scalars in the type the field declares."
+            )
+
+
 @experimental(feature_id=ExperimentalFeature.VECTOR_STORES)
 class InMemoryStore(BaseVectorStore):
     """Create in-memory collection clients that share process-local state."""