[Refactor] InternalEngine to always use soft deletes

qa/full-cluster-restart/src/test/java/org/opensearch/upgrades/FullClusterRestartIT.java

@@ -719,6 +719,12 @@ public void testRecovery() throws Exception {
              * or not we have one. */
             shouldHaveTranslog = randomBoolean();
 
+            Settings.Builder settings = Settings.builder();
+            if (minimumNodeVersion().before(Version.V_2_0_0) && randomBoolean()) {
+                settings.put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), randomBoolean());
+            }
+            createIndex(index, settings.build());
+
             indexRandomDocuments(count, true, true, i -> jsonBuilder().startObject().field("field", "value").endObject());
 
             // make sure all recoveries are done
@@ -1344,7 +1350,8 @@ public void testOperationBasedRecovery() throws Exception {
             if (minimumNodeVersion().before(Version.V_2_0_0)) {
                 settings.put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), randomBoolean());
             }
-            createIndex(index, settings.build());
+            final String mappings = randomBoolean() ? "\"_source\": { \"enabled\": false}" : null;
+            createIndex(index, settings.build(), mappings);
             ensureGreen(index);
             int committedDocs = randomIntBetween(100, 200);
             for (int i = 0; i < committedDocs; i++) {

server/src/internalClusterTest/java/org/opensearch/indices/stats/IndexStatsIT.java

@@ -1248,9 +1248,7 @@ public void testFilterCacheStats() throws Exception {
             client().prepareIndex("index", "type", "1").setSource("foo", "bar"),
             client().prepareIndex("index", "type", "2").setSource("foo", "baz")
         );
-        if (IndexSettings.INDEX_SOFT_DELETES_SETTING.get(settings)) {
-            persistGlobalCheckpoint("index"); // Need to persist the global checkpoint for the soft-deletes retention MP.
-        }
+        persistGlobalCheckpoint("index"); // Need to persist the global checkpoint for the soft-deletes retention MP.
         refresh();
         ensureGreen();
 
@@ -1287,22 +1285,20 @@ public void testFilterCacheStats() throws Exception {
         // Here we are testing that a fully deleted segment should be dropped and its cached is evicted.
         // In order to instruct the merge policy not to keep a fully deleted segment,
         // we need to flush and make that commit safe so that the SoftDeletesPolicy can drop everything.
-        if (IndexSettings.INDEX_SOFT_DELETES_SETTING.get(settings)) {
-            persistGlobalCheckpoint("index");
-            assertBusy(() -> {
-                for (final ShardStats shardStats : client().admin().indices().prepareStats("index").get().getIndex("index").getShards()) {
-                    final long maxSeqNo = shardStats.getSeqNoStats().getMaxSeqNo();
-                    assertTrue(
-                        shardStats.getRetentionLeaseStats()
-                            .retentionLeases()
-                            .leases()
-                            .stream()
-                            .allMatch(retentionLease -> retentionLease.retainingSequenceNumber() == maxSeqNo + 1)
-                    );
-                }
-            });
-            flush("index");
-        }
+        persistGlobalCheckpoint("index");
+        assertBusy(() -> {
+            for (final ShardStats shardStats : client().admin().indices().prepareStats("index").get().getIndex("index").getShards()) {
+                final long maxSeqNo = shardStats.getSeqNoStats().getMaxSeqNo();
+                assertTrue(
+                    shardStats.getRetentionLeaseStats()
+                        .retentionLeases()
+                        .leases()
+                        .stream()
+                        .allMatch(retentionLease -> retentionLease.retainingSequenceNumber() == maxSeqNo + 1)
+                );
+            }
+        });
+        flush("index");
         logger.info("--> force merging to a single segment");
         ForceMergeResponse forceMergeResponse = client().admin()
             .indices()

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

@@ -200,16 +200,13 @@ protected final OpenSearchDirectoryReader wrapReader(
         DirectoryReader reader,
         Function<DirectoryReader, DirectoryReader> readerWrapperFunction
     ) throws IOException {
-        if (engineConfig.getIndexSettings().isSoftDeleteEnabled()) {
-            reader = new SoftDeletesDirectoryReaderWrapper(reader, Lucene.SOFT_DELETES_FIELD);
-        }
         reader = readerWrapperFunction.apply(reader);
         return OpenSearchDirectoryReader.wrap(reader, engineConfig.getShardId());
     }
 
     protected DirectoryReader open(IndexCommit commit) throws IOException {
         assert Transports.assertNotTransportThread("opening index commit of a read-only engine");
-        return DirectoryReader.open(commit);
+        return new SoftDeletesDirectoryReaderWrapper(DirectoryReader.open(commit), Lucene.SOFT_DELETES_FIELD);
     }
 
     @Override
@@ -337,10 +334,7 @@ public Translog.Snapshot newChangesSnapshot(
         long fromSeqNo,
         long toSeqNo,
         boolean requiredFullRange
-    ) throws IOException {
-        if (engineConfig.getIndexSettings().isSoftDeleteEnabled() == false) {
-            throw new IllegalStateException("accessing changes snapshot requires soft-deletes enabled");
-        }
+    ) {
         return newEmptySnapshot();
     }
 

server/src/main/java/org/opensearch/index/mapper/SourceFieldMapper.java

@@ -190,7 +190,7 @@ public void preParse(ParseContext context) throws IOException {
             context.doc().add(new StoredField(fieldType().name(), ref.bytes, ref.offset, ref.length));
         }
 
-        if (originalSource != null && adaptedSource != originalSource && context.indexSettings().isSoftDeleteEnabled()) {
+        if (originalSource != null && adaptedSource != originalSource) {
             // if we omitted source or modified it we add the _recovery_source to ensure we have it for ops based recovery
             BytesRef ref = originalSource.toBytesRef();
             context.doc().add(new StoredField(RECOVERY_SOURCE_NAME, ref.bytes, ref.offset, ref.length));