Use Lucene soft-deletes in peer recovery

server/src/main/java/org/elasticsearch/index/engine/CombinedDeletionPolicy.java

@@ -46,14 +46,17 @@
 public final class CombinedDeletionPolicy extends IndexDeletionPolicy {
     private final Logger logger;
     private final TranslogDeletionPolicy translogDeletionPolicy;
+    private final SoftDeletesPolicy softDeletesPolicy;
     private final LongSupplier globalCheckpointSupplier;
     private final ObjectIntHashMap<IndexCommit> snapshottedCommits; // Number of snapshots held against each commit point.
     private volatile IndexCommit safeCommit; // the most recent safe commit point - its max_seqno at most the persisted global checkpoint.
     private volatile IndexCommit lastCommit; // the most recent commit point
 
-    CombinedDeletionPolicy(Logger logger, TranslogDeletionPolicy translogDeletionPolicy, LongSupplier globalCheckpointSupplier) {
+    CombinedDeletionPolicy(Logger logger, TranslogDeletionPolicy translogDeletionPolicy,
+                           SoftDeletesPolicy softDeletesPolicy, LongSupplier globalCheckpointSupplier) {
         this.logger = logger;
         this.translogDeletionPolicy = translogDeletionPolicy;
+        this.softDeletesPolicy = softDeletesPolicy;
         this.globalCheckpointSupplier = globalCheckpointSupplier;
         this.snapshottedCommits = new ObjectIntHashMap<>();
     }
@@ -80,7 +83,7 @@ public synchronized void onCommit(List<? extends IndexCommit> commits) throws IO
                 deleteCommit(commits.get(i));
             }
         }
-        updateTranslogDeletionPolicy();
+        updateRetentionPolicy();
     }
 
     private void deleteCommit(IndexCommit commit) throws IOException {
@@ -90,7 +93,7 @@ private void deleteCommit(IndexCommit commit) throws IOException {
         assert commit.isDeleted() : "Deletion commit [" + commitDescription(commit) + "] was suppressed";
     }
 
-    private void updateTranslogDeletionPolicy() throws IOException {
+    private void updateRetentionPolicy() throws IOException {
         assert Thread.holdsLock(this);
         logger.debug("Safe commit [{}], last commit [{}]", commitDescription(safeCommit), commitDescription(lastCommit));
         assert safeCommit.isDeleted() == false : "The safe commit must not be deleted";
@@ -101,6 +104,8 @@ private void updateTranslogDeletionPolicy() throws IOException {
         assert minRequiredGen <= lastGen : "minRequiredGen must not be greater than lastGen";
         translogDeletionPolicy.setTranslogGenerationOfLastCommit(lastGen);
         translogDeletionPolicy.setMinTranslogGenerationForRecovery(minRequiredGen);
+
+        softDeletesPolicy.setCheckpointOfSafeCommit(Long.parseLong(safeCommit.getUserData().get(SequenceNumbers.LOCAL_CHECKPOINT_KEY)));
     }
 
     /**

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

@@ -149,6 +149,7 @@ public class InternalEngine extends Engine {
     private final CounterMetric numDocUpdates = new CounterMetric();
     private final NumericDocValuesField softDeleteField = Lucene.newSoftDeleteField();
     private final boolean softDeleteEnabled;
+    private final SoftDeletesPolicy softDeletesPolicy;
     private final LastRefreshedCheckpointListener lastRefreshedCheckpointListener;
 
     /**
@@ -173,7 +174,6 @@ public InternalEngine(EngineConfig engineConfig) {
         if (engineConfig.isAutoGeneratedIDsOptimizationEnabled() == false) {
             maxUnsafeAutoIdTimestamp.set(Long.MAX_VALUE);
         }
-        this.softDeleteEnabled = engineConfig.getIndexSettings().isSoftDeleteEnabled();
         final TranslogDeletionPolicy translogDeletionPolicy = new TranslogDeletionPolicy(
                 engineConfig.getIndexSettings().getTranslogRetentionSize().getBytes(),
                 engineConfig.getIndexSettings().getTranslogRetentionAge().getMillis()
@@ -195,8 +195,11 @@ public InternalEngine(EngineConfig engineConfig) {
                 assert translog.getGeneration() != null;
                 this.translog = translog;
                 this.localCheckpointTracker = createLocalCheckpointTracker(localCheckpointTrackerSupplier);
+                this.softDeleteEnabled = engineConfig.getIndexSettings().isSoftDeleteEnabled();
+                this.softDeletesPolicy = new SoftDeletesPolicy(translog::getLastSyncedGlobalCheckpoint,
+                    engineConfig.getIndexSettings().getSoftDeleteRetentionOperations());
                 this.combinedDeletionPolicy =
-                    new CombinedDeletionPolicy(logger, translogDeletionPolicy, translog::getLastSyncedGlobalCheckpoint);
+                    new CombinedDeletionPolicy(logger, translogDeletionPolicy, softDeletesPolicy, translog::getLastSyncedGlobalCheckpoint);
                 writer = createWriter();
                 bootstrapAppendOnlyInfoFromWriter(writer);
                 historyUUID = loadHistoryUUID(writer);
@@ -2013,7 +2016,7 @@ private IndexWriterConfig getIndexWriterConfig() {
         MergePolicy mergePolicy = config().getMergePolicy();
         if (softDeleteEnabled) {
             iwc.setSoftDeletesField(Lucene.SOFT_DELETE_FIELD);
-            mergePolicy = new SoftDeletesRetentionMergePolicy(Lucene.SOFT_DELETE_FIELD, this::softDeletesRetentionQuery, mergePolicy);
+            mergePolicy = new SoftDeletesRetentionMergePolicy(Lucene.SOFT_DELETE_FIELD, softDeletesPolicy::retentionQuery, mergePolicy);
         }
         iwc.setMergePolicy(new ElasticsearchMergePolicy(mergePolicy));
         iwc.setSimilarity(engineConfig.getSimilarity());
@@ -2026,20 +2029,6 @@ private IndexWriterConfig getIndexWriterConfig() {
         return iwc;
     }
 
-    /**
-     * Documents including tombstones are soft-deleted and matched this query will be retained and won't cleaned up by merges.
-     */
-    private Query softDeletesRetentionQuery() {
-        ensureOpen();
-        // TODO: We send the safe commit in peer-recovery, thus we need to retain all operations after the local checkpoint of that commit.
-        final long retainedExtraOps = engineConfig.getIndexSettings().getSoftDeleteRetentionOperations();
-        // Prefer using the global checkpoint which is persisted on disk than an in-memory value.
-        // If we failed to fsync checkpoint but already used a higher global checkpoint value to clean up soft-deleted ops,
-        // then we may not have all required operations whose seq# greater than the global checkpoint after restarted.
-        final long persistedGlobalCheckpoint = translog.getLastSyncedGlobalCheckpoint();
-        return LongPoint.newRangeQuery(SeqNoFieldMapper.NAME, persistedGlobalCheckpoint + 1 - retainedExtraOps, Long.MAX_VALUE);
-    }
-
     /** Extended SearcherFactory that warms the segments if needed when acquiring a new searcher */
     static final class SearchFactory extends EngineSearcherFactory {
         private final Engine.Warmer warmer;
@@ -2281,6 +2270,8 @@ public void onSettingsChanged() {
         final IndexSettings indexSettings = engineConfig.getIndexSettings();
         translogDeletionPolicy.setRetentionAgeInMillis(indexSettings.getTranslogRetentionAge().getMillis());
         translogDeletionPolicy.setRetentionSizeInBytes(indexSettings.getTranslogRetentionSize().getBytes());
+
+        softDeletesPolicy.setRetentionOperations(indexSettings.getSoftDeleteRetentionOperations());
     }
 
     public MergeStats getMergeStats() {

server/src/main/java/org/elasticsearch/index/engine/SoftDeletesPolicy.java

@@ -0,0 +1,113 @@
+/*
+ * 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.index.engine;
+
+import org.apache.lucene.document.LongPoint;
+import org.apache.lucene.search.Query;
+import org.elasticsearch.common.lease.Releasable;
+import org.elasticsearch.index.mapper.SeqNoFieldMapper;
+import org.elasticsearch.index.seqno.SequenceNumbers;
+import org.elasticsearch.index.translog.Translog;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.LongSupplier;
+
+/**
+ * A policy that controls how many soft-deleted documents should be retained for peer-recovery and querying changes purpose.
+ */
+final class SoftDeletesPolicy {
+    private final LongSupplier globalCheckpointSupplier;
+    private int retentionLockCount;
+    private long checkpointOfSafeCommit;
+    private long minRequiredSeqNoForRecovery;
+    private long retentionOperations;
+
+    SoftDeletesPolicy(LongSupplier globalCheckpointSupplier, long retentionOperations) {
+        this.globalCheckpointSupplier = globalCheckpointSupplier;
+        this.retentionOperations = retentionOperations;
+        this.checkpointOfSafeCommit = SequenceNumbers.NO_OPS_PERFORMED;
+        this.minRequiredSeqNoForRecovery = checkpointOfSafeCommit;
+        this.retentionLockCount = 0;
+    }
+
+    /**
+     * Updates the number of soft-deleted prior to the global checkpoint to be retained
+     * See {@link org.elasticsearch.index.IndexSettings#INDEX_SOFT_DELETES_RETENTION_OPERATIONS_SETTING}
+     */
+    synchronized void setRetentionOperations(long retentionOperations) {
+        this.retentionOperations = retentionOperations;
+    }
+
+    /**
+     * Sets the local checkpoint of the current safe commit.
+     * All operations whose seqno are greater than this checkpoint will be retained until the new checkpoint is advanced.
+     */
+    synchronized void setCheckpointOfSafeCommit(long newCheckpoint) {
+        if (newCheckpoint < this.checkpointOfSafeCommit) {
+            throw new IllegalArgumentException("Local checkpoint can't go backwards; " +
+                "new checkpoint [" + newCheckpoint + "]," + "current checkpoint [" + checkpointOfSafeCommit + "]");
+        }
+        this.checkpointOfSafeCommit = newCheckpoint;
+        updateMinRequiredSeqNoForRecovery();
+    }
+
+    private void updateMinRequiredSeqNoForRecovery() {
+        assert Thread.holdsLock(this) : Thread.currentThread().getName();
+        if (retentionLockCount == 0) {
+            this.minRequiredSeqNoForRecovery = checkpointOfSafeCommit;
+        }
+    }
+
+    /**
+     * Acquires a lock on soft-deleted documents to prevent them from cleaning up in merge processes. This is necessary to
+     * make sure that all operations after the local checkpoint of the safe commit are retained until the lock is released.
+     * This is a analogy to the translog's retention lock; see {@link Translog#acquireRetentionLock()}
+     */
+    synchronized Releasable acquireRetentionLock() {
+        assert retentionLockCount >= 0 : "Invalid number of retention locks [" + retentionLockCount + "]";
+        retentionLockCount++;
+        final AtomicBoolean released = new AtomicBoolean();
+        return () -> {
+            if (released.compareAndSet(false, true)) {
+                releaseRetentionLock();
+            }
+        };
+    }
+
+    private synchronized void releaseRetentionLock() {
+        assert retentionLockCount > 0 : "Invalid number of retention locks [" + retentionLockCount + "]";
+        retentionLockCount--;
+        updateMinRequiredSeqNoForRecovery();
+    }
+
+    /**
+     * Returns a soft-deletes retention query that will be used in {@link org.apache.lucene.index.SoftDeletesRetentionMergePolicy}
+     * Documents including tombstones are soft-deleted and matched this query will be retained and won't cleaned up by merges.
+     */
+    Query retentionQuery() {
+        return LongPoint.newRangeQuery(SeqNoFieldMapper.NAME, getMinSeqNoToRetain(), Long.MAX_VALUE);
+    }
+
+    // Package-level for testing
+    synchronized long getMinSeqNoToRetain() {
+        final long minSeqNoForQueryingChanges = globalCheckpointSupplier.getAsLong() - retentionOperations;
+        return Math.min(minRequiredSeqNoForRecovery, minSeqNoForQueryingChanges) + 1;
+    }
+}