Reindex resolve indices early

Resolve indices before starting to reindex. This ensures that the list
of indices does not change when failing over (TBD). The one exception to
this is aliases, which we still need to access through the alias.

In addition, resolved index patterns are sorted by create-date and
otherwise the listed order is preserved. This ensures that once we
reindex one index at a time, we will get reasonable time locality for
time based indices.

The resolved list of indices will also by used to do searching one
index (or index group) at a time, improving search performance (since we
use sort) and allowing us to do more fine-grained checkpoint and track
progress (TBD).

Relates elastic#42612

Author: henningandersen

libs/x-content/src/main/java/org/elasticsearch/common/xcontent/AbstractObjectParser.java

@@ -225,6 +225,7 @@ private static <T> List<T> parseArray(XContentParser parser, IOSupplier<T> suppl
             while (parser.nextToken() != XContentParser.Token.END_ARRAY) {
                 if (parser.currentToken().isValue()
                         || parser.currentToken() == XContentParser.Token.VALUE_NULL
+                        || parser.currentToken() == XContentParser.Token.START_ARRAY
                         || parser.currentToken() == XContentParser.Token.START_OBJECT) {
                     list.add(supplier.get());
                 } else {

libs/x-content/src/test/java/org/elasticsearch/common/xcontent/ObjectParserTests.java

@@ -585,30 +585,39 @@ class TestStruct {
     public void testArraysOfGenericValues() throws IOException {
         XContentParser parser = createParser(
             JsonXContent.jsonXContent,
-            "{ \"test_array\": [ 1, null, \"3\", 4.2], \"int_array\":  [ 1, 2, 3] }"
+            "{ \"test_array\": [ 1, null, \"3\", 4.2], \"int_array\":  [ 1, 2, 3], \"multi_array\": [ [1, 2], [3, 4] ] }"
         );
         class TestStruct {
             List<Object> testArray = new ArrayList<>();
 
             List<Integer> ints = new ArrayList<>();
 
+            List<List<Integer> > multis = new ArrayList<>();
+
             public void setInts(List<Integer> ints) {
                 this.ints = ints;
             }
 
             public void setArray(List<Object> testArray) {
                 this.testArray = testArray;
             }
+
+            public void setMultis(List<List<Integer>> multis) {
+                this.multis = multis;
+            }
         }
         ObjectParser<TestStruct, Void> objectParser = new ObjectParser<>("foo");
         TestStruct s = new TestStruct();
 
         objectParser.declareFieldArray(TestStruct::setArray, (p, c) -> XContentParserUtils.parseFieldsValue(p),
             new ParseField("test_array"), ValueType.VALUE_ARRAY);
         objectParser.declareIntArray(TestStruct::setInts, new ParseField("int_array"));
+        objectParser.declareFieldArray(TestStruct::setMultis, (p, c) -> (List<Integer>) XContentParserUtils.parseFieldsValue(p),
+            new ParseField("multi_array"), ValueType.OBJECT_ARRAY);
         objectParser.parse(parser, s, null);
         assertEquals(s.testArray, Arrays.asList(1, null, "3", 4.2));
         assertEquals(s.ints, Arrays.asList(1, 2, 3));
+        assertEquals(s.multis, Arrays.asList(Arrays.asList(1, 2), Arrays.asList(3, 4)));
 
         parser = createParser(JsonXContent.jsonXContent, "{\"test_array\": 42}");
         s = new TestStruct();

modules/reindex/src/main/java/org/elasticsearch/index/reindex/ReindexTaskParams.java

@@ -29,6 +29,8 @@
 import org.elasticsearch.persistent.PersistentTaskParams;
 
 import java.io.IOException;
+import java.util.Collection;
+import java.util.List;
 import java.util.Map;
 
 public class ReindexTaskParams implements PersistentTaskParams {
@@ -37,26 +39,32 @@ public class ReindexTaskParams implements PersistentTaskParams {
 
     @SuppressWarnings("unchecked")
     public static final ConstructingObjectParser<ReindexTaskParams, Void> PARSER
-        = new ConstructingObjectParser<>(NAME, a -> new ReindexTaskParams((Boolean) a[0], (Map<String, String>) a[1]));
+        = new ConstructingObjectParser<>(NAME, a -> new ReindexTaskParams((Boolean) a[0], (List<List<String>>) a[1],
+        (Map<String, String>) a[2]));
 
     private static String STORE_RESULT = "store_result";
     private static String HEADERS = "headers";
+    private static String INDEX_GROUPS = "index_groups";
 
     static {
         PARSER.declareBoolean(ConstructingObjectParser.constructorArg(), new ParseField(STORE_RESULT));
+        PARSER.declareObjectArray(ConstructingObjectParser.constructorArg(), (p, c) -> p.list(), new ParseField(INDEX_GROUPS));
         PARSER.declareObject(ConstructingObjectParser.constructorArg(), (p, c) -> p.mapStrings(), new ParseField(HEADERS));
     }
 
     private final boolean storeResult;
+    private final List<? extends Collection<String>> indexGroups;
     private final Map<String, String> headers;
 
-    public ReindexTaskParams(boolean storeResult, Map<String, String> headers) {
+    public ReindexTaskParams(boolean storeResult, List<? extends Collection<String>> indexGroups, Map<String, String> headers) {
         this.storeResult = storeResult;
+        this.indexGroups = indexGroups;
         this.headers = headers;
     }
 
     public ReindexTaskParams(StreamInput in) throws IOException {
         storeResult = in.readBoolean();
+        indexGroups = in.readList(StreamInput::readStringList);
         headers = in.readMap(StreamInput::readString, StreamInput::readString);
     }
 
@@ -74,13 +82,15 @@ public Version getMinimalSupportedVersion() {
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         out.writeBoolean(storeResult);
+        out.writeCollection(indexGroups, StreamOutput::writeStringCollection);
         out.writeMap(headers, StreamOutput::writeString, StreamOutput::writeString);
     }
 
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject();
         builder.field(STORE_RESULT, storeResult);
+        builder.field(INDEX_GROUPS, indexGroups);
         builder.field(HEADERS, headers);
         return builder.endObject();
     }
@@ -89,6 +99,10 @@ public boolean shouldStoreResult() {
         return storeResult;
     }
 
+    public List<? extends Collection<String>> getIndexGroups() {
+        return indexGroups;
+    }
+
     public Map<String, String> getHeaders() {
         return headers;
     }

modules/reindex/src/main/java/org/elasticsearch/index/reindex/TransportStartReindexTaskAction.java

@@ -26,13 +26,17 @@
 import org.elasticsearch.action.support.AutoCreateIndex;
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.client.Client;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.metadata.AliasOrIndex;
+import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.persistent.PersistentTasksCustomMetaData;
 import org.elasticsearch.persistent.PersistentTasksService;
@@ -41,17 +45,27 @@
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.SortedMap;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 public class TransportStartReindexTaskAction
     extends HandledTransportAction<StartReindexTaskAction.Request, StartReindexTaskAction.Response> {
 
     private final List<String> headersToInclude;
     private final ThreadPool threadPool;
     private final PersistentTasksService persistentTasksService;
+    private final IndexNameExpressionResolver indexNameExpressionResolver;
+    private final ClusterService clusterService;
     private final ReindexValidator reindexValidator;
     private final ReindexIndexClient reindexIndexClient;
 
@@ -63,6 +77,8 @@ public TransportStartReindexTaskAction(Settings settings, Client client, Transpo
         super(StartReindexTaskAction.NAME, transportService, actionFilters, StartReindexTaskAction.Request::new);
         this.headersToInclude = ReindexHeaders.REINDEX_INCLUDED_HEADERS.get(settings);
         this.threadPool = threadPool;
+        this.indexNameExpressionResolver = indexNameExpressionResolver;
+        this.clusterService = clusterService;
         this.reindexValidator = new ReindexValidator(settings, clusterService, indexNameExpressionResolver, autoCreateIndex);
         this.persistentTasksService = persistentTasksService;
         this.reindexIndexClient = new ReindexIndexClient(client, clusterService, xContentRegistry);
@@ -87,7 +103,7 @@ protected void doExecute(Task task, StartReindexTaskAction.Request request, Acti
 
         // In the current implementation, we only need to store task results if we do not wait for completion
         boolean storeTaskResult = request.getWaitForCompletion() == false;
-        ReindexTaskParams job = new ReindexTaskParams(storeTaskResult, included);
+        ReindexTaskParams job = new ReindexTaskParams(storeTaskResult, resolveIndexPatterns(request.getReindexRequest()), included);
 
         ReindexTaskStateDoc reindexState = new ReindexTaskStateDoc(request.getReindexRequest());
         reindexIndexClient.createReindexTaskDoc(generatedId, reindexState, new ActionListener<>() {
@@ -212,4 +228,164 @@ private boolean isDone(ReindexPersistentTaskState state) {
             return state != null && state.isDone();
         }
     }
+
+    /**
+     * Resolve index patterns to ensure they do not start resolving differently during reindex failovers.
+     * Do not resolve aliases, since accessing the underlying indices is not semantically equivalent to accessing the alias.
+     * Within each index pattern, sort the resolved indices by create date, since this ensures that if we reindex from a pattern of indices,
+     * destination will receive oldest data first. This is in particular important if destination does rollover and it is time-based data.
+     *
+     * @return list of groups of indices/aliases that must be searched together.
+     */
+    private List<Set<String>> resolveIndexPatterns(ReindexRequest request) {
+        return resolveIndexPatterns(request, clusterService.state(), indexNameExpressionResolver);
+    }
+
+    // visible for testing
+    static List<Set<String>> resolveIndexPatterns(ReindexRequest request, ClusterState clusterState,
+                                                          IndexNameExpressionResolver indexNameResolver) {
+        if (request.getRemoteInfo() == null) {
+            return resolveIndexPatterns(request.getSearchRequest().indices(), clusterState, indexNameResolver);
+        } else {
+            return Collections.emptyList();
+        }
+    }
+
+    private static List<Set<String>> resolveIndexPatterns(String[] indices, ClusterState clusterState,
+                                                    IndexNameExpressionResolver indexNameResolver) {
+        Set<String> resolvedNames = indexNameResolver.resolveExpressions(clusterState, indices);
+
+        List<IndexGroup> groups = Arrays.stream(indices)
+            .flatMap(expression -> resolveSingleIndexExpression(expression, resolvedNames::contains,clusterState, indexNameResolver))
+            .collect(Collectors.toList());
+
+        return resolveGroups(groups).stream().map(IndexGroup::newResolvedGroup).collect(Collectors.toList());
+    }
+
+    private static List<IndexGroup> resolveGroups(List<IndexGroup> groups) {
+        List<IndexGroup> result = new ArrayList<>(groups);
+
+        // n^2, but OK since data volume is low.
+        // reverse order since we bubble data towards the lower index end.
+        for (int i = result.size() - 1; i >= 0; --i) {
+            IndexGroup current = result.get(i);
+            for (int j = i - 1; current != null && j >= 0; --j) {
+                IndexGroup earlier = result.get(j);
+                Tuple<IndexGroup, IndexGroup> collapsed = earlier.collapse(current);
+                result.set(j, collapsed.v1());
+                current = collapsed.v2();
+            }
+            result.set(i, current);
+        }
+
+        return result.stream().filter(Objects::nonNull).collect(Collectors.toList());
+    }
+
+    private static Stream<IndexGroup> resolveSingleIndexExpression(String expression, Predicate<String> predicate,
+                                                            ClusterState clusterState,
+                                                            IndexNameExpressionResolver indexNameExpressionResolver) {
+        SortedMap<String, AliasOrIndex> lookup = clusterState.getMetaData().getAliasAndIndexLookup();
+        Comparator<AliasOrIndex> createDateIndexOrder = (i1, i2) -> {
+            if (i1.isAlias() && i2.isAlias()) {
+                return ((AliasOrIndex.Alias) i1).getAliasName().compareTo(((AliasOrIndex.Alias) i2).getAliasName());
+            }
+            if (i1.isAlias() != i2.isAlias()) {
+                return Boolean.compare(i1.isAlias(), i2.isAlias());
+            }
+
+            assert i1.getIndices().size() == 1;
+            assert i2.getIndices().size() == 1;
+            IndexMetaData indexMetaData1 = i1.getIndices().get(0);
+            IndexMetaData indexMetaData2 = i2.getIndices().get(0);
+            int compare = Long.compare(indexMetaData1.getCreationDate(), indexMetaData2.getCreationDate());
+            return compare != 0 ? compare : indexMetaData1.getIndex().getName().compareTo(indexMetaData2.getIndex().getName());
+        };
+
+        return indexNameExpressionResolver.resolveExpressions(clusterState, expression).stream()
+            .filter(predicate).map(lookup::get).sorted(createDateIndexOrder).map(IndexGroup::create);
+    }
+
+    /**
+     * Immutable group of indices and aliases.
+     */
+    private static class IndexGroup {
+        private final Set<String> indices;
+        private final Set<String> allIndices;
+        private final Set<AliasOrIndex.Alias> aliases;
+        private final List<String> orderedIndices;
+
+        private IndexGroup(List<String> indices, Set<AliasOrIndex.Alias> aliases) {
+            orderedIndices = indices;
+            this.indices = indices.stream().collect(Collectors.toUnmodifiableSet());
+            this.aliases = aliases;
+            this.allIndices = Stream.concat(indices.stream(),
+                aliases.stream().flatMap(aliasOrIndex -> aliasOrIndex.getIndices().stream())
+                    .map(imd -> imd.getIndex().getName())
+            ).collect(Collectors.toUnmodifiableSet());
+        }
+
+        private IndexGroup(IndexGroup group1, IndexGroup group2) {
+            this(Stream.concat(group1.orderedIndices.stream(), group2.orderedIndices.stream()).collect(Collectors.toList()),
+                Stream.concat(group1.aliases.stream(), group2.aliases.stream())
+                    .collect(Collectors.toSet()));
+        }
+
+        public static IndexGroup create(AliasOrIndex aliasOrIndex) {
+            if (aliasOrIndex.isAlias()) {
+                return new IndexGroup(Collections.emptyList(), Collections.singleton((AliasOrIndex.Alias) aliasOrIndex));
+            } else {
+                return new IndexGroup(Collections.singletonList(aliasOrIndex.getIndices().get(0).getIndex().getName()),
+                    Collections.emptySet());
+            }
+        }
+
+        private Tuple<IndexGroup, IndexGroup> collapse(IndexGroup other) {
+            if (other == this) {
+                return Tuple.tuple(this, this);
+            }
+
+            if (aliasOverlap(this.aliases, other.allIndices) || aliasOverlap(other.aliases, this.allIndices)) {
+                return Tuple.tuple(new IndexGroup(this, other), null);
+            }
+
+            Set<String> intersection = Sets.intersection(indices, other.indices);
+            assert intersection.isEmpty() == false || Sets.intersection(allIndices, other.allIndices).isEmpty();
+            assert intersection.isEmpty() || Sets.intersection(allIndices, other.allIndices).isEmpty() == false;
+            return Tuple.tuple(this.add(intersection, orderedIndices), other.subtract(intersection));
+        }
+
+        private IndexGroup add(Set<String> intersection, List<String> order) {
+            if (intersection.isEmpty()) {
+                return this;
+            }
+
+            List<String> indices =
+                Stream.concat(this.orderedIndices.stream(), order.stream().filter(intersection::contains)).collect(Collectors.toList());
+            return new IndexGroup(indices, aliases);
+        }
+
+        private IndexGroup subtract(Set<String> intersection) {
+            if (intersection.isEmpty()) {
+                return this;
+            }
+
+            List<String> indices =
+                this.orderedIndices.stream().filter(Predicate.not(intersection::contains)).collect(Collectors.toList());
+
+            if (indices.isEmpty()) {
+                return null;
+            }
+            return new IndexGroup(indices, aliases);
+        }
+
+        private static boolean aliasOverlap(Set<AliasOrIndex.Alias> aliases, Set<String> indices) {
+            return aliases.stream()
+                .flatMap(aliasOrIndex -> aliasOrIndex.getIndices().stream()).map(imd -> imd.getIndex().getName())
+                .anyMatch(indices::contains);
+        }
+
+        public Set<String> newResolvedGroup() {
+            return Stream.concat(indices.stream(), aliases.stream().map(AliasOrIndex.Alias::getAliasName)).collect(Collectors.toSet());
+        }
+    }
 }