Add ability to split shards (#26931)

This change adds a new `_split` API that allows to split indices into a new
index with a power of two more shards that the source index.  This API works
alongside the `_shrink` API but doesn't require any shard relocation before
indices can be split.

The split operation is conceptually an inverse `_shrink` operation since we
initialize the index with a _syntetic_ number of routing shards that are used
for the consistent hashing at index time. Compared to indices created with
earlier versions this might produce slightly different shard distributions but
has no impact on the per-index backwards compatibility.  For now, the user is
required to prepare an index to be splittable by setting the
`index.number_of_routing_shards` at index creation time.  The setting allows the
user to prepare the index to be splittable in factors of
`index.number_of_routing_shards` ie. if the index is created with
`index.number_of_routing_shards: 16` and `index.number_of_shards: 2` it can be
split into `4, 8, 16` shards. This is an intermediate step until we can make
this the default. This also allows us to safely backport this change to 6.x.

The `_split` operation is implemented internally as a DeleteByQuery on the
lucene level that is executed while the primary shards execute their initial
recovery. Subsequent merges that are triggered due to this operation will not be
executed immediately. All merges will be deferred unti the shards are started
and will then be throttled accordingly.

This change is intended for the 6.1 feature release but will not support pre-6.1
indices to be split unless these indices have been shrunk before. In that case
these indices can be split backwards into their original number of shards.

core/src/main/java/org/elasticsearch/action/ActionModule.java

@@ -128,7 +128,9 @@
 import org.elasticsearch.action.admin.indices.settings.put.UpdateSettingsAction;
 import org.elasticsearch.action.admin.indices.shards.IndicesShardStoresAction;
 import org.elasticsearch.action.admin.indices.shards.TransportIndicesShardStoresAction;
+import org.elasticsearch.action.admin.indices.shrink.ResizeAction;
 import org.elasticsearch.action.admin.indices.shrink.ShrinkAction;
+import org.elasticsearch.action.admin.indices.shrink.TransportResizeAction;
 import org.elasticsearch.action.admin.indices.shrink.TransportShrinkAction;
 import org.elasticsearch.action.admin.indices.stats.IndicesStatsAction;
 import org.elasticsearch.action.admin.indices.stats.TransportIndicesStatsAction;
@@ -181,7 +183,6 @@
 import org.elasticsearch.action.search.TransportMultiSearchAction;
 import org.elasticsearch.action.search.TransportSearchAction;
 import org.elasticsearch.action.search.TransportSearchScrollAction;
-import org.elasticsearch.action.support.ActionFilter;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.AutoCreateIndex;
 import org.elasticsearch.action.support.DestructiveOperations;
@@ -199,7 +200,6 @@
 import org.elasticsearch.common.NamedRegistry;
 import org.elasticsearch.common.inject.AbstractModule;
 import org.elasticsearch.common.inject.multibindings.MapBinder;
-import org.elasticsearch.common.inject.multibindings.Multibinder;
 import org.elasticsearch.common.logging.ESLoggerFactory;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.IndexScopedSettings;
@@ -271,6 +271,7 @@
 import org.elasticsearch.rest.action.admin.indices.RestRefreshAction;
 import org.elasticsearch.rest.action.admin.indices.RestRolloverIndexAction;
 import org.elasticsearch.rest.action.admin.indices.RestShrinkIndexAction;
+import org.elasticsearch.rest.action.admin.indices.RestSplitIndexAction;
 import org.elasticsearch.rest.action.admin.indices.RestSyncedFlushAction;
 import org.elasticsearch.rest.action.admin.indices.RestUpdateSettingsAction;
 import org.elasticsearch.rest.action.admin.indices.RestUpgradeAction;
@@ -324,7 +325,6 @@
 import java.util.function.UnaryOperator;
 import java.util.stream.Collectors;
 
-import static java.util.Collections.unmodifiableList;
 import static java.util.Collections.unmodifiableMap;
 
 /**
@@ -438,6 +438,7 @@ public <Request extends ActionRequest, Response extends ActionResponse> void reg
         actions.register(IndicesShardStoresAction.INSTANCE, TransportIndicesShardStoresAction.class);
         actions.register(CreateIndexAction.INSTANCE, TransportCreateIndexAction.class);
         actions.register(ShrinkAction.INSTANCE, TransportShrinkAction.class);
+        actions.register(ResizeAction.INSTANCE, TransportResizeAction.class);
         actions.register(RolloverAction.INSTANCE, TransportRolloverAction.class);
         actions.register(DeleteIndexAction.INSTANCE, TransportDeleteIndexAction.class);
         actions.register(GetIndexAction.INSTANCE, TransportGetIndexAction.class);
@@ -554,6 +555,7 @@ public void initRestHandlers(Supplier<DiscoveryNodes> nodesInCluster) {
         registerHandler.accept(new RestIndicesAliasesAction(settings, restController));
         registerHandler.accept(new RestCreateIndexAction(settings, restController));
         registerHandler.accept(new RestShrinkIndexAction(settings, restController));
+        registerHandler.accept(new RestSplitIndexAction(settings, restController));
         registerHandler.accept(new RestRolloverIndexAction(settings, restController));
         registerHandler.accept(new RestDeleteIndexAction(settings, restController));
         registerHandler.accept(new RestCloseIndexAction(settings, restController));

core/src/main/java/org/elasticsearch/action/admin/indices/create/CreateIndexClusterStateUpdateRequest.java

@@ -20,6 +20,7 @@
 package org.elasticsearch.action.admin.indices.create;
 
 import org.elasticsearch.action.admin.indices.alias.Alias;
+import org.elasticsearch.action.admin.indices.shrink.ResizeType;
 import org.elasticsearch.action.support.ActiveShardCount;
 import org.elasticsearch.cluster.ack.ClusterStateUpdateRequest;
 import org.elasticsearch.cluster.block.ClusterBlock;
@@ -43,7 +44,8 @@ public class CreateIndexClusterStateUpdateRequest extends ClusterStateUpdateRequ
     private final String index;
     private final String providedName;
     private final boolean updateAllTypes;
-    private Index shrinkFrom;
+    private Index recoverFrom;
+    private ResizeType resizeType;
 
     private IndexMetaData.State state = IndexMetaData.State.OPEN;
 
@@ -59,7 +61,6 @@ public class CreateIndexClusterStateUpdateRequest extends ClusterStateUpdateRequ
 
     private ActiveShardCount waitForActiveShards = ActiveShardCount.DEFAULT;
 
-
     public CreateIndexClusterStateUpdateRequest(TransportMessage originalMessage, String cause, String index, String providedName,
                                                 boolean updateAllTypes) {
         this.originalMessage = originalMessage;
@@ -99,8 +100,8 @@ public CreateIndexClusterStateUpdateRequest state(IndexMetaData.State state) {
         return this;
     }
 
-    public CreateIndexClusterStateUpdateRequest shrinkFrom(Index shrinkFrom) {
-        this.shrinkFrom = shrinkFrom;
+    public CreateIndexClusterStateUpdateRequest recoverFrom(Index recoverFrom) {
+        this.recoverFrom = recoverFrom;
         return this;
     }
 
@@ -109,6 +110,11 @@ public CreateIndexClusterStateUpdateRequest waitForActiveShards(ActiveShardCount
         return this;
     }
 
+    public CreateIndexClusterStateUpdateRequest resizeType(ResizeType resizeType) {
+        this.resizeType = resizeType;
+        return this;
+    }
+
     public TransportMessage originalMessage() {
         return originalMessage;
     }
@@ -145,8 +151,8 @@ public Set<ClusterBlock> blocks() {
         return blocks;
     }
 
-    public Index shrinkFrom() {
-        return shrinkFrom;
+    public Index recoverFrom() {
+        return recoverFrom;
     }
 
     /** True if all fields that span multiple types should be updated, false otherwise */
@@ -165,4 +171,11 @@ public String getProvidedName() {
     public ActiveShardCount waitForActiveShards() {
         return waitForActiveShards;
     }
+
+    /**
+     * Returns the resize type or null if this is an ordinary create index request
+     */
+    public ResizeType resizeType() {
+        return resizeType;
+    }
 }

core/src/main/java/org/elasticsearch/action/admin/indices/shrink/ResizeAction.java

@@ -0,0 +1,45 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.action.admin.indices.shrink;
+
+import org.elasticsearch.Version;
+import org.elasticsearch.action.Action;
+import org.elasticsearch.client.ElasticsearchClient;
+
+public class ResizeAction extends Action<ResizeRequest, ResizeResponse, ResizeRequestBuilder> {
+
+    public static final ResizeAction INSTANCE = new ResizeAction();
+    public static final String NAME = "indices:admin/resize";
+    public static final Version COMPATIBILITY_VERSION = Version.V_7_0_0_alpha1; // TODO remove this once it's backported
+
+    private ResizeAction() {
+        super(NAME);
+    }
+
+    @Override
+    public ResizeResponse newResponse() {
+        return new ResizeResponse();
+    }
+
+    @Override
+    public ResizeRequestBuilder newRequestBuilder(ElasticsearchClient client) {
+        return new ResizeRequestBuilder(client, this);
+    }
+}

core/src/main/java/org/elasticsearch/action/admin/indices/shrink/ShrinkRequest.java → core/src/main/java/org/elasticsearch/action/admin/indices/shrink/ResizeRequest.java

@@ -18,12 +18,14 @@
  */
 package org.elasticsearch.action.admin.indices.shrink;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.IndicesRequest;
 import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
 import org.elasticsearch.action.support.ActiveShardCount;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.action.support.master.AcknowledgedRequest;
+import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.common.ParseField;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -37,37 +39,41 @@
 /**
  * Request class to shrink an index into a single shard
  */
-public class ShrinkRequest extends AcknowledgedRequest<ShrinkRequest> implements IndicesRequest {
+public class ResizeRequest extends AcknowledgedRequest<ResizeRequest> implements IndicesRequest {
 
-    public static final ObjectParser<ShrinkRequest, Void> PARSER = new ObjectParser<>("shrink_request", null);
+    public static final ObjectParser<ResizeRequest, Void> PARSER = new ObjectParser<>("resize_request", null);
     static {
-        PARSER.declareField((parser, request, context) -> request.getShrinkIndexRequest().settings(parser.map()),
+        PARSER.declareField((parser, request, context) -> request.getTargetIndexRequest().settings(parser.map()),
             new ParseField("settings"), ObjectParser.ValueType.OBJECT);
-        PARSER.declareField((parser, request, context) -> request.getShrinkIndexRequest().aliases(parser.map()),
+        PARSER.declareField((parser, request, context) -> request.getTargetIndexRequest().aliases(parser.map()),
             new ParseField("aliases"), ObjectParser.ValueType.OBJECT);
     }
 
-    private CreateIndexRequest shrinkIndexRequest;
+    private CreateIndexRequest targetIndexRequest;
     private String sourceIndex;
+    private ResizeType type = ResizeType.SHRINK;
 
-    ShrinkRequest() {}
+    ResizeRequest() {}
 
-    public ShrinkRequest(String targetIndex, String sourceindex) {
-        this.shrinkIndexRequest = new CreateIndexRequest(targetIndex);
-        this.sourceIndex = sourceindex;
+    public ResizeRequest(String targetIndex, String sourceIndex) {
+        this.targetIndexRequest = new CreateIndexRequest(targetIndex);
+        this.sourceIndex = sourceIndex;
     }
 
     @Override
     public ActionRequestValidationException validate() {
-        ActionRequestValidationException validationException = shrinkIndexRequest == null ? null : shrinkIndexRequest.validate();
+        ActionRequestValidationException validationException = targetIndexRequest == null ? null : targetIndexRequest.validate();
         if (sourceIndex == null) {
             validationException = addValidationError("source index is missing", validationException);
         }
-        if (shrinkIndexRequest == null) {
-            validationException = addValidationError("shrink index request is missing", validationException);
+        if (targetIndexRequest == null) {
+            validationException = addValidationError("target index request is missing", validationException);
         }
-        if (shrinkIndexRequest.settings().getByPrefix("index.sort.").isEmpty() == false) {
-            validationException = addValidationError("can't override index sort when shrinking index", validationException);
+        if (targetIndexRequest.settings().getByPrefix("index.sort.").isEmpty() == false) {
+            validationException = addValidationError("can't override index sort when resizing an index", validationException);
+        }
+        if (type == ResizeType.SPLIT && IndexMetaData.INDEX_NUMBER_OF_SHARDS_SETTING.exists(targetIndexRequest.settings()) == false) {
+            validationException = addValidationError("index.number_of_shards is required for split operations", validationException);
         }
         return validationException;
     }
@@ -79,16 +85,24 @@ public void setSourceIndex(String index) {
     @Override
     public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
-        shrinkIndexRequest = new CreateIndexRequest();
-        shrinkIndexRequest.readFrom(in);
+        targetIndexRequest = new CreateIndexRequest();
+        targetIndexRequest.readFrom(in);
         sourceIndex = in.readString();
+        if (in.getVersion().onOrAfter(ResizeAction.COMPATIBILITY_VERSION)) {
+            type = in.readEnum(ResizeType.class);
+        } else {
+            type = ResizeType.SHRINK; // BWC this used to be shrink only
+        }
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
-        shrinkIndexRequest.writeTo(out);
+        targetIndexRequest.writeTo(out);
         out.writeString(sourceIndex);
+        if (out.getVersion().onOrAfter(ResizeAction.COMPATIBILITY_VERSION)) {
+            out.writeEnum(type);
+        }
     }
 
     @Override
@@ -101,15 +115,15 @@ public IndicesOptions indicesOptions() {
         return IndicesOptions.lenientExpandOpen();
     }
 
-    public void setShrinkIndex(CreateIndexRequest shrinkIndexRequest) {
-        this.shrinkIndexRequest = Objects.requireNonNull(shrinkIndexRequest, "shrink index request must not be null");
+    public void setTargetIndex(CreateIndexRequest targetIndexRequest) {
+        this.targetIndexRequest = Objects.requireNonNull(targetIndexRequest, "target index request must not be null");
     }
 
     /**
      * Returns the {@link CreateIndexRequest} for the shrink index
      */
-    public CreateIndexRequest getShrinkIndexRequest() {
-        return shrinkIndexRequest;
+    public CreateIndexRequest getTargetIndexRequest() {
+        return targetIndexRequest;
     }
 
     /**
@@ -128,13 +142,13 @@ public String getSourceIndex() {
      * non-negative integer, up to the number of copies per shard (number of replicas + 1),
      * to wait for the desired amount of shard copies to become active before returning.
      * Index creation will only wait up until the timeout value for the number of shard copies
-     * to be active before returning.  Check {@link ShrinkResponse#isShardsAcked()} to
+     * to be active before returning.  Check {@link ResizeResponse#isShardsAcked()} to
      * determine if the requisite shard copies were all started before returning or timing out.
      *
      * @param waitForActiveShards number of active shard copies to wait on
      */
     public void setWaitForActiveShards(ActiveShardCount waitForActiveShards) {
-        this.getShrinkIndexRequest().waitForActiveShards(waitForActiveShards);
+        this.getTargetIndexRequest().waitForActiveShards(waitForActiveShards);
     }
 
     /**
@@ -145,4 +159,18 @@ public void setWaitForActiveShards(ActiveShardCount waitForActiveShards) {
     public void setWaitForActiveShards(final int waitForActiveShards) {
         setWaitForActiveShards(ActiveShardCount.from(waitForActiveShards));
     }
+
+    /**
+     * The type of the resize operation
+     */
+    public void setResizeType(ResizeType type) {
+        this.type = Objects.requireNonNull(type);
+    }
+
+    /**
+     * Returns the type of the resize operation
+     */
+    public ResizeType getResizeType() {
+        return type;
+    }
 }