Introducing indexing & deletion strategy planner interfaces

CHANGELOG.md

@@ -35,6 +35,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Added TopN selection logic for streaming terms aggregations ([#20481](https://github.com/opensearch-project/OpenSearch/pull/20481))
 - Added support for Intra Segment Search ([#19704](https://github.com/opensearch-project/OpenSearch/pull/19704))
 - Introduce AdditionalCodecs and EnginePlugin::getAdditionalCodecs hook to allow additional Codec registration ([#20411](https://github.com/opensearch-project/OpenSearch/pull/20411))
+- Introduced strategy planner interfaces for indexing and deletion ([#20585](https://github.com/opensearch-project/OpenSearch/pull/20585))
 
 ### Changed
 - Move Randomness from server to libs/common ([#20570](https://github.com/opensearch-project/OpenSearch/pull/20570))

server/src/main/java/org/opensearch/index/engine/DeletionStrategy.java

@@ -0,0 +1,110 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.index.engine;
+
+import org.opensearch.common.lucene.uid.Versions;
+import org.opensearch.index.seqno.SequenceNumbers;
+
+import java.util.Objects;
+
+/**
+ * The deletion strategy
+ *
+ * @opensearch.internal
+ */
+public final class DeletionStrategy extends OperationStrategy {
+
+    public final boolean currentlyDeleted;
+
+    private DeletionStrategy(
+        boolean deleteFromEngine,
+        boolean addStaleOpToEngine,
+        boolean currentlyDeleted,
+        long versionOfDeletion,
+        int reservedDocs,
+        Engine.DeleteResult earlyResultOnPreflightError
+    ) {
+        super(deleteFromEngine, addStaleOpToEngine, versionOfDeletion, earlyResultOnPreflightError, reservedDocs);
+        assert (deleteFromEngine && earlyResultOnPreflightError != null) == false
+            : "can only delete from engine or have a preflight result but not both."
+                + "deleteFromEngine: "
+                + deleteFromEngine
+                + "  earlyResultOnPreFlightError:"
+                + earlyResultOnPreflightError;
+        assert reservedDocs == 0 || deleteFromEngine || addStaleOpToEngine : reservedDocs;
+        this.currentlyDeleted = currentlyDeleted;
+    }
+
+    static DeletionStrategy skipDueToVersionConflict(VersionConflictEngineException e, long currentVersion, boolean currentlyDeleted) {
+        final Engine.DeleteResult deleteResult = new Engine.DeleteResult(
+            e,
+            currentVersion,
+            SequenceNumbers.UNASSIGNED_PRIMARY_TERM,
+            SequenceNumbers.UNASSIGNED_SEQ_NO,
+            currentlyDeleted == false
+        );
+        return new DeletionStrategy(false, false, currentlyDeleted, Versions.NOT_FOUND, 0, deleteResult);
+    }
+
+    static DeletionStrategy processNormally(boolean currentlyDeleted, long versionOfDeletion, int reservedDocs) {
+        return new DeletionStrategy(true, false, currentlyDeleted, versionOfDeletion, reservedDocs, null);
+
+    }
+
+    static DeletionStrategy processButSkipEngine(boolean currentlyDeleted, long versionOfDeletion) {
+        return new DeletionStrategy(false, false, currentlyDeleted, versionOfDeletion, 0, null);
+    }
+
+    static DeletionStrategy processAsStaleOp(long versionOfDeletion) {
+        return new DeletionStrategy(false, true, false, versionOfDeletion, 0, null);
+    }
+
+    static DeletionStrategy failAsTooManyDocs(Exception e) {
+        final Engine.DeleteResult deleteResult = new Engine.DeleteResult(
+            e,
+            Versions.NOT_FOUND,
+            SequenceNumbers.UNASSIGNED_PRIMARY_TERM,
+            SequenceNumbers.UNASSIGNED_SEQ_NO,
+            false
+        );
+        return new DeletionStrategy(false, false, false, Versions.NOT_FOUND, 0, deleteResult);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        if (!super.equals(o)) return false;
+        DeletionStrategy that = (DeletionStrategy) o;
+        return currentlyDeleted == that.currentlyDeleted;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(super.hashCode(), currentlyDeleted);
+    }
+
+    @Override
+    public String toString() {
+        return "DeletionStrategy{"
+            + "currentlyDeleted="
+            + currentlyDeleted
+            + ", executeOpOnEngine="
+            + executeOpOnEngine
+            + ", addStaleOpToEngine="
+            + addStaleOpToEngine
+            + ", version="
+            + version
+            + ", earlyResultOnPreFlightError="
+            + earlyResultOnPreFlightError
+            + ", reservedDocs="
+            + reservedDocs
+            + '}';
+    }
+}

server/src/main/java/org/opensearch/index/engine/DeletionStrategyPlanner.java

@@ -0,0 +1,146 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.index.engine;
+
+import org.opensearch.common.CheckedBiFunction;
+import org.opensearch.common.CheckedFunction;
+import org.opensearch.common.lucene.uid.Versions;
+import org.opensearch.core.index.shard.ShardId;
+import org.opensearch.index.IndexSettings;
+import org.opensearch.index.seqno.SequenceNumbers;
+
+import java.io.IOException;
+import java.util.function.BiFunction;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+
+/**
+ * Plans execution strategies for deletion operations.
+ * The planner produces {@link DeletionStrategy} instances that guide the engine's
+ * execution of delete operations on both primary and replica shards.
+ *
+ * @opensearch.internal
+ */
+public class DeletionStrategyPlanner implements OperationStrategyPlanner<Engine.Delete, DeletionStrategy> {
+
+    private final IndexSettings indexSettings;
+    private final ShardId shardId;
+    private final Predicate<Engine.Operation> hasBeenProcessedBefore;
+    private final CheckedFunction<Engine.Operation, OpVsEngineDocStatus, IOException> opVsEngineDocStatusFunction;
+    private final CheckedBiFunction<Engine.Operation, Boolean, VersionValue, IOException> docVersionSupplier;
+    private final BiFunction<Engine.Operation, Integer, Exception> tryAcquireInFlightDocs;
+    private final Supplier<Boolean> incrementVersionLookup;
+
+    public DeletionStrategyPlanner(
+        IndexSettings indexSettings,
+        ShardId shardId,
+        Predicate<Engine.Operation> hasBeenProcessedBefore,
+        CheckedFunction<Engine.Operation, OpVsEngineDocStatus, IOException> opVsEngineDocStatusFunction,
+        CheckedBiFunction<Engine.Operation, Boolean, VersionValue, IOException> docVersionSupplier,
+        BiFunction<Engine.Operation, Integer, Exception> tryAcquireInFlightDocs,
+        Supplier<Boolean> incrementVersionLookup
+    ) {
+        this.indexSettings = indexSettings;
+        this.shardId = shardId;
+        this.hasBeenProcessedBefore = hasBeenProcessedBefore;
+        this.opVsEngineDocStatusFunction = opVsEngineDocStatusFunction;
+        this.docVersionSupplier = docVersionSupplier;
+        this.tryAcquireInFlightDocs = tryAcquireInFlightDocs;
+        this.incrementVersionLookup = incrementVersionLookup;
+    }
+
+    @Override
+    public DeletionStrategy planOperationAsPrimary(Engine.Delete delete) throws IOException {
+        assert delete.origin() == Engine.Operation.Origin.PRIMARY : "planing as primary but got " + delete.origin();
+        // resolve operation from external to internal
+        final VersionValue versionValue = docVersionSupplier.apply(delete, delete.getIfSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO);
+        assert incrementVersionLookup.get();
+        final long currentVersion;
+        final boolean currentlyDeleted;
+        if (versionValue == null) {
+            currentVersion = Versions.NOT_FOUND;
+            currentlyDeleted = true;
+        } else {
+            currentVersion = versionValue.version;
+            currentlyDeleted = versionValue.isDelete();
+        }
+        final DeletionStrategy plan;
+        if (delete.getIfSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO && currentlyDeleted) {
+            final VersionConflictEngineException e = new VersionConflictEngineException(
+                shardId,
+                delete.id(),
+                delete.getIfSeqNo(),
+                delete.getIfPrimaryTerm(),
+                SequenceNumbers.UNASSIGNED_SEQ_NO,
+                SequenceNumbers.UNASSIGNED_PRIMARY_TERM
+            );
+            plan = DeletionStrategy.skipDueToVersionConflict(e, currentVersion, true);
+        } else if (delete.getIfSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO
+            && (versionValue.seqNo != delete.getIfSeqNo() || versionValue.term != delete.getIfPrimaryTerm())) {
+                final VersionConflictEngineException e = new VersionConflictEngineException(
+                    shardId,
+                    delete.id(),
+                    delete.getIfSeqNo(),
+                    delete.getIfPrimaryTerm(),
+                    versionValue.seqNo,
+                    versionValue.term
+                );
+                plan = DeletionStrategy.skipDueToVersionConflict(e, currentVersion, currentlyDeleted);
+            } else if (delete.versionType().isVersionConflictForWrites(currentVersion, delete.version(), currentlyDeleted)) {
+                final VersionConflictEngineException e = new VersionConflictEngineException(
+                    shardId,
+                    delete,
+                    currentVersion,
+                    currentlyDeleted
+                );
+                plan = DeletionStrategy.skipDueToVersionConflict(e, currentVersion, currentlyDeleted);
+            } else {
+                final Exception reserveError = tryAcquireInFlightDocs.apply(delete, 1);
+                if (reserveError != null) {
+                    plan = DeletionStrategy.failAsTooManyDocs(reserveError);
+                } else {
+                    final long versionOfDeletion = delete.versionType().updateVersion(currentVersion, delete.version());
+                    plan = DeletionStrategy.processNormally(currentlyDeleted, versionOfDeletion, 1);
+                }
+            }
+        return plan;
+    }
+
+    @Override
+    public DeletionStrategy planOperationAsNonPrimary(Engine.Delete delete) throws IOException {
+        assert assertNonPrimaryOrigin(delete);
+        final DeletionStrategy plan;
+        if (hasBeenProcessedBefore.test(delete)) {
+            // the operation seq# was processed thus this operation was already put into lucene
+            // this can happen during recovery where older operations are sent from the translog that are already
+            // part of the lucene commit (either from a peer recovery or a local translog)
+            // or due to concurrent indexing & recovery. For the former it is important to skip lucene as the operation in
+            // question may have been deleted in an out of order op that is not replayed.
+            // See testRecoverFromStoreWithOutOfOrderDelete for an example of local recovery
+            // See testRecoveryWithOutOfOrderDelete for an example of peer recovery
+            plan = DeletionStrategy.processButSkipEngine(false, delete.version());
+        } else {
+            boolean segRepEnabled = indexSettings.isSegRepEnabledOrRemoteNode();
+            final OpVsEngineDocStatus opVsLucene = opVsEngineDocStatusFunction.apply(delete);
+            if (opVsLucene == OpVsEngineDocStatus.OP_STALE_OR_EQUAL) {
+                if (segRepEnabled) {
+                    // For segrep based indices, we can't completely rely on localCheckpointTracker
+                    // as the preserved checkpoint may not have all the operations present in lucene
+                    // we don't need to index it again as stale op as it would create multiple documents for same seq no
+                    plan = DeletionStrategy.processButSkipEngine(false, delete.version());
+                } else {
+                    plan = DeletionStrategy.processAsStaleOp(delete.version());
+                }
+            } else {
+                plan = DeletionStrategy.processNormally(opVsLucene == OpVsEngineDocStatus.DOC_NOT_FOUND, delete.version(), 0);
+            }
+        }
+        return plan;
+    }
+}

server/src/main/java/org/opensearch/index/engine/IndexingStrategy.java

@@ -0,0 +1,122 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.index.engine;
+
+import org.opensearch.common.lucene.uid.Versions;
+
+import java.util.Objects;
+
+/**
+ * The indexing strategy
+ *
+ * @opensearch.internal
+ */
+public final class IndexingStrategy extends OperationStrategy {
+
+    public final boolean currentNotFoundOrDeleted;
+    public final boolean useUpdateDocument;
+
+    private IndexingStrategy(
+        boolean currentNotFoundOrDeleted,
+        boolean useUpdateDocument,
+        boolean indexIntoEngine,
+        boolean addStaleOpToEngine,
+        long versionForIndexing,
+        int reservedDocs,
+        Engine.IndexResult earlyResultOnPreFlightError
+    ) {
+        super(indexIntoEngine, addStaleOpToEngine, versionForIndexing, earlyResultOnPreFlightError, reservedDocs);
+        assert useUpdateDocument == false || indexIntoEngine : "use update is set to true, but we're not indexing into engine";
+        assert (indexIntoEngine && earlyResultOnPreFlightError != null) == false
+            : "can only index into engine or have a preflight result but not both."
+                + "indexIntoEngine: "
+                + indexIntoEngine
+                + "  earlyResultOnPreFlightError:"
+                + earlyResultOnPreFlightError;
+        assert reservedDocs == 0 || indexIntoEngine || addStaleOpToEngine : reservedDocs;
+        this.currentNotFoundOrDeleted = currentNotFoundOrDeleted;
+        this.useUpdateDocument = useUpdateDocument;
+    }
+
+    static IndexingStrategy optimizedAppendOnly(long versionForIndexing, int reservedDocs) {
+        return new IndexingStrategy(true, false, true, false, versionForIndexing, reservedDocs, null);
+    }
+
+    static IndexingStrategy skipDueToVersionConflict(
+        VersionConflictEngineException e,
+        boolean currentNotFoundOrDeleted,
+        long currentVersion
+    ) {
+        final Engine.IndexResult result = new Engine.IndexResult(e, currentVersion);
+        return new IndexingStrategy(currentNotFoundOrDeleted, false, false, false, Versions.NOT_FOUND, 0, result);
+    }
+
+    static IndexingStrategy processNormally(boolean currentNotFoundOrDeleted, long versionForIndexing, int reservedDocs) {
+        return new IndexingStrategy(
+            currentNotFoundOrDeleted,
+            currentNotFoundOrDeleted == false,
+            true,
+            false,
+            versionForIndexing,
+            reservedDocs,
+            null
+        );
+    }
+
+    static IndexingStrategy processButSkipEngine(boolean currentNotFoundOrDeleted, long versionForIndexing) {
+        return new IndexingStrategy(currentNotFoundOrDeleted, false, false, false, versionForIndexing, 0, null);
+    }
+
+    static IndexingStrategy processAsStaleOp(long versionForIndexing) {
+        return new IndexingStrategy(false, false, false, true, versionForIndexing, 0, null);
+    }
+
+    static IndexingStrategy failAsTooManyDocs(Exception e) {
+        final Engine.IndexResult result = new Engine.IndexResult(e, Versions.NOT_FOUND);
+        return new IndexingStrategy(false, false, false, false, Versions.NOT_FOUND, 0, result);
+    }
+
+    static IndexingStrategy failAsIndexAppendOnly(Engine.IndexResult result, long versionForIndexing, int reservedDocs) {
+        return new IndexingStrategy(false, false, false, true, versionForIndexing, reservedDocs, result);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        if (!super.equals(o)) return false;
+        IndexingStrategy that = (IndexingStrategy) o;
+        return currentNotFoundOrDeleted == that.currentNotFoundOrDeleted && useUpdateDocument == that.useUpdateDocument;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(super.hashCode(), currentNotFoundOrDeleted, useUpdateDocument);
+    }
+
+    @Override
+    public String toString() {
+        return "IndexingStrategy{"
+            + "currentNotFoundOrDeleted="
+            + currentNotFoundOrDeleted
+            + ", useUpdateDocument="
+            + useUpdateDocument
+            + ", executeOpOnEngine="
+            + executeOpOnEngine
+            + ", addStaleOpToEngine="
+            + addStaleOpToEngine
+            + ", version="
+            + version
+            + ", earlyResultOnPreFlightError="
+            + earlyResultOnPreFlightError
+            + ", reservedDocs="
+            + reservedDocs
+            + '}';
+    }
+}