ESQL: Initial support for unmapped fields

docs/changelog/119886.yaml

@@ -0,0 +1,5 @@
+pr: 119886
+summary: Initial support for unmapped fields
+area: ES|QL
+type: feature
+issues: []

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

@@ -22,6 +22,7 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Objects;
 
 /**
  * Loads values from {@code _source}. This whole process is very slow and cast-tastic,
@@ -230,7 +231,7 @@ private static class BytesRefs extends BlockSourceReader {
 
         @Override
         protected void append(BlockLoader.Builder builder, Object v) {
-            ((BlockLoader.BytesRefBuilder) builder).appendBytesRef(toBytesRef(scratch, (String) v));
+            ((BlockLoader.BytesRefBuilder) builder).appendBytesRef(toBytesRef(scratch, Objects.toString(v)));
         }
 
         @Override

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/type/EsField.java

@@ -27,11 +27,12 @@
 public class EsField implements Writeable {
 
     private static Map<String, Writeable.Reader<? extends EsField>> readers = Map.ofEntries(
-        Map.entry("EsField", EsField::new),
         Map.entry("DateEsField", DateEsField::new),
+        Map.entry("EsField", EsField::new),
         Map.entry("InvalidMappedField", InvalidMappedField::new),
         Map.entry("KeywordEsField", KeywordEsField::new),
         Map.entry("MultiTypeEsField", MultiTypeEsField::new),
+        Map.entry("PotentiallyUnmappedKeywordEsField", PotentiallyUnmappedKeywordEsField::new),
         Map.entry("TextEsField", TextEsField::new),
         Map.entry("UnsupportedEsField", UnsupportedEsField::new)
     );

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/type/PotentiallyUnmappedKeywordEsField.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+package org.elasticsearch.xpack.esql.core.type;
+
+import org.elasticsearch.common.io.stream.StreamInput;
+
+import java.io.IOException;
+
+/** Information about a field in an ES index that may not mapped. If it is unmapped, it should be loaded from source directly. */
+public class PotentiallyUnmappedKeywordEsField extends KeywordEsField {
+    public PotentiallyUnmappedKeywordEsField(String name) {
+        super(name);
+    }
+
+    public PotentiallyUnmappedKeywordEsField(StreamInput in) throws IOException {
+        super(in);
+    }
+
+    public String getWriteableName() {
+        return "PotentiallyUnmappedKeywordEsField";
+    }
+}

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/util/CollectionUtils.java

@@ -10,7 +10,9 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.OptionalInt;
 import java.util.Set;
+import java.util.function.Predicate;
 
 import static java.util.Collections.emptyList;
 
@@ -83,4 +85,13 @@ public static <T> List<T> nullSafeList(T... entries) {
         }
         return list;
     }
+
+    public static <T> OptionalInt findFirstIndex(List<T> list, Predicate<T> predicate) {
+        for (int i = 0; i < list.size(); i++) {
+            if (predicate.test(list.get(i))) {
+                return OptionalInt.of(i);
+            }
+        }
+        return OptionalInt.empty();
+    }
 }

x-pack/plugin/esql/qa/server/multi-clusters/src/javaRestTest/java/org/elasticsearch/xpack/esql/ccq/MultiClusterSpecIT.java

@@ -51,6 +51,7 @@
 import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.JOIN_LOOKUP_V12;
 import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.JOIN_PLANNING_V1;
 import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.METADATA_FIELDS_REMOTE_TEST;
+import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.UNMAPPED_FIELDS;
 import static org.elasticsearch.xpack.esql.qa.rest.EsqlSpecTestCase.Mode.SYNC;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doAnswer;
@@ -125,6 +126,8 @@ protected void shouldSkipTest(String testName) throws IOException {
         assumeFalse("INLINESTATS not yet supported in CCS", testCase.requiredCapabilities.contains(INLINESTATS_V2.capabilityName()));
         assumeFalse("INLINESTATS not yet supported in CCS", testCase.requiredCapabilities.contains(JOIN_PLANNING_V1.capabilityName()));
         assumeFalse("LOOKUP JOIN not yet supported in CCS", testCase.requiredCapabilities.contains(JOIN_LOOKUP_V12.capabilityName()));
+        // Unmapped fields require a coorect capability response from every cluster, which isn't currently implemented.
+        assumeFalse("UNMAPPED FIELDS not yet supported in CCS", testCase.requiredCapabilities.contains(UNMAPPED_FIELDS.capabilityName()));
     }
 
     private TestFeatureService remoteFeaturesService() throws IOException {

x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/CsvTestUtils.java

@@ -120,7 +120,9 @@ public static Tuple<Version, Version> skipVersionRange(String testName, String i
         return null;
     }
 
-    public static Tuple<Page, List<String>> loadPageFromCsv(URL source, Map<String, String> typeMapping) throws Exception {
+    public record PageColumn(String name, DataType dataType) {}
+
+    public static Tuple<Page, List<PageColumn>> loadPageFromCsv(URL source, Map<String, String> typeMapping) throws Exception {
 
         record CsvColumn(String name, Type type, BuilderWrapper builderWrapper) implements Releasable {
             void append(String stringValue) {
@@ -230,13 +232,13 @@ public void close() {
                 lineNumber++;
             }
         }
-        var columnNames = new ArrayList<String>(columns.length);
+        var pageColumns = new ArrayList<PageColumn>(columns.length);
         try {
             var blocks = Arrays.stream(columns)
-                .peek(b -> columnNames.add(b.name))
+                .peek(b -> pageColumns.add(new PageColumn(b.name, DataType.fromTypeName(b.type.name()))))
                 .map(b -> b.builderWrapper.builder().build())
                 .toArray(Block[]::new);
-            return new Tuple<>(new Page(blocks), columnNames);
+            return new Tuple<>(new Page(blocks), pageColumns);
         } finally {
             Releasables.closeExpectNoException(columns);
         }

x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/CsvTestsDataLoader.java

@@ -42,6 +42,8 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import static org.elasticsearch.common.logging.LoggerMessageFormat.format;
 import static org.elasticsearch.xpack.esql.CsvTestUtils.COMMA_ESCAPING_REGEX;
@@ -83,6 +85,22 @@ public class CsvTestsDataLoader {
         .withData("sample_data_ts_nanos.csv")
         .withTypeMapping(Map.of("@timestamp", "date_nanos"));
     private static final TestDataset MISSING_IP_SAMPLE_DATA = new TestDataset("missing_ip_sample_data");
+    private static final TestDataset SAMPLE_DATA_PARTIAL_MAPPING = new TestDataset("partial_mapping_sample_data");
+    private static final TestDataset SAMPLE_DATA_NO_MAPPING = new TestDataset(
+        "no_mapping_sample_data",
+        "mapping-no_mapping_sample_data.json",
+        "partial_mapping_sample_data.csv"
+    ).withTypeMapping(Stream.of("timestamp", "client_ip", "event_duration").collect(Collectors.toMap(k -> k, k -> "keyword")));
+    private static final TestDataset SAMPLE_DATA_PARTIAL_MAPPING_NO_SOURCE = new TestDataset(
+        "partial_mapping_no_source_sample_data",
+        "mapping-partial_mapping_no_source_sample_data.json",
+        "partial_mapping_sample_data.csv"
+    );
+    private static final TestDataset SAMPLE_DATA_PARTIAL_MAPPING_EXCLUDED_SOURCE = new TestDataset(
+        "partial_mapping_excluded_source_sample_data",
+        "mapping-partial_mapping_excluded_source_sample_data.json",
+        "partial_mapping_sample_data.csv"
+    );
     private static final TestDataset CLIENT_IPS = new TestDataset("clientips");
     private static final TestDataset CLIENT_IPS_LOOKUP = CLIENT_IPS.withIndex("clientips_lookup")
         .withSetting("clientips_lookup-settings.json");
@@ -128,6 +146,10 @@ public class CsvTestsDataLoader {
         Map.entry(LANGUAGES_NESTED_FIELDS.indexName, LANGUAGES_NESTED_FIELDS),
         Map.entry(UL_LOGS.indexName, UL_LOGS),
         Map.entry(SAMPLE_DATA.indexName, SAMPLE_DATA),
+        Map.entry(SAMPLE_DATA_PARTIAL_MAPPING.indexName, SAMPLE_DATA_PARTIAL_MAPPING),
+        Map.entry(SAMPLE_DATA_NO_MAPPING.indexName, SAMPLE_DATA_NO_MAPPING),
+        Map.entry(SAMPLE_DATA_PARTIAL_MAPPING_NO_SOURCE.indexName, SAMPLE_DATA_PARTIAL_MAPPING_NO_SOURCE),
+        Map.entry(SAMPLE_DATA_PARTIAL_MAPPING_EXCLUDED_SOURCE.indexName, SAMPLE_DATA_PARTIAL_MAPPING_EXCLUDED_SOURCE),
         Map.entry(MV_SAMPLE_DATA.indexName, MV_SAMPLE_DATA),
         Map.entry(ALERTS.indexName, ALERTS),
         Map.entry(SAMPLE_DATA_STR.indexName, SAMPLE_DATA_STR),

x-pack/plugin/esql/qa/testFixtures/src/main/resources/data/partial_mapping_sample_data.csv

@@ -0,0 +1,8 @@
+@timestamp:date,client_ip:ip,event_duration:long,message:keyword,unmapped_message:keyword,unmapped_event_duration:keyword
+2024-10-23T13:55:01.543Z,173.21.3.15,1756466,Connected to 10.1.0.1!,Disconnected from 10.1.0.1,1756468
+2024-10-23T13:53:55.832Z,173.21.3.15,5033754,Connection error?,Disconnection error,5033756
+2024-10-23T13:52:55.015Z,173.21.3.15,8268152,Connection error?,Disconnection error,8268154
+2024-10-23T13:51:54.732Z,173.21.3.15,725447,Connection error?,Disconnection error,725449
+2024-10-23T13:33:34.937Z,173.21.0.5,1232381,42,43,1232383
+2024-10-23T12:27:28.948Z,173.21.2.113,2764888,Connected to 10.1.0.2!,Disconnected from 10.1.0.2,2764890
+2024-10-23T12:15:03.360Z,173.21.2.162,3450232,Connected to 10.1.0.3!,Disconnected from 10.1.0.3,3450234

x-pack/plugin/esql/qa/testFixtures/src/main/resources/mapping-no_mapping_sample_data.json

@@ -0,0 +1,4 @@
+{
+    "dynamic": "false",
+    "properties": {}
+}

x-pack/plugin/esql/qa/testFixtures/src/main/resources/mapping-partial_mapping_excluded_source_sample_data.json

@@ -0,0 +1,13 @@
+{
+    "dynamic": "false",
+    "properties": {
+        "@timestamp": {
+            "type": "date"
+        }
+    },
+    "_source": {
+        "excludes": [
+            "message"
+        ]
+    }
+}

x-pack/plugin/esql/qa/testFixtures/src/main/resources/mapping-partial_mapping_no_source_sample_data.json

@@ -0,0 +1,11 @@
+{
+    "dynamic": "false",
+    "properties": {
+      "@timestamp": {
+        "type": "date"
+      }
+    },
+    "_source": {
+        "enabled": false
+    }
+}

x-pack/plugin/esql/qa/testFixtures/src/main/resources/mapping-partial_mapping_sample_data.json

@@ -0,0 +1,17 @@
+{
+    "dynamic": "false",
+    "properties": {
+        "@timestamp": {
+            "type": "date"
+        },
+        "client_ip": {
+            "type": "ip"
+        },
+        "event_duration": {
+            "type": "long"
+        },
+        "message": {
+            "type": "keyword"
+        }
+    }
+}