Expose indexing plans

This is an additional engine refactoring to expose indexing plans.

Relates #27375

Author: jasontedor

core/src/main/java/org/elasticsearch/index/engine/InternalEngine.java

@@ -72,6 +72,7 @@
 import org.elasticsearch.index.seqno.SequenceNumbers;
 import org.elasticsearch.index.seqno.SequenceNumbersService;
 import org.elasticsearch.index.shard.ElasticsearchMergePolicy;
+import org.elasticsearch.index.shard.IndexingStats;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.translog.Translog;
 import org.elasticsearch.index.translog.TranslogConfig;
@@ -680,8 +681,7 @@ private boolean canOptimizeAddDocument(Index index) {
                 + index.getAutoGeneratedIdTimestamp();
             switch (index.origin()) {
                 case PRIMARY:
-                    assert (index.version() == Versions.MATCH_ANY && index.versionType() == VersionType.INTERNAL)
-                        : "version: " + index.version() + " type: " + index.versionType();
+                    assertPrimaryCanOptimizeAddDocument(index);
                     return true;
                 case PEER_RECOVERY:
                 case REPLICA:
@@ -698,6 +698,12 @@ private boolean canOptimizeAddDocument(Index index) {
         return false;
     }
 
+    protected boolean assertPrimaryCanOptimizeAddDocument(final Index index) {
+        assert (index.version() == Versions.MATCH_ANY && index.versionType() == VersionType.INTERNAL)
+                : "version: " + index.version() + " type: " + index.versionType();
+        return true;
+    }
+
     private boolean assertVersionType(final Engine.Operation operation) {
         if (operation.origin() == Operation.Origin.REPLICA ||
                 operation.origin() == Operation.Origin.PEER_RECOVERY ||
@@ -718,16 +724,21 @@ private boolean assertIncomingSequenceNumber(final Engine.Operation.Origin origi
             assert seqNo == SequenceNumbers.UNASSIGNED_SEQ_NO : "old op recovering but it already has a seq no.;" +
                 " index version: " + engineConfig.getIndexSettings().getIndexVersionCreated() + ", seqNo: " + seqNo;
         } else if (origin == Operation.Origin.PRIMARY) {
-            // sequence number should not be set when operation origin is primary
-            assert seqNo == SequenceNumbers.UNASSIGNED_SEQ_NO
-                    : "primary operations must never have an assigned sequence number but was [" + seqNo + "]";
+            assertPrimaryIncomingSequenceNumber(origin, seqNo);
         } else if (engineConfig.getIndexSettings().getIndexVersionCreated().onOrAfter(Version.V_6_0_0_alpha1)) {
             // sequence number should be set when operation origin is not primary
             assert seqNo >= 0 : "recovery or replica ops should have an assigned seq no.; origin: " + origin;
         }
         return true;
     }
 
+    protected boolean assertPrimaryIncomingSequenceNumber(final Engine.Operation.Origin origin, final long seqNo) {
+        // sequence number should not be set when operation origin is primary
+        assert seqNo == SequenceNumbers.UNASSIGNED_SEQ_NO
+                : "primary operations must never have an assigned sequence number but was [" + seqNo + "]";
+        return true;
+    }
+
     private boolean assertSequenceNumberBeforeIndexing(final Engine.Operation.Origin origin, final long seqNo) {
         if (engineConfig.getIndexSettings().getIndexVersionCreated().onOrAfter(Version.V_6_0_0_alpha1) ||
             origin == Operation.Origin.PRIMARY) {
@@ -788,14 +799,7 @@ public IndexResult index(Index index) throws IOException {
                  *  if A arrives on the shard first we use addDocument since maxUnsafeAutoIdTimestamp is < 10. A` will then just be skipped or calls
                  *  updateDocument.
                  */
-                final IndexingStrategy plan;
-
-                if (index.origin() == Operation.Origin.PRIMARY) {
-                    plan = planIndexingAsPrimary(index);
-                } else {
-                    // non-primary mode (i.e., replica or recovery)
-                    plan = planIndexingAsNonPrimary(index);
-                }
+                final IndexingStrategy plan = indexingStrategyForOperation(index);
 
                 final IndexResult indexResult;
                 if (plan.earlyResultOnPreFlightError.isPresent()) {
@@ -836,7 +840,8 @@ public IndexResult index(Index index) throws IOException {
         }
     }
 
-    private IndexingStrategy planIndexingAsNonPrimary(Index index) throws IOException {
+    protected final IndexingStrategy planIndexingAsNonPrimary(Index index) throws IOException {
+        assertNonPrimaryOrigin(index);
         final IndexingStrategy plan;
         if (canOptimizeAddDocument(index) && mayHaveBeenIndexedBefore(index) == false) {
             // no need to deal with out of order delivery - we never saw this one
@@ -879,7 +884,16 @@ assert config().getIndexSettings().getIndexVersionCreated().before(Version.V_6_0
         return plan;
     }
 
-    private IndexingStrategy planIndexingAsPrimary(Index index) throws IOException {
+    protected IndexingStrategy indexingStrategyForOperation(final Index index) throws IOException {
+        if (index.origin() == Operation.Origin.PRIMARY) {
+            return planIndexingAsPrimary(index);
+        } else {
+            // non-primary mode (i.e., replica or recovery)
+            return planIndexingAsNonPrimary(index);
+        }
+    }
+
+    protected final IndexingStrategy planIndexingAsPrimary(Index index) throws IOException {
         assert index.origin() == Operation.Origin.PRIMARY : "planing as primary but origin isn't. got " + index.origin();
         final IndexingStrategy plan;
         // resolve an external operation into an internal one which is safe to replay
@@ -995,7 +1009,7 @@ private static void index(final List<ParseContext.Document> docs, final IndexWri
         }
     }
 
-    private static final class IndexingStrategy {
+    protected static final class IndexingStrategy {
         final boolean currentNotFoundOrDeleted;
         final boolean useLuceneUpdateDocument;
         final long seqNoForIndexing;
@@ -1089,12 +1103,7 @@ public DeleteResult delete(Delete delete) throws IOException {
         try (ReleasableLock ignored = readLock.acquire(); Releasable ignored2 = acquireLock(delete.uid())) {
             ensureOpen();
             lastWriteNanos = delete.startTime();
-            final DeletionStrategy plan;
-            if (delete.origin() == Operation.Origin.PRIMARY) {
-                plan = planDeletionAsPrimary(delete);
-            } else {
-                plan = planDeletionAsNonPrimary(delete);
-            }
+            final DeletionStrategy plan = deletionStrategyForOperation(delete);
 
             if (plan.earlyResultOnPreflightError.isPresent()) {
                 deleteResult = plan.earlyResultOnPreflightError.get();
@@ -1133,8 +1142,17 @@ public DeleteResult delete(Delete delete) throws IOException {
         return deleteResult;
     }
 
-    private DeletionStrategy planDeletionAsNonPrimary(Delete delete) throws IOException {
-        assert delete.origin() != Operation.Origin.PRIMARY : "planing as primary but got " + delete.origin();
+    protected DeletionStrategy deletionStrategyForOperation(final Delete delete) throws IOException {
+        if (delete.origin() == Operation.Origin.PRIMARY) {
+            return planDeletionAsPrimary(delete);
+        } else {
+            // non-primary mode (i.e., replica or recovery)
+            return planDeletionAsNonPrimary(delete);
+        }
+    }
+
+    protected final DeletionStrategy planDeletionAsNonPrimary(Delete delete) throws IOException {
+        assertNonPrimaryOrigin(delete);
         // drop out of order operations
         assert delete.versionType().versionTypeForReplicationAndRecovery() == delete.versionType() :
             "resolving out of order delivery based on versioning but version type isn't fit for it. got ["
@@ -1171,7 +1189,12 @@ assert config().getIndexSettings().getIndexVersionCreated().before(Version.V_6_0
         return plan;
     }
 
-    private DeletionStrategy planDeletionAsPrimary(Delete delete) throws IOException {
+    protected boolean assertNonPrimaryOrigin(final Operation operation) {
+        assert operation.origin() != Operation.Origin.PRIMARY : "planing as primary but got " + operation.origin();
+        return true;
+    }
+
+    protected final DeletionStrategy planDeletionAsPrimary(Delete delete) throws IOException {
         assert delete.origin() == Operation.Origin.PRIMARY : "planing as primary but got " + delete.origin();
         // resolve operation from external to internal
         final VersionValue versionValue = resolveDocVersion(delete);
@@ -1222,7 +1245,7 @@ private DeleteResult deleteInLucene(Delete delete, DeletionStrategy plan)
         }
     }
 
-    private static final class DeletionStrategy {
+    protected static final class DeletionStrategy {
         // of a rare double delete
         final boolean deleteFromLucene;
         final boolean currentlyDeleted;