Fix from_range in search_after in changes snapshot (#33335)

We can have multiple documents in Lucene with the same seq_no for
parent-child documents (or without rollback). In this case, the usage
"lastSeenSeqNo + 1" is an off-by-one error as it may miss some
documents. This error merely affects the `skippedOperations` contract.

See: #33222 (comment)

Closes #33318

Author: dnhatn

modules/parent-join/src/test/java/org/elasticsearch/join/query/ParentChildTestCase.java

@@ -25,7 +25,6 @@
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.index.IndexModule;
-import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.join.ParentJoinPlugin;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.test.ESIntegTestCase;
@@ -59,8 +58,6 @@ protected Collection<Class<? extends Plugin>> transportClientPlugins() {
     @Override
     public Settings indexSettings() {
         Settings.Builder builder =  Settings.builder().put(super.indexSettings())
-            // AwaitsFix: https://github.com/elastic/elasticsearch/issues/33318
-            .put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), false)
             // aggressive filter caching so that we can assert on the filter cache size
             .put(IndexModule.INDEX_QUERY_CACHE_ENABLED_SETTING.getKey(), true)
             .put(IndexModule.INDEX_QUERY_CACHE_EVERYTHING_SETTING.getKey(), true);

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

@@ -212,7 +212,7 @@ private void fillParallelArray(ScoreDoc[] scoreDocs, ParallelArray parallelArray
     }
 
     private TopDocs searchOperations(ScoreDoc after) throws IOException {
-        final Query rangeQuery = LongPoint.newRangeQuery(SeqNoFieldMapper.NAME, lastSeenSeqNo + 1, toSeqNo);
+        final Query rangeQuery = LongPoint.newRangeQuery(SeqNoFieldMapper.NAME, Math.max(fromSeqNo, lastSeenSeqNo), toSeqNo);
         final Sort sortedBySeqNoThenByTerm = new Sort(
             new SortField(SeqNoFieldMapper.NAME, SortField.Type.LONG),
             new SortField(SeqNoFieldMapper.PRIMARY_TERM_NAME, SortField.Type.LONG, true)

server/src/test/java/org/elasticsearch/index/engine/LuceneChangesSnapshotTests.java

@@ -182,12 +182,17 @@ public void testDedupByPrimaryTerm() throws Exception {
             }
         }
         long maxSeqNo = engine.getLocalCheckpointTracker().getMaxSeqNo();
-        try (Translog.Snapshot snapshot = engine.newChangesSnapshot("test", mapperService, 0, maxSeqNo, false)) {
+        engine.refresh("test");
+        Engine.Searcher searcher = engine.acquireSearcher("test", Engine.SearcherScope.INTERNAL);
+        try (Translog.Snapshot snapshot = new LuceneChangesSnapshot(searcher, mapperService, between(1, 100), 0, maxSeqNo, false)) {
+            searcher = null;
             Translog.Operation op;
             while ((op = snapshot.next()) != null) {
                 assertThat(op.toString(), op.primaryTerm(), equalTo(latestOperations.get(op.seqNo())));
             }
             assertThat(snapshot.skippedOperations(), equalTo(totalOps - latestOperations.size()));
+        } finally {
+            IOUtils.close(searcher);
         }
     }