feat: stream() and list() objects in collection

CursorSpliterator powers 2 patterns for iterating over objects:
- list() returns an Iterable that can be used in a for-loop
- stream() presents the internal Iterator via a familiar Stream API

Author: bevzzz

src/it/java/io/weaviate/integration/PaginationITest.java

@@ -0,0 +1,63 @@
+package io.weaviate.integration;
+
+import static org.junit.Assume.assumeTrue;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+
+import org.assertj.core.api.Assertions;
+import org.junit.Test;
+
+import io.weaviate.ConcurrentTest;
+import io.weaviate.client6.v1.api.WeaviateClient;
+import io.weaviate.client6.v1.api.collections.WeaviateMetadata;
+import io.weaviate.client6.v1.api.collections.WeaviateObject;
+import io.weaviate.containers.Container;
+
+public class PaginationITest extends ConcurrentTest {
+  private static WeaviateClient client = Container.WEAVIATE.getClient();
+
+  @Test
+  public void test_stream() throws IOException {
+    // Arrange
+    var nsThings = ns("Things");
+    var count = 10;
+
+    client.collections.create(nsThings);
+    var things = client.collections.use(nsThings);
+
+    var inserted = new ArrayList<String>();
+    for (var i = 0; i < count; i++) {
+      var object = things.data.insert(Collections.emptyMap());
+      inserted.add(object.metadata().uuid());
+    }
+    assumeTrue("all objects were inserted", inserted.size() == count);
+
+    // Act: stream
+    var gotStream = things.stream()
+        .map(WeaviateObject::metadata).map(WeaviateMetadata::uuid).toList();
+
+    // Assert
+    Assertions.assertThat(gotStream)
+        .as("stream fetched all objects")
+        .hasSize(inserted.size())
+        .containsAll(inserted);
+
+    // Act: list
+    var gotList = new ArrayList<String>();
+    for (var object : things.list()) {
+      gotList.add(object.metadata().uuid());
+    }
+
+    // Assert
+    Assertions.assertThat(gotList)
+        .as("list fetched all objects")
+        .hasSize(inserted.size())
+        .containsAll(inserted);
+
+    Assertions.assertThat(gotStream)
+        .as("stream and list return consistent order")
+        .containsExactlyElementsOf(gotList);
+  }
+}

src/main/java/io/weaviate/client6/v1/api/collections/CollectionHandle.java

@@ -1,8 +1,14 @@
 package io.weaviate.client6.v1.api.collections;
 
+import java.util.Spliterator;
+import java.util.Spliterators;
+import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
+
 import io.weaviate.client6.v1.api.collections.aggregate.WeaviateAggregateClient;
 import io.weaviate.client6.v1.api.collections.config.WeaviateConfigClient;
 import io.weaviate.client6.v1.api.collections.data.WeaviateDataClient;
+import io.weaviate.client6.v1.api.collections.query.QueryMetadata;
 import io.weaviate.client6.v1.api.collections.query.WeaviateQueryClient;
 import io.weaviate.client6.v1.internal.grpc.GrpcTransport;
 import io.weaviate.client6.v1.internal.orm.CollectionDescriptor;
@@ -24,4 +30,18 @@ public CollectionHandle(
     this.data = new WeaviateDataClient<>(collectionDescriptor, restTransport, this.query);
     this.aggregate = new WeaviateAggregateClient(collectionDescriptor, grpcTransport);
   }
+
+  public Stream<WeaviateObject<T, Object, QueryMetadata>> stream() {
+    return StreamSupport.stream(spliterator(2), false);
+  }
+
+  public Iterable<WeaviateObject<T, Object, QueryMetadata>> list() {
+    return () -> Spliterators.iterator(spliterator(2));
+  }
+
+  private Spliterator<WeaviateObject<T, Object, QueryMetadata>> spliterator(int batchSize) {
+    return new CursorSpliterator<>(batchSize,
+        (after, limit) -> this.query.fetchObjects(
+            query -> query.after(after).limit(limit)).objects());
+  }
 }

src/main/java/io/weaviate/client6/v1/api/collections/CursorSpliterator.java

@@ -0,0 +1,62 @@
+package io.weaviate.client6.v1.api.collections;
+
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Spliterator;
+import java.util.function.BiFunction;
+import java.util.function.Consumer;
+
+import io.weaviate.client6.v1.api.collections.query.QueryMetadata;
+
+class CursorSpliterator<T> implements Spliterator<WeaviateObject<T, Object, QueryMetadata>> {
+  private final int batchSize;
+  private final BiFunction<String, Integer, List<WeaviateObject<T, Object, QueryMetadata>>> fetch;
+
+  // Spliterators do not promise thread-safety, so there's no mechanism
+  // to protect access to its internal state.
+  private String cursor;
+  private Iterator<WeaviateObject<T, Object, QueryMetadata>> currentPage = Collections.emptyIterator();
+
+  public CursorSpliterator(int batchSize,
+      BiFunction<String, Integer, List<WeaviateObject<T, Object, QueryMetadata>>> fetch) {
+    this.batchSize = batchSize;
+    this.fetch = fetch;
+  }
+
+  @Override
+  public boolean tryAdvance(Consumer<? super WeaviateObject<T, Object, QueryMetadata>> action) {
+    // Happy path: there are remaining objects in the current page.
+    if (currentPage.hasNext()) {
+      action.accept(currentPage.next());
+      return true;
+    }
+
+    // It's OK for the cursor to be null, because it's String (object).
+    var nextPage = fetch.apply(cursor, batchSize);
+    if (nextPage.isEmpty()) {
+      return false;
+    }
+    cursor = nextPage.get(nextPage.size() - 1).metadata().uuid();
+    currentPage = nextPage.iterator();
+    return tryAdvance(action);
+  }
+
+  @Override
+  public Spliterator<WeaviateObject<T, Object, QueryMetadata>> trySplit() {
+    // Do not support splitting just now;
+    return null;
+  }
+
+  @Override
+  public long estimateSize() {
+    // CursorSpliterator does not have SIZED characteristic, so this is our
+    // best-effort estimate. The number of objects in the db is unbounded.
+    return Long.MAX_VALUE;
+  }
+
+  @Override
+  public int characteristics() {
+    return ORDERED | NONNULL;
+  }
+}