Add split recovery source

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

@@ -185,6 +185,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion ESQL_THREAD_NAME_IN_DRIVER_PROFILE = def(9_027_0_00);
     public static final TransportVersion INFERENCE_CONTEXT = def(9_028_0_00);
     public static final TransportVersion ML_INFERENCE_DEEPSEEK = def(9_029_00_0);
+    public static final TransportVersion SOURCE_PRIMARY_TERM_IN_START_SHARD = def(9_030_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/cluster/action/shard/ShardStateAction.java

@@ -41,6 +41,7 @@
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.TimeValue;
@@ -580,6 +581,31 @@ public void shardStarted(
         );
     }
 
+    public void shardSplit(
+        final ShardRouting shardRouting,
+        final long primaryTerm,
+        final String message,
+        final ShardLongFieldRange timestampRange,
+        final ShardLongFieldRange eventIngestedRange,
+        final long sourcePrimaryTerm,
+        final ActionListener<Void> listener
+    ) {
+        ClusterState currentState = clusterService.state();
+        remoteShardStateUpdateDeduplicator.executeOnce(
+            new StartedShardEntry(
+                shardRouting.shardId(),
+                shardRouting.allocationId().getId(),
+                primaryTerm,
+                message,
+                timestampRange,
+                eventIngestedRange,
+                new ShardSplit(sourcePrimaryTerm)
+            ),
+            listener,
+            (req, l) -> sendShardAction(SHARD_STARTED_ACTION_NAME, currentState, req, l)
+        );
+    }
+
     // TODO: Make this a TransportMasterNodeAction and remove duplication of master failover retrying from upstream code
     private static class ShardStartedTransportHandler implements TransportRequestHandler<StartedShardEntry> {
         private final MasterServiceTaskQueue<StartedShardUpdateTask> taskQueue;
@@ -691,6 +717,12 @@ public ClusterState execute(BatchExecutionContext<StartedShardUpdateTask> batchE
                                 matched
                             );
                             tasksToBeApplied.add(taskContext);
+                        } else if (invalidShardSplit(startedShardEntry, projectId, initialState)) {
+                            logger.debug("{} failing shard started task because split validation failed", startedShardEntry.shardId);
+                            // TODO: Currently invalid shard split triggers if the primary term changes, the source primary term changes or
+                            // is >= the target primary term or if the source is relocating. In the second and third scenario this will be
+                            // swallow currently. In the split process we will need to handle this.
+                            taskContext.success(task::onSuccess);
                         } else {
                             logger.debug(
                                 "{} starting shard {} (shard started task: [{}])",
@@ -789,6 +821,30 @@ public ClusterState execute(BatchExecutionContext<StartedShardUpdateTask> batchE
             return maybeUpdatedState;
         }
 
+        private static boolean invalidShardSplit(StartedShardEntry startedShardEntry, ProjectId projectId, ClusterState clusterState) {
+            ShardSplit shardSplit = startedShardEntry.shardSplit;
+            if (shardSplit == null) {
+                return false;
+            }
+            IndexRoutingTable routingTable = clusterState.routingTable(projectId).index(startedShardEntry.shardId.getIndex());
+            // TODO: Splits only double atm. However, eventually there will be a reshard object in the index metadatata indicate the
+            // split specifics
+            int sourceShardId = startedShardEntry.shardId.getId() % (routingTable.size() / 2);
+            final IndexMetadata indexMetadata = clusterState.metadata().getProject(projectId).index(startedShardEntry.shardId.getIndex());
+            assert indexMetadata != null;
+            long currentSourcePrimaryTerm = indexMetadata.primaryTerm(sourceShardId);
+            long primaryTermDiff = startedShardEntry.primaryTerm - currentSourcePrimaryTerm;
+            // The source primary term must not have changed, the target primary term must at least be 1 greater and the source cannot be
+            // relocating.
+            if (startedShardEntry.shardSplit.sourcePrimaryTerm() != currentSourcePrimaryTerm
+                || primaryTermDiff < 1
+                || routingTable.shard(sourceShardId).primaryShard().relocating()) {
+                return true;
+            } else {
+                return false;
+            }
+        }
+
         private static boolean assertStartedIndicesHaveCompleteTimestampRanges(ClusterState clusterState) {
             for (ProjectId projectId : clusterState.metadata().projects().keySet()) {
                 for (Map.Entry<String, IndexRoutingTable> cursor : clusterState.routingTable(projectId).getIndicesRouting().entrySet()) {
@@ -827,13 +883,26 @@ public void clusterStatePublished(ClusterState newClusterState) {
         }
     }
 
+    record ShardSplit(long sourcePrimaryTerm) implements Writeable {
+
+        ShardSplit(StreamInput in) throws IOException {
+            this(in.readVLong());
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            out.writeVLong(sourcePrimaryTerm);
+        }
+    }
+
     public static class StartedShardEntry extends TransportRequest {
         final ShardId shardId;
         final String allocationId;
         final long primaryTerm;
         final String message;
         final ShardLongFieldRange timestampRange;
         final ShardLongFieldRange eventIngestedRange;
+        final ShardSplit shardSplit;
 
         StartedShardEntry(StreamInput in) throws IOException {
             super(in);
@@ -847,6 +916,11 @@ public static class StartedShardEntry extends TransportRequest {
             } else {
                 this.eventIngestedRange = ShardLongFieldRange.UNKNOWN;
             }
+            if (in.getTransportVersion().onOrAfter(TransportVersions.SOURCE_PRIMARY_TERM_IN_START_SHARD)) {
+                this.shardSplit = in.readOptionalWriteable(ShardSplit::new);
+            } else {
+                this.shardSplit = null;
+            }
         }
 
         public StartedShardEntry(
@@ -856,13 +930,26 @@ public StartedShardEntry(
             final String message,
             final ShardLongFieldRange timestampRange,
             final ShardLongFieldRange eventIngestedRange
+        ) {
+            this(shardId, allocationId, primaryTerm, message, timestampRange, eventIngestedRange, null);
+        }
+
+        public StartedShardEntry(
+            final ShardId shardId,
+            final String allocationId,
+            final long primaryTerm,
+            final String message,
+            final ShardLongFieldRange timestampRange,
+            final ShardLongFieldRange eventIngestedRange,
+            @Nullable final ShardSplit shardSplit
         ) {
             this.shardId = shardId;
             this.allocationId = allocationId;
             this.primaryTerm = primaryTerm;
             this.message = message;
             this.timestampRange = timestampRange;
             this.eventIngestedRange = eventIngestedRange;
+            this.shardSplit = shardSplit;
         }
 
         @Override
@@ -876,6 +963,9 @@ public void writeTo(StreamOutput out) throws IOException {
             if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_15_0)) {
                 eventIngestedRange.writeTo(out);
             }
+            if (out.getTransportVersion().onOrAfter(TransportVersions.SOURCE_PRIMARY_TERM_IN_START_SHARD)) {
+                out.writeOptionalWriteable(shardSplit);
+            }
         }
 
         @Override
@@ -891,20 +981,20 @@ public String toString() {
 
         @Override
         public boolean equals(Object o) {
-            if (this == o) return true;
             if (o == null || getClass() != o.getClass()) return false;
             StartedShardEntry that = (StartedShardEntry) o;
             return primaryTerm == that.primaryTerm
-                && shardId.equals(that.shardId)
-                && allocationId.equals(that.allocationId)
-                && message.equals(that.message)
-                && timestampRange.equals(that.timestampRange)
-                && eventIngestedRange.equals(that.eventIngestedRange);
+                && Objects.equals(shardId, that.shardId)
+                && Objects.equals(allocationId, that.allocationId)
+                && Objects.equals(message, that.message)
+                && Objects.equals(timestampRange, that.timestampRange)
+                && Objects.equals(eventIngestedRange, that.eventIngestedRange)
+                && Objects.equals(shardSplit, that.shardSplit);
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(shardId, allocationId, primaryTerm, message, timestampRange, eventIngestedRange);
+            return Objects.hash(shardId, allocationId, primaryTerm, message, timestampRange, eventIngestedRange, shardSplit);
         }
     }
 
@@ -946,7 +1036,5 @@ public NoLongerPrimaryShardException(ShardId shardId, String msg) {
         public NoLongerPrimaryShardException(StreamInput in) throws IOException {
             super(in);
         }
-
     }
-
 }

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

@@ -893,8 +893,18 @@ public IndexMetadata withInSyncAllocationIds(int shardId, Set<String> inSyncSet)
      * @return updated instance with incremented primary term
      */
     public IndexMetadata withIncrementedPrimaryTerm(int shardId) {
+        return withSetPrimaryTerm(shardId, this.primaryTerms[shardId] + 1);
+    }
+
+    /**
+     * Creates a copy of this instance that has the primary term for the given shard id set to the value provided.
+     * @param shardId shard id to set primary term for
+     * @param primaryTerm primary term to set
+     * @return updated instance with set primary term
+     */
+    public IndexMetadata withSetPrimaryTerm(int shardId, long primaryTerm) {
         final long[] incremented = this.primaryTerms.clone();
-        incremented[shardId]++;
+        incremented[shardId] = primaryTerm;
         return new IndexMetadata(
             this.index,
             this.version,

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

@@ -58,6 +58,10 @@ public static boolean shouldReserveSpaceForInitializingShard(ShardRouting shard,
             // shrink/split/clone operation is going to clone existing locally placed shards using file system hard links
             // so no additional space is going to be used until future merges
             case LOCAL_SHARDS -> false;
+
+            // Split currently does not require space locally as it maps to existing store. Future implementations might require the space
+            // locally.
+            case SPLIT -> false;
         };
     }
 

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

@@ -57,6 +57,7 @@ public static RecoverySource readFrom(StreamInput in) throws IOException {
             case PEER -> PeerRecoverySource.INSTANCE;
             case SNAPSHOT -> new SnapshotRecoverySource(in);
             case LOCAL_SHARDS -> LocalShardsRecoverySource.INSTANCE;
+            case SPLIT -> SplitRecoverySource.INSTANCE;
         };
     }
 
@@ -78,7 +79,8 @@ public enum Type {
         EXISTING_STORE,
         PEER,
         SNAPSHOT,
-        LOCAL_SHARDS
+        LOCAL_SHARDS,
+        SPLIT
     }
 
     public abstract Type getType();
@@ -319,4 +321,36 @@ public boolean expectEmptyRetentionLeases() {
             return false;
         }
     }
+
+    /**
+     * peer recovery from a primary shard
+     */
+    public static class SplitRecoverySource extends RecoverySource {
+
+        public static final SplitRecoverySource INSTANCE = new SplitRecoverySource();
+
+        private SplitRecoverySource() {}
+
+        @Override
+        public Type getType() {
+            return Type.SPLIT;
+        }
+
+        @Override
+        public String toString() {
+            return "split recovery";
+        }
+
+        @Override
+        protected void writeAdditionalFields(StreamOutput out) throws IOException {
+            super.writeAdditionalFields(out);
+        }
+
+        @Override
+        public void addAdditionalFields(XContentBuilder builder, Params params) throws IOException {
+            super.addAdditionalFields(builder, params);
+        }
+
+        // TODO: Expect empty retention leases?
+    }
 }

server/src/main/java/org/elasticsearch/cluster/routing/allocation/IndexMetadataUpdater.java

@@ -15,6 +15,7 @@
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.metadata.ProjectMetadata;
 import org.elasticsearch.cluster.routing.GlobalRoutingTable;
+import org.elasticsearch.cluster.routing.IndexRoutingTable;
 import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
 import org.elasticsearch.cluster.routing.RecoverySource;
 import org.elasticsearch.cluster.routing.RoutingChangesObserver;
@@ -127,15 +128,18 @@ public Metadata applyChanges(Metadata oldMetadata, GlobalRoutingTable newRouting
                 for (Map.Entry<ShardId, Updates> shardEntry : indexChanges) {
                     ShardId shardId = shardEntry.getKey();
                     Updates updates = shardEntry.getValue();
-                    updatedIndexMetadata = updateInSyncAllocations(
-                        newRoutingTable.routingTable(projectMetadata.id()),
-                        oldIndexMetadata,
-                        updatedIndexMetadata,
-                        shardId,
-                        updates
-                    );
+                    RoutingTable routingTable = newRoutingTable.routingTable(projectMetadata.id());
+                    updatedIndexMetadata = updateInSyncAllocations(routingTable, oldIndexMetadata, updatedIndexMetadata, shardId, updates);
+                    IndexRoutingTable indexRoutingTable = routingTable.index(shardEntry.getKey().getIndex());
+                    RecoverySource recoverySource = indexRoutingTable.shard(shardEntry.getKey().id()).primaryShard().recoverySource();
+                    boolean split = recoverySource != null && recoverySource.getType() == RecoverySource.Type.SPLIT;
                     updatedIndexMetadata = updates.increaseTerm
-                        ? updatedIndexMetadata.withIncrementedPrimaryTerm(shardId.id())
+                        ? split
+                            ? updatedIndexMetadata.withSetPrimaryTerm(
+                                shardId.id(),
+                                splitPrimaryTerm(updatedIndexMetadata, shardId, indexRoutingTable.size())
+                            )
+                            : updatedIndexMetadata.withIncrementedPrimaryTerm(shardId.id())
                         : updatedIndexMetadata;
                 }
                 if (updatedIndexMetadata != oldIndexMetadata) {
@@ -147,6 +151,18 @@ public Metadata applyChanges(Metadata oldMetadata, GlobalRoutingTable newRouting
         return updatedMetadata.build();
     }
 
+    private static long splitPrimaryTerm(IndexMetadata updatedIndexMetadata, ShardId shardId, int shardCount) {
+        // TODO: Splits only double atm. However, eventually there will be a reshard object in the index metadatata indicate the
+        // split specifics
+
+        // We take the max of the source and target primary terms and increment by 1. This guarantees that the target primary term stays
+        // greater than the source.
+        return Math.max(
+            updatedIndexMetadata.primaryTerm(shardId.getId() % (shardCount / 2)),
+            updatedIndexMetadata.primaryTerm(shardId.id())
+        ) + 1;
+    }
+
     /**
      * Updates in-sync allocations with routing changes that were made to the routing table.
      */

server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -3328,7 +3328,12 @@ public void startRecovery(
         // }
         assert recoveryState.getRecoverySource().equals(shardRouting.recoverySource());
         switch (recoveryState.getRecoverySource().getType()) {
-            case EMPTY_STORE, EXISTING_STORE -> executeRecovery("from store", recoveryState, recoveryListener, this::recoverFromStore);
+            case EMPTY_STORE, EXISTING_STORE, SPLIT -> executeRecovery(
+                "from store",
+                recoveryState,
+                recoveryListener,
+                this::recoverFromStore
+            );
             case PEER -> {
                 try {
                     markAsRecovering("from " + recoveryState.getSourceNode(), recoveryState);

server/src/main/java/org/elasticsearch/index/shard/StoreRecovery.java

@@ -89,8 +89,9 @@ public final class StoreRecovery {
     void recoverFromStore(final IndexShard indexShard, ActionListener<Boolean> listener) {
         if (canRecover(indexShard)) {
             RecoverySource.Type recoveryType = indexShard.recoveryState().getRecoverySource().getType();
-            assert recoveryType == RecoverySource.Type.EMPTY_STORE || recoveryType == RecoverySource.Type.EXISTING_STORE
-                : "expected store recovery type but was: " + recoveryType;
+            assert recoveryType == RecoverySource.Type.EMPTY_STORE
+                || recoveryType == RecoverySource.Type.EXISTING_STORE
+                || recoveryType == RecoverySource.Type.SPLIT : "expected store recovery type but was: " + recoveryType;
             logger.debug("starting recovery from store ...");
             final var recoveryListener = recoveryListener(indexShard, listener);
             try {