Merge branch 'rharing-adding_sortField_in_preparation_of_incremental_reindex'

Author: awislowski

README.md

@@ -67,6 +67,13 @@ With segmentation by prefix on string field:
  cluster_name1 -segmentationField userId -segmentationPrefixes 1,2,3,4,5,6,7`
 
  In this example index querying will divide data into segments based on the first character of the userId field: 1,2,3,4,5,6,7
+ 
+With query option:
+
+`./run.sh -s http://host:9300/index/type -t http://host1:9300/index1/type1  -sc cluster_name -tc
+ cluster_name1 "\"{range\": {\"_timestamp\" : {\"gte\" : 1447142880000}}}\"" -sort _timestamp -sortOrder DESC`
+ 
+ In this example index querying will be filtered with query and reindex will take place ordered by sort field and sortOrder 
 
 Options:
 
@@ -84,7 +91,13 @@ Options:
        Segmentation prefixes (comma-separated)
     -segmentationThresholds
        Segmentation thresholds (only double type)
-
+    -query
+       Give a query to filter data 
+    -sort
+       Give field to sort on (if query option in use)
+    -sortOrder
+       Give sortOrder  (if query option in use)    
+       
 `segmentationField`, `segmentationThreshold` and `segmentationPrefixes` are optional parameters, allowing to spread
 querying for field with double values or prefix for string field
 

src/main/java/pl/allegro/tech/search/elasticsearch/tools/reindex/ReindexCommandParser.java

@@ -4,18 +4,19 @@
 import com.beust.jcommander.ParameterException;
 import pl.allegro.tech.search.elasticsearch.tools.reindex.command.ReindexCommand;
 import pl.allegro.tech.search.elasticsearch.tools.reindex.connection.ElasticDataPointer;
-import pl.allegro.tech.search.elasticsearch.tools.reindex.connection.ParsingElasticsearchAddressException;
-import pl.allegro.tech.search.elasticsearch.tools.reindex.query.QuerySegmentationFactory;
 import pl.allegro.tech.search.elasticsearch.tools.reindex.connection.ElasticDataPointerBuilder;
+import pl.allegro.tech.search.elasticsearch.tools.reindex.connection.ElasticSearchQuery;
+import pl.allegro.tech.search.elasticsearch.tools.reindex.connection.ElasticSearchQueryBuilder;
+import pl.allegro.tech.search.elasticsearch.tools.reindex.connection.ParsingElasticsearchAddressException;
 import pl.allegro.tech.search.elasticsearch.tools.reindex.query.QuerySegmentation;
+import pl.allegro.tech.search.elasticsearch.tools.reindex.query.QuerySegmentationFactory;
 
 public class ReindexCommandParser {
 
   private ElasticDataPointer sourcePointer;
   private ElasticDataPointer targetPointer;
   private QuerySegmentation segmentation;
 
-
   public boolean tryParse(String... args) {
     ReindexCommand command = new ReindexCommand();
     JCommander jCommander = new JCommander(command);
@@ -62,4 +63,5 @@ public ElasticDataPointer getTargetPointer() {
   public QuerySegmentation getSegmentation() {
     return segmentation;
   }
+
 }

src/main/java/pl/allegro/tech/search/elasticsearch/tools/reindex/ReindexInvoker.java

@@ -5,6 +5,7 @@
 import org.slf4j.LoggerFactory;
 import pl.allegro.tech.search.elasticsearch.tools.reindex.connection.ElasticDataPointer;
 import pl.allegro.tech.search.elasticsearch.tools.reindex.connection.ElasticSearchClientFactory;
+import pl.allegro.tech.search.elasticsearch.tools.reindex.connection.ElasticSearchQuery;
 import pl.allegro.tech.search.elasticsearch.tools.reindex.process.IndexingComponent;
 import pl.allegro.tech.search.elasticsearch.tools.reindex.process.IndexingProcessBuilder;
 import pl.allegro.tech.search.elasticsearch.tools.reindex.process.ProcessConfiguration;
@@ -79,11 +80,12 @@ private void startQueriesProcesses(Client client, ElasticDataPointer sourcePoint
                     .setDataPointer(sourcePointer)
                     .setSegmentationField(segmentation.getFieldName())
                     .setBound(segmentation.getThreshold(i))
+                    .setQuery(segmentation.getQuery())
                     .createQueryComponent()
         ).map(
         queryComponent -> new QueryProcess(processSynchronizer, queryComponent)
     ).forEach(
-        queryProcess -> processExecutor.startProcess(queryProcess)
+        processExecutor::startProcess
     );
   }
 

src/main/java/pl/allegro/tech/search/elasticsearch/tools/reindex/command/ReindexCommand.java

@@ -22,6 +22,15 @@ public class ReindexCommand {
   @Parameter(names = { "-segmentationField" }, description = "Segmentation field")
   private String segmentationField;
 
+  @Parameter(names = { "-query" }, description = "Give a query to filter data")
+  private String query;
+
+  @Parameter(names = { "-sort" }, description = "Give field to sort on (if query option in use)")
+  private String sort;
+
+  @Parameter(names = { "-sortOrder" }, description = "Give sortOrder (if query option in use)")
+  private String sortOrder;
+
   @Parameter(names = { "-segmentationThresholds" }, description = "Segmentation thresholds (only double type)")
   private List<Double> segmentationThresholds;
 
@@ -38,6 +47,18 @@ public String getSegmentationField() {
     return segmentationField;
   }
 
+  public String getQuery() {
+    return query;
+  }
+
+  public String getSort() {
+    return sort;
+  }
+
+  public String getSortOrder() {
+    return sortOrder;
+  }
+
   public List<Double> getSegmentationThresholds() {
     return segmentationThresholds;
   }

src/main/java/pl/allegro/tech/search/elasticsearch/tools/reindex/connection/ElasticSearchQuery.java

@@ -0,0 +1,27 @@
+package pl.allegro.tech.search.elasticsearch.tools.reindex.connection;
+
+import org.elasticsearch.search.sort.SortOrder;
+
+public class ElasticSearchQuery {
+  private final String query;
+  private final String sortField;
+  private final SortOrder sortOrder;
+
+  ElasticSearchQuery(String query, String sortField, SortOrder sortOrder) {
+    this.query = query;
+    this.sortField = sortField;
+    this.sortOrder = sortOrder;
+  }
+
+  public String getQuery() {
+    return query;
+  }
+
+  public SortOrder getSortOrder() {
+    return sortOrder;
+  }
+
+  public String getSortField() {
+    return sortField;
+  }
+}

src/main/java/pl/allegro/tech/search/elasticsearch/tools/reindex/connection/ElasticSearchQueryBuilder.java

@@ -0,0 +1,45 @@
+package pl.allegro.tech.search.elasticsearch.tools.reindex.connection;
+
+import com.google.common.base.Strings;
+import org.elasticsearch.search.sort.SortOrder;
+
+public class ElasticSearchQueryBuilder {
+  private String query;
+  private String sortField;
+  private SortOrder sortOrder = SortOrder.ASC;
+
+
+  private ElasticSearchQueryBuilder() {
+  }
+
+  public ElasticSearchQueryBuilder setSortOrder(String sortOrder) {
+    if (!Strings.isNullOrEmpty(sortOrder)) {
+      try {
+        this.sortOrder = SortOrder.valueOf(sortOrder);
+      } catch (IllegalArgumentException e) {
+        throw new ParsingElasticsearchAddressException("SortOrder can be only ASC or DESC, not " + sortOrder);
+      }
+    }
+    return this;
+  }
+
+  public ElasticSearchQueryBuilder setQuery(String query) {
+    this.query = query;
+    return this;
+  }
+
+  public ElasticSearchQueryBuilder setSortByField(String orderByField) {
+    this.sortField = orderByField;
+    return this;
+  }
+
+  public ElasticSearchQuery build() {
+    return new ElasticSearchQuery(query, sortField, sortOrder);
+  }
+
+  public static ElasticSearchQueryBuilder builder() {
+    return new ElasticSearchQueryBuilder();
+  }
+
+
+}

src/main/java/pl/allegro/tech/search/elasticsearch/tools/reindex/process/QueryComponent.java

@@ -1,13 +1,16 @@
 package pl.allegro.tech.search.elasticsearch.tools.reindex.process;
 
-import pl.allegro.tech.search.elasticsearch.tools.reindex.connection.ElasticDataPointer;
-import pl.allegro.tech.search.elasticsearch.tools.reindex.query.BoundedSegment;
-import pl.allegro.tech.search.elasticsearch.tools.reindex.query.filter.BoundedFilterFactory;
+import com.google.common.base.Strings;
 import org.elasticsearch.action.search.SearchRequestBuilder;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.search.SearchType;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.search.sort.FieldSortBuilder;
+import pl.allegro.tech.search.elasticsearch.tools.reindex.connection.ElasticDataPointer;
+import pl.allegro.tech.search.elasticsearch.tools.reindex.connection.ElasticSearchQuery;
+import pl.allegro.tech.search.elasticsearch.tools.reindex.query.BoundedSegment;
+import pl.allegro.tech.search.elasticsearch.tools.reindex.query.filter.BoundedFilterFactory;
 
 import java.util.Optional;
 
@@ -21,13 +24,15 @@ public class QueryComponent {
   private Optional<String> segmentationField;
   private ElasticDataPointer dataPointer;
   private Optional<BoundedSegment> bound;
+  private ElasticSearchQuery query;
   private BoundedFilterFactory boundedFilterFactory = new BoundedFilterFactory();
 
-  QueryComponent(Client client, ElasticDataPointer dataPointer, Optional<String> segmentationField, Optional<BoundedSegment> bound) {
+  QueryComponent(Client client, ElasticDataPointer dataPointer, Optional<String> segmentationField, Optional<BoundedSegment> bound, ElasticSearchQuery query) {
     this.client = client;
     this.dataPointer = dataPointer;
     this.segmentationField = segmentationField;
     this.bound = bound;
+    this.query = query;
   }
 
   public SearchResponse prepareSearchScrollRequest() {
@@ -38,6 +43,13 @@ public SearchResponse prepareSearchScrollRequest() {
         .setScroll(new TimeValue(SCROLL_TIME_LIMIT))
         .setSize(SCROLL_SHARD_LIMIT);
 
+    if (!Strings.isNullOrEmpty(query.getQuery())) {
+      searchRequestBuilder.setQuery(query.getQuery());
+    }
+    if (!Strings.isNullOrEmpty(query.getSortField())) {
+      searchRequestBuilder.addSort(new FieldSortBuilder(query.getSortField()).order(query.getSortOrder()));
+    }
+
     bound.map(resolvedBound -> boundedFilterFactory.createBoundedFilter(segmentationField.get(), resolvedBound))
         .ifPresent(searchRequestBuilder::setQuery);
 

src/main/java/pl/allegro/tech/search/elasticsearch/tools/reindex/process/QueryComponentBuilder.java

@@ -1,6 +1,7 @@
 package pl.allegro.tech.search.elasticsearch.tools.reindex.process;
 
 import pl.allegro.tech.search.elasticsearch.tools.reindex.connection.ElasticDataPointer;
+import pl.allegro.tech.search.elasticsearch.tools.reindex.connection.ElasticSearchQuery;
 import pl.allegro.tech.search.elasticsearch.tools.reindex.query.BoundedSegment;
 import org.elasticsearch.client.Client;
 
@@ -11,6 +12,7 @@ public final class QueryComponentBuilder {
   private ElasticDataPointer dataPointer;
   private Optional<String> segmentationField = Optional.empty();
   private Optional<BoundedSegment> bound = Optional.empty();
+  private ElasticSearchQuery query;
 
   private QueryComponentBuilder() {
   }
@@ -35,11 +37,18 @@ public QueryComponentBuilder setBound(Optional<BoundedSegment> bound) {
     return this;
   }
 
+  public QueryComponentBuilder setQuery(ElasticSearchQuery query) {
+    this.query = query;
+    return this;
+  }
+
+
   public static QueryComponentBuilder builder() {
     return new QueryComponentBuilder();
   }
 
   public QueryComponent createQueryComponent() {
-    return new QueryComponent(client, dataPointer, segmentationField, bound);
+    return new QueryComponent(client, dataPointer, segmentationField, bound, query);
   }
+
 }

src/main/java/pl/allegro/tech/search/elasticsearch/tools/reindex/query/DoubleFieldSegmentation.java

@@ -1,16 +1,19 @@
 package pl.allegro.tech.search.elasticsearch.tools.reindex.query;
 
+import pl.allegro.tech.search.elasticsearch.tools.reindex.connection.ElasticSearchQuery;
+
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 
-public final class DoubleFieldSegmentation implements QuerySegmentation {
+public final class DoubleFieldSegmentation extends SegmentationQueryTrait implements QuerySegmentation {
 
   private final List<Double> thresholds;
 
   private final Optional<String> fieldName;
 
-  private DoubleFieldSegmentation(String fieldName, List<Double> thresholds) {
+  private DoubleFieldSegmentation(String fieldName, List<Double> thresholds, ElasticSearchQuery query) {
+    super(query);
     this.fieldName = Optional.of(fieldName);
     this.thresholds = Collections.unmodifiableList(thresholds);
   }
@@ -35,7 +38,7 @@ public Optional<BoundedSegment> getThreshold(int i) {
     return Optional.of(segmentation);
   }
 
-  public static DoubleFieldSegmentation create(String fieldName, List<Double> thresholds) {
-    return new DoubleFieldSegmentation(fieldName, thresholds);
+  public static DoubleFieldSegmentation create(String fieldName, List<Double> thresholds, ElasticSearchQuery query) {
+    return new DoubleFieldSegmentation(fieldName, thresholds, query);
   }
 }

src/main/java/pl/allegro/tech/search/elasticsearch/tools/reindex/query/EmptySegmentation.java

@@ -1,12 +1,13 @@
 package pl.allegro.tech.search.elasticsearch.tools.reindex.query;
 
-import java.util.Optional;
+import pl.allegro.tech.search.elasticsearch.tools.reindex.connection.ElasticSearchQuery;
 
-public final class EmptySegmentation implements QuerySegmentation {
+import java.util.Optional;
 
-  public static final EmptySegmentation EMPTY_SEGMENTATION_INSTANCE = new EmptySegmentation();
+public final class EmptySegmentation extends SegmentationQueryTrait implements QuerySegmentation {
 
-  private EmptySegmentation() {
+  private EmptySegmentation(ElasticSearchQuery query) {
+    super(query);
   }
 
   @Override
@@ -24,8 +25,8 @@ public Optional<BoundedSegment> getThreshold(int i) {
     return Optional.empty();
   }
 
-  public static EmptySegmentation createEmptySegmentation() {
-    return EMPTY_SEGMENTATION_INSTANCE;
+  public static EmptySegmentation createEmptySegmentation(ElasticSearchQuery query) {
+    return new EmptySegmentation(query);
   }
 
 }