diff --git a/integration/test_collection_config.py b/integration/test_collection_config.py index 814c5d41a..542eeaa1d 100644 --- a/integration/test_collection_config.py +++ b/integration/test_collection_config.py @@ -7,6 +7,7 @@ import weaviate import weaviate.classes as wvc from integration.conftest import ( + AsyncCollectionFactory, CollectionFactory, OpenAICollection, _sanitize_collection_name, @@ -40,6 +41,9 @@ _NamedVectorConfigCreate, _VectorizerConfigCreate, IndexName, + InvertedIndexState, + InvertedIndexTaskStatus, + InvertedIndexType, ) from weaviate.collections.classes.tenants import Tenant from weaviate.exceptions import ( @@ -2678,3 +2682,236 @@ def test_text_analyzer_roundtrip_from_dict( assert config == new assert config.to_dict() == new.to_dict() client.collections.delete(name) + + +def test_property_reindex_searchable_lifecycle(collection_factory: CollectionFactory) -> None: + """Test the full runtime lifecycle of a searchable index: create, no-op, rebuild, cancel, delete.""" + collection_dummy = collection_factory("dummy") + if collection_dummy._connection._weaviate_version.is_lower_than(1, 39, 0): + pytest.skip("Runtime property reindex requires Weaviate >= 1.39.0") + + collection = collection_factory( + properties=[ + Property( + name="name", + data_type=DataType.TEXT, + index_filterable=True, + index_searchable=False, + ) + ], + ) + collection.data.insert_many([{"name": f"object {i}"} for i in range(10)]) + + # create the searchable index declaratively and wait for it to become ready + status = collection.config.update_property_index( + "name", + InvertedIndexType.SEARCHABLE, + tokenization=Tokenization.WORD, + wait_for_completion=True, + ) + assert status.type == "searchable" + assert status.status == InvertedIndexState.READY + assert status.tokenization == Tokenization.WORD + + # re-putting the matching configuration is a no-op + task = collection.config.update_property_index( + "name", InvertedIndexType.SEARCHABLE, tokenization=Tokenization.WORD + ) + assert task.status == InvertedIndexTaskStatus.NO_OP + assert task.task_id is None + + # the status endpoint reports the index as ready + indexes = collection.config.get_property_indexes() + assert indexes.collection == collection.name + entry = next( + index + for prop in indexes.properties + if prop.name == "name" + for index in prop.indexes + if index.type == "searchable" + ) + assert entry.status == InvertedIndexState.READY + + # rebuild the index from scratch + status = collection.config.rebuild_property_index( + "name", InvertedIndexType.SEARCHABLE, wait_for_completion=True + ) + assert status.type == "searchable" + assert status.status == InvertedIndexState.READY + + # cancelling when no task is live is an idempotent no-op + task = collection.config.cancel_property_index_task("name", InvertedIndexType.SEARCHABLE) + assert task.status == InvertedIndexTaskStatus.NO_OP + + # the pre-existing delete API removes the index again + assert collection.config.delete_property_index("name", "searchable") is True + + +def test_property_reindex_range_filters(collection_factory: CollectionFactory) -> None: + """Test creating a rangeFilters index on an int property via an empty request body.""" + collection_dummy = collection_factory("dummy") + if collection_dummy._connection._weaviate_version.is_lower_than(1, 39, 0): + pytest.skip("Runtime property reindex requires Weaviate >= 1.39.0") + + collection = collection_factory( + properties=[ + Property( + name="age", + data_type=DataType.INT, + index_filterable=True, + index_range_filters=False, + ) + ], + ) + collection.data.insert_many([{"age": i} for i in range(10)]) + + status = collection.config.update_property_index( + "age", InvertedIndexType.RANGE_FILTERS, wait_for_completion=True + ) + assert status.type == "rangeFilters" + assert status.status == InvertedIndexState.READY + + entry = next( + index + for prop in collection.config.get_property_indexes().properties + if prop.name == "age" + for index in prop.indexes + if index.type == "rangeFilters" + ) + assert entry.status == InvertedIndexState.READY + + +def test_property_reindex_coupled_tokenization_change( + collection_factory: CollectionFactory, +) -> None: + """Test that a tokenization change on searchable is coupled with the filterable index as one task.""" + collection_dummy = collection_factory("dummy") + if collection_dummy._connection._weaviate_version.is_lower_than(1, 39, 0): + pytest.skip("Runtime property reindex requires Weaviate >= 1.39.0") + + collection = collection_factory( + properties=[ + Property( + name="name", + data_type=DataType.TEXT, + index_filterable=True, + index_searchable=True, + tokenization=Tokenization.WORD, + ) + ], + ) + collection.data.insert_many([{"name": f"object {i}"} for i in range(100)]) + + task = collection.config.update_property_index( + "name", InvertedIndexType.SEARCHABLE, tokenization=Tokenization.FIELD + ) + assert task.status == InvertedIndexTaskStatus.STARTED + assert task.task_id is not None + + prop = next(p for p in collection.config.get_property_indexes().properties if p.name == "name") + searchable = next(i for i in prop.indexes if i.type == "searchable") + filterable = next(i for i in prop.indexes if i.type == "filterable") + if searchable.task_id is not None: + # both entries are driven by the one coupled task while it is in flight + assert searchable.task_id == task.task_id + assert filterable.task_id == task.task_id + assert searchable.target_tokenization == Tokenization.FIELD + assert filterable.target_tokenization == Tokenization.FIELD + else: + # the task already finalized before the first poll + assert searchable.tokenization == Tokenization.FIELD + + # poll the status endpoint (via the wait path of a NO_OP upsert) until the migration is done + status = collection.config.update_property_index( + "name", + InvertedIndexType.SEARCHABLE, + tokenization=Tokenization.FIELD, + wait_for_completion=True, + ) + assert status.status == InvertedIndexState.READY + assert status.tokenization == Tokenization.FIELD + + prop = next(p for p in collection.config.get_property_indexes().properties if p.name == "name") + filterable = next(i for i in prop.indexes if i.type == "filterable") + assert filterable.status == InvertedIndexState.READY + assert filterable.tokenization == Tokenization.FIELD + + +def test_property_reindex_multi_tenant(collection_factory: CollectionFactory) -> None: + """Test rangeFilters creation and rebuild with a tenants selection on a multi-tenant collection.""" + collection_dummy = collection_factory("dummy") + if collection_dummy._connection._weaviate_version.is_lower_than(1, 39, 0): + pytest.skip("Runtime property reindex requires Weaviate >= 1.39.0") + + collection = collection_factory( + properties=[ + Property( + name="age", + data_type=DataType.INT, + index_filterable=True, + index_range_filters=False, + ) + ], + multi_tenancy_config=Configure.multi_tenancy(enabled=True), + ) + collection.tenants.create([Tenant(name="tenant1"), Tenant(name="tenant2")]) + collection.with_tenant("tenant1").data.insert_many([{"age": i} for i in range(5)]) + + status = collection.config.update_property_index( + "age", + InvertedIndexType.RANGE_FILTERS, + tenants=["tenant1", "tenant2"], + wait_for_completion=True, + ) + assert status.type == "rangeFilters" + assert status.status == InvertedIndexState.READY + + status = collection.config.rebuild_property_index( + "age", InvertedIndexType.RANGE_FILTERS, tenants=["tenant1"], wait_for_completion=True + ) + assert status.type == "rangeFilters" + assert status.status == InvertedIndexState.READY + + +@pytest.mark.asyncio +async def test_property_reindex_async(async_collection_factory: AsyncCollectionFactory) -> None: + """Test the runtime property reindex lifecycle through the async client.""" + collection = await async_collection_factory( + properties=[ + Property( + name="name", + data_type=DataType.TEXT, + index_filterable=True, + index_searchable=False, + ) + ], + ) + if collection._connection._weaviate_version.is_lower_than(1, 39, 0): + pytest.skip("Runtime property reindex requires Weaviate >= 1.39.0") + + await collection.data.insert_many([{"name": f"object {i}"} for i in range(10)]) + + status = await collection.config.update_property_index( + "name", + InvertedIndexType.SEARCHABLE, + tokenization=Tokenization.WORD, + wait_for_completion=True, + ) + assert status.type == "searchable" + assert status.status == InvertedIndexState.READY + + task = await collection.config.update_property_index( + "name", InvertedIndexType.SEARCHABLE, tokenization=Tokenization.WORD + ) + assert task.status == InvertedIndexTaskStatus.NO_OP + + indexes = await collection.config.get_property_indexes() + assert indexes.collection == collection.name + + status = await collection.config.rebuild_property_index( + "name", InvertedIndexType.SEARCHABLE, wait_for_completion=True + ) + assert status.status == InvertedIndexState.READY + + task = await collection.config.cancel_property_index_task("name", InvertedIndexType.SEARCHABLE) + assert task.status == InvertedIndexTaskStatus.NO_OP diff --git a/mock_tests/test_property_reindex.py b/mock_tests/test_property_reindex.py new file mode 100644 index 000000000..c6813ea13 --- /dev/null +++ b/mock_tests/test_property_reindex.py @@ -0,0 +1,603 @@ +import json +from typing import Generator, Union + +import grpc +import pytest +from pytest_httpserver import HTTPServer +from werkzeug.wrappers import Response + +import weaviate +from mock_tests.conftest import MOCK_IP, MOCK_PORT, MOCK_PORT_GRPC +from weaviate.collections.classes.config import ( + DataType, + InvertedIndexState, + InvertedIndexTaskStatus, + InvertedIndexType, + Tokenization, +) +from weaviate.exceptions import ( + ReindexCanceledError, + ReindexFailedError, + WeaviateUnsupportedFeatureError, +) + +COLLECTION = "TestCollection" +SCHEMA_PATH = f"/v1/schema/{COLLECTION}" +TASK_ID = "00000000-0000-0000-0000-000000000001" + + +@pytest.fixture(scope="function") +def weaviate_139_mock(ready_mock: HTTPServer) -> Generator[HTTPServer, None, None]: + """A mock server advertising Weaviate 1.39.0, which supports runtime property reindexing.""" + ready_mock.expect_request("/v1/meta").respond_with_json({"version": "1.39.0"}) + ready_mock.expect_request("/v1/nodes").respond_with_json( + {"nodes": [{"gitHash": "ABC", "status": "HEALTHY"}]} + ) + ready_mock.expect_request("/v1/.well-known/openid-configuration").respond_with_response( + Response(json.dumps({}), status=404) + ) + yield ready_mock + + +@pytest.fixture(scope="function") +def client_139( + weaviate_139_mock: HTTPServer, start_grpc_server: grpc.Server +) -> Generator[weaviate.WeaviateClient, None, None]: + client = weaviate.connect_to_local(port=MOCK_PORT, host=MOCK_IP, grpc_port=MOCK_PORT_GRPC) + yield client + client.close() + + +def test_update_property_index_started( + weaviate_139_mock: HTTPServer, client_139: weaviate.WeaviateClient +) -> None: + weaviate_139_mock.expect_request( + f"{SCHEMA_PATH}/properties/name/index/searchable", + method="PUT", + json={"tokenization": "word", "algorithm": "blockmax"}, + ).respond_with_json({"taskId": TASK_ID, "status": "STARTED"}, status=202) + + task = client_139.collections.use(COLLECTION).config.update_property_index( + "name", + "searchable", + tokenization=Tokenization.WORD, + algorithm="blockmax", + ) + assert task.task_id == TASK_ID + assert task.status == InvertedIndexTaskStatus.STARTED + weaviate_139_mock.check_assertions() + + +def test_update_property_index_no_op( + weaviate_139_mock: HTTPServer, client_139: weaviate.WeaviateClient +) -> None: + weaviate_139_mock.expect_request( + f"{SCHEMA_PATH}/properties/name/index/searchable", + method="PUT", + json={"tokenization": "word"}, + ).respond_with_json({"status": "NO_OP"}, status=200) + + task = client_139.collections.use(COLLECTION).config.update_property_index( + "name", "searchable", tokenization=Tokenization.WORD + ) + assert task.task_id is None + assert task.status == InvertedIndexTaskStatus.NO_OP + weaviate_139_mock.check_assertions() + + +def test_update_property_index_range_filters_with_tenants( + weaviate_139_mock: HTTPServer, client_139: weaviate.WeaviateClient +) -> None: + """A rangeFilters creation sends an empty body and encodes tenants as a csv query param.""" + weaviate_139_mock.expect_request( + f"{SCHEMA_PATH}/properties/age/index/rangeFilters", + method="PUT", + query_string={"tenants": "tenant1,tenant2"}, + json={}, + ).respond_with_json({"taskId": TASK_ID, "status": "STARTED"}, status=202) + + task = client_139.collections.use(COLLECTION).config.update_property_index( + "age", "rangeFilters", tenants=["tenant1", "tenant2"] + ) + assert task.task_id == TASK_ID + assert task.status == InvertedIndexTaskStatus.STARTED + weaviate_139_mock.check_assertions() + + +def test_update_property_index_wait_for_completion( + weaviate_139_mock: HTTPServer, client_139: weaviate.WeaviateClient +) -> None: + weaviate_139_mock.expect_request( + f"{SCHEMA_PATH}/properties/name/index/searchable", + method="PUT", + json={"tokenization": "word"}, + ).respond_with_json({"taskId": TASK_ID, "status": "STARTED"}, status=202) + weaviate_139_mock.expect_request(f"{SCHEMA_PATH}/indexes", method="GET").respond_with_json( + { + "collection": COLLECTION, + "properties": [ + { + "name": "name", + "dataType": "text", + "indexes": [{"type": "searchable", "status": "ready", "tokenization": "word"}], + } + ], + } + ) + + status = client_139.collections.use(COLLECTION).config.update_property_index( + "name", "searchable", tokenization=Tokenization.WORD, wait_for_completion=True + ) + assert status.type == "searchable" + assert status.status == InvertedIndexState.READY + assert status.tokenization == Tokenization.WORD + weaviate_139_mock.check_assertions() + + +def test_update_property_index_bare_str_tenant( + weaviate_139_mock: HTTPServer, client_139: weaviate.WeaviateClient +) -> None: + """A bare string tenant is normalized to a single csv value, not exploded into characters.""" + weaviate_139_mock.expect_request( + f"{SCHEMA_PATH}/properties/age/index/rangeFilters", + method="PUT", + query_string={"tenants": "tenant1"}, + json={}, + ).respond_with_json({"taskId": TASK_ID, "status": "STARTED"}, status=202) + + task = client_139.collections.use(COLLECTION).config.update_property_index( + "age", "rangeFilters", tenants="tenant1" + ) + assert task.status == InvertedIndexTaskStatus.STARTED + weaviate_139_mock.check_assertions() + + +@pytest.mark.parametrize( + "index_status,exception", + [("failed", ReindexFailedError), ("cancelled", ReindexCanceledError)], +) +def test_update_property_index_wait_raises( + weaviate_139_mock: HTTPServer, + client_139: weaviate.WeaviateClient, + index_status: str, + exception: type, +) -> None: + weaviate_139_mock.expect_request( + f"{SCHEMA_PATH}/properties/name/index/searchable", + method="PUT", + json={"tokenization": "word"}, + ).respond_with_json({"taskId": TASK_ID, "status": "STARTED"}, status=202) + weaviate_139_mock.expect_request(f"{SCHEMA_PATH}/indexes", method="GET").respond_with_json( + { + "collection": COLLECTION, + "properties": [ + { + "name": "name", + "dataType": "text", + "indexes": [ + { + "type": "searchable", + "status": index_status, + "progress": 0.42, + "taskId": TASK_ID, + "tokenization": "word", + } + ], + } + ], + } + ) + + with pytest.raises(exception): + client_139.collections.use(COLLECTION).config.update_property_index( + "name", "searchable", tokenization=Tokenization.WORD, wait_for_completion=True + ) + weaviate_139_mock.check_assertions() + + +@pytest.mark.parametrize( + "index_status,exception", + [("failed", ReindexFailedError), ("cancelled", ReindexCanceledError)], +) +def test_rebuild_property_index_wait_raises( + weaviate_139_mock: HTTPServer, + client_139: weaviate.WeaviateClient, + index_status: str, + exception: type, +) -> None: + weaviate_139_mock.expect_request( + f"{SCHEMA_PATH}/properties/name/index/searchable/rebuild", + method="POST", + json={}, + ).respond_with_json({"taskId": TASK_ID, "status": "STARTED"}, status=202) + weaviate_139_mock.expect_request(f"{SCHEMA_PATH}/indexes", method="GET").respond_with_json( + { + "collection": COLLECTION, + "properties": [ + { + "name": "name", + "dataType": "text", + "indexes": [ + { + "type": "searchable", + "status": index_status, + "progress": 0.42, + "taskId": TASK_ID, + "tokenization": "word", + } + ], + } + ], + } + ) + + with pytest.raises(exception): + client_139.collections.use(COLLECTION).config.rebuild_property_index( + "name", "searchable", wait_for_completion=True + ) + weaviate_139_mock.check_assertions() + + +def test_rebuild_property_index( + weaviate_139_mock: HTTPServer, client_139: weaviate.WeaviateClient +) -> None: + weaviate_139_mock.expect_request( + f"{SCHEMA_PATH}/properties/name/index/searchable/rebuild", + method="POST", + json={}, + ).respond_with_json({"taskId": TASK_ID, "status": "STARTED"}, status=202) + + task = client_139.collections.use(COLLECTION).config.rebuild_property_index( + "name", "searchable" + ) + assert task.task_id == TASK_ID + assert task.status == InvertedIndexTaskStatus.STARTED + weaviate_139_mock.check_assertions() + + +def test_rebuild_property_index_with_tenants( + weaviate_139_mock: HTTPServer, client_139: weaviate.WeaviateClient +) -> None: + weaviate_139_mock.expect_request( + f"{SCHEMA_PATH}/properties/age/index/rangeFilters/rebuild", + method="POST", + query_string={"tenants": "tenant1,tenant2"}, + json={}, + ).respond_with_json({"taskId": TASK_ID, "status": "STARTED"}, status=202) + + task = client_139.collections.use(COLLECTION).config.rebuild_property_index( + "age", "rangeFilters", tenants=["tenant1", "tenant2"] + ) + assert task.task_id == TASK_ID + assert task.status == InvertedIndexTaskStatus.STARTED + weaviate_139_mock.check_assertions() + + +def test_cancel_property_index_task_cancelled( + weaviate_139_mock: HTTPServer, client_139: weaviate.WeaviateClient +) -> None: + weaviate_139_mock.expect_request( + f"{SCHEMA_PATH}/properties/name/index/searchable/cancel", + method="POST", + json={}, + ).respond_with_json({"taskId": TASK_ID, "status": "CANCELLED"}, status=202) + + task = client_139.collections.use(COLLECTION).config.cancel_property_index_task( + "name", "searchable" + ) + assert task.task_id == TASK_ID + assert task.status == InvertedIndexTaskStatus.CANCELLED + weaviate_139_mock.check_assertions() + + +def test_cancel_property_index_task_no_op( + weaviate_139_mock: HTTPServer, client_139: weaviate.WeaviateClient +) -> None: + weaviate_139_mock.expect_request( + f"{SCHEMA_PATH}/properties/name/index/searchable/cancel", + method="POST", + json={}, + ).respond_with_json({"status": "NO_OP"}, status=202) + + task = client_139.collections.use(COLLECTION).config.cancel_property_index_task( + "name", "searchable" + ) + assert task.task_id is None + assert task.status == InvertedIndexTaskStatus.NO_OP + weaviate_139_mock.check_assertions() + + +def test_get_property_indexes( + weaviate_139_mock: HTTPServer, client_139: weaviate.WeaviateClient +) -> None: + """A coupled tokenization change renders both entries with a shared taskId and progress.""" + weaviate_139_mock.expect_request(f"{SCHEMA_PATH}/indexes", method="GET").respond_with_json( + { + "collection": COLLECTION, + "properties": [ + { + "name": "name", + "dataType": "text", + "description": "a text property", + "indexes": [ + { + "type": "searchable", + "status": "indexing", + "progress": 0.5, + "taskId": TASK_ID, + "tokenization": "word", + "targetTokenization": "field", + "algorithm": "wand", + "targetAlgorithm": "blockmax", + }, + { + "type": "filterable", + "status": "indexing", + "progress": 0.5, + "taskId": TASK_ID, + "tokenization": "word", + "targetTokenization": "field", + }, + ], + }, + { + "name": "age", + "dataType": "int", + "indexes": [{"type": "rangeFilters", "status": "ready"}], + }, + ], + } + ) + + indexes = client_139.collections.use(COLLECTION).config.get_property_indexes() + assert indexes.collection == COLLECTION + assert len(indexes.properties) == 2 + + name = indexes.properties[0] + assert name.name == "name" + assert name.data_type == "text" + assert name.description == "a text property" + assert len(name.indexes) == 2 + searchable, filterable = name.indexes + assert searchable.type == "searchable" + assert searchable.status == InvertedIndexState.INDEXING + assert searchable.progress == 0.5 + assert searchable.task_id == TASK_ID + assert searchable.tokenization == Tokenization.WORD + assert searchable.target_tokenization == Tokenization.FIELD + assert searchable.algorithm == "wand" + assert searchable.target_algorithm == "blockmax" + assert filterable.type == "filterable" + assert filterable.task_id == TASK_ID # coupled change: one task drives both entries + assert filterable.target_tokenization == Tokenization.FIELD + assert filterable.algorithm is None + assert filterable.target_algorithm is None + + age = indexes.properties[1] + assert age.name == "age" + assert age.data_type == "int" + assert age.description is None + assert len(age.indexes) == 1 + assert age.indexes[0].type == "rangeFilters" + assert age.indexes[0].status == InvertedIndexState.READY + assert age.indexes[0].progress is None + assert age.indexes[0].task_id is None + assert age.indexes[0].tokenization is None + + # the nested dataclasses serialize all the way down to a JSON-compatible dict + out = json.loads(json.dumps(indexes.to_dict())) + assert out["collection"] == COLLECTION + assert out["properties"][0]["indexes"][0]["taskId"] == TASK_ID + assert out["properties"][0]["indexes"][0]["targetTokenization"] == "field" + assert out["properties"][1]["dataType"] == "int" + assert out["properties"][1]["indexes"][0]["status"] == "ready" + + weaviate_139_mock.check_assertions() + + +@pytest.mark.parametrize( + "index_name,wire", + [ + (InvertedIndexType.SEARCHABLE, "searchable"), + ("searchable", "searchable"), + (InvertedIndexType.FILTERABLE, "filterable"), + ("filterable", "filterable"), + (InvertedIndexType.RANGE_FILTERS, "rangeFilters"), + ("rangeFilters", "rangeFilters"), + ], +) +def test_update_property_index_enum_and_literal_hit_same_route( + weaviate_139_mock: HTTPServer, + client_139: weaviate.WeaviateClient, + index_name: Union[InvertedIndexType, str], + wire: str, +) -> None: + """The enum and literal forms of index_name hit the exact same wire route.""" + weaviate_139_mock.expect_request( + f"{SCHEMA_PATH}/properties/name/index/{wire}", + method="PUT", + json={}, + ).respond_with_json({"taskId": TASK_ID, "status": "STARTED"}, status=202) + + task = client_139.collections.use(COLLECTION).config.update_property_index( + "name", + # runtime leniency pin: raw strings must keep hitting the same route + index_name, # type: ignore + ) + assert task.status == InvertedIndexTaskStatus.STARTED + weaviate_139_mock.check_assertions() + + +@pytest.mark.parametrize( + "index_name,wire", + [ + (InvertedIndexType.SEARCHABLE, "searchable"), + ("searchable", "searchable"), + (InvertedIndexType.FILTERABLE, "filterable"), + ("filterable", "filterable"), + (InvertedIndexType.RANGE_FILTERS, "rangeFilters"), + ("rangeFilters", "rangeFilters"), + ], +) +def test_delete_property_index_enum_and_literal_hit_same_route( + weaviate_139_mock: HTTPServer, + client_139: weaviate.WeaviateClient, + index_name: Union[InvertedIndexType, str], + wire: str, +) -> None: + """The enum and literal forms of index_name hit the exact same wire route.""" + weaviate_139_mock.expect_request( + f"{SCHEMA_PATH}/properties/name/index/{wire}", + method="DELETE", + ).respond_with_json({}, status=200) + + assert ( + client_139.collections.use(COLLECTION).config.delete_property_index( + "name", + # runtime leniency pin: raw strings must keep hitting the same route + index_name, # type: ignore + ) + is True + ) + weaviate_139_mock.check_assertions() + + +@pytest.mark.parametrize( + "index_name", + [InvertedIndexType.RANGE_FILTERS, "rangeFilters"], +) +def test_rebuild_property_index_enum_and_literal_hit_same_route( + weaviate_139_mock: HTTPServer, + client_139: weaviate.WeaviateClient, + index_name: Union[InvertedIndexType, str], +) -> None: + """The enum and literal forms of index_name hit the exact same rebuild route.""" + weaviate_139_mock.expect_request( + f"{SCHEMA_PATH}/properties/age/index/rangeFilters/rebuild", + method="POST", + json={}, + ).respond_with_json({"taskId": TASK_ID, "status": "STARTED"}, status=202) + + task = client_139.collections.use(COLLECTION).config.rebuild_property_index( + "age", + # runtime leniency pin: raw strings must keep hitting the same route + index_name, # type: ignore + ) + assert task.status == InvertedIndexTaskStatus.STARTED + weaviate_139_mock.check_assertions() + + +@pytest.mark.parametrize( + "index_name", + [InvertedIndexType.RANGE_FILTERS, "rangeFilters"], +) +def test_cancel_property_index_task_enum_and_literal_hit_same_route( + weaviate_139_mock: HTTPServer, + client_139: weaviate.WeaviateClient, + index_name: Union[InvertedIndexType, str], +) -> None: + """The enum and literal forms of index_name hit the exact same cancel route.""" + weaviate_139_mock.expect_request( + f"{SCHEMA_PATH}/properties/age/index/rangeFilters/cancel", + method="POST", + json={}, + ).respond_with_json({"taskId": TASK_ID, "status": "CANCELLED"}, status=202) + + task = client_139.collections.use(COLLECTION).config.cancel_property_index_task( + "age", + # runtime leniency pin: raw strings must keep hitting the same route + index_name, # type: ignore + ) + assert task.status == InvertedIndexTaskStatus.CANCELLED + weaviate_139_mock.check_assertions() + + +def test_get_property_indexes_reference_property( + weaviate_139_mock: HTTPServer, client_139: weaviate.WeaviateClient +) -> None: + """Reference properties carry the target collection name as dataType and still parse.""" + weaviate_139_mock.expect_request(f"{SCHEMA_PATH}/indexes", method="GET").respond_with_json( + { + "collection": COLLECTION, + "properties": [ + { + "name": "title", + "dataType": "text", + "indexes": [{"type": "searchable", "status": "ready", "tokenization": "word"}], + }, + { + "name": "ofArticle", + "dataType": "Article", + "indexes": [{"type": "filterable", "status": "ready"}], + }, + ], + } + ) + + indexes = client_139.collections.use(COLLECTION).config.get_property_indexes() + title, ref = indexes.properties + # primitive values match the DataType str-enum + assert title.data_type == DataType.TEXT + # reference properties carry the qualified target collection name instead + assert ref.name == "ofArticle" + assert ref.data_type == "Article" + assert ref.indexes[0].type == "filterable" + + out = json.loads(json.dumps(indexes.to_dict())) + assert out["properties"][1]["dataType"] == "Article" + + weaviate_139_mock.check_assertions() + + +def test_delete_property_index_surfaces_server_message( + weaviate_139_mock: HTTPServer, client_139: weaviate.WeaviateClient +) -> None: + """A DELETE rejection surfaces the server's cause behind a neutral prefix. + + The 422 mutation guard (in-flight reindex task) carries an actionable server message; + the client prefix must only name the failed operation, not assert a cause. + """ + server_message = "cannot delete index: a reindex task is in progress for property 'name'" + weaviate_139_mock.expect_request( + f"{SCHEMA_PATH}/properties/name/index/searchable", + method="DELETE", + ).respond_with_json({"error": [{"message": server_message}]}, status=422) + + with pytest.raises(weaviate.exceptions.UnexpectedStatusCodeError) as e: + client_139.collections.use(COLLECTION).config.delete_property_index("name", "searchable") + assert e.value.status_code == 422 + assert "Property index may not have been deleted." in e.value.message + assert server_message in e.value.message + assert "may not exist" not in e.value.message + weaviate_139_mock.check_assertions() + + +def test_property_reindex_invalid_input( + weaviate_139_mock: HTTPServer, client_139: weaviate.WeaviateClient +) -> None: + """Invalid argument types raise WeaviateInvalidInputError before any request is sent.""" + config = client_139.collections.use(COLLECTION).config + + with pytest.raises(weaviate.exceptions.WeaviateInvalidInputError): + config.update_property_index("age", "rangeFilters", tenants=123) # type: ignore + with pytest.raises(weaviate.exceptions.WeaviateInvalidInputError): + config.rebuild_property_index("age", "rangeFilters", tenants=[1, 2]) # type: ignore + with pytest.raises(weaviate.exceptions.WeaviateInvalidInputError): + config.cancel_property_index_task(123, "searchable") # type: ignore + weaviate_139_mock.check_assertions() + + +def test_property_reindex_unsupported_version( + weaviate_client: weaviate.WeaviateClient, +) -> None: + """Every new method raises against a server older than 1.39.0 (the mock advertises 1.36).""" + config = weaviate_client.collections.use(COLLECTION).config + + with pytest.raises(WeaviateUnsupportedFeatureError): + config.update_property_index("name", "searchable", tokenization=Tokenization.WORD) + with pytest.raises(WeaviateUnsupportedFeatureError): + config.rebuild_property_index("name", "searchable") + with pytest.raises(WeaviateUnsupportedFeatureError): + config.cancel_property_index_task("name", "searchable") + with pytest.raises(WeaviateUnsupportedFeatureError): + config.get_property_indexes() diff --git a/weaviate/classes/config.py b/weaviate/classes/config.py index c154062d3..a806aa04a 100644 --- a/weaviate/classes/config.py +++ b/weaviate/classes/config.py @@ -4,6 +4,7 @@ DataType, GenerativeSearches, IndexName, + InvertedIndexType, PQEncoderDistribution, PQEncoderType, Property, @@ -37,6 +38,7 @@ "MultiVectorAggregation", "ReplicationDeletionStrategy", "Property", + "InvertedIndexType", "PQEncoderDistribution", "PQEncoderType", "ReferenceProperty", diff --git a/weaviate/collections/classes/config.py b/weaviate/collections/classes/config.py index 0f6d974c0..9026dd693 100644 --- a/weaviate/collections/classes/config.py +++ b/weaviate/collections/classes/config.py @@ -117,6 +117,20 @@ ] +class InvertedIndexType(str, BaseEnum): + """The available property index types in Weaviate. + + Attributes: + SEARCHABLE: The searchable index, used for keyword (BM25) searches over text properties. + FILTERABLE: The filterable index, used for exact-match filtering. + RANGE_FILTERS: The rangeFilters index, used for range filtering. + """ + + SEARCHABLE = "searchable" + FILTERABLE = "filterable" + RANGE_FILTERS = "rangeFilters" + + class ConsistencyLevel(str, BaseEnum): """The consistency levels when writing to Weaviate with replication enabled. @@ -2207,6 +2221,96 @@ class _ShardStatus: ShardStatus = _ShardStatus +class InvertedIndexTaskStatus(str, BaseEnum): + """The status of a runtime property index task submission. + + Attributes: + STARTED: A reindexing task was submitted and started. + CANCELLED: A live reindexing task was cancelled. + NO_OP: No work was necessary, e.g. the index configuration already matched the request + or there was no live task to cancel. + """ + + STARTED = "STARTED" + CANCELLED = "CANCELLED" + NO_OP = "NO_OP" + + +class InvertedIndexState(str, BaseEnum): + """The state of a property index as reported by the index status endpoint. + + Attributes: + READY: The index is ready to serve queries. + PENDING: A reindexing task for the index is queued but has not started yet. + INDEXING: A reindexing task for the index is in progress. + FAILED: The reindexing task for the index failed. + CANCELLED: The reindexing task for the index was cancelled. + """ + + READY = "ready" + PENDING = "pending" + INDEXING = "indexing" + FAILED = "failed" + CANCELLED = "cancelled" + + +@dataclass +class _InvertedIndexTask(_ConfigBase): + task_id: Optional[str] + status: InvertedIndexTaskStatus + + +InvertedIndexTask = _InvertedIndexTask + + +@dataclass +class _InvertedIndexStatus(_ConfigBase): + type: IndexName # noqa: A003 + status: InvertedIndexState + progress: Optional[float] + task_id: Optional[str] + tokenization: Optional[Tokenization] + target_tokenization: Optional[Tokenization] + algorithm: Optional[str] + target_algorithm: Optional[str] + + +InvertedIndexStatus = _InvertedIndexStatus + + +@dataclass +class _PropertyInvertedIndexes(_ConfigBase): + name: str + # For primitive properties the value matches the `DataType` str-enum (comparisons like + # `data_type == DataType.TEXT` work); reference properties instead carry the qualified + # target collection name (e.g. "Article"), which is why this is a plain str, not `DataType`. + data_type: str + description: Optional[str] + indexes: List[InvertedIndexStatus] + + def to_dict(self) -> Dict[str, Any]: + out = super().to_dict() + out["indexes"] = [index.to_dict() for index in self.indexes] + return out + + +PropertyInvertedIndexes = _PropertyInvertedIndexes + + +@dataclass +class _CollectionInvertedIndexes(_ConfigBase): + collection: str + properties: List[PropertyInvertedIndexes] + + def to_dict(self) -> Dict[str, Any]: + out = super().to_dict() + out["properties"] = [prop.to_dict() for prop in self.properties] + return out + + +CollectionInvertedIndexes = _CollectionInvertedIndexes + + class _TextAnalyzerConfigCreate(_ConfigCreateModel): """Text analysis options for a property. diff --git a/weaviate/collections/classes/config_methods.py b/weaviate/collections/classes/config_methods.py index 691cf208d..dbb7762d2 100644 --- a/weaviate/collections/classes/config_methods.py +++ b/weaviate/collections/classes/config_methods.py @@ -4,6 +4,9 @@ from weaviate.collections.classes.config import ( DataType, GenerativeSearches, + IndexName, + InvertedIndexState, + InvertedIndexTaskStatus, PQEncoderDistribution, PQEncoderType, ReplicationDeletionStrategy, @@ -19,8 +22,11 @@ _BQConfig, _CollectionConfig, _CollectionConfigSimple, + _CollectionInvertedIndexes, _GenerativeConfig, _InvertedIndexConfig, + _InvertedIndexStatus, + _InvertedIndexTask, _MultiTenancyConfig, _MultiVectorConfig, _MuveraConfig, @@ -31,6 +37,7 @@ _PQConfig, _PQEncoderConfig, _Property, + _PropertyInvertedIndexes, _PropertyVectorizerConfig, _ReferenceProperty, _ReplicationConfig, @@ -558,3 +565,44 @@ def _references_from_config(schema: Dict[str, Any]) -> List[_ReferenceProperty]: for prop in schema["properties"] if not _is_primitive(prop["dataType"]) ] + + +def _inverted_index_task_from_json(response: Dict[str, Any]) -> _InvertedIndexTask: + return _InvertedIndexTask( + task_id=response.get("taskId"), + status=InvertedIndexTaskStatus(response["status"]), + ) + + +def _inverted_index_status_from_json(index: Dict[str, Any]) -> _InvertedIndexStatus: + tokenization = index.get("tokenization") + target_tokenization = index.get("targetTokenization") + return _InvertedIndexStatus( + type=cast(IndexName, index["type"]), + status=InvertedIndexState(index["status"]), + progress=index.get("progress"), + task_id=index.get("taskId"), + tokenization=Tokenization(tokenization) if tokenization is not None else None, + target_tokenization=( + Tokenization(target_tokenization) if target_tokenization is not None else None + ), + algorithm=index.get("algorithm"), + target_algorithm=index.get("targetAlgorithm"), + ) + + +def _collection_inverted_indexes_from_json(response: Dict[str, Any]) -> _CollectionInvertedIndexes: + return _CollectionInvertedIndexes( + collection=response["collection"], + properties=[ + _PropertyInvertedIndexes( + name=prop["name"], + data_type=prop["dataType"], + description=prop.get("description"), + indexes=[ + _inverted_index_status_from_json(index) for index in prop.get("indexes") or [] + ], + ) + for prop in response.get("properties") or [] + ], + ) diff --git a/weaviate/collections/config/async_.pyi b/weaviate/collections/config/async_.pyi index 015b70dab..035ae95e7 100644 --- a/weaviate/collections/config/async_.pyi +++ b/weaviate/collections/config/async_.pyi @@ -5,11 +5,16 @@ from typing_extensions import deprecated from weaviate.collections.classes.config import ( CollectionConfig, CollectionConfigSimple, + CollectionInvertedIndexes, IndexName, + InvertedIndexStatus, + InvertedIndexTask, + InvertedIndexType, Property, ReferenceProperty, ShardStatus, ShardTypes, + Tokenization, _GenerativeProvider, _InvertedIndexConfigUpdate, _MultiTenancyConfigUpdate, @@ -89,4 +94,50 @@ class _ConfigCollectionAsync(_ConfigCollectionExecutor[ConnectionAsync]): async def add_vector( self, *, vector_config: Union[_VectorConfigCreate, List[_VectorConfigCreate]] ) -> None: ... - async def delete_property_index(self, property_name: str, index_name: IndexName) -> bool: ... + async def delete_property_index( + self, property_name: str, index_name: Union[InvertedIndexType, IndexName] + ) -> bool: ... + @overload + async def update_property_index( + self, + property_name: str, + index_name: InvertedIndexType, + *, + tokenization: Optional[Tokenization] = None, + algorithm: Optional[Literal["blockmax"]] = None, + tenants: Union[List[str], str, None] = None, + wait_for_completion: Literal[True], + ) -> InvertedIndexStatus: ... + @overload + async def update_property_index( + self, + property_name: str, + index_name: InvertedIndexType, + *, + tokenization: Optional[Tokenization] = None, + algorithm: Optional[Literal["blockmax"]] = None, + tenants: Union[List[str], str, None] = None, + wait_for_completion: Literal[False] = False, + ) -> InvertedIndexTask: ... + @overload + async def rebuild_property_index( + self, + property_name: str, + index_name: InvertedIndexType, + *, + tenants: Union[List[str], str, None] = None, + wait_for_completion: Literal[True], + ) -> InvertedIndexStatus: ... + @overload + async def rebuild_property_index( + self, + property_name: str, + index_name: InvertedIndexType, + *, + tenants: Union[List[str], str, None] = None, + wait_for_completion: Literal[False] = False, + ) -> InvertedIndexTask: ... + async def cancel_property_index_task( + self, property_name: str, index_name: InvertedIndexType + ) -> InvertedIndexTask: ... + async def get_property_indexes(self) -> CollectionInvertedIndexes: ... diff --git a/weaviate/collections/config/executor.py b/weaviate/collections/config/executor.py index 103ab70ac..7fb371bce 100644 --- a/weaviate/collections/config/executor.py +++ b/weaviate/collections/config/executor.py @@ -1,4 +1,5 @@ import asyncio +import time from typing import ( Any, Dict, @@ -20,12 +21,18 @@ from weaviate.collections.classes.config import ( CollectionConfig, CollectionConfigSimple, + CollectionInvertedIndexes, IndexName, + InvertedIndexState, + InvertedIndexStatus, + InvertedIndexTask, + InvertedIndexType, Property, PropertyType, ReferenceProperty, ShardStatus, ShardTypes, + Tokenization, _CollectionConfigUpdate, _GenerativeProvider, _InvertedIndexConfigUpdate, @@ -45,6 +52,8 @@ from weaviate.collections.classes.config_methods import ( _collection_config_from_json, _collection_config_simple_from_json, + _collection_inverted_indexes_from_json, + _inverted_index_task_from_json, ) from weaviate.collections.classes.config_object_ttl import _ObjectTTLConfigUpdate from weaviate.collections.classes.config_vector_index import ( @@ -53,6 +62,8 @@ from weaviate.connect import executor from weaviate.connect.v4 import ConnectionAsync, ConnectionType, _ExpectedStatusCodes from weaviate.exceptions import ( + ReindexCanceledError, + ReindexFailedError, WeaviateInvalidInputError, WeaviateUnsupportedFeatureError, ) @@ -79,6 +90,37 @@ def _property_has_text_analyzer(prop: Property) -> bool: return any(_property_has_text_analyzer(np) for np in nested_list) +def _find_property_index_status( + indexes: CollectionInvertedIndexes, property_name: str, index_name: IndexName +) -> Optional[InvertedIndexStatus]: + for prop in indexes.properties: + if prop.name != property_name: + continue + for index in prop.indexes: + if index.type == index_name: + return index + return None + + +def _terminal_property_index_status( + entry: Optional[InvertedIndexStatus], property_name: str, index_name: IndexName +) -> Optional[InvertedIndexStatus]: + """Return the entry once it is ready, raise on failure/cancellation, or return None to keep polling.""" + if entry is None: + return None + if entry.status == InvertedIndexState.READY: + return entry + if entry.status == InvertedIndexState.FAILED: + raise ReindexFailedError( + f"Reindexing the '{index_name}' index of property '{property_name}' failed." + ) + if entry.status == InvertedIndexState.CANCELLED: + raise ReindexCanceledError( + f"Reindexing the '{index_name}' index of property '{property_name}' was cancelled." + ) + return None + + class _ConfigCollectionExecutor(Generic[ConnectionType]): def __init__( self, @@ -629,7 +671,7 @@ async def _execute() -> None: def delete_property_index( self, property_name: str, - index_name: IndexName, + index_name: Union[InvertedIndexType, IndexName], ) -> executor.Result[bool]: """Delete a property index from the collection in Weaviate. @@ -638,13 +680,16 @@ def delete_property_index( Args: property_name: The property name from which to delete the index. - index_name: The type of the index to delete. + index_name: The type of the index to delete, an `InvertedIndexType` value or one of + the literals `searchable`, `filterable` or `rangeFilters`. Raises: weaviate.exceptions.WeaviateConnectionError: If the network connection to Weaviate fails. weaviate.exceptions.UnexpectedStatusCodeError: If Weaviate reports a non-OK status. weaviate.exceptions.WeaviateInvalidInputError: If the property or index does not exist. """ + if isinstance(index_name, InvertedIndexType): + index_name = cast(IndexName, index_name.value) _validate_input( [_ValidateArgument(expected=[str], name="property_name", value=property_name)] ) @@ -663,6 +708,370 @@ def resp(res: Response) -> bool: response_callback=resp, method=self._connection.delete, path=path, - error_msg="Property may not exist", - status_codes=_ExpectedStatusCodes(ok_in=[200], error="property exists"), + error_msg="Property index may not have been deleted.", + status_codes=_ExpectedStatusCodes(ok_in=[200], error="Delete property index"), + ) + + def __check_property_reindex_support(self, feature: str) -> None: + if not self._connection._weaviate_version.is_at_least(1, 39, 0): + raise WeaviateUnsupportedFeatureError( + feature, + str(self._connection._weaviate_version), + "1.39.0", + ) + + def __property_index_path(self, property_name: str, index_name: IndexName) -> str: + return ( + f"/schema/{_capitalize_first_letter(self._name)}" + + f"/properties/{property_name}" + + f"/index/{index_name}" + ) + + def __wait_for_property_index( + self, property_name: str, index_name: IndexName + ) -> executor.Result[InvertedIndexStatus]: + if isinstance(self._connection, ConnectionAsync): + + async def _execute() -> InvertedIndexStatus: + while True: + indexes = await executor.aresult(self.get_property_indexes()) + entry = _find_property_index_status(indexes, property_name, index_name) + done = _terminal_property_index_status(entry, property_name, index_name) + if done is not None: + return done + await asyncio.sleep(1) + + return _execute() + while True: + indexes = executor.result(self.get_property_indexes()) + entry = _find_property_index_status(indexes, property_name, index_name) + done = _terminal_property_index_status(entry, property_name, index_name) + if done is not None: + return done + time.sleep(1) + + @overload + def update_property_index( + self, + property_name: str, + index_name: InvertedIndexType, + *, + tokenization: Optional[Tokenization] = None, + algorithm: Optional[Literal["blockmax"]] = None, + tenants: Union[List[str], str, None] = None, + wait_for_completion: Literal[True], + ) -> executor.Result[InvertedIndexStatus]: ... + + @overload + def update_property_index( + self, + property_name: str, + index_name: InvertedIndexType, + *, + tokenization: Optional[Tokenization] = None, + algorithm: Optional[Literal["blockmax"]] = None, + tenants: Union[List[str], str, None] = None, + wait_for_completion: Literal[False] = False, + ) -> executor.Result[InvertedIndexTask]: ... + + def update_property_index( + self, + property_name: str, + index_name: InvertedIndexType, + *, + tokenization: Optional[Tokenization] = None, + algorithm: Optional[Literal["blockmax"]] = None, + tenants: Union[List[str], str, None] = None, + wait_for_completion: bool = False, + ) -> executor.Result[Union[InvertedIndexTask, InvertedIndexStatus]]: + """Create or migrate a property index in this collection. + + Note: This method is a declarative upsert operation. If the index does not exist, it is created + with the requested configuration. If it exists, it is migrated towards the requested + configuration. If the configuration already matches, no work is submitted and the returned + task reports a `NO_OP` status. The server accepts at most one configuration change per request. + + Caution: changing `tokenization` via the `searchable` index ALSO retokenizes the property's + `filterable` index when one exists. Both indexes are migrated by a single coupled task (their + status entries share one `taskId`), and the retokenization changes how filters match on that + property. To retokenize only the filterable bucket, target the `filterable` index instead. + Cancelling via either index type cancels the whole coupled task. + + Args: + property_name: The property whose index to create or migrate. + index_name: The type of the index, an `InvertedIndexType` value. + tokenization: The tokenization of the index. Required when creating a `searchable` index; + optional as a change on an existing `searchable` or `filterable` index. Not valid for + `rangeFilters`. + algorithm: The search algorithm of a `searchable` index. Only `blockmax` may be requested. + tenants: The tenant/list of tenants for which to create the index. Only valid when + creating a `rangeFilters` index on a multi-tenant collection. If not provided, all + tenants are affected. + wait_for_completion: Whether to wait until the index reports `ready`. By default False. + + Returns: + A `InvertedIndexTask` when `wait_for_completion=False`, or the final `InvertedIndexStatus` + of the index when `wait_for_completion=True`. + + Raises: + weaviate.exceptions.WeaviateInvalidInputError: If the input parameters are invalid. + weaviate.exceptions.WeaviateConnectionError: If the network connection to Weaviate fails. + weaviate.exceptions.UnexpectedStatusCodeError: If Weaviate reports a non-OK status. + weaviate.exceptions.ReindexFailedError: If `wait_for_completion=True` and the reindexing task failed. + weaviate.exceptions.ReindexCanceledError: If `wait_for_completion=True` and the reindexing task was cancelled. + """ + self.__check_property_reindex_support("Collection config update_property_index") + index = cast( + IndexName, + index_name.value if isinstance(index_name, InvertedIndexType) else index_name, + ) + _validate_input( + [_ValidateArgument(expected=[str], name="property_name", value=property_name)] + ) + _validate_input([_ValidateArgument(expected=[str], name="index_name", value=index)]) + _validate_input( + [_ValidateArgument(expected=[str, List[str], None], name="tenants", value=tenants)] + ) + + path = self.__property_index_path(property_name, index) + body: Dict[str, Any] = {} + if tokenization is not None: + body["tokenization"] = ( + tokenization.value if isinstance(tokenization, Tokenization) else tokenization + ) + if algorithm is not None: + body["algorithm"] = algorithm + if isinstance(tenants, str): + tenants = [tenants] + params: Optional[Dict[str, Any]] = ( + {"tenants": ",".join(tenants)} if tenants is not None else None + ) + + def resp(res: Response) -> InvertedIndexTask: + response = _decode_json_response_dict(res, "Update property index") + assert response is not None + return _inverted_index_task_from_json(response) + + if isinstance(self._connection, ConnectionAsync): + + async def _execute() -> Union[InvertedIndexTask, InvertedIndexStatus]: + res = await executor.aresult( + self._connection.put( + path=path, + weaviate_object=body, + params=params, + error_msg="Property index may not have been updated.", + status_codes=_ExpectedStatusCodes( + ok_in=[200, 202], error="Update property index" + ), + ) + ) + task = resp(res) + if wait_for_completion: + return await executor.aresult( + self.__wait_for_property_index(property_name, index) + ) + return task + + return _execute() + res = executor.result( + self._connection.put( + path=path, + weaviate_object=body, + params=params, + error_msg="Property index may not have been updated.", + status_codes=_ExpectedStatusCodes(ok_in=[200, 202], error="Update property index"), + ) + ) + task = resp(res) + if wait_for_completion: + return executor.result(self.__wait_for_property_index(property_name, index)) + return task + + @overload + def rebuild_property_index( + self, + property_name: str, + index_name: InvertedIndexType, + *, + tenants: Union[List[str], str, None] = None, + wait_for_completion: Literal[True], + ) -> executor.Result[InvertedIndexStatus]: ... + + @overload + def rebuild_property_index( + self, + property_name: str, + index_name: InvertedIndexType, + *, + tenants: Union[List[str], str, None] = None, + wait_for_completion: Literal[False] = False, + ) -> executor.Result[InvertedIndexTask]: ... + + def rebuild_property_index( + self, + property_name: str, + index_name: InvertedIndexType, + *, + tenants: Union[List[str], str, None] = None, + wait_for_completion: bool = False, + ) -> executor.Result[Union[InvertedIndexTask, InvertedIndexStatus]]: + """Rebuild an existing property index from scratch with its current configuration. + + Args: + property_name: The property whose index to rebuild. + index_name: The type of the index, an `InvertedIndexType` value. + tenants: The tenant/list of tenants for which to rebuild the index on a multi-tenant + collection. If not provided, all tenants are affected. + wait_for_completion: Whether to wait until the index reports `ready`. By default False. + + Returns: + A `InvertedIndexTask` when `wait_for_completion=False`, or the final `InvertedIndexStatus` + of the index when `wait_for_completion=True`. + + Raises: + weaviate.exceptions.WeaviateInvalidInputError: If the input parameters are invalid. + weaviate.exceptions.WeaviateConnectionError: If the network connection to Weaviate fails. + weaviate.exceptions.UnexpectedStatusCodeError: If Weaviate reports a non-OK status. + weaviate.exceptions.ReindexFailedError: If `wait_for_completion=True` and the reindexing task failed. + weaviate.exceptions.ReindexCanceledError: If `wait_for_completion=True` and the reindexing task was cancelled. + """ + self.__check_property_reindex_support("Collection config rebuild_property_index") + index = cast( + IndexName, + index_name.value if isinstance(index_name, InvertedIndexType) else index_name, + ) + _validate_input( + [_ValidateArgument(expected=[str], name="property_name", value=property_name)] + ) + _validate_input([_ValidateArgument(expected=[str], name="index_name", value=index)]) + _validate_input( + [_ValidateArgument(expected=[str, List[str], None], name="tenants", value=tenants)] + ) + + path = self.__property_index_path(property_name, index) + "/rebuild" + if isinstance(tenants, str): + tenants = [tenants] + params: Optional[Dict[str, Any]] = ( + {"tenants": ",".join(tenants)} if tenants is not None else None + ) + + def resp(res: Response) -> InvertedIndexTask: + response = _decode_json_response_dict(res, "Rebuild property index") + assert response is not None + return _inverted_index_task_from_json(response) + + if isinstance(self._connection, ConnectionAsync): + + async def _execute() -> Union[InvertedIndexTask, InvertedIndexStatus]: + res = await executor.aresult( + self._connection.post( + path=path, + weaviate_object={}, + params=params, + error_msg="Property index may not have been rebuilt.", + status_codes=_ExpectedStatusCodes( + ok_in=[202], error="Rebuild property index" + ), + ) + ) + task = resp(res) + if wait_for_completion: + return await executor.aresult( + self.__wait_for_property_index(property_name, index) + ) + return task + + return _execute() + res = executor.result( + self._connection.post( + path=path, + weaviate_object={}, + params=params, + error_msg="Property index may not have been rebuilt.", + status_codes=_ExpectedStatusCodes(ok_in=[202], error="Rebuild property index"), + ) + ) + task = resp(res) + if wait_for_completion: + return executor.result(self.__wait_for_property_index(property_name, index)) + return task + + def cancel_property_index_task( + self, + property_name: str, + index_name: InvertedIndexType, + ) -> executor.Result[InvertedIndexTask]: + """Cancel the live reindexing task of a property index. + + This operation is idempotent: if there is no live task for the index, the returned task + reports a `NO_OP` status. Note that a coupled tokenization change (a `searchable` change on a + property that also has a `filterable` index) is a single task; cancelling it via either index + type cancels the whole task. + + Args: + property_name: The property whose reindexing task to cancel. + index_name: The type of the index, an `InvertedIndexType` value. + + Returns: + A `InvertedIndexTask` with status `CANCELLED` if a live task was cancelled or `NO_OP` otherwise. + + Raises: + weaviate.exceptions.WeaviateInvalidInputError: If the input parameters are invalid. + weaviate.exceptions.WeaviateConnectionError: If the network connection to Weaviate fails. + weaviate.exceptions.UnexpectedStatusCodeError: If Weaviate reports a non-OK status. + """ + self.__check_property_reindex_support("Collection config cancel_property_index_task") + index = cast( + IndexName, + index_name.value if isinstance(index_name, InvertedIndexType) else index_name, + ) + _validate_input( + [_ValidateArgument(expected=[str], name="property_name", value=property_name)] + ) + _validate_input([_ValidateArgument(expected=[str], name="index_name", value=index)]) + + path = self.__property_index_path(property_name, index) + "/cancel" + + def resp(res: Response) -> InvertedIndexTask: + response = _decode_json_response_dict(res, "Cancel property index task") + assert response is not None + return _inverted_index_task_from_json(response) + + return executor.execute( + response_callback=resp, + method=self._connection.post, + path=path, + weaviate_object={}, + error_msg="Property index task may not have been cancelled.", + status_codes=_ExpectedStatusCodes(ok_in=[202], error="Cancel property index task"), + ) + + def get_property_indexes(self) -> executor.Result[CollectionInvertedIndexes]: + """Get the statuses of the property indexes of this collection. + + The response includes the state of any in-flight reindexing tasks, e.g. their progress and + target configuration. Poll this endpoint for a `ready` status to detect the completion of a + reindexing task. + + Returns: + A `CollectionInvertedIndexes` object containing the index statuses grouped by property. + + Raises: + weaviate.exceptions.WeaviateConnectionError: If the network connection to Weaviate fails. + weaviate.exceptions.UnexpectedStatusCodeError: If Weaviate reports a non-OK status. + """ + self.__check_property_reindex_support("Collection config get_property_indexes") + + def resp(res: Response) -> CollectionInvertedIndexes: + response = _decode_json_response_dict(res, "Get property indexes") + assert response is not None + return _collection_inverted_indexes_from_json(response) + + return executor.execute( + response_callback=resp, + method=self._connection.get, + path=f"/schema/{_capitalize_first_letter(self._name)}/indexes", + error_msg="Property index statuses could not be retrieved.", + status_codes=_ExpectedStatusCodes(ok_in=[200], error="Get property indexes"), ) diff --git a/weaviate/collections/config/sync.pyi b/weaviate/collections/config/sync.pyi index e54d8c8fc..eb2112c9d 100644 --- a/weaviate/collections/config/sync.pyi +++ b/weaviate/collections/config/sync.pyi @@ -5,11 +5,16 @@ from typing_extensions import deprecated from weaviate.collections.classes.config import ( CollectionConfig, CollectionConfigSimple, + CollectionInvertedIndexes, IndexName, + InvertedIndexStatus, + InvertedIndexTask, + InvertedIndexType, Property, ReferenceProperty, ShardStatus, ShardTypes, + Tokenization, _GenerativeProvider, _InvertedIndexConfigUpdate, _MultiTenancyConfigUpdate, @@ -87,4 +92,50 @@ class _ConfigCollection(_ConfigCollectionExecutor[ConnectionSync]): def add_vector( self, *, vector_config: Union[_VectorConfigCreate, List[_VectorConfigCreate]] ) -> None: ... - def delete_property_index(self, property_name: str, index_name: IndexName) -> bool: ... + def delete_property_index( + self, property_name: str, index_name: Union[InvertedIndexType, IndexName] + ) -> bool: ... + @overload + def update_property_index( + self, + property_name: str, + index_name: InvertedIndexType, + *, + tokenization: Optional[Tokenization] = None, + algorithm: Optional[Literal["blockmax"]] = None, + tenants: Union[List[str], str, None] = None, + wait_for_completion: Literal[True], + ) -> InvertedIndexStatus: ... + @overload + def update_property_index( + self, + property_name: str, + index_name: InvertedIndexType, + *, + tokenization: Optional[Tokenization] = None, + algorithm: Optional[Literal["blockmax"]] = None, + tenants: Union[List[str], str, None] = None, + wait_for_completion: Literal[False] = False, + ) -> InvertedIndexTask: ... + @overload + def rebuild_property_index( + self, + property_name: str, + index_name: InvertedIndexType, + *, + tenants: Union[List[str], str, None] = None, + wait_for_completion: Literal[True], + ) -> InvertedIndexStatus: ... + @overload + def rebuild_property_index( + self, + property_name: str, + index_name: InvertedIndexType, + *, + tenants: Union[List[str], str, None] = None, + wait_for_completion: Literal[False] = False, + ) -> InvertedIndexTask: ... + def cancel_property_index_task( + self, property_name: str, index_name: InvertedIndexType + ) -> InvertedIndexTask: ... + def get_property_indexes(self) -> CollectionInvertedIndexes: ... diff --git a/weaviate/exceptions.py b/weaviate/exceptions.py index ce0fe6f7e..a1bcaf6a9 100644 --- a/weaviate/exceptions.py +++ b/weaviate/exceptions.py @@ -149,6 +149,14 @@ class ExportCanceledError(WeaviateBaseError): """Export Canceled Exception.""" +class ReindexFailedError(WeaviateBaseError): + """Reindex Failed Exception.""" + + +class ReindexCanceledError(WeaviateBaseError): + """Reindex Canceled Exception.""" + + class EmptyResponseError(WeaviateBaseError): """Occurs when an HTTP request unexpectedly returns an empty response.""" diff --git a/weaviate/outputs/config.py b/weaviate/outputs/config.py index 17ebebf0e..7f156ce50 100644 --- a/weaviate/outputs/config.py +++ b/weaviate/outputs/config.py @@ -3,15 +3,21 @@ BM25Config, CollectionConfig, CollectionConfigSimple, + CollectionInvertedIndexes, GenerativeConfig, GenerativeSearches, InvertedIndexConfig, + InvertedIndexState, + InvertedIndexStatus, + InvertedIndexTask, + InvertedIndexTaskStatus, MultiTenancyConfig, PQConfig, PQEncoderConfig, PQEncoderDistribution, PQEncoderType, PropertyConfig, + PropertyInvertedIndexes, PropertyType, ReferencePropertyConfig, ReplicationConfig, @@ -35,6 +41,7 @@ "BM25Config", "CollectionConfig", "CollectionConfigSimple", + "CollectionInvertedIndexes", "GenerativeConfig", "GenerativeSearches", "InvertedIndexConfig", @@ -45,6 +52,11 @@ "PQEncoderDistribution", "PQEncoderType", "PropertyConfig", + "PropertyInvertedIndexes", + "InvertedIndexState", + "InvertedIndexStatus", + "InvertedIndexTask", + "InvertedIndexTaskStatus", "PropertyType", "ReferencePropertyConfig", "ReplicationConfig",