From 12f3e21d78e83c7f8914f9ed25a5c21c7013eedc Mon Sep 17 00:00:00 2001 From: Weijian He <230282609+Madin-H23@users.noreply.github.com> Date: Wed, 21 Oct 2026 18:00:55 +0800 Subject: [PATCH 1/5] Python: fix(vectors): compare in-memory filter operands in the stored representation Closes #9438 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, and the two sides were compared as they were: an equality filter matched nothing and an ordered comparison raised, even though the connector accepts those operand types. The operand now goes through the same serializer before the comparison, so both sides are in the stored representation. The filter tree is rebuilt with the normalized operands, so nested groups are covered too. The change is scoped to the in-memory connector, which is the one that keeps the serialized form; the other connectors store native values and compare them natively. --- .../core/agent_framework/_in_memory.py | 42 +++++++ .../core/tests/core/test_in_memory.py | 114 ++++++++++++++++++ 2 files changed, 156 insertions(+) diff --git a/python/packages/core/agent_framework/_in_memory.py b/python/packages/core/agent_framework/_in_memory.py index cc8f8046cb7..c211dcf186b 100644 --- a/python/packages/core/agent_framework/_in_memory.py +++ b/python/packages/core/agent_framework/_in_memory.py @@ -5,6 +5,7 @@ from __future__ import annotations import math +import msgspec from collections.abc import Mapping, Sequence from copy import deepcopy from dataclasses import dataclass @@ -20,6 +21,8 @@ from ._vector_filters import ( Filter, FilterExpression, + FilterGroup, + Param, validate_filter, ) from ._vectors import ( @@ -34,6 +37,7 @@ Vector, VectorStoreCollectionDefinition, _evaluate_filter, # pyright: ignore[reportPrivateUsage] + _msgspec_enc_hook, # pyright: ignore[reportPrivateUsage] _validate_vector_dimensions, # pyright: ignore[reportPrivateUsage] ) from .exceptions import IntegrationException @@ -152,6 +156,42 @@ def _calculate_score(left: Vector, right: Vector, distance_function: DistanceFun return score +def _to_stored_filter_value(value: Any) -> Any: + """Return a filter operand in the builtin form records are stored in. + + In-memory records are stored through the shared serializer, so a native + scalar such as a datetime is held as its serialized value. An operand is + normalized the same way, so that the comparison sees one representation on + both sides. + """ + if isinstance(value, Param): + return value + try: + return msgspec.to_builtins( + value, + str_keys=True, + builtin_types=(bytes, bytearray), + enc_hook=_msgspec_enc_hook, + ) + except (TypeError, ValueError): + return value + + +def _normalize_in_memory_filter(expression: FilterExpression) -> FilterExpression: + """Return the filter with its operand values in the stored representation.""" + if isinstance(expression, Filter): + value = expression.value + if isinstance(value, Filter | FilterGroup): + normalized: Any = _normalize_in_memory_filter(value) + else: + normalized = _to_stored_filter_value(value) + return Filter(expression.field_name, expression.operator, normalized) + return FilterGroup( + expression.operator, + tuple(_normalize_in_memory_filter(item) for item in expression.filters), + ) + + 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,6 +347,7 @@ async def _inner_get( records = list(self._state.records.values()) if filter is not None: _validate_in_memory_filter(filter, self.definition) + filter = _normalize_in_memory_filter(filter) records = [record for record in records if _evaluate_filter(filter, record, self.definition)] if order_by: for field_name, ascending in reversed(tuple(order_by.items())): @@ -362,6 +403,7 @@ async def _inner_search( _validate_vector_dimensions(query_vector, vector_field) if filter is not None: _validate_in_memory_filter(filter, self.definition) + filter = _normalize_in_memory_filter(filter) distance_function = vector_field.distance_function or "DEFAULT" if distance_function == "DEFAULT": diff --git a/python/packages/core/tests/core/test_in_memory.py b/python/packages/core/tests/core/test_in_memory.py index 1253a189519..e9c1ad99b48 100644 --- a/python/packages/core/tests/core/test_in_memory.py +++ b/python/packages/core/tests/core/test_in_memory.py @@ -8,7 +8,10 @@ import warnings from collections.abc import Iterator, Sequence from dataclasses import dataclass +from datetime import date, datetime, timedelta, timezone +from decimal import Decimal from typing import Annotated, Any, cast +from uuid import UUID import pytest @@ -866,3 +869,114 @@ def test_in_memory_filter_implementation_does_not_compile_or_evaluate_source() - def test_in_memory_and_filter_apis_are_experimental() -> None: for api in (Filter, FilterGroup, Param, InMemoryCollection, InMemoryStore): assert getattr(api, "__feature_stage__", None) == "experimental" + + +@vectorstoremodel(collection_name="scalar_filter_records") +@dataclass +class ScalarFilterRecord: + id: Annotated[str, VectorStoreField("key")] + created: Annotated[datetime, VectorStoreField("data")] + day: Annotated[date, VectorStoreField("data")] + token: Annotated[UUID, VectorStoreField("data")] + amount: Annotated[Decimal, VectorStoreField("data")] + span: Annotated[timedelta, VectorStoreField("data")] + label: Annotated[str, VectorStoreField("data")] + count: Annotated[int, VectorStoreField("data")] + flag: Annotated[bool, VectorStoreField("data")] + vector: Annotated[list[float] | None, VectorStoreField("vector", dimensions=2)] = None + + +_SCALAR_FILTER_RECORD = ScalarFilterRecord( + "one", + datetime(2026, 10, 16, 1, 0, tzinfo=timezone.utc), + date(2026, 10, 16), + UUID("00000000-0000-0000-0000-000000000001"), + Decimal("1.5"), + timedelta(hours=2), + "featured", + 5, + True, + [1.0, 0.0], +) + + +async def _scalar_filter_collection() -> InMemoryCollection[str, ScalarFilterRecord]: + collection = InMemoryCollection(ScalarFilterRecord) + await collection.ensure_collection_exists() + await collection.upsert([_SCALAR_FILTER_RECORD], generate_vectors=False) + return collection + + +@pytest.mark.parametrize( + ("field_name", "value"), + [ + ("created", datetime(2026, 10, 16, 1, 0, tzinfo=timezone.utc)), + ("day", date(2026, 10, 16)), + ("token", UUID("00000000-0000-0000-0000-000000000001")), + ("amount", Decimal("1.5")), + ("span", timedelta(hours=2)), + ], +) +async def test_in_memory_filter_matches_a_native_scalar_against_the_stored_value( + field_name: str, + value: Any, +) -> None: + """A native scalar filter must match the record it was stored from. + + The stored value is the serialized form of the field, so the operand has to + be compared in the same representation. + """ + collection = await _scalar_filter_collection() + + matched = await collection.get(filter=Filter(field_name, "eq", value)) + + assert [record.id for record in matched] == ["one"] + + +@pytest.mark.parametrize( + ("field_name", "value"), + [ + ("label", "featured"), + ("count", 5), + ("flag", True), + ], +) +async def test_in_memory_filter_still_matches_the_scalars_it_stored_unchanged( + field_name: str, + value: Any, +) -> None: + """The controls the issue lists as working must keep working.""" + collection = await _scalar_filter_collection() + + matched = await collection.get(filter=Filter(field_name, "eq", value)) + + assert [record.id for record in matched] == ["one"] + + +async def test_in_memory_vector_search_matches_a_native_scalar_filter() -> None: + """The search path compares the operands the same way as the get path.""" + collection = await _scalar_filter_collection() + value = datetime(2026, 10, 16, 1, 0, tzinfo=timezone.utc) + + results = await collection.search(vector=[1.0, 0.0], filter=Filter("created", "eq", value), top=10) + + assert [result["record"].id async for result in results] == ["one"] + + +async def test_in_memory_ordered_filter_compares_a_native_scalar() -> None: + """An ordered comparison on a native scalar must compare, not raise.""" + collection = await _scalar_filter_collection() + value = datetime(2026, 10, 16, 1, 0, tzinfo=timezone.utc) + + matched = await collection.get(filter=Filter("created", "gte", value)) + + assert [record.id for record in matched] == ["one"] + + +async def test_in_memory_filter_does_not_match_a_different_native_scalar() -> None: + """A different value must still not match.""" + collection = await _scalar_filter_collection() + + matched = await collection.get(filter=Filter("created", "eq", datetime(2025, 1, 1, tzinfo=timezone.utc))) + + assert matched == [] From da2885236725cf35cfdcf43fa8ca81e579953e23 Mon Sep 17 00:00:00 2001 From: Weijian He <230282609+Madin-H23@users.noreply.github.com> Date: Thu, 22 Oct 2026 10:30:56 +0800 Subject: [PATCH 2/5] Python: fix(vectors): compare in-memory filters in the field's declared type The first version of this change normalized the filter operand to the representation the records are stored in. That fixed the equality case, but the comparison then ran on serialized text, so an ordered comparison on a decimal or a duration compared digits instead of numbers. The stored value is now restored to the type its field declares before the comparison, so the comparison runs in the field's own semantics and the operands stay untouched. The restore is scoped to the in-memory connector, which is the one that keeps the serialized form; a store that holds native values is unaffected. --- .../core/agent_framework/_in_memory.py | 118 +++++++++++++----- .../core/tests/core/test_in_memory.py | 43 +++++++ 2 files changed, 127 insertions(+), 34 deletions(-) diff --git a/python/packages/core/agent_framework/_in_memory.py b/python/packages/core/agent_framework/_in_memory.py index c211dcf186b..f7189ea77cc 100644 --- a/python/packages/core/agent_framework/_in_memory.py +++ b/python/packages/core/agent_framework/_in_memory.py @@ -5,13 +5,13 @@ from __future__ import annotations import math -import msgspec +import re from collections.abc import Mapping, Sequence from copy import deepcopy from dataclasses import dataclass from datetime import date, datetime, time, timedelta from decimal import Decimal -from typing import Any, ClassVar, Generic, cast +from typing import Any, Callable, ClassVar, Generic, cast from uuid import UUID, uuid4 from typing_extensions import TypeVar @@ -21,8 +21,6 @@ from ._vector_filters import ( Filter, FilterExpression, - FilterGroup, - Param, validate_filter, ) from ._vectors import ( @@ -37,7 +35,6 @@ Vector, VectorStoreCollectionDefinition, _evaluate_filter, # pyright: ignore[reportPrivateUsage] - _msgspec_enc_hook, # pyright: ignore[reportPrivateUsage] _validate_vector_dimensions, # pyright: ignore[reportPrivateUsage] ) from .exceptions import IntegrationException @@ -156,40 +153,80 @@ def _calculate_score(left: Vector, right: Vector, distance_function: DistanceFun return score -def _to_stored_filter_value(value: Any) -> Any: - """Return a filter operand in the builtin form records are stored in. +_DURATION_PATTERN = re.compile( + r"^(?P-)?P(?:(?P[0-9]+(?:\.[0-9]+)?)D)?" + r"(?:T(?:(?P[0-9]+(?:\.[0-9]+)?)H)?(?:(?P[0-9]+(?:\.[0-9]+)?)M)?" + r"(?:(?P[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.""" + match = _DURATION_PATTERN.match(value) + if match is None: + raise ValueError(f"Unsupported duration '{value}'.") + parts = match.groupdict() + seconds = ( + float(parts["days"] or 0) * 86400 + + float(parts["hours"] or 0) * 3600 + + float(parts["minutes"] or 0) * 60 + + float(parts["seconds"] or 0) + ) + delta = timedelta(seconds=seconds) + return -delta if parts["sign"] else delta + + +_RESTORE_IN_MEMORY_VALUE_TYPES: dict[str, Callable[[str], Any]] = { + "datetime": _restore_datetime, + "date": date.fromisoformat, + "time": time.fromisoformat, + "timedelta": _restore_timedelta, + "UUID": UUID, + "Decimal": Decimal, +} + + +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. An operand is - normalized the same way, so that the comparison sees one representation on - both sides. + 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 isinstance(value, Param): + 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 msgspec.to_builtins( - value, - str_keys=True, - builtin_types=(bytes, bytearray), - enc_hook=_msgspec_enc_hook, - ) + return restore(value) except (TypeError, ValueError): return value -def _normalize_in_memory_filter(expression: FilterExpression) -> FilterExpression: - """Return the filter with its operand values in the stored representation.""" - if isinstance(expression, Filter): - value = expression.value - if isinstance(value, Filter | FilterGroup): - normalized: Any = _normalize_in_memory_filter(value) - else: - normalized = _to_stored_filter_value(value) - return Filter(expression.field_name, expression.operator, normalized) - return FilterGroup( - expression.operator, - tuple(_normalize_in_memory_filter(item) for item in expression.filters), - ) +def _restore_in_memory_record( + record: Mapping[str, Any], + definition: VectorStoreCollectionDefinition, + field_names: set[str], +) -> dict[str, Any]: + """Return the record with the filtered fields restored to their declared type.""" + restored = dict(record) + for field_name in field_names: + field = definition.try_get_field(field_name) + if field is None: + continue + 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 restored def _validate_in_memory_filter_value(value: Any) -> None: @@ -347,8 +384,16 @@ async def _inner_get( records = list(self._state.records.values()) if filter is not None: _validate_in_memory_filter(filter, self.definition) - filter = _normalize_in_memory_filter(filter) - records = [record for record in records if _evaluate_filter(filter, record, self.definition)] + filtered = {item.field_name for item in _walk_filters(filter)} + records = [ + record + for record in records + if _evaluate_filter( + filter, + _restore_in_memory_record(record, self.definition, filtered), + self.definition, + ) + ] if order_by: for field_name, ascending in reversed(tuple(order_by.items())): if not isinstance(ascending, bool): @@ -401,9 +446,10 @@ async def _inner_search( if vector_field is None: raise ValueError("InMemoryCollection vector search requires a vector field.") _validate_vector_dimensions(query_vector, vector_field) + filtered: set[str] = set() if filter is not None: _validate_in_memory_filter(filter, self.definition) - filter = _normalize_in_memory_filter(filter) + filtered = {item.field_name for item in _walk_filters(filter)} distance_function = vector_field.distance_function or "DEFAULT" if distance_function == "DEFAULT": @@ -417,7 +463,11 @@ 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_filter( + filter, + _restore_in_memory_record(record, self.definition, filtered), + self.definition, + ): continue stored_vector = record.get(storage_name) if stored_vector is None: diff --git a/python/packages/core/tests/core/test_in_memory.py b/python/packages/core/tests/core/test_in_memory.py index e9c1ad99b48..c3c5fa22db7 100644 --- a/python/packages/core/tests/core/test_in_memory.py +++ b/python/packages/core/tests/core/test_in_memory.py @@ -973,6 +973,49 @@ async def test_in_memory_ordered_filter_compares_a_native_scalar() -> None: assert [record.id for record in matched] == ["one"] +async def test_in_memory_ordered_filter_compares_decimals_numerically() -> None: + """An ordered comparison on a decimal must compare numbers, not digits. + + The stored form of the two values sorts the other way round as text, so a + comparison on the serialized form would pick the wrong record. + """ + collection = InMemoryCollection(ScalarFilterRecord) + await collection.ensure_collection_exists() + await collection.upsert( + [ + ScalarFilterRecord( + id="two", + created=datetime(2026, 10, 16, 1, 0, tzinfo=timezone.utc), + day=date(2026, 10, 16), + token=UUID("00000000-0000-0000-0000-000000000002"), + amount=Decimal("2"), + span=timedelta(hours=2), + label="small", + count=2, + flag=True, + vector=[1.0, 0.0], + ), + ScalarFilterRecord( + id="ten", + created=datetime(2026, 10, 17, 1, 0, tzinfo=timezone.utc), + day=date(2026, 10, 17), + token=UUID("00000000-0000-0000-0000-000000000010"), + amount=Decimal("10"), + span=timedelta(hours=10), + label="large", + count=10, + flag=True, + vector=[0.0, 1.0], + ), + ], + generate_vectors=False, + ) + + matched = await collection.get(filter=Filter("amount", "gte", Decimal("10"))) + + assert [record.id for record in matched] == ["ten"] + + async def test_in_memory_filter_does_not_match_a_different_native_scalar() -> None: """A different value must still not match.""" collection = await _scalar_filter_collection() From a296683416f9d34da9520fadb9080990af5d990b Mon Sep 17 00:00:00 2001 From: Weijian He <230282609+Madin-H23@users.noreply.github.com> Date: Thu, 22 Oct 2026 12:08:49 +0800 Subject: [PATCH 3/5] Python: fix(vectors): reject in-memory filter operands of the wrong type The previous commit restored the stored value to the type its field declares, so the comparison runs in the field's own semantics. The issue allows the other option as well: an operand type whose semantics cannot be preserved should be rejected explicitly rather than silently returning no matches. The operand validator accepts every native scalar, so a string operand for a datetime field, a datetime operand for a string field, or an integer operand for a decimal field was accepted and then matched nothing. Such an operand is now rejected with a TypeError that names the field, its declared type and the operand type. The check is limited to the scalar types the shared serializer writes as text, because outside that set the stored value is the native value and the comparison keeps the ordinary Python semantics the framework already defines. --- .../core/agent_framework/_in_memory.py | 52 +++++++ .../core/tests/core/test_in_memory.py | 135 +++++++++++++++++- 2 files changed, 180 insertions(+), 7 deletions(-) diff --git a/python/packages/core/agent_framework/_in_memory.py b/python/packages/core/agent_framework/_in_memory.py index f7189ea77cc..798f83abe54 100644 --- a/python/packages/core/agent_framework/_in_memory.py +++ b/python/packages/core/agent_framework/_in_memory.py @@ -191,6 +191,9 @@ def _restore_timedelta(value: str) -> 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: @@ -384,6 +387,7 @@ async def _inner_get( records = list(self._state.records.values()) if filter is not None: _validate_in_memory_filter(filter, self.definition) + _reject_incompatible_filter_operands(filter, self.definition) filtered = {item.field_name for item in _walk_filters(filter)} records = [ record @@ -449,6 +453,7 @@ async def _inner_search( filtered: set[str] = set() if filter is not None: _validate_in_memory_filter(filter, self.definition) + _reject_incompatible_filter_operands(filter, self.definition) filtered = {item.field_name for item in _walk_filters(filter)} distance_function = vector_field.distance_function or "DEFAULT" @@ -526,6 +531,53 @@ 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. + """ + for item in _walk_filters(expression): + field = definition.try_get_field(item.field_name) + if field is None or field.type_ is None: + 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.""" diff --git a/python/packages/core/tests/core/test_in_memory.py b/python/packages/core/tests/core/test_in_memory.py index c3c5fa22db7..0395feff8a3 100644 --- a/python/packages/core/tests/core/test_in_memory.py +++ b/python/packages/core/tests/core/test_in_memory.py @@ -973,19 +973,15 @@ async def test_in_memory_ordered_filter_compares_a_native_scalar() -> None: assert [record.id for record in matched] == ["one"] -async def test_in_memory_ordered_filter_compares_decimals_numerically() -> None: - """An ordered comparison on a decimal must compare numbers, not digits. - - The stored form of the two values sorts the other way round as text, so a - comparison on the serialized form would pick the wrong record. - """ +async def _scalar_amount_collection() -> InMemoryCollection[str, ScalarFilterRecord]: + """Two records whose amounts and timestamps sort the other way round as text.""" collection = InMemoryCollection(ScalarFilterRecord) await collection.ensure_collection_exists() await collection.upsert( [ ScalarFilterRecord( id="two", - created=datetime(2026, 10, 16, 1, 0, tzinfo=timezone.utc), + created=datetime(2026, 10, 16, 23, 0, tzinfo=timezone(timedelta(hours=2))), day=date(2026, 10, 16), token=UUID("00000000-0000-0000-0000-000000000002"), amount=Decimal("2"), @@ -1010,6 +1006,16 @@ async def test_in_memory_ordered_filter_compares_decimals_numerically() -> None: ], generate_vectors=False, ) + return collection + + +async def test_in_memory_ordered_filter_compares_decimals_numerically() -> None: + """An ordered comparison on a decimal must compare numbers, not digits. + + The stored form of the two values sorts the other way round as text, so a + comparison on the serialized form would pick the wrong record. + """ + collection = await _scalar_amount_collection() matched = await collection.get(filter=Filter("amount", "gte", Decimal("10"))) @@ -1023,3 +1029,118 @@ async def test_in_memory_filter_does_not_match_a_different_native_scalar() -> No matched = await collection.get(filter=Filter("created", "eq", datetime(2025, 1, 1, tzinfo=timezone.utc))) assert matched == [] + + +async def test_in_memory_filter_does_not_match_a_different_decimal() -> None: + """A different decimal must not match the stored one.""" + collection = await _scalar_amount_collection() + + matched = await collection.get(filter=Filter("amount", "eq", Decimal("2"))) + + assert [record.id for record in matched] == ["two"] + + +async def test_in_memory_filter_matches_the_same_instant_written_in_another_offset() -> None: + """The same instant written with another UTC offset is still the same value.""" + collection = await _scalar_filter_collection() + value = datetime(2026, 10, 16, 3, 0, tzinfo=timezone(timedelta(hours=2))) + + matched = await collection.get(filter=Filter("created", "eq", value)) + + assert [record.id for record in matched] == ["one"] + + +async def test_in_memory_ordered_filter_compares_timestamps_written_in_another_offset() -> None: + """The order of two timestamps must not depend on their UTC offsets. + + The stored form of the earlier timestamp sorts after the later one as text, + so a comparison on the serialized form would pick the wrong record. + """ + collection = await _scalar_amount_collection() + value = datetime(2026, 10, 16, 22, 0, tzinfo=timezone.utc) + + matched = await collection.get(filter=Filter("created", "gte", value)) + + assert [record.id for record in matched] == ["ten"] + + +@pytest.mark.parametrize( + ("field_name", "value"), + [ + ("created", "2026-10-16T01:00:00+00:00"), + ("day", datetime(2026, 10, 16)), + ("amount", 1.5), + ("span", 7200), + ("token", "00000000-0000-0000-0000-000000000001"), + ], +) +async def test_in_memory_filter_rejects_an_operand_of_another_type(field_name: str, value: Any) -> None: + """An operand that cannot be compared in the field's own type is rejected. + + The operand validator accepts every one of these operands, so without a + rejection the filter would quietly match no record at all. + """ + collection = await _scalar_filter_collection() + + with pytest.raises(TypeError, match="cannot compare field"): + await collection.get(filter=Filter(field_name, "eq", value)) + + +@pytest.mark.parametrize( + ("field_name", "value"), + [ + ("label", datetime(2026, 10, 16, 1, 0, tzinfo=timezone.utc)), + ("count", Decimal("5")), + ], +) +async def test_in_memory_filter_rejects_a_native_scalar_operand_for_another_field_type( + field_name: str, + value: Any, +) -> None: + """A native scalar operand only fits the field that declares it.""" + collection = await _scalar_filter_collection() + + with pytest.raises(TypeError, match="cannot compare field"): + await collection.get(filter=Filter(field_name, "eq", value)) + + +async def test_in_memory_search_rejects_an_operand_of_another_type() -> None: + """The search path rejects the operands the get path rejects.""" + collection = await _scalar_filter_collection() + + with pytest.raises(TypeError, match="cannot compare field"): + await collection.search(vector=[1.0, 0.0], filter=Filter("created", "eq", "2026-10-16"), top=10) + + +async def test_in_memory_filter_rejects_an_incompatible_operand_inside_a_collection() -> None: + """Every operand of a collection filter is checked.""" + collection = await _scalar_filter_collection() + value = datetime(2026, 10, 16, 1, 0, tzinfo=timezone.utc) + + with pytest.raises(TypeError, match="cannot compare field"): + await collection.get(filter=Filter("created", "in", (value, "2026-10-16"))) + with pytest.raises(TypeError, match="cannot compare field"): + await collection.get(filter=Filter("created", "between", (value, "2026-10-17"))) + + +async def test_in_memory_filter_rejects_an_incompatible_operand_inside_a_group() -> None: + """Every operand of a filter group is checked.""" + collection = await _scalar_filter_collection() + + with pytest.raises(TypeError, match="cannot compare field"): + await collection.get( + filter=FilterGroup( + "and", + (Filter("label", "eq", "featured"), Filter("created", "eq", "2026-10-16")), + ) + ) + + +async def test_in_memory_filter_still_matches_a_collection_of_native_scalars() -> None: + """A collection of operands that fit the field still matches.""" + collection = await _scalar_filter_collection() + value = datetime(2026, 10, 16, 1, 0, tzinfo=timezone.utc) + + matched = await collection.get(filter=Filter("created", "in", (value,))) + + assert [record.id for record in matched] == ["one"] From e31ca1b96db09e877b2317b4389f11273e5738ea Mon Sep 17 00:00:00 2001 From: Weijian He <230282609+Madin-H23@users.noreply.github.com> Date: Thu, 22 Oct 2026 17:48:42 +0800 Subject: [PATCH 4/5] Python: fix(vectors): import Callable from collections.abc `pyupgrade` runs with `--py310-plus` in the repository's pre-commit hooks, so it rewrites `Callable` out of the `typing` import. The import the previous commit added therefore failed the Pre-commit Hooks check, and `merge-gatekeeper` with it. --- python/packages/core/agent_framework/_in_memory.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/python/packages/core/agent_framework/_in_memory.py b/python/packages/core/agent_framework/_in_memory.py index 798f83abe54..c11337730fe 100644 --- a/python/packages/core/agent_framework/_in_memory.py +++ b/python/packages/core/agent_framework/_in_memory.py @@ -6,12 +6,12 @@ import math import re -from collections.abc import Mapping, Sequence +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 typing import Any, Callable, ClassVar, Generic, cast +from typing import Any, ClassVar, Generic, cast from uuid import UUID, uuid4 from typing_extensions import TypeVar From 27e48d0d013f943a30efa3540a6a1cb9e7c3f74f Mon Sep 17 00:00:00 2001 From: Weijian He <230282609+Madin-H23@users.noreply.github.com> Date: Fri, 23 Oct 2026 12:26:43 +0800 Subject: [PATCH 5/5] Python: fix(vectors): restore per filter leaf and keep durations exact Address review feedback on the in-memory filter restore: - Evaluate each filter leaf against the record as stored instead of rewriting the record up front. A comparison operator restores its own leaf's field to the declared type; the text operators keep reading the serialized string, so Filter("created", "starts_with", "2026") works again and a group can mix a text operator and a native comparison over one field. - Sum the ISO duration as exact decimals of microseconds. A float of the total seconds loses microseconds on large durations and rounds timedelta.max out of the type's own range. - Normalize the trailing "Z" the serializer writes for a UTC time before time.fromisoformat, which accepts it only from Python 3.11. --- .../core/agent_framework/_in_memory.py | 104 +++++++++------ .../core/tests/core/test_in_memory.py | 121 +++++++++++++++++- 2 files changed, 185 insertions(+), 40 deletions(-) diff --git a/python/packages/core/agent_framework/_in_memory.py b/python/packages/core/agent_framework/_in_memory.py index c11337730fe..665c2918455 100644 --- a/python/packages/core/agent_framework/_in_memory.py +++ b/python/packages/core/agent_framework/_in_memory.py @@ -10,7 +10,7 @@ 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 @@ -19,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 ( @@ -168,25 +170,40 @@ def _restore_datetime(value: str) -> datetime: def _restore_timedelta(value: str) -> timedelta: - """Parse the ISO 8601 duration the shared serializer wrote.""" + """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() - seconds = ( - float(parts["days"] or 0) * 86400 - + float(parts["hours"] or 0) * 3600 - + float(parts["minutes"] or 0) * 60 - + float(parts["seconds"] or 0) + 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(seconds=seconds) + 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": time.fromisoformat, + "time": _restore_time, "timedelta": _restore_timedelta, "UUID": UUID, "Decimal": Decimal, @@ -215,21 +232,41 @@ def _restore_in_memory_value(value: Any, type_tag: str | None) -> Any: return value -def _restore_in_memory_record( +def _evaluate_in_memory_filter( + expression: FilterExpression, record: Mapping[str, Any], definition: VectorStoreCollectionDefinition, - field_names: set[str], -) -> dict[str, Any]: - """Return the record with the filtered fields restored to their declared type.""" +) -> 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) - for field_name in field_names: - field = definition.try_get_field(field_name) - if field is None: - continue - 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 restored + 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: @@ -388,16 +425,7 @@ async def _inner_get( if filter is not None: _validate_in_memory_filter(filter, self.definition) _reject_incompatible_filter_operands(filter, self.definition) - filtered = {item.field_name for item in _walk_filters(filter)} - records = [ - record - for record in records - if _evaluate_filter( - filter, - _restore_in_memory_record(record, self.definition, filtered), - 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): @@ -450,11 +478,9 @@ async def _inner_search( if vector_field is None: raise ValueError("InMemoryCollection vector search requires a vector field.") _validate_vector_dimensions(query_vector, vector_field) - filtered: set[str] = set() if filter is not None: _validate_in_memory_filter(filter, self.definition) _reject_incompatible_filter_operands(filter, self.definition) - filtered = {item.field_name for item in _walk_filters(filter)} distance_function = vector_field.distance_function or "DEFAULT" if distance_function == "DEFAULT": @@ -468,11 +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, - _restore_in_memory_record(record, self.definition, filtered), - 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: @@ -562,12 +584,16 @@ def _reject_incompatible_filter_operands( 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. + 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 diff --git a/python/packages/core/tests/core/test_in_memory.py b/python/packages/core/tests/core/test_in_memory.py index 0395feff8a3..8a30897a040 100644 --- a/python/packages/core/tests/core/test_in_memory.py +++ b/python/packages/core/tests/core/test_in_memory.py @@ -8,7 +8,7 @@ import warnings from collections.abc import Iterator, Sequence from dataclasses import dataclass -from datetime import date, datetime, timedelta, timezone +from datetime import date, datetime, time, timedelta, timezone from decimal import Decimal from typing import Annotated, Any, cast from uuid import UUID @@ -1144,3 +1144,122 @@ async def test_in_memory_filter_still_matches_a_collection_of_native_scalars() - matched = await collection.get(filter=Filter("created", "in", (value,))) assert [record.id for record in matched] == ["one"] + + +@vectorstoremodel(collection_name="time_filter_records") +@dataclass +class TimeFilterRecord: + id: Annotated[str, VectorStoreField("key")] + starts: Annotated[time, VectorStoreField("data")] + vector: Annotated[list[float] | None, VectorStoreField("vector", dimensions=2)] = None + + +async def _time_filter_collection() -> InMemoryCollection[str, TimeFilterRecord]: + collection = InMemoryCollection(TimeFilterRecord) + await collection.ensure_collection_exists() + await collection.upsert([TimeFilterRecord("one", time(12, 30, tzinfo=timezone.utc))], generate_vectors=False) + return collection + + +async def test_in_memory_filter_matches_a_utc_time_stored_with_a_z_suffix() -> None: + """A UTC time is stored with a trailing "Z" and must still match. + + ``time.fromisoformat`` accepts the "Z" suffix only from Python 3.11, so + the restore normalizes it the way the datetime restore does. + """ + collection = await _time_filter_collection() + + matched = await collection.get(filter=Filter("starts", "eq", time(12, 30, tzinfo=timezone.utc))) + + assert [record.id for record in matched] == ["one"] + + +async def test_in_memory_text_operator_compares_the_stored_scalar_string() -> None: + """A text operator reads the stored string, not the restored scalar. + + The serialized timestamp is what the text operators have always compared + these fields by. + """ + collection = await _scalar_filter_collection() + + matched = await collection.get(filter=Filter("created", "starts_with", "2026-10-16T01")) + + assert [record.id for record in matched] == ["one"] + + +async def test_in_memory_group_mixes_a_text_operator_and_a_native_comparison() -> None: + """One field can carry a text operator and a native comparison in a group. + + The two leaves need different forms of the stored value: the text operator + reads the serialized string while the comparison restores the scalar. + """ + collection = await _scalar_amount_collection() + + matched = await collection.get( + filter=FilterGroup( + "and", + ( + Filter("created", "starts_with", "2026-10-17"), + Filter("created", "gte", datetime(2026, 10, 17, 1, 0, tzinfo=timezone.utc)), + ), + ) + ) + + assert [record.id for record in matched] == ["ten"] + + +async def _long_span_collection(span: timedelta) -> InMemoryCollection[str, ScalarFilterRecord]: + collection = InMemoryCollection(ScalarFilterRecord) + await collection.ensure_collection_exists() + await collection.upsert( + [ + ScalarFilterRecord( + id="long", + created=datetime(2026, 10, 16, 1, 0, tzinfo=timezone.utc), + day=date(2026, 10, 16), + token=UUID("00000000-0000-0000-0000-000000000001"), + amount=Decimal("1.5"), + span=span, + label="long", + count=1, + flag=True, + ) + ], + generate_vectors=False, + ) + return collection + + +async def test_in_memory_restores_a_large_duration_exactly() -> None: + """A microsecond inside a large duration survives the restore. + + Summing the duration through ``float`` loses this microsecond, so the + equality would quietly match nothing. + """ + collection = await _long_span_collection(timedelta(days=100_000, microseconds=1)) + + matched = await collection.get(filter=Filter("span", "eq", timedelta(days=100_000, microseconds=1))) + + assert [record.id for record in matched] == ["long"] + + +async def test_in_memory_distinguishes_durations_a_microsecond_apart() -> None: + """Two durations a microsecond apart must not collapse into one.""" + collection = await _long_span_collection(timedelta(days=100_000, microseconds=1)) + + matched = await collection.get(filter=Filter("span", "eq", timedelta(days=100_000, microseconds=2))) + + assert matched == [] + + +async def test_in_memory_restores_the_largest_duration_exactly() -> None: + """``timedelta.max`` is restored, not rounded out of the type's range. + + A ``float`` of the total seconds rounds the stored duration past the + largest span ``timedelta`` supports. + """ + collection = await _long_span_collection(timedelta.max) + + matched = await collection.get(filter=Filter("span", "eq", timedelta.max)) + + assert [record.id for record in matched] == ["long"]