Merge pull request #121 from SpineEventEngine/aggregate-storage-unique-index

Return unique aggregate IDs from the `DsAggregateStorage.index()`

Author: armiol

datastore/src/main/java/io/spine/server/storage/datastore/DatastoreWrapper.java

@@ -255,27 +255,29 @@ public Entity read(Key key) {
      * <p>The Datastore may return a partial result set, so an execution of this method may
      * result in several Datastore queries.
      *
-     * <p>The limit included in the {@link StructuredQuery}, will be a maximum count of entities
+     * <p>The limit included in the {@link StructuredQuery}, will be a maximum count of objects
      * in the returned iterator.
      *
      * <p>The returned {@link DsQueryIterator} allows to {@linkplain DsQueryIterator#nextPageQuery()
-     * create a query} to the next page of entities reusing an existing cursor.
+     * create a query} to the next page of results reusing an existing cursor.
      *
      * <p>The resulting {@code Iterator} is evaluated lazily. A call to
      * {@link Iterator#remove() Iterator.remove()} causes an {@link UnsupportedOperationException}.
      *
      * @param query
      *         {@link Query} to execute upon the Datastore
+     * @param <R>
+     *         the type of queried objects
      * @return results fo the query as a lazily evaluated {@link Iterator}
      * @see DatastoreReader#run(Query)
      */
-    public DsQueryIterator read(StructuredQuery<Entity> query) {
+    public <R> DsQueryIterator<R> read(StructuredQuery<R> query) {
         Namespace namespace = currentNamespace();
-        StructuredQuery<Entity> queryWithNamespace =
+        StructuredQuery<R> queryWithNamespace =
                 query.toBuilder()
                      .setNamespace(namespace.getValue())
                      .build();
-        DsQueryIterator result = new DsQueryIterator(queryWithNamespace, actor);
+        DsQueryIterator<R> result = new DsQueryIterator<>(queryWithNamespace, actor);
         return result;
     }
 
@@ -292,11 +294,13 @@ public DsQueryIterator read(StructuredQuery<Entity> query) {
      *         {@link Query} to execute upon the Datastore
      * @param pageSize
      *         a non-zero number of elements to be returned per a single read from Datastore
+     * @param <R>
+     *         the type of queried objects
      * @return results fo the query as a lazily evaluated {@link Iterator}
      * @throws IllegalArgumentException
      *         if the provided {@linkplain StructuredQuery#getLimit() query includes a limit}
      */
-    Iterator<Entity> readAll(StructuredQuery<Entity> query, int pageSize) {
+    <R> Iterator<R> readAll(StructuredQuery<R> query, int pageSize) {
         return readAllPageByPage(query, pageSize);
     }
 
@@ -311,11 +315,13 @@ Iterator<Entity> readAll(StructuredQuery<Entity> query, int pageSize) {
      *
      * @param query
      *         {@link Query} to execute upon the Datastore
+     * @param <R>
+     *         the type of queried objects
      * @return results fo the query as a lazily evaluated {@link Iterator}
      * @throws IllegalArgumentException
      *         if the provided {@linkplain StructuredQuery#getLimit() query includes a limit}
      */
-    Iterator<Entity> readAll(StructuredQuery<Entity> query) {
+    <R> Iterator<R> readAll(StructuredQuery<R> query) {
         return readAllPageByPage(query, null);
     }
 
@@ -333,26 +339,29 @@ Iterator<Entity> readAll(StructuredQuery<Entity> query) {
      * @param pageSize
      *         a non-zero number of elements to be returned per a single read from Datastore;
      *         if {@code null} the page size will be dictated by the Datastore
+     * @param <R>
+     *         the type of queried objects
      * @return results fo the query as a lazily evaluated {@link Iterator}
      * @throws IllegalArgumentException
      *         if the provided {@linkplain StructuredQuery#getLimit() query includes a limit} or
      *         the provided {@code batchSize} is 0
      */
-    private Iterator<Entity> readAllPageByPage(StructuredQuery<Entity> query,
-                                               @Nullable Integer pageSize) {
+    @SuppressWarnings("unchecked") // Checked logically.
+    private <R> Iterator<R>
+    readAllPageByPage(StructuredQuery<R> query, @Nullable Integer pageSize) {
         checkArgument(query.getLimit() == null,
                       "Cannot limit a number of entities for \"read all\" operation.");
         checkArgument(pageSize == null || pageSize != 0,
                       "The size of a single read operation cannot be 0.");
 
-        StructuredQuery<Entity> limitedQuery = limit(query, pageSize);
-        return stream(new DsQueryPageIterator(limitedQuery, this))
+        StructuredQuery<R> limitedQuery = limit(query, pageSize);
+        return stream(new DsQueryPageIterator<>(limitedQuery, this))
                 .flatMap(Streams::stream)
                 .iterator();
     }
 
-    private static StructuredQuery<Entity> limit(StructuredQuery<Entity> query,
-                                                 @Nullable Integer batchSize) {
+    private static <R> StructuredQuery<R> limit(StructuredQuery<R> query,
+                                                @Nullable Integer batchSize) {
         return batchSize == null
                ? query
                : query.toBuilder()

datastore/src/main/java/io/spine/server/storage/datastore/DsAggregateStorage.java

@@ -23,6 +23,7 @@
 import com.google.cloud.datastore.Entity;
 import com.google.cloud.datastore.EntityQuery;
 import com.google.cloud.datastore.Key;
+import com.google.cloud.datastore.ProjectionEntity;
 import com.google.cloud.datastore.Query;
 import com.google.cloud.datastore.StructuredQuery;
 import com.google.common.annotations.VisibleForTesting;
@@ -306,10 +307,12 @@ public void writeLifecycleFlags(I id, LifecycleFlags flags) {
     public Iterator<I> index() {
         checkNotClosed();
 
-        StructuredQuery<Entity> allQuery = Query.newEntityQueryBuilder()
-                                                .setKind(stateTypeName.value())
-                                                .build();
-        Iterator<I> index = stream(datastore.readAll(allQuery))
+        StructuredQuery<ProjectionEntity> query = Query.newProjectionEntityQueryBuilder()
+                                                       .setKind(stateTypeName.value())
+                                                       .setProjection(aggregate_id.name())
+                                                       .setDistinctOn(aggregate_id.name())
+                                                       .build();
+        Iterator<I> index = stream(datastore.readAll(query))
                 .map(new IndexTransformer<>(idClass))
                 .iterator();
         return index;
@@ -333,7 +336,7 @@ private Key toLifecycleRecordKey(I id) {
      * @param <I>
      *         the generic ID type
      */
-    private static class IndexTransformer<I> implements Function<Entity, I> {
+    private static class IndexTransformer<I> implements Function<ProjectionEntity, I> {
 
         private final Class<I> idClass;
 
@@ -342,7 +345,7 @@ private IndexTransformer(Class<I> idClass) {
         }
 
         @Override
-        public I apply(@Nullable Entity entity) {
+        public I apply(@Nullable ProjectionEntity entity) {
             checkNotNull(entity);
             String stringId = entity.getString(aggregate_id.toString());
             return Stringifiers.fromString(stringId, idClass);

datastore/src/main/java/io/spine/server/storage/datastore/DsQueryIterator.java

@@ -22,7 +22,6 @@
 
 import com.google.cloud.datastore.Cursor;
 import com.google.cloud.datastore.DatastoreReaderWriter;
-import com.google.cloud.datastore.Entity;
 import com.google.cloud.datastore.QueryResults;
 import com.google.cloud.datastore.StructuredQuery;
 import com.google.common.collect.UnmodifiableIterator;
@@ -44,18 +43,21 @@
  * <p>A call to {@link #next() next()} may not cause a Datastore query.
  *
  * <p>The {@link #remove() remove()} method throws an {@link UnsupportedOperationException}.
+ *
+ * @param <R>
+ *         the type of queried objects
  */
-final class DsQueryIterator extends UnmodifiableIterator<Entity> {
+final class DsQueryIterator<R> extends UnmodifiableIterator<R> {
 
-    private final StructuredQuery<Entity> query;
-    private final QueryResults<Entity> currentPage;
+    private final StructuredQuery<R> query;
+    private final QueryResults<R> currentPage;
 
     private final Integer limit;
     private int readCount = 0;
 
     private boolean terminated;
 
-    DsQueryIterator(StructuredQuery<Entity> query, DatastoreReaderWriter datastore) {
+    DsQueryIterator(StructuredQuery<R> query, DatastoreReaderWriter datastore) {
         super();
         this.query = query;
         this.limit = query.getLimit();
@@ -92,21 +94,21 @@ private void terminate() {
      * <p>The query is built utilizing the {@linkplain Cursor Datastore Cursor} from the current
      * query results.
      */
-    StructuredQuery<Entity> nextPageQuery() {
+    StructuredQuery<R> nextPageQuery() {
         Cursor cursorAfter = currentPage.getCursorAfter();
-        StructuredQuery<Entity> queryForMoreResults =
+        StructuredQuery<R> queryForMoreResults =
                 query.toBuilder()
                      .setStartCursor(cursorAfter)
                      .build();
         return queryForMoreResults;
     }
 
     @Override
-    public Entity next() {
+    public R next() {
         if (!hasNext()) {
             throw new NoSuchElementException("The query results Iterator is empty.");
         }
-        Entity result = currentPage.next();
+        R result = currentPage.next();
         readCount++;
         return result;
     }

datastore/src/main/java/io/spine/server/storage/datastore/DsQueryPageIterator.java

@@ -20,7 +20,6 @@
 
 package io.spine.server.storage.datastore;
 
-import com.google.cloud.datastore.Entity;
 import com.google.cloud.datastore.StructuredQuery;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
@@ -39,15 +38,18 @@
  *
  * <p>If the limit is not specified, then the page size is determined by the Datastore
  * query restrictions.
+ *
+ * @param <R>
+ *         the type of queried objects
  */
-final class DsQueryPageIterator implements Iterator<DsQueryIterator> {
+final class DsQueryPageIterator<R> implements Iterator<DsQueryIterator> {
 
     private final DatastoreWrapper datastore;
 
-    private DsQueryIterator currentPage;
-    private @Nullable DsQueryIterator nextPage;
+    private DsQueryIterator<R> currentPage;
+    private @Nullable DsQueryIterator<R> nextPage;
 
-    DsQueryPageIterator(StructuredQuery<Entity> query, DatastoreWrapper datastore) {
+    DsQueryPageIterator(StructuredQuery<R> query, DatastoreWrapper datastore) {
         this.datastore = datastore;
         this.currentPage = datastore.read(query);
     }
@@ -61,7 +63,7 @@ public boolean hasNext() {
     }
 
     @Override
-    public DsQueryIterator next() {
+    public DsQueryIterator<R> next() {
         if (nextPage == null) {
             currentPage = loadNextPage();
         } else {
@@ -74,8 +76,8 @@ public DsQueryIterator next() {
         return currentPage;
     }
 
-    private DsQueryIterator loadNextPage() {
-        StructuredQuery<Entity> nextPageQuery = currentPage.nextPageQuery();
+    private DsQueryIterator<R> loadNextPage() {
+        StructuredQuery<R> nextPageQuery = currentPage.nextPageQuery();
         return datastore.read(nextPageQuery);
     }
 }

datastore/src/main/java/io/spine/server/storage/datastore/Indexes.java

@@ -20,10 +20,9 @@
 
 package io.spine.server.storage.datastore;
 
-import com.google.cloud.datastore.Entity;
-import com.google.cloud.datastore.EntityQuery;
 import com.google.cloud.datastore.Key;
 import com.google.cloud.datastore.Query;
+import com.google.cloud.datastore.StructuredQuery;
 import com.google.common.collect.Streams;
 import io.spine.string.Stringifiers;
 import org.checkerframework.checker.nullness.qual.Nullable;
@@ -63,19 +62,19 @@ static <I> Iterator<I> indexIterator(DatastoreWrapper datastore, Kind kind, Clas
         checkNotNull(kind);
         checkNotNull(idType);
 
-        EntityQuery.Builder query = Query.newEntityQueryBuilder()
-                                         .setKind(kind.getValue());
-        Iterator<Entity> allEntities = datastore.read(query.build());
+        StructuredQuery<Key> query = Query.newKeyQueryBuilder()
+                                          .setKind(kind.getValue())
+                                          .build();
+        Iterator<Key> allEntities = datastore.read(query);
         Iterator<I> idIterator = Streams.stream(allEntities)
                                         .map(idExtractor(idType))
                                         .iterator();
         return idIterator;
     }
 
-    private static <I> Function<Entity, @Nullable I> idExtractor(Class<I> idType) {
-        return input -> {
-            checkNotNull(input);
-            Key key = input.getKey();
+    private static <I> Function<Key, @Nullable I> idExtractor(Class<I> idType) {
+        return key -> {
+            checkNotNull(key);
             String stringId = key.getName();
             I id = Stringifiers.fromString(stringId, idType);
             return id;

license-report.md

@@ -578,10 +578,10 @@
      * **POM Project URL:** [https://javacc.dev.java.net/](https://javacc.dev.java.net/)
      * **POM License: Berkeley Software Distribution (BSD) License** - [http://www.opensource.org/licenses/bsd-license.html](http://www.opensource.org/licenses/bsd-license.html)
 
-1. **Group:** net.sourceforge.pmd **Name:** pmd-core **Version:** 6.13.0
+1. **Group:** net.sourceforge.pmd **Name:** pmd-core **Version:** 6.16.0
      * **POM License: BSD-style** - [http://pmd.sourceforge.net/license.html](http://pmd.sourceforge.net/license.html)
 
-1. **Group:** net.sourceforge.pmd **Name:** pmd-java **Version:** 6.13.0
+1. **Group:** net.sourceforge.pmd **Name:** pmd-java **Version:** 6.16.0
      * **POM License: BSD-style** - [http://pmd.sourceforge.net/license.html](http://pmd.sourceforge.net/license.html)
 
 1. **Group:** net.sourceforge.saxon **Name:** saxon **Version:** 9.1.0.8
@@ -698,9 +698,9 @@
      * **POM Project URL:** [https://junit.org/junit5/](https://junit.org/junit5/)
      * **POM License: Eclipse Public License v2.0** - [https://www.eclipse.org/legal/epl-v20.html](https://www.eclipse.org/legal/epl-v20.html)
 
-1. **Group:** org.junit.jupiter **Name:** junit-jupiter-engine **Version:** 5.4.2
+1. **Group:** org.junit.jupiter **Name:** junit-jupiter-engine **Version:** 5.5.1
      * **POM Project URL:** [https://junit.org/junit5/](https://junit.org/junit5/)
-     * **POM License: Eclipse Public License v2.0** - [http://www.eclipse.org/legal/epl-v20.html](http://www.eclipse.org/legal/epl-v20.html)
+     * **POM License: Eclipse Public License v2.0** - [https://www.eclipse.org/legal/epl-v20.html](https://www.eclipse.org/legal/epl-v20.html)
 
 1. **Group:** org.junit.jupiter **Name:** junit-jupiter-params **Version:** 5.5.1
      * **POM Project URL:** [https://junit.org/junit5/](https://junit.org/junit5/)
@@ -710,9 +710,9 @@
      * **POM Project URL:** [https://junit.org/junit5/](https://junit.org/junit5/)
      * **POM License: Eclipse Public License v2.0** - [https://www.eclipse.org/legal/epl-v20.html](https://www.eclipse.org/legal/epl-v20.html)
 
-1. **Group:** org.junit.platform **Name:** junit-platform-engine **Version:** 1.4.2
+1. **Group:** org.junit.platform **Name:** junit-platform-engine **Version:** 1.5.1
      * **POM Project URL:** [https://junit.org/junit5/](https://junit.org/junit5/)
-     * **POM License: Eclipse Public License v2.0** - [http://www.eclipse.org/legal/epl-v20.html](http://www.eclipse.org/legal/epl-v20.html)
+     * **POM License: Eclipse Public License v2.0** - [https://www.eclipse.org/legal/epl-v20.html](https://www.eclipse.org/legal/epl-v20.html)
 
 1. **Group:** org.mockito **Name:** mockito-core **Version:** 2.12.0
      * **POM Project URL:** [https://github.com/mockito/mockito](https://github.com/mockito/mockito)
@@ -780,7 +780,7 @@
  The dependencies distributed under several licenses, are used according their commercial-use-friendly license.
 
 
-This report was generated on **Thu Jul 25 16:54:10 EEST 2019** using [Gradle-License-Report plugin](https://github.com/jk1/Gradle-License-Report) by Evgeny Naumenko, licensed under [Apache 2.0 License](https://github.com/jk1/Gradle-License-Report/blob/master/LICENSE).
+This report was generated on **Fri Aug 02 19:17:57 EEST 2019** using [Gradle-License-Report plugin](https://github.com/jk1/Gradle-License-Report) by Evgeny Naumenko, licensed under [Apache 2.0 License](https://github.com/jk1/Gradle-License-Report/blob/master/LICENSE).
 
 
 
@@ -1347,10 +1347,10 @@ This report was generated on **Thu Jul 25 16:54:10 EEST 2019** using [Gradle-Lic
      * **POM Project URL:** [https://javacc.dev.java.net/](https://javacc.dev.java.net/)
      * **POM License: Berkeley Software Distribution (BSD) License** - [http://www.opensource.org/licenses/bsd-license.html](http://www.opensource.org/licenses/bsd-license.html)
 
-1. **Group:** net.sourceforge.pmd **Name:** pmd-core **Version:** 6.13.0
+1. **Group:** net.sourceforge.pmd **Name:** pmd-core **Version:** 6.16.0
      * **POM License: BSD-style** - [http://pmd.sourceforge.net/license.html](http://pmd.sourceforge.net/license.html)
 
-1. **Group:** net.sourceforge.pmd **Name:** pmd-java **Version:** 6.13.0
+1. **Group:** net.sourceforge.pmd **Name:** pmd-java **Version:** 6.16.0
      * **POM License: BSD-style** - [http://pmd.sourceforge.net/license.html](http://pmd.sourceforge.net/license.html)
 
 1. **Group:** net.sourceforge.saxon **Name:** saxon **Version:** 9.1.0.8
@@ -1467,9 +1467,9 @@ This report was generated on **Thu Jul 25 16:54:10 EEST 2019** using [Gradle-Lic
      * **POM Project URL:** [https://junit.org/junit5/](https://junit.org/junit5/)
      * **POM License: Eclipse Public License v2.0** - [https://www.eclipse.org/legal/epl-v20.html](https://www.eclipse.org/legal/epl-v20.html)
 
-1. **Group:** org.junit.jupiter **Name:** junit-jupiter-engine **Version:** 5.4.2
+1. **Group:** org.junit.jupiter **Name:** junit-jupiter-engine **Version:** 5.5.1
      * **POM Project URL:** [https://junit.org/junit5/](https://junit.org/junit5/)
-     * **POM License: Eclipse Public License v2.0** - [http://www.eclipse.org/legal/epl-v20.html](http://www.eclipse.org/legal/epl-v20.html)
+     * **POM License: Eclipse Public License v2.0** - [https://www.eclipse.org/legal/epl-v20.html](https://www.eclipse.org/legal/epl-v20.html)
 
 1. **Group:** org.junit.jupiter **Name:** junit-jupiter-params **Version:** 5.5.1
      * **POM Project URL:** [https://junit.org/junit5/](https://junit.org/junit5/)
@@ -1479,9 +1479,9 @@ This report was generated on **Thu Jul 25 16:54:10 EEST 2019** using [Gradle-Lic
      * **POM Project URL:** [https://junit.org/junit5/](https://junit.org/junit5/)
      * **POM License: Eclipse Public License v2.0** - [https://www.eclipse.org/legal/epl-v20.html](https://www.eclipse.org/legal/epl-v20.html)
 
-1. **Group:** org.junit.platform **Name:** junit-platform-engine **Version:** 1.4.2
+1. **Group:** org.junit.platform **Name:** junit-platform-engine **Version:** 1.5.1
      * **POM Project URL:** [https://junit.org/junit5/](https://junit.org/junit5/)
-     * **POM License: Eclipse Public License v2.0** - [http://www.eclipse.org/legal/epl-v20.html](http://www.eclipse.org/legal/epl-v20.html)
+     * **POM License: Eclipse Public License v2.0** - [https://www.eclipse.org/legal/epl-v20.html](https://www.eclipse.org/legal/epl-v20.html)
 
 1. **Group:** org.mockito **Name:** mockito-core **Version:** 2.12.0
      * **POM Project URL:** [https://github.com/mockito/mockito](https://github.com/mockito/mockito)
@@ -1549,4 +1549,4 @@ This report was generated on **Thu Jul 25 16:54:10 EEST 2019** using [Gradle-Lic
  The dependencies distributed under several licenses, are used according their commercial-use-friendly license.
 
 
-This report was generated on **Thu Jul 25 16:54:11 EEST 2019** using [Gradle-License-Report plugin](https://github.com/jk1/Gradle-License-Report) by Evgeny Naumenko, licensed under [Apache 2.0 License](https://github.com/jk1/Gradle-License-Report/blob/master/LICENSE).
+This report was generated on **Fri Aug 02 19:18:10 EEST 2019** using [Gradle-License-Report plugin](https://github.com/jk1/Gradle-License-Report) by Evgeny Naumenko, licensed under [Apache 2.0 License](https://github.com/jk1/Gradle-License-Report/blob/master/LICENSE).