dev/1.39

.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:

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])

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)
 

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

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"})

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())

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)

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