diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 2fa0d9b8..ee9e1702 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -16,6 +16,9 @@ jobs: build: name: Build runs-on: ubuntu-latest + env: + QDRANT_SERVERLESS_URL: ${{ secrets.QDRANT_SERVERLESS_URL }} + QDRANT_SERVERLESS_API_KEY: ${{ secrets.QDRANT_SERVERLESS_API_KEY }} steps: - name: Checkout diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 7f4db1e3..28c5cad4 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -107,6 +107,16 @@ Since the API reference is published at , 5. Submit your pull request and get those approvals. +### Qdrant Serverless + +[`QdrantServerlessClient`](https://github.com/qdrant/java-client/blob/master/src/main/java/io/qdrant/client/QdrantServerlessClient.java) uses a copy of the [serverless collections proto](https://github.com/qdrant/qdrant-cloud-public-api/blob/main/proto/qdrant/serverless/collections.proto) in `src/main/proto`. To update it: + +```bash +./gradlew syncServerlessProto +``` + +Its tests run against a live serverless space when `QDRANT_SERVERLESS_URL` and `QDRANT_SERVERLESS_API_KEY` are set, and are skipped otherwise. + ### Releasing a New Version Once the new Qdrant version is live: diff --git a/build.gradle b/build.gradle index b9a20f6f..eefedbd5 100644 --- a/build.gradle +++ b/build.gradle @@ -156,6 +156,44 @@ tasks.register('downloadProtos') { } } +tasks.register('syncServerlessProto') { + def protoFile = file('src/main/proto/serverless_collections.proto') + outputs.file(protoFile) + outputs.upToDateWhen { false } + + doLast { + def url = 'https://raw.githubusercontent.com/qdrant/qdrant-cloud-public-api/main/proto/qdrant/serverless/collections.proto' + String source + try (def httpClient = HttpClients.createDefault()) { + def response = httpClient.execute(new HttpGet(url)) + if (response.getStatusLine().getStatusCode() != 200) { + throw new GradleException("failed to download ${url}: ${response.getStatusLine()}") + } + source = response.getEntity().getContent().getText('UTF-8') + } + + // buf.validate is server-side only and would need an extra proto dependency. Only plain + // single-option lists are stripped. Anything else is left in place and fails the check. + def javaPackage = 'option java_package = "io.qdrant.client.grpc.serverless";' + source = source + .replace('import "buf/validate/validate.proto";\n', '') + .replaceAll(/\s*\[\(buf\.validate\.[^\[\]"',]*\]/, '') + .replace('package qdrant.serverless;\n', "package qdrant.serverless;\n\n${javaPackage}\n") + if (source.contains('buf.validate') || source.contains('buf/validate') || !source.contains(javaPackage)) { + throw new GradleException('unexpected serverless proto, update syncServerlessProto') + } + + def header = '''\ +// Source: https://github.com/qdrant/qdrant-cloud-public-api/blob/main/proto/qdrant/serverless/collections.proto +// Renamed because the protobuf registry already has the regular qdrant "collections.proto". +// buf.validate options are stripped: they are server-side only. +// Regenerate with: ./gradlew syncServerlessProto +''' + protoFile.parentFile.mkdirs() + protoFile.text = header + '\n' + source + } +} + processResources { dependsOn downloadProtos } diff --git a/src/main/java/io/qdrant/client/QdrantClient.java b/src/main/java/io/qdrant/client/QdrantClient.java index 7d59e4dd..5498ce26 100644 --- a/src/main/java/io/qdrant/client/QdrantClient.java +++ b/src/main/java/io/qdrant/client/QdrantClient.java @@ -2857,8 +2857,31 @@ public ListenableFuture countAsync( requestBuilder.setFilter(filter); } + return countAsync(requestBuilder.build(), timeout); + } + + /** + * Count the points in a collection. + * + * @param request The count request. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture countAsync(CountPoints request) { + return countAsync(request, null); + } + + /** + * Count the points in a collection. + * + * @param request The count request. + * @param timeout The timeout for the call. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture countAsync(CountPoints request, @Nullable Duration timeout) { + String collectionName = request.getCollectionName(); + Preconditions.checkArgument(!collectionName.isEmpty(), "Collection name must not be empty"); logger.debug("Count on '{}'", collectionName); - ListenableFuture future = getPoints(timeout).count(requestBuilder.build()); + ListenableFuture future = getPoints(timeout).count(request); addLogFailureCallback(future, "Count"); return Futures.transform( future, response -> response.getResult().getCount(), MoreExecutors.directExecutor()); @@ -2943,9 +2966,34 @@ public ListenableFuture> queryBatchAsync( requestBuilder.setReadConsistency(readConsistency); } + return queryBatchAsync(requestBuilder.build(), timeout); + } + + /** + * Universally query points in batch. Covers all capabilities of search, recommend, discover, + * filters. Also enables hybrid and multi-stage queries. + * + * @param request The batch query request. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture> queryBatchAsync(QueryBatchPoints request) { + return queryBatchAsync(request, null); + } + + /** + * Universally query points in batch. Covers all capabilities of search, recommend, discover, + * filters. Also enables hybrid and multi-stage queries. + * + * @param request The batch query request. + * @param timeout The timeout for the call. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture> queryBatchAsync( + QueryBatchPoints request, @Nullable Duration timeout) { + String collectionName = request.getCollectionName(); + Preconditions.checkArgument(!collectionName.isEmpty(), "Collection name must not be empty"); logger.debug("Query batch on '{}'", collectionName); - ListenableFuture future = - getPoints(timeout).queryBatch(requestBuilder.build()); + ListenableFuture future = getPoints(timeout).queryBatch(request); addLogFailureCallback(future, "Query batch"); return Futures.transform( future, QueryBatchResponse::getResultList, MoreExecutors.directExecutor()); @@ -3261,6 +3309,10 @@ public void close() { } private void addLogFailureCallback(ListenableFuture future, String message) { + addLogFailureCallback(logger, future, message); + } + + static void addLogFailureCallback(Logger logger, ListenableFuture future, String message) { Futures.addCallback( future, new FutureCallback() { diff --git a/src/main/java/io/qdrant/client/QdrantGrpcClient.java b/src/main/java/io/qdrant/client/QdrantGrpcClient.java index 5db34253..4a9882d4 100644 --- a/src/main/java/io/qdrant/client/QdrantGrpcClient.java +++ b/src/main/java/io/qdrant/client/QdrantGrpcClient.java @@ -11,6 +11,8 @@ import io.qdrant.client.grpc.PointsGrpc.PointsFutureStub; import io.qdrant.client.grpc.QdrantGrpc.QdrantFutureStub; import io.qdrant.client.grpc.SnapshotsGrpc.SnapshotsFutureStub; +import io.qdrant.client.grpc.serverless.CollectionsServiceGrpc; +import io.qdrant.client.grpc.serverless.CollectionsServiceGrpc.CollectionsServiceFutureStub; import java.time.Duration; import java.util.Map; import java.util.concurrent.TimeUnit; @@ -183,6 +185,18 @@ public SnapshotsFutureStub snapshots() { timeout != null ? Deadline.after(timeout.toMillis(), TimeUnit.MILLISECONDS) : null); } + /** + * Gets the client for Qdrant Serverless collections + * + * @return a new instance of {@link CollectionsServiceFutureStub} + */ + public CollectionsServiceFutureStub serverlessCollections() { + return CollectionsServiceGrpc.newFutureStub(interceptedChannel) + .withCallCredentials(callCredentials) + .withDeadline( + timeout != null ? Deadline.after(timeout.toMillis(), TimeUnit.MILLISECONDS) : null); + } + @Override public void close() { if (shutdownChannelOnClose && !channel.isShutdown() && !channel.isTerminated()) { diff --git a/src/main/java/io/qdrant/client/QdrantServerlessClient.java b/src/main/java/io/qdrant/client/QdrantServerlessClient.java new file mode 100644 index 00000000..5074cbe4 --- /dev/null +++ b/src/main/java/io/qdrant/client/QdrantServerlessClient.java @@ -0,0 +1,597 @@ +package io.qdrant.client; + +import com.google.common.base.Preconditions; +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.util.concurrent.MoreExecutors; +import io.qdrant.client.grpc.Points.BatchResult; +import io.qdrant.client.grpc.Points.ClearPayloadPoints; +import io.qdrant.client.grpc.Points.CountPoints; +import io.qdrant.client.grpc.Points.DeletePayloadPoints; +import io.qdrant.client.grpc.Points.DeletePointVectors; +import io.qdrant.client.grpc.Points.DeletePoints; +import io.qdrant.client.grpc.Points.GetPoints; +import io.qdrant.client.grpc.Points.PointGroup; +import io.qdrant.client.grpc.Points.QueryBatchPoints; +import io.qdrant.client.grpc.Points.QueryPointGroups; +import io.qdrant.client.grpc.Points.QueryPoints; +import io.qdrant.client.grpc.Points.RetrievedPoint; +import io.qdrant.client.grpc.Points.ScoredPoint; +import io.qdrant.client.grpc.Points.ScrollPoints; +import io.qdrant.client.grpc.Points.ScrollResponse; +import io.qdrant.client.grpc.Points.SetPayloadPoints; +import io.qdrant.client.grpc.Points.UpdateBatchPoints; +import io.qdrant.client.grpc.Points.UpdatePointVectors; +import io.qdrant.client.grpc.Points.UpdateResult; +import io.qdrant.client.grpc.Points.UpsertPoints; +import io.qdrant.client.grpc.serverless.CollectionsServiceGrpc.CollectionsServiceFutureStub; +import io.qdrant.client.grpc.serverless.ServerlessCollections.CollectionConfig; +import io.qdrant.client.grpc.serverless.ServerlessCollections.CreateCollectionRequest; +import io.qdrant.client.grpc.serverless.ServerlessCollections.CreateCollectionResponse; +import io.qdrant.client.grpc.serverless.ServerlessCollections.DeleteCollectionRequest; +import io.qdrant.client.grpc.serverless.ServerlessCollections.DeleteCollectionResponse; +import io.qdrant.client.grpc.serverless.ServerlessCollections.GetCollectionRequest; +import io.qdrant.client.grpc.serverless.ServerlessCollections.GetCollectionResponse; +import io.qdrant.client.grpc.serverless.ServerlessCollections.ListCollectionsRequest; +import io.qdrant.client.grpc.serverless.ServerlessCollections.ListCollectionsResponse; +import java.time.Duration; +import java.util.List; +import java.util.concurrent.TimeUnit; +import javax.annotation.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Client for Qdrant Serverless. Experimental: the API may change without notice. + * + *

Point operations take the same requests as {@link QdrantClient}. + */ +public class QdrantServerlessClient implements AutoCloseable { + private static final Logger logger = LoggerFactory.getLogger(QdrantServerlessClient.class); + + private static final int GRPC_PORT = 443; + + private final QdrantGrpcClient grpcClient; + private final QdrantClient points; + + /** + * Creates a new instance of {@link QdrantServerlessClient}. Build the gRPC client with {@link + * #newGrpcClientBuilder(String)}: serverless has no server version to check compatibility with. + * + * @param grpcClient The low-level gRPC client to use. + */ + public QdrantServerlessClient(QdrantGrpcClient grpcClient) { + this.grpcClient = grpcClient; + this.points = new QdrantClient(grpcClient); + } + + /** + * Creates a builder for a gRPC client connecting to Qdrant Serverless over TLS on port 443, + * without the server version compatibility check. + * + * @param host The host of the serverless space. + * @return a new instance of {@link QdrantGrpcClient.Builder} + */ + public static QdrantGrpcClient.Builder newGrpcClientBuilder(String host) { + return QdrantGrpcClient.newBuilder(host, GRPC_PORT, true, false); + } + + /** + * Gets the low-level gRPC client. + * + * @return The low-level gRPC client + */ + public QdrantGrpcClient grpcClient() { + return grpcClient; + } + + // region Collections + + /** + * Creates a collection. An existing collection fails with {@code ALREADY_EXISTS}, or is reported + * in the response's {@code result}. + * + * @param collectionName The name of the collection. + * @param config The configuration of the collection. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture createCollectionAsync( + String collectionName, CollectionConfig config) { + return createCollectionAsync(collectionName, config, null); + } + + /** + * Creates a collection. An existing collection fails with {@code ALREADY_EXISTS}, or is reported + * in the response's {@code result}. + * + * @param collectionName The name of the collection. + * @param config The configuration of the collection. + * @param timeout The timeout for the call. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture createCollectionAsync( + String collectionName, CollectionConfig config, @Nullable Duration timeout) { + Preconditions.checkArgument(!collectionName.isEmpty(), "Collection name must not be empty"); + logger.debug("Create collection '{}'", collectionName); + ListenableFuture future = + getCollections(timeout) + .createCollection( + CreateCollectionRequest.newBuilder() + .setCollectionName(collectionName) + .setConfig(config) + .build()); + QdrantClient.addLogFailureCallback(logger, future, "Create collection"); + return future; + } + + /** + * Deletes a collection and all of its data. + * + * @param collectionName The name of the collection. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture deleteCollectionAsync(String collectionName) { + return deleteCollectionAsync(collectionName, null); + } + + /** + * Deletes a collection and all of its data. + * + * @param collectionName The name of the collection. + * @param timeout The timeout for the call. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture deleteCollectionAsync( + String collectionName, @Nullable Duration timeout) { + Preconditions.checkArgument(!collectionName.isEmpty(), "Collection name must not be empty"); + logger.debug("Delete collection '{}'", collectionName); + ListenableFuture future = + getCollections(timeout) + .deleteCollection( + DeleteCollectionRequest.newBuilder().setCollectionName(collectionName).build()); + QdrantClient.addLogFailureCallback(logger, future, "Delete collection"); + return future; + } + + /** + * Gets a collection's configuration and point count. A missing collection is reported by {@link + * GetCollectionResponse#getExists()}, not as an error. + * + * @param collectionName The name of the collection. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture getCollectionAsync(String collectionName) { + return getCollectionAsync(collectionName, null); + } + + /** + * Gets a collection's configuration and point count. A missing collection is reported by {@link + * GetCollectionResponse#getExists()}, not as an error. + * + * @param collectionName The name of the collection. + * @param timeout The timeout for the call. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture getCollectionAsync( + String collectionName, @Nullable Duration timeout) { + Preconditions.checkArgument(!collectionName.isEmpty(), "Collection name must not be empty"); + logger.debug("Get collection info for '{}'", collectionName); + ListenableFuture future = + getCollections(timeout) + .getCollection( + GetCollectionRequest.newBuilder().setCollectionName(collectionName).build()); + QdrantClient.addLogFailureCallback(logger, future, "Get collection info"); + return future; + } + + /** + * Checks whether a collection exists. + * + * @param collectionName The name of the collection. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture collectionExistsAsync(String collectionName) { + return collectionExistsAsync(collectionName, null); + } + + /** + * Checks whether a collection exists. + * + * @param collectionName The name of the collection. + * @param timeout The timeout for the call. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture collectionExistsAsync( + String collectionName, @Nullable Duration timeout) { + return Futures.transform( + getCollectionAsync(collectionName, timeout), + GetCollectionResponse::getExists, + MoreExecutors.directExecutor()); + } + + /** + * Lists the first page of collections. + * + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture listCollectionsAsync() { + return listCollectionsAsync(ListCollectionsRequest.getDefaultInstance(), null); + } + + /** + * Lists a page of collections. Pass {@link ListCollectionsResponse#getNextOffsetToken()} as the + * offset token to get the next page. + * + * @param request The list collections request. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture listCollectionsAsync( + ListCollectionsRequest request) { + return listCollectionsAsync(request, null); + } + + /** + * Lists a page of collections. Pass {@link ListCollectionsResponse#getNextOffsetToken()} as the + * offset token to get the next page. + * + * @param request The list collections request. + * @param timeout The timeout for the call. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture listCollectionsAsync( + ListCollectionsRequest request, @Nullable Duration timeout) { + logger.debug("List collections"); + ListenableFuture future = + getCollections(timeout).listCollections(request); + QdrantClient.addLogFailureCallback(logger, future, "List collections"); + return future; + } + + // endregion + + // region Points + + /** + * Queries points. + * + * @param request The query request. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture> queryAsync(QueryPoints request) { + return points.queryAsync(request); + } + + /** + * Queries points. + * + * @param request The query request. + * @param timeout The timeout for the call. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture> queryAsync( + QueryPoints request, @Nullable Duration timeout) { + return points.queryAsync(request, timeout); + } + + /** + * Runs several queries in one request. + * + * @param request The query batch request. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture> queryBatchAsync(QueryBatchPoints request) { + return queryBatchAsync(request, null); + } + + /** + * Runs several queries in one request. + * + * @param request The query batch request. + * @param timeout The timeout for the call. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture> queryBatchAsync( + QueryBatchPoints request, @Nullable Duration timeout) { + return points.queryBatchAsync(request, timeout); + } + + /** + * Queries points grouped by a payload field. + * + * @param request The query groups request. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture> queryGroupsAsync(QueryPointGroups request) { + return points.queryGroupsAsync(request); + } + + /** + * Queries points grouped by a payload field. + * + * @param request The query groups request. + * @param timeout The timeout for the call. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture> queryGroupsAsync( + QueryPointGroups request, @Nullable Duration timeout) { + return points.queryGroupsAsync(request, timeout); + } + + /** + * Retrieves points by IDs. + * + * @param request The get points request. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture> retrieveAsync(GetPoints request) { + return points.retrieveAsync(request, null); + } + + /** + * Retrieves points by IDs. + * + * @param request The get points request. + * @param timeout The timeout for the call. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture> retrieveAsync( + GetPoints request, @Nullable Duration timeout) { + return points.retrieveAsync(request, timeout); + } + + /** + * Gets a page of points. + * + * @param request The scroll request. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture scrollAsync(ScrollPoints request) { + return points.scrollAsync(request); + } + + /** + * Gets a page of points. + * + * @param request The scroll request. + * @param timeout The timeout for the call. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture scrollAsync( + ScrollPoints request, @Nullable Duration timeout) { + return points.scrollAsync(request, timeout); + } + + /** + * Counts points. + * + * @param request The count request. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture countAsync(CountPoints request) { + return countAsync(request, null); + } + + /** + * Counts points. + * + * @param request The count request. + * @param timeout The timeout for the call. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture countAsync(CountPoints request, @Nullable Duration timeout) { + return points.countAsync(request, timeout); + } + + /** + * Inserts or overwrites points. + * + * @param request The upsert request. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture upsertAsync(UpsertPoints request) { + return points.upsertAsync(request); + } + + /** + * Inserts or overwrites points. + * + * @param request The upsert request. + * @param timeout The timeout for the call. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture upsertAsync( + UpsertPoints request, @Nullable Duration timeout) { + return points.upsertAsync(request, timeout); + } + + /** + * Updates vectors of points. + * + * @param request The update vectors request. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture updateVectorsAsync(UpdatePointVectors request) { + return points.updateVectorsAsync(request); + } + + /** + * Updates vectors of points. + * + * @param request The update vectors request. + * @param timeout The timeout for the call. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture updateVectorsAsync( + UpdatePointVectors request, @Nullable Duration timeout) { + return points.updateVectorsAsync(request, timeout); + } + + /** + * Removes vectors from points. + * + * @param request The delete vectors request. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture deleteVectorsAsync(DeletePointVectors request) { + return points.deleteVectorsAsync(request, null); + } + + /** + * Removes vectors from points. + * + * @param request The delete vectors request. + * @param timeout The timeout for the call. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture deleteVectorsAsync( + DeletePointVectors request, @Nullable Duration timeout) { + return points.deleteVectorsAsync(request, timeout); + } + + /** + * Deletes points. + * + * @param request The delete request. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture deleteAsync(DeletePoints request) { + return points.deleteAsync(request); + } + + /** + * Deletes points. + * + * @param request The delete request. + * @param timeout The timeout for the call. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture deleteAsync( + DeletePoints request, @Nullable Duration timeout) { + return points.deleteAsync(request, timeout); + } + + /** + * Sets payload fields of points. + * + * @param request The set payload request. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture setPayloadAsync(SetPayloadPoints request) { + return points.setPayloadAsync(request, null); + } + + /** + * Sets payload fields of points. + * + * @param request The set payload request. + * @param timeout The timeout for the call. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture setPayloadAsync( + SetPayloadPoints request, @Nullable Duration timeout) { + return points.setPayloadAsync(request, timeout); + } + + /** + * Replaces the entire payload of points. + * + * @param request The set payload request. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture overwritePayloadAsync(SetPayloadPoints request) { + return points.overwritePayloadAsync(request, null); + } + + /** + * Replaces the entire payload of points. + * + * @param request The set payload request. + * @param timeout The timeout for the call. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture overwritePayloadAsync( + SetPayloadPoints request, @Nullable Duration timeout) { + return points.overwritePayloadAsync(request, timeout); + } + + /** + * Removes payload fields from points. + * + * @param request The delete payload request. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture deletePayloadAsync(DeletePayloadPoints request) { + return points.deletePayloadAsync(request, null); + } + + /** + * Removes payload fields from points. + * + * @param request The delete payload request. + * @param timeout The timeout for the call. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture deletePayloadAsync( + DeletePayloadPoints request, @Nullable Duration timeout) { + return points.deletePayloadAsync(request, timeout); + } + + /** + * Removes the entire payload of points. + * + * @param request The clear payload request. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture clearPayloadAsync(ClearPayloadPoints request) { + return points.clearPayloadAsync(request, null); + } + + /** + * Removes the entire payload of points. + * + * @param request The clear payload request. + * @param timeout The timeout for the call. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture clearPayloadAsync( + ClearPayloadPoints request, @Nullable Duration timeout) { + return points.clearPayloadAsync(request, timeout); + } + + /** + * Runs several update operations in one request. + * + * @param request The update batch request. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture> batchUpdateAsync(UpdateBatchPoints request) { + return points.batchUpdateAsync(request, null); + } + + /** + * Runs several update operations in one request. + * + * @param request The update batch request. + * @param timeout The timeout for the call. + * @return a new instance of {@link ListenableFuture} + */ + public ListenableFuture> batchUpdateAsync( + UpdateBatchPoints request, @Nullable Duration timeout) { + return points.batchUpdateAsync(request, timeout); + } + + // endregion + + @Override + public void close() { + grpcClient.close(); + } + + private CollectionsServiceFutureStub getCollections(@Nullable Duration timeout) { + return timeout != null + ? grpcClient + .serverlessCollections() + .withDeadlineAfter(timeout.toMillis(), TimeUnit.MILLISECONDS) + : grpcClient.serverlessCollections(); + } +} diff --git a/src/main/java/io/qdrant/client/ServerlessPayloadIndexFactory.java b/src/main/java/io/qdrant/client/ServerlessPayloadIndexFactory.java new file mode 100644 index 00000000..790fbfe4 --- /dev/null +++ b/src/main/java/io/qdrant/client/ServerlessPayloadIndexFactory.java @@ -0,0 +1,145 @@ +package io.qdrant.client; + +import io.qdrant.client.grpc.serverless.ServerlessCollections.BoolIndex; +import io.qdrant.client.grpc.serverless.ServerlessCollections.DatetimeIndex; +import io.qdrant.client.grpc.serverless.ServerlessCollections.DisabledStemmer; +import io.qdrant.client.grpc.serverless.ServerlessCollections.FloatIndex; +import io.qdrant.client.grpc.serverless.ServerlessCollections.GeoIndex; +import io.qdrant.client.grpc.serverless.ServerlessCollections.IntegerIndex; +import io.qdrant.client.grpc.serverless.ServerlessCollections.KeywordIndex; +import io.qdrant.client.grpc.serverless.ServerlessCollections.PayloadIndexConfig; +import io.qdrant.client.grpc.serverless.ServerlessCollections.SnowballParams; +import io.qdrant.client.grpc.serverless.ServerlessCollections.StemmingAlgorithm; +import io.qdrant.client.grpc.serverless.ServerlessCollections.TextIndex; +import io.qdrant.client.grpc.serverless.ServerlessCollections.UuidIndex; + +/** + * Convenience methods for constructing Qdrant Serverless {@link PayloadIndexConfig} and {@link + * StemmingAlgorithm}. + */ +public final class ServerlessPayloadIndexFactory { + private ServerlessPayloadIndexFactory() {} + + /** + * Creates a keyword index with default options. + * + * @return a new instance of {@link PayloadIndexConfig} + */ + public static PayloadIndexConfig keyword() { + return keyword(KeywordIndex.getDefaultInstance()); + } + + /** + * Creates a keyword index. + * + * @param index The keyword index options. + * @return a new instance of {@link PayloadIndexConfig} + */ + public static PayloadIndexConfig keyword(KeywordIndex index) { + return PayloadIndexConfig.newBuilder().setKeyword(index).build(); + } + + /** + * Creates an integer index with default options. + * + * @return a new instance of {@link PayloadIndexConfig} + */ + public static PayloadIndexConfig integer() { + return integer(IntegerIndex.getDefaultInstance()); + } + + /** + * Creates an integer index. + * + * @param index The integer index options. + * @return a new instance of {@link PayloadIndexConfig} + */ + public static PayloadIndexConfig integer(IntegerIndex index) { + return PayloadIndexConfig.newBuilder().setInteger(index).build(); + } + + /** + * Creates a float index. + * + * @return a new instance of {@link PayloadIndexConfig} + */ + public static PayloadIndexConfig floatIndex() { + return PayloadIndexConfig.newBuilder().setFloat(FloatIndex.getDefaultInstance()).build(); + } + + /** + * Creates a UUID index. + * + * @return a new instance of {@link PayloadIndexConfig} + */ + public static PayloadIndexConfig uuid() { + return PayloadIndexConfig.newBuilder().setUuid(UuidIndex.getDefaultInstance()).build(); + } + + /** + * Creates a datetime index. + * + * @return a new instance of {@link PayloadIndexConfig} + */ + public static PayloadIndexConfig datetime() { + return PayloadIndexConfig.newBuilder().setDatetime(DatetimeIndex.getDefaultInstance()).build(); + } + + /** + * Creates a full-text index with default options. + * + * @return a new instance of {@link PayloadIndexConfig} + */ + public static PayloadIndexConfig text() { + return text(TextIndex.getDefaultInstance()); + } + + /** + * Creates a full-text index. + * + * @param index The full-text index options. + * @return a new instance of {@link PayloadIndexConfig} + */ + public static PayloadIndexConfig text(TextIndex index) { + return PayloadIndexConfig.newBuilder().setText(index).build(); + } + + /** + * Creates a geo index. + * + * @return a new instance of {@link PayloadIndexConfig} + */ + public static PayloadIndexConfig geo() { + return PayloadIndexConfig.newBuilder().setGeo(GeoIndex.getDefaultInstance()).build(); + } + + /** + * Creates a bool index. + * + * @return a new instance of {@link PayloadIndexConfig} + */ + public static PayloadIndexConfig bool() { + return PayloadIndexConfig.newBuilder().setBool(BoolIndex.getDefaultInstance()).build(); + } + + /** + * Creates a Snowball stemmer for a full-text index. + * + * @param language The language whose words should be stemmed, e.g. "english". + * @return a new instance of {@link StemmingAlgorithm} + */ + public static StemmingAlgorithm snowballStemmer(String language) { + return StemmingAlgorithm.newBuilder() + .setSnowball(SnowballParams.newBuilder().setLanguage(language).build()) + .build(); + } + + /** + * Explicitly disables stemming for a full-text index. + * + * @return a new instance of {@link StemmingAlgorithm} + */ + public static StemmingAlgorithm disabledStemmer() { + return StemmingAlgorithm.newBuilder().setDisabled(DisabledStemmer.getDefaultInstance()).build(); + } +} diff --git a/src/main/proto/serverless_collections.proto b/src/main/proto/serverless_collections.proto new file mode 100644 index 00000000..d7910e46 --- /dev/null +++ b/src/main/proto/serverless_collections.proto @@ -0,0 +1,296 @@ +// Source: https://github.com/qdrant/qdrant-cloud-public-api/blob/main/proto/qdrant/serverless/collections.proto +// Renamed because the protobuf registry already has the regular qdrant "collections.proto". +// buf.validate options are stripped: they are server-side only. +// Regenerate with: ./gradlew syncServerlessProto + +syntax = "proto3"; + +package qdrant.serverless; + +option java_package = "io.qdrant.client.grpc.serverless"; + + +// CollectionsService manages the collections of a qdrant serverless space. +// Unlike the qdrant server API, it exposes only a simplified configuration: +// the manager turns it into concrete qdrant settings, which are never +// exposed to the tenant. +service CollectionsService { + // Creates a collection with the given tenant-facing configuration. + rpc CreateCollection(CreateCollectionRequest) returns (CreateCollectionResponse); + // Deletes a collection and all of its data. + rpc DeleteCollection(DeleteCollectionRequest) returns (DeleteCollectionResponse); + // Returns a single collection's configuration and stats. + rpc GetCollection(GetCollectionRequest) returns (GetCollectionResponse); + // Lists the caller's collections. + rpc ListCollections(ListCollectionsRequest) returns (ListCollectionsResponse); +} + +// Distance metric used to compare dense vectors. +enum Distance { + // Unset; a concrete metric is required. + DISTANCE_UNSPECIFIED = 0; + // Cosine similarity. + COSINE = 1; + // Euclidean (L2) distance. + EUCLID = 2; + // Dot product. + DOT = 3; + // Manhattan (L1) distance. + MANHATTAN = 4; +} + +// How much of the original vector precision may be traded for cost. The manager +// turns this into a concrete quantization / datatype choice; the tenant never +// picks scalar/product/binary quantization or its parameters directly. +enum PrecisionTier { + // Unset: HIGH. + PRECISION_TIER_UNSPECIFIED = 0; + // Aggressive compression, lowest cost, approximate results. + LOW = 1; + // Moderate compression with a small accuracy trade-off. + MEDIUM = 2; + // Light compression, near-exact results. + HIGH = 3; + // No lossy compression: exact vectors, fully in memory. Highest cost. + FULL = 4; +} + +// Configuration of a single dense (embedding) vector. +message DenseVectorConfig { + // Dimensionality of the embedding, e.g. 512 (CLIP), 1536, 3072. + uint64 size = 1; + // Distance metric used to compare vectors. + Distance distance = 2; + // Store several sub-vectors per point and compare with max-sim, for + // late-interaction models (ColBERT, ColPali, ...). + bool multivector = 3; + // Precision/cost trade-off for this vector. Unset: HIGH. + optional PrecisionTier precision_tier = 4; +} + +// Configuration of a single sparse vector. +message SparseVectorConfig { + // Apply the IDF modifier at query time. Enable for BM25-style models that + // expect inverse-document-frequency weighting. + bool use_idf = 1; + // Precision/cost trade-off for this vector. Unset: HIGH. + optional PrecisionTier precision_tier = 2; +} + +// Full-text tokenizer, mirrors qdrant's `TokenizerType`. +enum Tokenizer { + // Unset: WHITESPACE. + TOKENIZER_UNSPECIFIED = 0; + // Index every prefix of each token. + PREFIX = 1; + // Split on whitespace. + WHITESPACE = 2; + // Split on word boundaries. + WORD = 3; + // Language-aware tokenization. + MULTILINGUAL = 4; +} + +// Exact match on string values, e.g. `color: "red"`. +message KeywordIndex { + // If set, enable prefix matching (`match: { "prefix": ... }`) on this field. + // Presence of this message enables prefix matching; it has no options yet. + optional KeywordPrefixParams prefix = 1; +} + +// Prefix matching options for the keyword index. Has no options yet: +// presence of this message enables prefix matching. +message KeywordPrefixParams {} + +// Exact match and/or range filters on integers, e.g. `age: 25`. Both are on +// by default; turning one off shrinks the index. +message IntegerIndex { + // Support exact-match filters. + optional bool lookup = 1; + // Support range filters. + optional bool range = 2; +} + +// Range filters on floating point (and integer) numbers, e.g. `price: 99.5`. +message FloatIndex {} + +// Exact match on UUID strings; like keyword but stored compactly. +message UuidIndex {} + +// Range filters on RFC 3339 datetimes, e.g. `created_at: "2023-02-08T10:49:00Z"`. +message DatetimeIndex {} + +// Tokens ignored by a full-text index. Language names match qdrant (e.g. +// "english"); predefined lists and custom tokens are merged. +message StopwordsSet { + // Languages whose predefined stopword lists to apply. + repeated string languages = 1; + // Extra stopwords to ignore, merged with the language lists. + repeated string custom = 2; +} + +// Snowball stemming for a full-text index. +message SnowballParams { + // Language for the snowball algorithm, e.g. "english". + string language = 1; +} + +// Explicitly disable stemming (overrides any language default). +message DisabledStemmer {} + +// Stemming algorithm for a full-text index. Unset: no stemming. +message StemmingAlgorithm { + // Which stemming algorithm to use. + oneof stemming_params { + // Snowball stemmer for the given language. + SnowballParams snowball = 1; + // Explicitly disable stemming. + DisabledStemmer disabled = 2; + } +} + +// Full-text filtering on string values. +message TextIndex { + // Tokenizer to split text with. Unset: WHITESPACE. + optional Tokenizer tokenizer = 1; + // Lowercase text before indexing. Default true. + optional bool lowercase = 2; + // Support phrase queries; extra index structure. Default true. + optional bool phrase_matching = 3; + // Minimum token length to index. + optional uint64 min_token_len = 4; + // Maximum token length to index. + optional uint64 max_token_len = 5; + // Fold accented characters to ASCII. Default false. + optional bool ascii_folding = 6; + // Tokens to ignore at index and query time. + optional StopwordsSet stopwords = 7; + // Stemming algorithm. Unset: engine default (no stemming). + optional StemmingAlgorithm stemmer = 8; +} + +// Geo radius / bounding box / polygon filters on `{lon, lat}` values. +message GeoIndex {} + +// Exact match on booleans, e.g. `is_active: true`. +message BoolIndex {} + +// One payload index. Only the *kind* of filter the field supports is chosen +// here; storage placement of the index is the manager's decision. +message PayloadIndexConfig { + // The kind of filter the field supports. + oneof index { + // Exact match on strings. + KeywordIndex keyword = 1; + // Exact match and/or range filters on integers. + IntegerIndex integer = 2; + // Range filters on numbers. + FloatIndex float = 3; + // Exact match on UUIDs. + UuidIndex uuid = 4; + // Range filters on datetimes. + DatetimeIndex datetime = 5; + // Full-text filtering. + TextIndex text = 6; + // Geo filters. + GeoIndex geo = 7; + // Exact match on booleans. + BoolIndex bool = 8; + } +} + +// The tenant's collection config. Persisted verbatim. At least one dense or +// sparse vector is required. +message CollectionConfig { + // Keyed by vector name. The empty name "" is the unnamed default vector, as + // in qdrant; a collection either has the single unnamed vector or named ones. + map dense_vectors = 1; + // Keyed by vector name. + map sparse_vectors = 2; + // Keyed by payload field name (JSON path, e.g. `user_id` or `meta.tags`). + map payload_indexes = 3; +} + +// Every request names the collection by its tenant-facing name only. The +// tenant (`x-account-id`, `x-space-id`) travels in gRPC metadata, injected by +// auth. The storage id is the manager's: minted on create, resolved +// internally on get/delete. +message CreateCollectionRequest { + // Tenant-facing name of the collection to create. + string collection_name = 1; + // The collection's configuration. + CollectionConfig config = 2; +} + +// Result of a create. +message CreateCollectionResponse { + // Tenant-facing name of the collection. + string collection_name = 1; + // Outcome, e.g. "created", "already exists". + string result = 2; + // Time spent to process + double time = 3; +} + +// Names the collection to delete. +message DeleteCollectionRequest { + // Tenant-facing name of the collection to delete. + string collection_name = 1; +} + +// Result of a delete. +message DeleteCollectionResponse { + // Whether the collection existed and was deleted. + bool deleted = 1; + // Time spent to process + double time = 2; +} + +// Names the collection to fetch. +message GetCollectionRequest { + // Tenant-facing name of the collection. + string collection_name = 1; +} + +// The collection's configuration and stats, if it exists. +message GetCollectionResponse { + // Whether the collection exists. + bool exists = 1; + // The configuration the collection was created with. + optional CollectionConfig config = 2; + // Available points as of the last applied write (eventually consistent); + // absent until the updater has written stats for the collection. + optional uint64 point_count = 3; + // Time spent to process + double time = 4; +} + +// Lists the caller's collections. The tenant travels in metadata. +message ListCollectionsRequest { + // Maximum number of collections to return. Defaults to 20 and must not + // exceed 100. + optional uint32 limit = 1; + // Opaque token returned as `next_offset_token` by the previous page. Clients + // must not interpret this value. + optional string offset_token = 2; +} + +// One collection in a listing. +message CollectionSummary { + // Tenant-facing name of the collection. + string collection_name = 1; + // Available points as of the last applied write (eventually consistent); + // absent until the updater has written stats for the collection. + optional uint64 point_count = 2; +} + +// The caller's collections. +message ListCollectionsResponse { + // Collections in this page. + repeated CollectionSummary collections = 1; + // Opaque token to pass as `offset_token` to retrieve the next page. Absent + // when there are no more results. + optional string next_offset_token = 2; + // Time spent to process + double time = 3; +} diff --git a/src/test/java/io/qdrant/client/QdrantServerlessClientTest.java b/src/test/java/io/qdrant/client/QdrantServerlessClientTest.java new file mode 100644 index 00000000..090c231b --- /dev/null +++ b/src/test/java/io/qdrant/client/QdrantServerlessClientTest.java @@ -0,0 +1,267 @@ +package io.qdrant.client; + +import static io.qdrant.client.ConditionFactory.matchKeyword; +import static io.qdrant.client.PointIdFactory.id; +import static io.qdrant.client.QueryFactory.nearest; +import static io.qdrant.client.ValueFactory.value; +import static io.qdrant.client.VectorsFactory.vectors; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +import io.grpc.Status; +import io.qdrant.client.grpc.Common.Filter; +import io.qdrant.client.grpc.Points.CountPoints; +import io.qdrant.client.grpc.Points.DeletePoints; +import io.qdrant.client.grpc.Points.GetPoints; +import io.qdrant.client.grpc.Points.PointStruct; +import io.qdrant.client.grpc.Points.PointsIdsList; +import io.qdrant.client.grpc.Points.PointsSelector; +import io.qdrant.client.grpc.Points.QueryPoints; +import io.qdrant.client.grpc.Points.RetrievedPoint; +import io.qdrant.client.grpc.Points.ScoredPoint; +import io.qdrant.client.grpc.Points.SetPayloadPoints; +import io.qdrant.client.grpc.Points.UpdateResult; +import io.qdrant.client.grpc.Points.UpdateStatus; +import io.qdrant.client.grpc.Points.UpsertPoints; +import io.qdrant.client.grpc.serverless.ServerlessCollections.CollectionConfig; +import io.qdrant.client.grpc.serverless.ServerlessCollections.CollectionSummary; +import io.qdrant.client.grpc.serverless.ServerlessCollections.DenseVectorConfig; +import io.qdrant.client.grpc.serverless.ServerlessCollections.Distance; +import io.qdrant.client.grpc.serverless.ServerlessCollections.GetCollectionResponse; +import io.qdrant.client.grpc.serverless.ServerlessCollections.IntegerIndex; +import io.qdrant.client.grpc.serverless.ServerlessCollections.ListCollectionsRequest; +import io.qdrant.client.grpc.serverless.ServerlessCollections.ListCollectionsResponse; +import io.qdrant.client.grpc.serverless.ServerlessCollections.PrecisionTier; +import io.qdrant.client.grpc.serverless.ServerlessCollections.SparseVectorConfig; +import io.qdrant.client.grpc.serverless.ServerlessCollections.StopwordsSet; +import io.qdrant.client.grpc.serverless.ServerlessCollections.TextIndex; +import io.qdrant.client.grpc.serverless.ServerlessCollections.Tokenizer; +import java.net.URI; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutionException; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +/** Runs against the serverless space at QDRANT_SERVERLESS_URL. Skipped if it is not configured. */ +class QdrantServerlessClientTest { + private QdrantServerlessClient client; + private String collectionName; + + @BeforeEach + void setup() { + String url = System.getenv("QDRANT_SERVERLESS_URL"); + String apiKey = System.getenv("QDRANT_SERVERLESS_API_KEY"); + // CI sets unavailable secrets to empty strings, e.g. on pull requests from forks. + assumeTrue( + url != null && !url.isEmpty() && apiKey != null && !apiKey.isEmpty(), + "QDRANT_SERVERLESS_URL and QDRANT_SERVERLESS_API_KEY are not set"); + client = + new QdrantServerlessClient( + QdrantServerlessClient.newGrpcClientBuilder(URI.create(url).getHost()) + .withApiKey(apiKey) + .withTimeout(Duration.ofSeconds(30)) + .build()); + collectionName = "java-client-test-" + System.nanoTime(); + } + + @AfterEach + void teardown() throws Exception { + if (client != null) { + try { + client.deleteCollectionAsync(collectionName).get(); + } finally { + client.close(); + } + } + } + + @Test + void collections() throws Exception { + CollectionConfig config = + CollectionConfig.newBuilder() + .putDenseVectors( + "", + DenseVectorConfig.newBuilder() + .setSize(4) + .setDistance(Distance.COSINE) + .setPrecisionTier(PrecisionTier.FULL) + .build()) + .putSparseVectors("bm25", SparseVectorConfig.newBuilder().setUseIdf(true).build()) + .putPayloadIndexes("color", ServerlessPayloadIndexFactory.keyword()) + .putPayloadIndexes( + "age", + ServerlessPayloadIndexFactory.integer( + IntegerIndex.newBuilder().setRange(false).build())) + .putPayloadIndexes("price", ServerlessPayloadIndexFactory.floatIndex()) + .putPayloadIndexes("user_id", ServerlessPayloadIndexFactory.uuid()) + .putPayloadIndexes("created", ServerlessPayloadIndexFactory.datetime()) + .putPayloadIndexes("geo", ServerlessPayloadIndexFactory.geo()) + .putPayloadIndexes("active", ServerlessPayloadIndexFactory.bool()) + .putPayloadIndexes( + "title", + ServerlessPayloadIndexFactory.text( + TextIndex.newBuilder() + .setTokenizer(Tokenizer.WORD) + .setStopwords(StopwordsSet.newBuilder().addLanguages("english")) + .setStemmer(ServerlessPayloadIndexFactory.snowballStemmer("english")) + .build())) + .putPayloadIndexes( + "body", + ServerlessPayloadIndexFactory.text( + TextIndex.newBuilder() + .setStemmer(ServerlessPayloadIndexFactory.disabledStemmer()) + .build())) + .build(); + + assertEquals("created", client.createCollectionAsync(collectionName, config).get().getResult()); + + // The API may report an existing collection as an error or through the result. + try { + String result = client.createCollectionAsync(collectionName, config).get().getResult(); + assertNotEquals("created", result); + } catch (ExecutionException e) { + assertEquals(Status.Code.ALREADY_EXISTS, Status.fromThrowable(e.getCause()).getCode()); + } + + GetCollectionResponse info = client.getCollectionAsync(collectionName).get(); + assertTrue(info.getExists()); + assertEquals(config, info.getConfig()); + assertTrue(client.collectionExistsAsync(collectionName).get()); + assertFalse(client.getCollectionAsync(collectionName + "-missing").get().getExists()); + + List names = new ArrayList<>(); + ListCollectionsRequest.Builder request = ListCollectionsRequest.newBuilder().setLimit(100); + ListCollectionsResponse page; + do { + page = client.listCollectionsAsync(request.build()).get(); + for (CollectionSummary summary : page.getCollectionsList()) { + names.add(summary.getCollectionName()); + } + request.setOffsetToken(page.getNextOffsetToken()); + } while (page.hasNextOffsetToken()); + assertTrue(names.contains(collectionName)); + } + + @Test + void points() throws Exception { + client + .createCollectionAsync( + collectionName, + CollectionConfig.newBuilder() + .putDenseVectors( + "", DenseVectorConfig.newBuilder().setSize(4).setDistance(Distance.DOT).build()) + .putPayloadIndexes("color", ServerlessPayloadIndexFactory.keyword()) + .build()) + .get(); + + client + .upsertAsync( + UpsertPoints.newBuilder() + .setCollectionName(collectionName) + .setWait(true) + .addPoints( + PointStruct.newBuilder() + .setId(id(1)) + .setVectors(vectors(1, 0, 0, 0)) + .putPayload("color", value("red"))) + .addPoints( + PointStruct.newBuilder() + .setId(id(2)) + .setVectors(vectors(0, 1, 0, 0)) + .putPayload("color", value("blue"))) + .build()) + .get(); + + // Reads can lag a few seconds behind writes. + eventually( + () -> + client + .countAsync( + CountPoints.newBuilder() + .setCollectionName(collectionName) + .setExact(true) + .setFilter(Filter.newBuilder().addMust(matchKeyword("color", "red"))) + .build()) + .get() + == 1); + + eventually( + () -> { + List hits = + client + .queryAsync( + QueryPoints.newBuilder() + .setCollectionName(collectionName) + .setQuery(nearest(1, 0, 0, 0)) + .build()) + .get(); + return hits.size() == 2 && hits.get(0).getId().getNum() == 1; + }); + + PointsSelector first = + PointsSelector.newBuilder().setPoints(PointsIdsList.newBuilder().addIds(id(1))).build(); + UpdateResult setPayload = + client + .setPayloadAsync( + SetPayloadPoints.newBuilder() + .setCollectionName(collectionName) + .setWait(true) + .setPointsSelector(first) + .putPayload("size", value(10)) + .build()) + .get(); + assertEquals(UpdateStatus.Completed, setPayload.getStatus()); + + List points = + client + .retrieveAsync( + GetPoints.newBuilder() + .setCollectionName(collectionName) + .addIds(id(1)) + .setWithPayload(WithPayloadSelectorFactory.enable(true)) + .build()) + .get(); + assertEquals(1, points.size()); + + UpdateResult delete = + client + .deleteAsync( + DeletePoints.newBuilder() + .setCollectionName(collectionName) + .setWait(true) + .setPoints( + PointsSelector.newBuilder() + .setPoints(PointsIdsList.newBuilder().addIds(id(2)))) + .build()) + .get(); + assertEquals(UpdateStatus.Completed, delete.getStatus()); + } + + // Retries on false and on errors, e.g. while a new collection is not readable yet. + private static void eventually(Callable condition) throws Exception { + long deadline = System.nanoTime() + Duration.ofMinutes(1).toNanos(); + while (true) { + try { + if (condition.call()) { + return; + } + if (System.nanoTime() >= deadline) { + fail("condition not met within 1 minute"); + } + } catch (Exception e) { + if (System.nanoTime() >= deadline) { + throw e; + } + } + Thread.sleep(250); + } + } +}