diff --git a/.github/workflows/main.yaml b/.github/workflows/main.yaml index d62caf8c2..489ff9504 100644 --- a/.github/workflows/main.yaml +++ b/.github/workflows/main.yaml @@ -25,11 +25,12 @@ env: WEAVIATE_131: 1.31.22 WEAVIATE_132: 1.32.27 WEAVIATE_133: 1.33.18 - WEAVIATE_134: 1.34.19 - WEAVIATE_135: 1.35.18 - WEAVIATE_136: 1.36.12 - WEAVIATE_137: 1.37.5-e0fe0d5.amd64 - WEAVIATE_139: 1.39.0-rc.0-b41225e.amd64 + WEAVIATE_134: 1.34.20 + WEAVIATE_135: 1.35.23 + WEAVIATE_136: 1.36.23 + WEAVIATE_137: 1.37.14 + WEAVIATE_138: 1.38.9 + WEAVIATE_139: 1.39.0 jobs: lint-and-format: @@ -219,11 +220,11 @@ jobs: fail-fast: false matrix: versions: [ - { py: "3.10", weaviate: $WEAVIATE_136}, - { py: "3.11", weaviate: $WEAVIATE_136}, - { py: "3.12", weaviate: $WEAVIATE_136}, - { py: "3.13", weaviate: $WEAVIATE_136}, - { py: "3.14", weaviate: $WEAVIATE_136} + { py: "3.10", weaviate: $WEAVIATE_139}, + { py: "3.11", weaviate: $WEAVIATE_139}, + { py: "3.12", weaviate: $WEAVIATE_139}, + { py: "3.13", weaviate: $WEAVIATE_139}, + { py: "3.14", weaviate: $WEAVIATE_139} ] optional_dependencies: [false] steps: @@ -322,6 +323,7 @@ jobs: $WEAVIATE_135, $WEAVIATE_136, $WEAVIATE_137, + $WEAVIATE_138, $WEAVIATE_139 ] steps: diff --git a/integration/test_collection.py b/integration/test_collection.py index d2745112d..f4016f930 100644 --- a/integration/test_collection.py +++ b/integration/test_collection.py @@ -36,6 +36,10 @@ ) from weaviate.collections.classes.internal import Object, ReferenceToMulti, _CrossReference from weaviate.collections.classes.types import PhoneNumber, WeaviateProperties, _PhoneNumber +from weaviate.collections.grpc.shared import ( + _BM25_AND_CROSS_MIN_VERSIONS, + _BM25_AND_CROSS_MIN_VERSIONS_STR, +) from weaviate.exceptions import ( UnexpectedStatusCodeError, WeaviateInsertInvalidPropertyError, @@ -1756,3 +1760,24 @@ def test_bm25_operators(collection_factory: CollectionFactory) -> None: assert len(objs.objects) == 4 assert objs.objects[0].uuid == uuid2 assert sorted(obj.uuid for obj in objs.objects[1:]) == sorted([uuid1, uuid3, uuid4]) + + +def test_bm25_operator_and_cross(collection_factory: CollectionFactory) -> None: + collection = collection_factory( + properties=[ + Property(name="title", data_type=DataType.TEXT), + Property(name="body", data_type=DataType.TEXT), + ], + vectorizer_config=Configure.Vectorizer.none(), + ) + + if not collection._connection._weaviate_version.is_at_least_any(*_BM25_AND_CROSS_MIN_VERSIONS): + pytest.skip(f"bm25 cross-property AND requires {_BM25_AND_CROSS_MIN_VERSIONS_STR}") + + # Neither of `split_across`'s properties holds both tokens, so only cross-property AND matches it. + split_across = collection.data.insert({"title": "banana", "body": "split"}) + single_property = collection.data.insert({"title": "banana split", "body": "dessert"}) + collection.data.insert({"title": "banana", "body": "bread"}) + + objs = collection.query.bm25("banana split", operator=wvc.query.BM25Operator.and_cross()) + assert sorted(obj.uuid for obj in objs.objects) == sorted([split_across, single_property]) diff --git a/integration/test_collection_config.py b/integration/test_collection_config.py index 786b486cd..b1bd30e12 100644 --- a/integration/test_collection_config.py +++ b/integration/test_collection_config.py @@ -54,11 +54,11 @@ def _expected_async_enabled(version: _ServerVersion, factor: int) -> bool: """Whether the server reports async replication as enabled for a collection. - Up to 1.38 the server stores whatever `async_enabled` the collection was created with. From - 1.39 it ignores that and derives the field as `factor > 1 and not globally disabled` instead, + Up to 1.38.9 the server stores whatever `async_enabled` the collection was created with. From + 1.38.9 it ignores that and derives the field as `factor > 1 and not globally disabled` instead, so a collection with a single replica always reports `False`. """ - if version.is_at_least(1, 39, 0): + if version.is_at_least(1, 38, 9): return factor > 1 return version.is_at_least(1, 26, 0) diff --git a/mock_tests/test_batch.py b/mock_tests/test_batch.py index c00bd08db..ad7357f04 100644 --- a/mock_tests/test_batch.py +++ b/mock_tests/test_batch.py @@ -1,7 +1,8 @@ -from typing import Generator +from typing import AsyncGenerator, Generator, List import grpc import pytest +import pytest_asyncio import weaviate from weaviate.proto.v1 import batch_pb2, weaviate_pb2_grpc from .conftest import MOCK_IP, MOCK_PORT, MOCK_PORT_GRPC, mock_class, HTTPServer @@ -60,3 +61,102 @@ def test_ssb_canceled_stream( for i in range(HOW_MANY): batch.add_object({"name": f"Object {i}"}) assert len(service.uuids) == HOW_MANY + + +class MockFailedObjectWeaviateService(weaviate_pb2_grpc.WeaviateServicer): + """Rejects every other object, starting with the first. + + A batch of 1 object gives 1 error and 0 uuids; a batch of 4 gives 2 errors and 2 uuids. + """ + + def __init__(self) -> None: + self.seen = 0 + + def BatchStream( + self, + request_iterator: Generator[batch_pb2.BatchStreamRequest, None, None], + context: grpc.ServicerContext, + ) -> Generator[batch_pb2.BatchStreamReply, None, None]: + yield batch_pb2.BatchStreamReply(started=batch_pb2.BatchStreamReply.Started()) + for request in request_iterator: + if request.HasField("data"): + uuids: List[str] = [] + errors: List[batch_pb2.BatchStreamReply.Results.Error] = [] + successes: List[batch_pb2.BatchStreamReply.Results.Success] = [] + for obj in request.data.objects.values: + uuids.append(obj.uuid) + if self.seen % 2 == 0: + errors.append( + batch_pb2.BatchStreamReply.Results.Error( + uuid=obj.uuid, error="mock failure" + ) + ) + else: + successes.append(batch_pb2.BatchStreamReply.Results.Success(uuid=obj.uuid)) + self.seen += 1 + yield batch_pb2.BatchStreamReply(acks=batch_pb2.BatchStreamReply.Acks(uuids=uuids)) + yield batch_pb2.BatchStreamReply( + results=batch_pb2.BatchStreamReply.Results(errors=errors, successes=successes) + ) + if request.HasField("stop"): + return + + +@pytest.fixture(scope="function") +def failed_object_stream( + canceled_stream_client: weaviate.WeaviateClient, start_grpc_server: grpc.Server +): + service = MockFailedObjectWeaviateService() + weaviate_pb2_grpc.add_WeaviateServicer_to_server(service, start_grpc_server) + return canceled_stream_client.collections.use(mock_class["class"]) + + +@pytest_asyncio.fixture +async def failed_object_stream_async( + weaviate_mock: HTTPServer, start_grpc_server: grpc.Server +) -> AsyncGenerator[weaviate.collections.CollectionAsync, None]: + weaviate_mock.expect_request(f"/v1/schema/{mock_class['class']}").respond_with_json(mock_class) + weaviate_pb2_grpc.add_WeaviateServicer_to_server( + MockFailedObjectWeaviateService(), start_grpc_server + ) + client = weaviate.use_async_with_local(port=MOCK_PORT, host=MOCK_IP, grpc_port=MOCK_PORT_GRPC) + await client.connect() + yield client.collections.use(mock_class["class"]) + await client.close() + + +def test_ingest_has_errors_on_failed_object( + failed_object_stream: weaviate.collections.Collection, +): + result = failed_object_stream.data.ingest([{"name": "Object 1"}]) + assert result.has_errors is True + assert len(result.errors) == 1 + + +def test_ssb_ingest_reports_has_errors( + failed_object_stream: weaviate.collections.Collection, +) -> None: + result = failed_object_stream.data.ingest({"name": f"Object {i}"} for i in range(4)) + assert len(result.errors) == 2 + assert len(result.uuids) == 2 + assert result.has_errors + + +@pytest.mark.asyncio +async def test_ssb_ingest_reports_has_errors_async( + failed_object_stream_async: weaviate.collections.CollectionAsync, +) -> None: + result = await failed_object_stream_async.data.ingest({"name": f"Object {i}"} for i in range(4)) + assert len(result.errors) == 2 + assert len(result.uuids) == 2 + assert result.has_errors + + +def test_ssb_stream_reports_has_errors( + failed_object_stream: weaviate.collections.Collection, +) -> None: + with failed_object_stream.batch.stream() as batch: + for i in range(4): + batch.add_object({"name": f"Object {i}"}) + assert len(failed_object_stream.batch.failed_objects) == 2 + assert failed_object_stream.batch.results.objs.has_errors diff --git a/test/collection/test_batch.py b/test/collection/test_batch.py index 0a2cda954..3837317ab 100644 --- a/test/collection/test_batch.py +++ b/test/collection/test_batch.py @@ -3,10 +3,38 @@ import pytest from weaviate.collections.batch.grpc_batch import _validate_props -from weaviate.collections.classes.batch import MAX_STORED_RESULTS, BatchObjectReturn +from weaviate.collections.classes.batch import ( + MAX_STORED_RESULTS, + BatchObject, + BatchObjectReturn, + BatchReference, + BatchReferenceReturn, + ErrorObject, + ErrorReference, +) from weaviate.exceptions import WeaviateInsertInvalidPropertyError +def _error_object(index: int) -> ErrorObject: + return ErrorObject( + message="something went wrong", + object_=BatchObject(collection="Test", properties={"name": "test"}, index=index), + ) + + +def _error_reference(index: int) -> ErrorReference: + return ErrorReference( + message="something went wrong", + reference=BatchReference( + from_object_collection="Test", + from_object_uuid=uuid.uuid4(), + from_property_name="other", + to_object_uuid=uuid.uuid4(), + index=index, + ), + ) + + def test_batch_object_return_add() -> None: lhs_uuids = [uuid.uuid4() for _ in range(MAX_STORED_RESULTS)] lhs = BatchObjectReturn( @@ -36,6 +64,42 @@ def test_batch_object_return_add() -> None: } +def test_batch_object_return_has_errors_when_constructed_with_errors() -> None: + err = _error_object(0) + result = BatchObjectReturn(_all_responses=[err], errors={0: err}) + assert result.has_errors + + +def test_batch_object_return_add_sets_has_errors() -> None: + err = _error_object(1) + result = BatchObjectReturn() + result += BatchObjectReturn(_all_responses=[uuid.uuid4()], uuids={0: uuid.uuid4()}) + result += BatchObjectReturn(_all_responses=[err], errors={1: err}) + assert result.has_errors + assert len(result.errors) == 1 + + +def test_batch_object_return_has_no_errors_when_all_succeed() -> None: + uid = uuid.uuid4() + result = BatchObjectReturn() + result += BatchObjectReturn(_all_responses=[uid], uuids={0: uid}) + assert not result.has_errors + + +def test_batch_reference_return_has_errors_when_constructed_with_errors() -> None: + err = _error_reference(0) + result = BatchReferenceReturn(errors={0: err}) + assert result.has_errors + + +def test_batch_reference_return_add_sets_has_errors() -> None: + err = _error_reference(0) + result = BatchReferenceReturn() + result += BatchReferenceReturn(errors={0: err}) + assert result.has_errors + assert len(result.errors) == 1 + + def test_validate_props_raises_for_top_level_id() -> None: with pytest.raises(WeaviateInsertInvalidPropertyError): _validate_props({"id": "abc123"}) diff --git a/test/collection/test_bm25_operator.py b/test/collection/test_bm25_operator.py new file mode 100644 index 000000000..9fa188b22 --- /dev/null +++ b/test/collection/test_bm25_operator.py @@ -0,0 +1,37 @@ +import pytest + +from weaviate.classes.query import BM25Operator +from weaviate.collections.grpc.query import _QueryGRPC +from weaviate.exceptions import WeaviateUnsupportedFeatureError +from weaviate.proto.v1 import base_search_pb2 +from weaviate.util import _ServerVersion + +_AND_CROSS = base_search_pb2.SearchOperatorOptions.OPERATOR_AND_CROSS + + +def _builder(version: str = "1.39.0") -> _QueryGRPC: + return _QueryGRPC( + weaviate_version=_ServerVersion.from_string(version), + name="Dummy", + tenant=None, + consistency_level=None, + validate_arguments=True, + uses_125_api=True, + uses_127_api=True, + ) + + +def test_and_cross_wired_into_request() -> None: + bm25 = _builder().bm25(query="banana split", operator=BM25Operator.and_cross()) + assert bm25.bm25_search.search_operator.operator == _AND_CROSS + + hybrid = _builder().hybrid( + query="banana split", alpha=0.0, bm25_operator=BM25Operator.and_cross() + ) + assert hybrid.hybrid_search.bm25_search_operator.operator == _AND_CROSS + + +@pytest.mark.parametrize("version", ["1.37.14", "1.38.7"]) +def test_and_cross_rejected_on_unsupported_versions(version: str) -> None: + with pytest.raises(WeaviateUnsupportedFeatureError): + _builder(version).bm25(query="banana split", operator=BM25Operator.and_cross()) diff --git a/test/test_server_version.py b/test/test_server_version.py index 1cf5e6097..ab7ed54a1 100644 --- a/test/test_server_version.py +++ b/test/test_server_version.py @@ -60,6 +60,25 @@ def test_server_version_is_at_least(is_valid: bool) -> None: assert is_valid +@pytest.mark.parametrize( + "version,expected", + [ + ("1.36.9", False), + ("1.37.14", False), + ("1.37.15", True), + ("1.38.7", False), + ("1.38.8", True), + ("1.39.0", True), + ("1.40.0", True), + ], +) +def test_server_version_is_at_least_any(version: str, expected: bool) -> None: + assert ( + _ServerVersion.from_string(version).is_at_least_any((1, 37, 15), (1, 38, 8), (1, 39, 0)) + is expected + ) + + def test_server_version_magic_methods() -> None: # Test __eq__ assert _ServerVersion(1, 2, 3) == _ServerVersion(1, 2, 3) diff --git a/weaviate/collections/classes/batch.py b/weaviate/collections/classes/batch.py index eb8d0181a..b90bebfad 100644 --- a/weaviate/collections/classes/batch.py +++ b/weaviate/collections/classes/batch.py @@ -216,6 +216,9 @@ class BatchObjectReturn: uuids: Dict[int, uuid_package.UUID] = field(default_factory=dict) has_errors: bool = False + def __post_init__(self) -> None: + self.has_errors = self.has_errors or len(self.errors) > 0 + @property def all_responses(self) -> List[Union[uuid_package.UUID, ErrorObject]]: """@deprecated: A list of all the responses from the batch operation. Each response is either a `uuid_package.UUID` object or an `Error` object. @@ -284,6 +287,9 @@ class BatchReferenceReturn: errors: Dict[int, ErrorReference] = field(default_factory=dict) has_errors: bool = False + def __post_init__(self) -> None: + self.has_errors = self.has_errors or len(self.errors) > 0 + def __add__(self, other: "BatchReferenceReturn") -> "BatchReferenceReturn": self.elapsed_seconds += other.elapsed_seconds prev_max = max(self.errors.keys()) if len(self.errors) > 0 else -1 diff --git a/weaviate/collections/classes/grpc.py b/weaviate/collections/classes/grpc.py index f5c8fe772..5cba315f7 100644 --- a/weaviate/collections/classes/grpc.py +++ b/weaviate/collections/classes/grpc.py @@ -672,6 +672,13 @@ class BM25OperatorAnd(BM25OperatorOptions): operator = base_search_pb2.SearchOperatorOptions.OPERATOR_AND +@dataclass +class BM25OperatorAndCross(BM25OperatorOptions): + """Define the cross-property 'And' operator for keyword queries.""" + + operator = base_search_pb2.SearchOperatorOptions.OPERATOR_AND_CROSS + + class BM25OperatorFactory: """Define how the BM25 query's token matching should be performed.""" @@ -697,6 +704,20 @@ def and_() -> BM25OperatorOptions: """ return BM25OperatorAnd() + @staticmethod + def and_cross() -> BM25OperatorOptions: + """Use the cross-property 'And' operator for keyword queries, where all query tokens must match across the searched properties combined. + + Unlike `and_()`, which requires every token to occur within a single property, a token may be + matched by any of the searched properties, as long as each token is matched by at least one. + + All searched properties must share the same tokenization and analyzer settings; the server + rejects the query otherwise. + + Requires Weaviate `1.37.15`, `1.38.8`, `1.39.0` or higher. + """ + return BM25OperatorAndCross() + OneDimensionalVectorType = Sequence[NUMBER] """Represents a one-dimensional vector, e.g. one produced by the `Configure.Vectors.text2vec_jinaai()` module""" diff --git a/weaviate/collections/grpc/query.py b/weaviate/collections/grpc/query.py index 46593da10..281da6e2d 100644 --- a/weaviate/collections/grpc/query.py +++ b/weaviate/collections/grpc/query.py @@ -24,7 +24,6 @@ REFERENCE, REFERENCES, BM25OperatorOptions, - BM25OperatorOr, HybridFusion, HybridVectorType, Move, @@ -243,14 +242,7 @@ def bm25( base_search_pb2.BM25( query=query, properties=properties if properties is not None else [], - search_operator=base_search_pb2.SearchOperatorOptions( - operator=operator.operator, - minimum_or_tokens_match=operator.minimum_should_match - if isinstance(operator, BM25OperatorOr) - else None, - ) - if operator is not None - else None, + search_operator=self._bm25_operator_to_grpc(operator), ) if query is not None else None diff --git a/weaviate/collections/grpc/shared.py b/weaviate/collections/grpc/shared.py index 51e2a65c7..3b39a611b 100644 --- a/weaviate/collections/grpc/shared.py +++ b/weaviate/collections/grpc/shared.py @@ -18,6 +18,7 @@ from weaviate.collections.classes.config import ConsistencyLevel from weaviate.collections.classes.grpc import ( MMR, + BM25OperatorAndCross, BM25OperatorOptions, BM25OperatorOr, HybridFusion, @@ -35,6 +36,7 @@ ) from weaviate.exceptions import ( WeaviateInvalidInputError, + WeaviateUnsupportedFeatureError, ) from weaviate.proto.v1 import base_pb2, base_search_pb2 from weaviate.types import NUMBER, UUID @@ -49,6 +51,12 @@ UINT32_LEN = 4 UINT64_LEN = 8 +# Cross-property AND was backported to the 1.37 and 1.38 branches after landing in 1.39. +_BM25_AND_CROSS_MIN_VERSIONS = ((1, 37, 15), (1, 38, 8), (1, 39, 0)) +_BM25_AND_CROSS_MIN_VERSIONS_STR = " or ".join( + f"{major}.{minor}.{patch}" for major, minor, patch in _BM25_AND_CROSS_MIN_VERSIONS +) + class _BaseGRPC: def __init__( @@ -76,6 +84,30 @@ def _get_consistency_level( assert consistency_level.value == ConsistencyLevel.ALL return base_pb2.ConsistencyLevel.CONSISTENCY_LEVEL_ALL + def _bm25_operator_to_grpc( + self, bm25_operator: Optional[BM25OperatorOptions] + ) -> Optional["base_search_pb2.SearchOperatorOptions"]: + if bm25_operator is None: + return None + + if isinstance( + bm25_operator, BM25OperatorAndCross + ) and not self._weaviate_version.is_at_least_any(*_BM25_AND_CROSS_MIN_VERSIONS): + raise WeaviateUnsupportedFeatureError( + "BM25Operator.and_cross()", + str(self._weaviate_version), + _BM25_AND_CROSS_MIN_VERSIONS_STR, + ) + + return base_search_pb2.SearchOperatorOptions( + operator=bm25_operator.operator, + minimum_or_tokens_match=( + bm25_operator.minimum_should_match + if isinstance(bm25_operator, BM25OperatorOr) + else None + ), + ) + def _recompute_target_vector_to_grpc( self, target_vector: Optional[TargetVectorJoinType], @@ -741,14 +773,7 @@ def _parse_hybrid( vector_distance=distance, vectors=vectors, selection=self._diversity_selection_to_grpc(diversity_selection), - bm25_search_operator=base_search_pb2.SearchOperatorOptions( - operator=bm25_operator.operator, - minimum_or_tokens_match=bm25_operator.minimum_should_match - if isinstance(bm25_operator, BM25OperatorOr) - else None, - ) - if bm25_operator is not None - else None, + bm25_search_operator=self._bm25_operator_to_grpc(bm25_operator), ) if query is not None or vector is not None else None diff --git a/weaviate/outputs/query.py b/weaviate/outputs/query.py index 1113c5e4a..e4fca0e7f 100644 --- a/weaviate/outputs/query.py +++ b/weaviate/outputs/query.py @@ -9,6 +9,7 @@ from weaviate.collections.classes.grpc import ( MMR, BM25OperatorAnd, + BM25OperatorAndCross, BM25OperatorOr, ListOfVectorsQuery, NearVectorInputType, @@ -62,6 +63,7 @@ "GenerativeSearchReturnType", "GeoCoordinate", "BM25OperatorAnd", + "BM25OperatorAndCross", "BM25OperatorOr", "ListOfVectorsQuery", "MMR", diff --git a/weaviate/util.py b/weaviate/util.py index a5b35e6aa..0da8c2905 100644 --- a/weaviate/util.py +++ b/weaviate/util.py @@ -607,6 +607,17 @@ def from_string(cls, version: str) -> "_ServerVersion": f"Unable to parse a version from the input string: {initial}. Is it in the format '(v)x.y.z' (e.g. 'v1.18.2' or '1.18.0')?" ) + def is_at_least_any(self, *minimums: Tuple[int, int, int]) -> bool: + """Check a minimum that was backported to several release branches. + + Each entry is the first version on its own minor branch to carry the feature, in ascending + order; the server is checked against the newest branch that is not newer than itself. + """ + for major, minor, patch in reversed(minimums): + if (self.major, self.minor) >= (major, minor): + return self >= _ServerVersion(major, minor, patch) + return False + def check_is_at_least_1_25_0(self, feature: str) -> None: if not self >= _ServerVersion(1, 25, 0): raise WeaviateUnsupportedFeatureError(feature, str(self), "1.25.0")