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/InvalidMappedField.java

@@ -45,7 +45,7 @@ public InvalidMappedField(String name, String errorMessage) {
      * Constructor supporting union types, used in ES|QL.
      */
     public InvalidMappedField(String name, Map<String, Set<String>> typesToIndices) {
-        this(name, makeErrorMessage(typesToIndices), new TreeMap<>(), typesToIndices);
+        this(name, makeErrorMessage(typesToIndices, false), new TreeMap<>(), typesToIndices);
     }
 
     private InvalidMappedField(String name, String errorMessage, Map<String, EsField> properties, Map<String, Set<String>> typesToIndices) {
@@ -107,12 +107,21 @@ public Map<String, Set<String>> getTypesToIndices() {
         return typesToIndices;
     }
 
-    private static String makeErrorMessage(Map<String, Set<String>> typesToIndices) {
+    public static String makeErrorsMessageIncludingInsistKeyword(Map<String, Set<String>> typesToIndices) {
+        return makeErrorMessage(typesToIndices, true);
+    }
+
+    private static String makeErrorMessage(Map<String, Set<String>> typesToIndices, boolean includeInsistKeyword) {
         StringBuilder errorMessage = new StringBuilder();
+        var isInsistKeywordOnlyKeyword = includeInsistKeyword && typesToIndices.containsKey(DataType.KEYWORD.typeName()) == false;
         errorMessage.append("mapped as [");
-        errorMessage.append(typesToIndices.size());
+        errorMessage.append(typesToIndices.size() + (isInsistKeywordOnlyKeyword ? 1 : 0));
         errorMessage.append("] incompatible types: ");
         boolean first = true;
+        if (isInsistKeywordOnlyKeyword) {
+            first = false;
+            errorMessage.append("[keyword] enforced by INSIST command");
+        }
         for (Map.Entry<String, Set<String>> e : typesToIndices.entrySet()) {
             if (first) {
                 first = false;
@@ -121,7 +130,12 @@ private static String makeErrorMessage(Map<String, Set<String>> typesToIndices)
             }
             errorMessage.append("[");
             errorMessage.append(e.getKey());
-            errorMessage.append("] in ");
+            errorMessage.append("] ");
+            if (e.getKey().equals(DataType.KEYWORD.typeName()) && includeInsistKeyword) {
+                errorMessage.append("enforced by INSIST command and in ");
+            } else {
+                errorMessage.append("in ");
+            }
             if (e.getValue().size() <= 3) {
                 errorMessage.append(e.getValue());
             } else {

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

@@ -0,0 +1,30 @@
+/*
+ * 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;
+
+/**
+ * This class is used as a marker for fields that may be unmapped, where an unmapped field is a field which exists in the _source but is not
+ * mapped in the index. Note that this field may be mapped for some indices, but is unmapped in at least one of them.
+ * For indices where the field is unmapped, we will try to load them directly from _source.
+ */
+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/qa/server/mixed-cluster/src/javaRestTest/java/org/elasticsearch/xpack/esql/qa/mixed/MixedClusterEsqlSpecIT.java

@@ -21,6 +21,7 @@
 
 import static org.elasticsearch.xpack.esql.CsvTestUtils.isEnabled;
 import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.JOIN_LOOKUP_V12;
+import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.SOURCE_FIELD_MAPPING;
 
 public class MixedClusterEsqlSpecIT extends EsqlSpecTestCase {
     @ClassRule
@@ -90,6 +91,11 @@ protected boolean supportsIndexModeLookup() throws IOException {
         return hasCapabilities(List.of(JOIN_LOOKUP_V12.capabilityName()));
     }
 
+    @Override
+    protected boolean supportsSourceFieldMapping() throws IOException {
+        return hasCapabilities(List.of(SOURCE_FIELD_MAPPING.capabilityName()));
+    }
+
     @Override
     protected boolean deduplicateExactWarnings() {
         /*

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

@@ -52,6 +52,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;
@@ -127,6 +128,8 @@ protected void shouldSkipTest(String testName) throws IOException {
         assumeFalse("INLINESTATS not yet supported in CCS", testCase.requiredCapabilities.contains(JOIN_PLANNING_V1.capabilityName()));
         assumeFalse("INLINESTATS not yet supported in CCS", testCase.requiredCapabilities.contains(INLINESTATS_V3.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()));
     }
 
     @Override
@@ -294,4 +297,9 @@ protected boolean supportsIndexModeLookup() throws IOException {
         // return hasCapabilities(List.of(JOIN_LOOKUP_V10.capabilityName()));
         return false;
     }
+
+    @Override
+    protected boolean supportsSourceFieldMapping() {
+        return false;
+    }
 }

x-pack/plugin/esql/qa/server/multi-node/src/javaRestTest/java/org/elasticsearch/xpack/esql/qa/multi_node/EsqlSpecIT.java

@@ -12,6 +12,8 @@
 import org.elasticsearch.xpack.esql.qa.rest.EsqlSpecTestCase;
 import org.junit.ClassRule;
 
+import java.io.IOException;
+
 public class EsqlSpecIT extends EsqlSpecTestCase {
     @ClassRule
     public static ElasticsearchCluster cluster = Clusters.testCluster(spec -> spec.plugin("inference-service-test"));
@@ -39,7 +41,7 @@ protected boolean enableRoundingDoubleValuesOnAsserting() {
     }
 
     @Override
-    protected boolean shouldSkipTestsWithSemanticTextFields() {
-        return true;
+    protected boolean supportsSourceFieldMapping() throws IOException {
+        return false;
     }
 }

x-pack/plugin/esql/qa/server/single-node/src/javaRestTest/java/org/elasticsearch/xpack/esql/qa/single_node/EsqlSpecIT.java

@@ -47,4 +47,9 @@ protected boolean enableRoundingDoubleValuesOnAsserting() {
     protected boolean shouldSkipTestsWithSemanticTextFields() {
         return cluster.getNumNodes() > 1;
     }
+
+    @Override
+    protected boolean supportsSourceFieldMapping() {
+        return cluster.getNumNodes() == 1;
+    }
 }

x-pack/plugin/esql/qa/server/src/main/java/org/elasticsearch/xpack/esql/qa/rest/EsqlSpecTestCase.java

@@ -71,6 +71,7 @@
 import static org.elasticsearch.xpack.esql.CsvTestsDataLoader.loadDataSetIntoEs;
 import static org.elasticsearch.xpack.esql.EsqlTestUtils.classpathResources;
 import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.SEMANTIC_TEXT_TYPE;
+import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.SOURCE_FIELD_MAPPING;
 
 // This test can run very long in serverless configurations
 @TimeoutSuite(millis = 30 * TimeUnits.MINUTE)
@@ -135,8 +136,11 @@ public void setup() throws IOException {
         if (supportsInferenceTestService() && clusterHasInferenceEndpoint(client()) == false) {
             createInferenceEndpoint(client());
         }
-        if (indexExists(availableDatasetsForEs(client(), supportsIndexModeLookup()).iterator().next().indexName()) == false) {
-            loadDataSetIntoEs(client(), supportsIndexModeLookup());
+
+        boolean supportsLookup = supportsIndexModeLookup();
+        boolean supportsSourceMapping = supportsSourceFieldMapping();
+        if (indexExists(availableDatasetsForEs(client(), supportsLookup, supportsSourceMapping).iterator().next().indexName()) == false) {
+            loadDataSetIntoEs(client(), supportsLookup, supportsSourceMapping);
         }
     }
 
@@ -183,6 +187,9 @@ protected void shouldSkipTest(String testName) throws IOException {
         if (shouldSkipTestsWithSemanticTextFields()) {
             assumeFalse("semantic_text tests are muted", testCase.requiredCapabilities.contains(SEMANTIC_TEXT_TYPE.capabilityName()));
         }
+        if (supportsSourceFieldMapping() == false) {
+            assumeFalse("source mapping tests are muted", testCase.requiredCapabilities.contains(SOURCE_FIELD_MAPPING.capabilityName()));
+        }
     }
 
     protected static void checkCapabilities(RestClient client, TestFeatureService testFeatureService, String testName, CsvTestCase testCase)
@@ -240,6 +247,10 @@ protected boolean supportsIndexModeLookup() throws IOException {
         return true;
     }
 
+    protected boolean supportsSourceFieldMapping() throws IOException {
+        return true;
+    }
+
     protected final void doTest() throws Throwable {
         RequestObjectBuilder builder = new RequestObjectBuilder(randomFrom(XContentType.values()));
 

x-pack/plugin/esql/qa/server/src/main/java/org/elasticsearch/xpack/esql/qa/rest/generative/GenerativeRestTest.java

@@ -57,7 +57,7 @@ public abstract class GenerativeRestTest extends ESRestTestCase {
     @Before
     public void setup() throws IOException {
         if (indexExists(CSV_DATASET_MAP.keySet().iterator().next()) == false) {
-            loadDataSetIntoEs(client(), true);
+            loadDataSetIntoEs(client(), true, true);
         }
     }