Use new source loader when lower docId is accessed

When using synthetic source, runtime fields data may come from
doc values. Doc values iterators can only be read once,
and in increasing docId order. But if a runtime field is referenced
multiple times in a query, currently the same doc value iterator
will be used. This causes an error, as the second field reference
will attempt to read the same iterator from a lower docId than was
previously used. The fix is to create a new source loader, and
thus a new doc value iterator, if the requested docId is lower than
the last seen docId.

Author: parkertimmins

server/src/main/java/org/elasticsearch/search/lookup/ConcurrentSegmentSourceProvider.java

@@ -29,11 +29,13 @@ class ConcurrentSegmentSourceProvider implements SourceProvider {
     private final SourceLoader sourceLoader;
     private final StoredFieldLoader storedFieldLoader;
     private final Map<Object, Leaf> leaves = ConcurrentCollections.newConcurrentMap();
+    private final boolean isStoredSource;
 
-    ConcurrentSegmentSourceProvider(SourceLoader loader, boolean loadSource) {
+    ConcurrentSegmentSourceProvider(SourceLoader loader, boolean isStoredSource) {
         this.sourceLoader = loader;
         // we force a sequential reader here since it is used during query execution where documents are scanned sequentially
-        this.storedFieldLoader = StoredFieldLoader.create(loadSource, sourceLoader.requiredStoredFields(), true);
+        this.storedFieldLoader = StoredFieldLoader.create(isStoredSource, sourceLoader.requiredStoredFields(), true);
+        this.isStoredSource = isStoredSource;
     }
 
     @Override
@@ -44,6 +46,11 @@ public Source getSource(LeafReaderContext ctx, int doc) throws IOException {
             leaf = new Leaf(sourceLoader.leaf(ctx.reader(), null), storedFieldLoader.getLoader(ctx, null));
             var existing = leaves.put(id, leaf);
             assert existing == null : "unexpected source provider [" + existing + "]";
+        } else if (isStoredSource == false && doc < leaf.doc) {
+            // For synthetic source, if a runtime field is used more than once, a new source loader must be used
+            // for each use of the field, as doc value iterators may only be read once in increasing docId order.
+            leaf = new Leaf(sourceLoader.leaf(ctx.reader(), null), storedFieldLoader.getLoader(ctx, null));
+            leaves.put(id, leaf);
         }
         return leaf.getSource(ctx, doc);
     }

x-pack/plugin/logsdb/src/javaRestTest/java/org/elasticsearch/xpack/logsdb/LogsdbRestIT.java

@@ -340,4 +340,108 @@ public void testLogsdbDefaultWithRecoveryUseSyntheticSource() throws IOException
         assertNull(settings.get("index.mapping.source.mode"));
         assertEquals("true", settings.get(IndexSettings.LOGSDB_SORT_ON_HOST_NAME.getKey()));
     }
+
+    public void testSyntheticSourceRuntimeFieldQueries() throws IOException {
+        String mappings = """
+            {
+                "runtime": {
+                    "message_length": {
+                        "type": "long"
+                    }
+                },
+                "dynamic": false,
+                "properties": {
+                    "@timestamp": {
+                        "type": "date"
+                    }
+                }
+            }
+            """;
+        String indexName = "test-foo";
+        createIndex(indexName, Settings.builder().put("index.mode", "logsdb").build(), mappings);
+
+        int numDocs = 100_000;
+        var sb = new StringBuilder();
+        var now = Instant.now();
+
+        for (int i = 0; i < numDocs; i++) {
+            String msg = randomAlphaOfLength(20);
+            String messageLength = Integer.toString(msg.length());
+            sb.append("{ \"create\": {} }").append('\n');
+            sb.append("""
+                {"@timestamp": "$now", "message_length": $l}
+                """.replace("$now", formatInstant(now)).replace("$l", messageLength));
+            sb.append('\n');
+            if (i != numDocs - 1) {
+                now = now.plusSeconds(1);
+            }
+
+            if (i % 1000 == 0) {
+                var bulkRequest = new Request("POST", "/" + indexName + "/_bulk");
+                bulkRequest.setJsonEntity(sb.toString());
+                var bulkResponse = client().performRequest(bulkRequest);
+                var bulkResponseBody = responseAsMap(bulkResponse);
+                assertThat(bulkResponseBody, Matchers.hasEntry("errors", false));
+                sb = new StringBuilder();
+            }
+        }
+
+        var bulkRequest = new Request("POST", "/" + indexName + "/_bulk");
+        bulkRequest.setJsonEntity(sb.toString());
+        bulkRequest.addParameter("refresh", "true");
+        var bulkResponse = client().performRequest(bulkRequest);
+        var bulkResponseBody = responseAsMap(bulkResponse);
+        assertThat(bulkResponseBody, Matchers.hasEntry("errors", false));
+
+        var forceMergeRequest = new Request("POST", "/" + indexName + "/_forcemerge");
+        var forceMergeResponse = client().performRequest(forceMergeRequest);
+        assertOK(forceMergeResponse);
+
+        var searchRequest = new Request("POST", "/" + indexName + "/_search");
+        searchRequest.setJsonEntity("""
+            {
+                "size": 1,
+                "query": {
+                    "bool": {
+                        "should": [
+                            {
+                                "range": {
+                                    "message_length": {
+                                        "gte": 1,
+                                        "lt": 900000
+                                    }
+                                }
+                            },
+                            {
+                                "range": {
+                                    "message_length": {
+                                        "gte": 900000,
+                                        "lt": 1000000
+                                    }
+                                }
+                            }
+                        ],
+                        "minimum_should_match": "1",
+                        "must_not": [
+                            {
+                                "range": {
+                                    "message_length": {
+                                        "lt": 0
+                                    }
+                                }
+                            }
+                        ]
+                    }
+                }
+            }
+            """);
+        var searchResponse = client().performRequest(searchRequest);
+        assertOK(searchResponse);
+        var searchResponseBody = responseAsMap(searchResponse);
+        var shardsHeader = (Map<?, ?>) searchResponseBody.get("_shards");
+        assertThat(shardsHeader.get("failed"), equalTo(0));
+        assertThat(shardsHeader.get("successful"), equalTo(1));
+        assertThat(shardsHeader.get("skipped"), equalTo(0));
+        logger.info("searchResponse: {}", searchResponseBody);
+    }
 }