Resharding - Adding shards to an existing index (elastic#121082)

This POC attempts to double the number of shards of an existing index.

Author: ankikuma

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -183,7 +183,8 @@ static TransportVersion def(int id) {
  public static final TransportVersion STORED_SCRIPT_CONTENT_LENGTH = def(9_019_0_00);
  public static final TransportVersion JINA_AI_EMBEDDING_TYPE_SUPPORT_ADDED = def(9_020_0_00);
  public static final TransportVersion RE_REMOVE_MIN_COMPATIBLE_SHARD_NODE = def(9_021_0_00);
- public static final TransportVersion INCLUDE_INDEX_MODE_IN_GET_DATA_STREAM = def(9_022_0_00);
+ public static final TransportVersion UNASSIGENEDINFO_RESHARD_ADDED = def(9_022_0_00);
+ public static final TransportVersion INCLUDE_INDEX_MODE_IN_GET_DATA_STREAM = def(9_023_0_00);
 
  /*
  * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/cluster/metadata/IndexMetadata.java

@@ -1947,6 +1947,39 @@ public Builder numberOfShards(int numberOfShards) {
  return this;
  }
 
+ /**
+ * Builder to create IndexMetadata that has an increased shard count (used for re-shard).
+ * The new shard count must be a multiple of the original shardcount.
+ * We do not support shrinking the shard count.
+ * @param shardCount updated shardCount
+ *
+ * TODO: Check if this.version needs to be incremented
+ */
+ public Builder reshardAddShards(int shardCount) {
+ // Assert routingNumShards is null ?
+ // Assert numberOfShards > 0
+ if (shardCount % numberOfShards() != 0) {
+ throw new IllegalArgumentException(
+ "New shard count ["
+ + shardCount
+ + "] should be a multiple"
+ + " of current shard count ["
+ + numberOfShards()
+ + "] for ["
+ + index
+ + "]"
+ );
+ }
+ IndexVersion indexVersionCreated = indexCreatedVersion(settings);
+ settings = Settings.builder().put(settings).put(SETTING_NUMBER_OF_SHARDS, shardCount).build();
+ var newPrimaryTerms = new long[shardCount];
+ Arrays.fill(newPrimaryTerms, this.primaryTerms.length, newPrimaryTerms.length, SequenceNumbers.UNASSIGNED_PRIMARY_TERM);
+ System.arraycopy(primaryTerms, 0, newPrimaryTerms, 0, this.primaryTerms.length);
+ primaryTerms = newPrimaryTerms;
+ routingNumShards = MetadataCreateIndexService.calculateNumRoutingShards(shardCount, indexVersionCreated);
+ return this;
+ }
+
  /**
  * Sets the number of shards that should be used for routing. This should only be used if the number of shards in
  * an index has changed ie if the index is shrunk.

server/src/main/java/org/elasticsearch/cluster/routing/IndexRoutingTable.java

@@ -662,7 +662,7 @@ public Builder addShard(ShardRouting shard) {
  return this;
  }
 
- void ensureShardArray(int shardCount) {
+ public void ensureShardArray(int shardCount) {
  if (shards == null) {
  shards = new IndexShardRoutingTable.Builder[shardCount];
  } else if (shards.length < shardCount) {

server/src/main/java/org/elasticsearch/cluster/routing/RoutingTable.java

@@ -436,6 +436,14 @@ public Builder(ShardRoutingRoleStrategy shardRoutingRoleStrategy, RoutingTable r
  this.indicesRouting = ImmutableOpenMap.builder(routingTable.indicesRouting);
  }
 
+ public IndexRoutingTable getIndexRoutingTable(String index) {
+ return indicesRouting.get(index);
+ }
+
+ public ShardRoutingRoleStrategy getShardRoutingRoleStrategy() {
+ return shardRoutingRoleStrategy;
+ }
+
  private static void addShard(
  final Map<String, IndexRoutingTable.Builder> indexRoutingTableBuilders,
  final ShardRouting shardRoutingEntry

server/src/main/java/org/elasticsearch/cluster/routing/UnassignedInfo.java

@@ -175,7 +175,11 @@ public enum Reason {
  /**
  * Replica is unpromotable and the primary failed.
  */
- UNPROMOTABLE_REPLICA
+ UNPROMOTABLE_REPLICA,
+ /**
+ * New shard added as part of index re-sharding operation
+ */
+ RESHARD_ADDED
  }
 
  /**
@@ -335,9 +339,14 @@ public void writeTo(StreamOutput out) throws IOException {
  out.writeByte((byte) Reason.NODE_LEFT.ordinal());
  } else if (reason.equals(Reason.UNPROMOTABLE_REPLICA) && out.getTransportVersion().before(VERSION_UNPROMOTABLE_REPLICA_ADDED)) {
  out.writeByte((byte) Reason.PRIMARY_FAILED.ordinal());
- } else {
- out.writeByte((byte) reason.ordinal());
- }
+ } else if (reason.equals(Reason.RESHARD_ADDED)
+ && out.getTransportVersion().before(TransportVersions.UNASSIGENEDINFO_RESHARD_ADDED)) {
+ // We should have protection to ensure we do not reshard in mixed clusters
+ assert false;
+ out.writeByte((byte) Reason.FORCED_EMPTY_PRIMARY.ordinal());
+ } else {
+ out.writeByte((byte) reason.ordinal());
+ }
  out.writeLong(unassignedTimeMillis);
  // Do not serialize unassignedTimeNanos as System.nanoTime() cannot be compared across different JVMs
  out.writeBoolean(delayed);

server/src/test/java/org/elasticsearch/cluster/routing/UnassignedInfoTests.java

@@ -88,7 +88,8 @@ public void testReasonOrdinalOrder() {
  UnassignedInfo.Reason.MANUAL_ALLOCATION,
  UnassignedInfo.Reason.INDEX_CLOSED,
  UnassignedInfo.Reason.NODE_RESTARTING,
- UnassignedInfo.Reason.UNPROMOTABLE_REPLICA };
+ UnassignedInfo.Reason.UNPROMOTABLE_REPLICA,
+ UnassignedInfo.Reason.RESHARD_ADDED };
  for (int i = 0; i < order.length; i++) {
  assertThat(order[i].ordinal(), equalTo(i));
  }