Address mapping and compute engine runtime field issues

server/src/main/java/org/elasticsearch/index/mapper/BlockDocValuesReader.java

@@ -82,6 +82,29 @@ public SortedSetDocValues ordinals(LeafReaderContext context) throws IOException
         }
     }
 
+    public abstract static class RowStrideOnlyDocValuesBlockLoader implements BlockLoader {
+
+        @Override
+        public final ColumnAtATimeReader columnAtATimeReader(LeafReaderContext context) throws IOException {
+            return null;
+        }
+
+        @Override
+        public boolean supportsOrdinals() {
+            return false;
+        }
+
+        @Override
+        public SortedSetDocValues ordinals(LeafReaderContext context) throws IOException {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public final StoredFieldsSpec rowStrideStoredFieldSpec() {
+            return StoredFieldsSpec.NEEDS_SOURCE;
+        }
+    }
+
     public static class LongsBlockLoader extends DocValuesBlockLoader {
         private final String fieldName;
 

server/src/main/java/org/elasticsearch/index/mapper/DocumentParser.java

@@ -941,7 +941,9 @@ public Query termQuery(Object value, SearchExecutionContext context) {
             protected void parseCreateField(DocumentParserContext context) {
                 // Run-time fields are mapped to this mapper, so it needs to handle storing values for use in synthetic source.
                 // #parseValue calls this method once the run-time field is created.
-                if (context.dynamic() == ObjectMapper.Dynamic.RUNTIME && context.canAddIgnoredField()) {
+                var fieldType = context.mappingLookup().getFieldType(path);
+                boolean isRuntimeField = fieldType instanceof AbstractScriptFieldType;
+                if ((context.dynamic() == ObjectMapper.Dynamic.RUNTIME || isRuntimeField) && context.canAddIgnoredField()) {
                     try {
                         context.addIgnoredField(
                             IgnoredSourceFieldMapper.NameValue.fromContext(context, path, context.encodeFlattenedToken())

server/src/main/java/org/elasticsearch/index/mapper/LongScriptBlockDocValuesReader.java

@@ -36,6 +36,25 @@ public AllReader reader(LeafReaderContext context) throws IOException {
         }
     }
 
+    static class RowStrideLongScriptBlockLoader extends RowStrideOnlyDocValuesBlockLoader {
+        private final LongFieldScript.LeafFactory factory;
+
+        RowStrideLongScriptBlockLoader(LongFieldScript.LeafFactory factory) {
+            this.factory = factory;
+        }
+
+        @Override
+        public Builder builder(BlockFactory factory, int expectedCount) {
+            return factory.longs(expectedCount);
+        }
+
+        @Override
+        public RowStrideReader rowStrideReader(LeafReaderContext context) throws IOException {
+            return new LongScriptBlockDocValuesReader(factory.newInstance(context));
+        }
+
+    }
+
     private final LongFieldScript script;
     private int docId;
 

server/src/main/java/org/elasticsearch/index/mapper/LongScriptFieldType.java

@@ -108,7 +108,12 @@ public DocValueFormat docValueFormat(String format, ZoneId timeZone) {
 
     @Override
     public BlockLoader blockLoader(BlockLoaderContext blContext) {
-        return new LongScriptBlockDocValuesReader.LongScriptBlockLoader(leafFactory(blContext.lookup()));
+        boolean isSyntheticSource = SourceFieldMapper.isSynthetic(blContext.indexSettings());
+        if (isSyntheticSource) {
+            return new LongScriptBlockDocValuesReader.RowStrideLongScriptBlockLoader(leafFactory(blContext.lookup()));
+        } else {
+            return new LongScriptBlockDocValuesReader.LongScriptBlockLoader(leafFactory(blContext.lookup()));
+        }
     }
 
     @Override

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

@@ -10,13 +10,16 @@
 import org.elasticsearch.client.Request;
 import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.time.DateFormatter;
+import org.elasticsearch.common.time.FormatNames;
 import org.elasticsearch.test.cluster.ElasticsearchCluster;
 import org.elasticsearch.test.cluster.local.distribution.DistributionType;
 import org.elasticsearch.test.rest.ESRestTestCase;
 import org.hamcrest.Matchers;
 import org.junit.ClassRule;
 
 import java.io.IOException;
+import java.time.Instant;
 import java.util.List;
 import java.util.Map;
 
@@ -108,4 +111,118 @@ public void testLogsdbSourceModeForLogsIndex() throws IOException {
         assertNull(settings.get("index.mapping.source.mode"));
     }
 
+    public void testEsqlRuntimeFields() throws IOException {
+        String mappings = """
+            {
+                "runtime": {
+                    "message_length": {
+                        "type": "long"
+                    },
+                    "log.offset": {
+                        "type": "long"
+                    }
+                },
+                "dynamic": false,
+                "properties": {
+                    "@timestamp": {
+                        "type": "date"
+                    },
+                    "log" : {
+                        "properties": {
+                            "level": {
+                                "type": "keyword"
+                            },
+                            "file": {
+                                "type": "keyword"
+                            }
+                        }
+                    }
+                }
+            }
+            """;
+        String indexName = "test-foo";
+        createIndex(indexName, Settings.builder().put("index.mode", "logsdb").build(), mappings);
+
+        int numDocs = 500;
+        var sb = new StringBuilder();
+        var now = Instant.now();
+
+        var expectedMinTimestamp = now;
+        for (int i = 0; i < numDocs; i++) {
+            String level = randomBoolean() ? "info" : randomBoolean() ? "warning" : randomBoolean() ? "error" : "fatal";
+            String msg = randomAlphaOfLength(20);
+            String path = randomAlphaOfLength(8);
+            String messageLength = Integer.toString(msg.length());
+            String offset = Integer.toString(randomNonNegativeInt());
+            sb.append("{ \"create\": {} }").append('\n');
+            if (randomBoolean()) {
+                sb.append(
+                    """
+                        {"@timestamp":"$now","message":"$msg","message_length":$l,"file":{"level":"$level","offset":5,"file":"$path"}}
+                        """.replace("$now", formatInstant(now))
+                        .replace("$level", level)
+                        .replace("$msg", msg)
+                        .replace("$path", path)
+                        .replace("$l", messageLength)
+                        .replace("$o", offset)
+                );
+            } else {
+                sb.append("""
+                    {"@timestamp": "$now", "message": "$msg", "message_length": $l}
+                    """.replace("$now", formatInstant(now)).replace("$msg", msg).replace("$l", messageLength));
+            }
+            sb.append('\n');
+            if (i != numDocs - 1) {
+                now = now.plusSeconds(1);
+            }
+        }
+        var expectedMaxTimestamp = now;
+
+        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");
+        forceMergeRequest.addParameter("max_num_segments", "1");
+        var forceMergeResponse = client().performRequest(forceMergeRequest);
+        assertOK(forceMergeResponse);
+
+        String query = "FROM test-foo | STATS count(*), min(@timestamp), max(@timestamp), min(message_length), max(message_length)"
+            + " ,sum(message_length), avg(message_length), min(log.offset), max(log.offset) | LIMIT 1";
+        final Request esqlRequest = new Request("POST", "/_query");
+        esqlRequest.setJsonEntity("""
+            {
+                "query": "$query"
+            }
+            """.replace("$query", query));
+        var esqlResponse = client().performRequest(esqlRequest);
+        assertOK(esqlResponse);
+        Map<String, Object> esqlResponseBody = responseAsMap(esqlResponse);
+
+        List<?> values = (List<?>) esqlResponseBody.get("values");
+        assertThat(values, Matchers.not(Matchers.empty()));
+        var count = ((List<?>) values.getFirst()).get(0);
+        assertThat(count, equalTo(numDocs));
+        logger.warn("VALUES: {}", values);
+
+        var minTimestamp = ((List<?>) values.getFirst()).get(1);
+        assertThat(minTimestamp, equalTo(formatInstant(expectedMinTimestamp)));
+        var maxTimestamp = ((List<?>) values.getFirst()).get(2);
+        assertThat(maxTimestamp, equalTo(formatInstant(expectedMaxTimestamp)));
+
+        var minLength = ((List<?>) values.getFirst()).get(3);
+        assertThat(minLength, equalTo(20));
+        var maxLength = ((List<?>) values.getFirst()).get(4);
+        assertThat(maxLength, equalTo(20));
+        var sumLength = ((List<?>) values.getFirst()).get(5);
+        assertThat(sumLength, equalTo(20 * numDocs));
+    }
+
+    static String formatInstant(Instant instant) {
+        return DateFormatter.forPattern(FormatNames.STRICT_DATE_OPTIONAL_TIME.getName()).format(instant);
+    }
+
 }