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

@@ -230,7 +230,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, v.toString()));
         }
 
         @Override

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

@@ -30,6 +30,7 @@ public class EsField implements Writeable {
         Map.entry("EsField", EsField::new),
         Map.entry("DateEsField", DateEsField::new),
         Map.entry("InvalidMappedField", InvalidMappedField::new),
+        Map.entry("UnmappedEsField", UnmappedEsField::new),
         Map.entry("KeywordEsField", KeywordEsField::new),
         Map.entry("MultiTypeEsField", MultiTypeEsField::new),
         Map.entry("TextEsField", TextEsField::new),

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

@@ -107,6 +107,10 @@ public Map<String, Set<String>> getTypesToIndices() {
         return typesToIndices;
     }
 
+    public InvalidMappedField withTypesToIndices(Map<String, Set<String>> typesToIndices) {
+        return new InvalidMappedField(getName(), makeErrorMessage(typesToIndices), getProperties(), typesToIndices);
+    }
+
     private static String makeErrorMessage(Map<String, Set<String>> typesToIndices) {
         StringBuilder errorMessage = new StringBuilder();
         errorMessage.append("mapped as [");

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

@@ -0,0 +1,165 @@
+/*
+ * 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 org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.core.Strings;
+import org.elasticsearch.xpack.esql.core.expression.Expression;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+import static org.elasticsearch.xpack.esql.core.type.DataType.KEYWORD;
+import static org.elasticsearch.xpack.esql.core.util.PlanStreamInput.readCachedStringWithVersionCheck;
+import static org.elasticsearch.xpack.esql.core.util.PlanStreamOutput.writeCachedStringWithVersionCheck;
+
+/// A special marker for fields explicitly marked as unmapped in the query. These need to be explicitly marked, so we know to search for
+/// during data loading. By default, the [DataType] of these fields is [DataType#KEYWORD], since we can always convert a [DataType#KEYWORD]
+/// to any other type. However, this can cause type conflicts with partially mapped types, i.e, types which appear in some indices but not
+/// all. In that sense, this field is very similar to the [MultiTypeEsField] feature, except we allow for unmapped fields.
+///
+/// Check out the [State] algebraic data type for the different kinds of unmapped fields.
+public class UnmappedEsField extends EsField {
+    private UnmappedEsField(State state, String name, DataType dataType, Map<String, EsField> properties) {
+        super(name, dataType, properties, true /* aggregatable */);
+        this.state = state;
+    }
+
+    public sealed interface State {}
+
+    /// A field which is either unmapped in all indices, or mapped to [DataType#KEYWORD] in all indices where it is mapped.
+    public enum NoConflicts implements State {
+        INSTANCE;
+    }
+
+    /// A field which is mapped to a single type, which is not a [DataType#KEYWORD]. This can be resolved using a cast, similar to union
+    /// types. This is treated differently from [InvalidMappedField], since resolving this conflict doesn't use [MultiTypeEsField].
+    public record SimpleConflict(DataType otherType) implements State {
+        public SimpleConflict {
+            if (otherType == KEYWORD) {
+                throw new IllegalArgumentException("Use NoConflicts.INSTANCE for KEYWORD");
+            }
+        }
+    }
+
+    /// A field which is mapped to a single type, but that type is not [DataType#KEYWORD]. This is resolved using the specified conversions.
+    public record SimpleResolution(Expression unmappedConversion, Expression mappedConversion) implements State {}
+
+    /// A field which is mapped to more than one type in multiple indices, *in addition* to the unmapped case which is always treated as
+    /// [DataType#KEYWORD]. This can be resolved using a cast, similar to union types.
+    public record Invalid(InvalidMappedField invalidMappedField) implements State {}
+
+    /// A field which is mapped to different types in different indices, but resolved using union types. In mapped indices, we treat this
+    /// as a union type, and use the specified conversion for unmapped indices.
+    public record MultiType(Expression conversionFromKeyword, MultiTypeEsField multiTypeEsField) implements State {}
+
+    private final State state;
+
+    public State getState() {
+        return state;
+    }
+
+    public static UnmappedEsField fromField(EsField f) {
+        State state = switch (f) {
+            case InvalidMappedField imf -> {
+                var newTypesToIndices = new TreeMap<>(imf.getTypesToIndices());
+                newTypesToIndices.compute(KEYWORD.typeName(), (k, v) -> v == null ? new TreeSet<>() : new TreeSet<>(v))
+                    .add("unmapped field");
+                yield new Invalid(imf.withTypesToIndices(newTypesToIndices));
+            }
+            case MultiTypeEsField mf -> throw new IllegalArgumentException("Use fromMultiType for MultiTypeEsField");
+            default -> f.getDataType() == KEYWORD ? UnmappedEsField.NoConflicts.INSTANCE : new SimpleConflict(f.getDataType());
+        };
+        return new UnmappedEsField(state, f.getName(), f.getDataType(), f.getProperties());
+    }
+
+    public static UnmappedEsField fromStandalone(String name) {
+        return new UnmappedEsField(NoConflicts.INSTANCE, name, KEYWORD, Map.of());
+    }
+
+    public static UnmappedEsField fromMultiType(Expression expression, MultiTypeEsField multiTypeEsField) {
+        return new UnmappedEsField(
+            new MultiType(expression, multiTypeEsField),
+            multiTypeEsField.getName(),
+            multiTypeEsField.getDataType(),
+            multiTypeEsField.getProperties()
+        );
+    }
+
+    public static UnmappedEsField simpleResolution(Expression unmappedConv, Expression mappedConv, String name) {
+        assert unmappedConv.dataType() == mappedConv.dataType()
+            : Strings.format(
+                "Both conversions must have the same target type, but got [%s, %s]",
+                unmappedConv.dataType(),
+                mappedConv.dataType()
+            );
+        assert unmappedConv.children().get(0).dataType() == KEYWORD
+            : Strings.format("Unmapped conversion must be from keyword, but got [%s]", unmappedConv.children().get(0).dataType());
+        assert mappedConv.children().get(0).dataType() != KEYWORD : Strings.format("Unmapped conversion must be from non-keyword");
+        return new UnmappedEsField(new SimpleResolution(unmappedConv, mappedConv), name, unmappedConv.dataType(), Map.of());
+    }
+
+    @Override
+    public void writeContent(StreamOutput out) throws IOException {
+        switch (state) {
+            case NoConflicts unused -> {
+                out.writeInt(0);
+            }
+            case SimpleConflict sf -> {
+                out.writeInt(1);
+                sf.otherType().writeTo(out);
+            }
+            case SimpleResolution sr -> {
+                out.writeInt(2);
+                out.writeNamedWriteable(sr.unmappedConversion());
+                out.writeNamedWriteable(sr.mappedConversion());
+            }
+            case Invalid invalid -> {
+                out.writeInt(3);
+                invalid.invalidMappedField().writeTo(out);
+            }
+            case MultiType mt -> {
+                out.writeInt(4);
+                out.writeNamedWriteable(mt.conversionFromKeyword());
+                mt.multiTypeEsField.writeTo(out);
+            }
+        }
+        writeCachedStringWithVersionCheck(out, getName());
+        getDataType().writeTo(out);
+        out.writeMap(getProperties(), (o, x) -> x.writeTo(out));
+    }
+
+    UnmappedEsField(StreamInput in) throws IOException {
+        this(readState(in), readCachedStringWithVersionCheck(in), DataType.readFrom(in), in.readImmutableMap(EsField::readFrom));
+    }
+
+    private static State readState(StreamInput in) throws IOException {
+        var ordinal = in.readInt();
+        return switch (ordinal) {
+            case 0 -> NoConflicts.INSTANCE;
+            case 1 -> new SimpleConflict(DataType.readFrom(in));
+            case 2 -> new SimpleResolution(in.readNamedWriteable(Expression.class), in.readNamedWriteable(Expression.class));
+            case 3 -> new Invalid(InvalidMappedField.readFrom(in));
+            case 4 -> new MultiType(in.readNamedWriteable(Expression.class), MultiTypeEsField.readFrom(in));
+            default -> throw new AssertionError("Unexpected ordinal: " + ordinal);
+        };
+    }
+
+    @Override
+    public String getWriteableName() {
+        return "UnmappedEsField";
+    }
+
+    @Override
+    public String toString() {
+        return Strings.format("UnmappedEsField{state=%s}", state);
+    }
+}

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,21 @@ public static <T> List<T> nullSafeList(T... entries) {
         }
         return list;
     }
+
+    public static <T, S extends T> List<S> collectType(Collection<T> list, Class<S> clazz) {
+        return list.stream().<S>mapMulti((e, c) -> {
+            if (clazz.isInstance(e)) {
+                c.accept(clazz.cast(e));
+            }
+        }).toList();
+    }
+
+    public static <T> OptionalInt findIndex(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-core/src/test/java/org/elasticsearch/xpack/esql/core/util/TestUtils.java

@@ -15,6 +15,7 @@
 import org.elasticsearch.xpack.esql.core.type.DataType;
 import org.elasticsearch.xpack.esql.core.type.EsField;
 
+import java.util.Collection;
 import java.util.regex.Pattern;
 
 import static java.util.Collections.emptyMap;
@@ -24,6 +25,8 @@
 import static org.elasticsearch.test.ESTestCase.randomZone;
 import static org.elasticsearch.xpack.esql.core.tree.Source.EMPTY;
 import static org.elasticsearch.xpack.esql.core.type.DataType.INTEGER;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasSize;
 
 public final class TestUtils {
     private TestUtils() {}
@@ -68,4 +71,9 @@ public static FieldAttribute getFieldAttribute(String name, DataType dataType) {
     public static String stripThrough(String input) {
         return WS_PATTERN.matcher(input).replaceAll(StringUtils.EMPTY);
     }
+
+    public static <T> T assertSingleton(Collection<T> collection) {
+        assertThat(collection, hasSize(1));
+        return collection.iterator().next();
+    }
 }

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,12 @@ 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 CLIENT_IPS = new TestDataset("clientips");
     private static final TestDataset CLIENT_IPS_LOOKUP = CLIENT_IPS.withIndex("clientips_lookup")
         .withSetting("clientips_lookup-settings.json");
@@ -128,6 +136,8 @@ 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(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_sample_data.json

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