ESQL: Initial support for unmapped fields (elastic#119886)

This PR adds initial support for unmapped fields, using the INSIST clause. For starters, this unmapped fields without a cast.
Note that the INSIST keyword is potentially a placeholder, as the method of defining an unmapped field might change in the future, e.g., use a special magic function. As this is currently under development, the actual syntax is INSIST_🐔.

First stage of elastic#120072.

Specifically, the following features are implemented in this PR:

* Support for INSIST keyword without a cast. In particular, if the type being INSISTed upon is mapped to anything other than KEYWORD, it will result in an InvalidMappedField. There is no support for union type resolution on top of INSIST. Future PRs will handle these conflicts. There is support for multiple parameters, or INSIST on top of INSIST which is equivalent.
* Enforcing that INSIST must always be on top of a FROM or another INSIST. While this may change in the future, e.g., handling cases like `FROM foo | EVAL x = 1 | INSIST bar` will not be done in this PR, as it makes handling INSIST too complicated.

Author: GalLalouche

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

@@ -85,6 +85,11 @@ protected boolean supportsIndexModeLookup() throws IOException {
         return hasCapabilities(List.of(JOIN_LOOKUP_V12.capabilityName()));
     }
 
+    @Override
+    protected boolean supportsSourceFieldMapping() {
+        return false;
+    }
+
     @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()));
     }
 
     private TestFeatureService remoteFeaturesService() throws IOException {
@@ -289,4 +292,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

@@ -37,4 +37,9 @@ public EsqlSpecIT(
     protected boolean enableRoundingDoubleValuesOnAsserting() {
         return true;
     }
+
+    @Override
+    protected boolean supportsSourceFieldMapping() {
+        return false;
+    }
 }

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

@@ -42,4 +42,5 @@ protected boolean enableRoundingDoubleValuesOnAsserting() {
         // This suite runs with more than one node and three shards in serverless
         return cluster.getNumNodes() > 1;
     }
+
 }

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

@@ -29,6 +29,7 @@
 import org.elasticsearch.xpack.esql.AssertWarnings;
 import org.elasticsearch.xpack.esql.CsvSpecReader.CsvTestCase;
 import org.elasticsearch.xpack.esql.CsvTestUtils;
+import org.elasticsearch.xpack.esql.CsvTestUtils.ExpectedResults;
 import org.elasticsearch.xpack.esql.EsqlTestUtils;
 import org.elasticsearch.xpack.esql.SpecReader;
 import org.elasticsearch.xpack.esql.plugin.EsqlFeatures;
@@ -61,7 +62,6 @@
 import static org.elasticsearch.xpack.esql.CsvAssert.assertData;
 import static org.elasticsearch.xpack.esql.CsvAssert.assertMetadata;
 import static org.elasticsearch.xpack.esql.CsvSpecReader.specParser;
-import static org.elasticsearch.xpack.esql.CsvTestUtils.ExpectedResults;
 import static org.elasticsearch.xpack.esql.CsvTestUtils.isEnabled;
 import static org.elasticsearch.xpack.esql.CsvTestUtils.loadCsvSpecValues;
 import static org.elasticsearch.xpack.esql.CsvTestsDataLoader.availableDatasetsForEs;
@@ -70,6 +70,7 @@
 import static org.elasticsearch.xpack.esql.CsvTestsDataLoader.deleteInferenceEndpoint;
 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.SOURCE_FIELD_MAPPING;
 
 // This test can run very long in serverless configurations
 @TimeoutSuite(millis = 30 * TimeUnits.MINUTE)
@@ -132,8 +133,10 @@ public void setup() throws IOException {
             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);
         }
     }
 
@@ -172,6 +175,9 @@ protected void shouldSkipTest(String testName) throws IOException {
         }
         checkCapabilities(adminClient(), testFeatureService, testName, testCase);
         assumeTrue("Test " + testName + " is not enabled", isEnabled(testName, instructions, Version.CURRENT));
+        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)
@@ -229,6 +235,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()));