From f2ffbb2c5bb5852712257e360a3bbd276af43c54 Mon Sep 17 00:00:00 2001 From: Robin Bygrave Date: Mon, 7 Mar 2016 13:49:23 +1300 Subject: [PATCH] #589 - ElasticSearch - Add copyIndex() via Query, Add findEachWhile() --- .../java/com/avaje/ebean/DocumentStore.java | 58 +++++++++++++++++-- .../server/core/DefaultServer.java | 2 +- .../docstore/none/NoneDocStore.java | 15 ++++- 3 files changed, 67 insertions(+), 8 deletions(-) diff --git a/src/main/java/com/avaje/ebean/DocumentStore.java b/src/main/java/com/avaje/ebean/DocumentStore.java index ed7e63838..d8e267d4a 100644 --- a/src/main/java/com/avaje/ebean/DocumentStore.java +++ b/src/main/java/com/avaje/ebean/DocumentStore.java @@ -48,7 +48,7 @@ public interface DocumentStore { * If the document is not found null is returned. */ @Nullable - T getById(Class beanType, Object id); + T find(Class beanType, Object id); /** * Execute the query against the document store returning the list. @@ -72,6 +72,18 @@ public interface DocumentStore { */ void findEach(Query query, QueryEachConsumer consumer); + /** + * Execute the query against the document store with the expectation of a large set of results + * that are processed in a scrolling resultSet fashion. + *

+ * Unlike findEach() this provides the opportunity to stop iterating through the large query. + *

+ *

+ * For example, with the ElasticSearch doc store this uses SCROLL. + *

+ */ + void findEachWhile(Query query, QueryEachWhileConsumer consumer); + /** * Process the queue entries sending updates to the document store or queuing them for later processing. */ @@ -87,9 +99,8 @@ public interface DocumentStore { * * @param indexName the name of the new index * @param alias the alias of the index - * @param mappingResource the path of the mapping file as a resource in the classpath */ - void createIndex(String indexName, String alias, String mappingResource); + void createIndex(String indexName, String alias); /** * Copy the index to a new index. @@ -97,9 +108,15 @@ public interface DocumentStore { * This copy process does not use the database but instead will copy from the source index to a destination index. *

* + * 
{@code
+   *
+   *  long copyCount = documentStore.copyIndex(Product.class, "product_copy");
+   *
+   * }
+ * + * * @param beanType The bean type of the source index * @param newIndex The name of the index to copy to - * * @return the number of documents copied to the new index */ long copyIndex(Class beanType, String newIndex); @@ -111,11 +128,42 @@ public interface DocumentStore { * To support this the document needs to have a @WhenModified property. *

* + * 
{@code
+   *
+   *  long copyCount = documentStore.copyIndex(Product.class, "product_copy", sinceMillis);
+   *
+   * }
+ * + * * @param beanType The bean type of the source index * @param newIndex The name of the index to copy to - * * @return the number of documents copied to the new index */ long copyIndex(Class beanType, String newIndex, long sinceEpochMillis); + /** + * Copy from a source index to a new index taking only the documents + * matching the given query. + * + * 
{@code
+   *
+   *  // predicates to select the source documents to copy
+   *  Query query = server.find(Product.class)
+   *    .where()
+   *      .ge("whenModified", new Timestamp(since))
+   *      .ge("name", "A")
+   *      .lt("name", "D")
+   *      .query();
+   *
+   *  // copy from the source index to "product_copy" index
+   *  long copyCount = documentStore.copyIndex(query, "product_copy", 1000);
+   *
+   * }
+ * + * @param query The query to select the source documents to copy + * @param newIndex The target index to copy the documents to + * @param bulkBatchSize The ElasticSearch bulk batch size, if 0 uses the default. + * @return The number of documents copied to the new index. + */ + long copyIndex(Query query, String newIndex, int bulkBatchSize); } diff --git a/src/main/java/com/avaje/ebeaninternal/server/core/DefaultServer.java b/src/main/java/com/avaje/ebeaninternal/server/core/DefaultServer.java index f2599708f..ea48ce575 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/core/DefaultServer.java +++ b/src/main/java/com/avaje/ebeaninternal/server/core/DefaultServer.java @@ -1140,7 +1140,7 @@ public final class DefaultServer implements SpiServer, SpiEbeanServer { SpiOrmQueryRequest request = createQueryRequest(spiQuery, t); if (request.isUseDocStore()) { - return docStore().getById(query.getBeanType(), query.getId()); + return docStore().find(query.getBeanType(), query.getId()); } try { request.initTransIfRequired(); diff --git a/src/main/java/com/avaje/ebeanservice/docstore/none/NoneDocStore.java b/src/main/java/com/avaje/ebeanservice/docstore/none/NoneDocStore.java index e9b3d9285..0b4662d10 100644 --- a/src/main/java/com/avaje/ebeanservice/docstore/none/NoneDocStore.java +++ b/src/main/java/com/avaje/ebeanservice/docstore/none/NoneDocStore.java @@ -5,6 +5,7 @@ import com.avaje.ebean.DocumentStore; import com.avaje.ebean.PagedList; import com.avaje.ebean.Query; import com.avaje.ebean.QueryEachConsumer; +import com.avaje.ebean.QueryEachWhileConsumer; import org.jetbrains.annotations.Nullable; import java.io.IOException; @@ -25,7 +26,7 @@ public class NoneDocStore implements DocumentStore { } @Override - public void createIndex(String indexName, String alias, String mappingResource) { + public void createIndex(String indexName, String alias) { throw implementationNotInClassPath(); } @@ -44,6 +45,11 @@ public class NoneDocStore implements DocumentStore { throw implementationNotInClassPath(); } + @Override + public long copyIndex(Query query, String newIndex, int bulkBatchSize) { + throw implementationNotInClassPath(); + } + @Override public void indexByQuery(Query query) { throw implementationNotInClassPath(); @@ -56,7 +62,7 @@ public class NoneDocStore implements DocumentStore { @Nullable @Override - public T getById(Class beanType, Object id) { + public T find(Class beanType, Object id) { throw implementationNotInClassPath(); } @@ -75,6 +81,11 @@ public class NoneDocStore implements DocumentStore { throw implementationNotInClassPath(); } + @Override + public void findEachWhile(Query query, QueryEachWhileConsumer consumer) { + throw implementationNotInClassPath(); + } + @Override public long process(List queueEntries) throws IOException { throw implementationNotInClassPath();