ESQL: Report original_types (#124913)

Adds the `original_types` to the description of ESQL's `unsupported`
fields. This looks like:

```
    {
      "name" : "a",
      "type" : "unsupported",
      "original_types" : [
        "long",
        "text"
      ]
    }
```

for union types. And like:

```
    {
      "name" : "a",
      "type" : "unsupported",
      "original_types" : [
        "date_range"
      ]
    }
```

for truly unsupported types.

This information is useful for the UI. For union types it can suggest
that users append a cast.

docs/changelog/124913.yaml

@@ -0,0 +1,5 @@
+pr: 124913
+summary: Report `original_types`
+area: ES|QL
+type: enhancement
+issues: []

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -203,6 +203,7 @@ public class TransportVersions {
     public static final TransportVersion ESQL_AGGREGATE_METRIC_DOUBLE_LITERAL = def(9_035_0_00);
     public static final TransportVersion INDEX_METADATA_INCLUDES_RECENT_WRITE_LOAD = def(9_036_0_00);
     public static final TransportVersion RERANK_COMMON_OPTIONS_ADDED = def(9_037_0_00);
+    public static final TransportVersion ESQL_REPORT_ORIGINAL_TYPES = def(9_038_00_0);
 
     /*
      * STOP! READ THIS FIRST! No, really,

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/Attribute.java

@@ -137,4 +137,14 @@ public abstract class Attribute extends NamedExpression {
     }
 
     protected abstract String label();
+
+    /**
+     * If this field is unsupported this contains the underlying ES types. If there
+     * is a type conflict this will have many elements, some or all of which may
+     * be actually supported types.
+     */
+    @Nullable
+    public List<String> originalTypes() {
+        return null;
+    }
 }

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

@@ -6,10 +6,14 @@
  */
 package org.elasticsearch.xpack.esql.core.type;
 
+import org.elasticsearch.TransportVersions;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.xpack.esql.core.util.PlanStreamInput;
+import org.elasticsearch.xpack.esql.core.util.PlanStreamOutput;
 
 import java.io.IOException;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.TreeMap;
@@ -23,32 +27,39 @@ import static org.elasticsearch.xpack.esql.core.util.PlanStreamOutput.writeCache
  */
 public class UnsupportedEsField extends EsField {
 
-    private final String originalType;
+    private final List<String> originalTypes;
     private final String inherited; // for fields belonging to parents (or grandparents) that have an unsupported type
 
-    public UnsupportedEsField(String name, String originalType) {
-        this(name, originalType, null, new TreeMap<>());
+    public UnsupportedEsField(String name, List<String> originalTypes) {
+        this(name, originalTypes, null, new TreeMap<>());
     }
 
-    public UnsupportedEsField(String name, String originalType, String inherited, Map<String, EsField> properties) {
+    public UnsupportedEsField(String name, List<String> originalTypes, String inherited, Map<String, EsField> properties) {
         super(name, DataType.UNSUPPORTED, properties, false);
-        this.originalType = originalType;
+        this.originalTypes = originalTypes;
         this.inherited = inherited;
     }
 
     public UnsupportedEsField(StreamInput in) throws IOException {
-        this(
-            readCachedStringWithVersionCheck(in),
-            readCachedStringWithVersionCheck(in),
-            in.readOptionalString(),
-            in.readImmutableMap(EsField::readFrom)
-        );
+        this(readCachedStringWithVersionCheck(in), readOriginalTypes(in), in.readOptionalString(), in.readImmutableMap(EsField::readFrom));
+    }
+
+    private static List<String> readOriginalTypes(StreamInput in) throws IOException {
+        if (in.getTransportVersion().onOrAfter(TransportVersions.ESQL_REPORT_ORIGINAL_TYPES)) {
+            return in.readCollectionAsList(i -> ((PlanStreamInput) i).readCachedString());
+        } else {
+            return List.of(readCachedStringWithVersionCheck(in).split(","));
+        }
     }
 
     @Override
     public void writeContent(StreamOutput out) throws IOException {
         writeCachedStringWithVersionCheck(out, getName());
-        writeCachedStringWithVersionCheck(out, getOriginalType());
+        if (out.getTransportVersion().onOrAfter(TransportVersions.ESQL_REPORT_ORIGINAL_TYPES)) {
+            out.writeCollection(getOriginalTypes(), (o, s) -> ((PlanStreamOutput) o).writeCachedString(s));
+        } else {
+            writeCachedStringWithVersionCheck(out, String.join(",", getOriginalTypes()));
+        }
         out.writeOptionalString(getInherited());
         out.writeMap(getProperties(), (o, x) -> x.writeTo(out));
     }
@@ -57,8 +68,8 @@ public class UnsupportedEsField extends EsField {
         return "UnsupportedEsField";
     }
 
-    public String getOriginalType() {
-        return originalType;
+    public List<String> getOriginalTypes() {
+        return originalTypes;
     }
 
     public String getInherited() {
@@ -81,11 +92,11 @@ public class UnsupportedEsField extends EsField {
             return false;
         }
         UnsupportedEsField that = (UnsupportedEsField) o;
-        return Objects.equals(originalType, that.originalType) && Objects.equals(inherited, that.inherited);
+        return Objects.equals(originalTypes, that.originalTypes) && Objects.equals(inherited, that.inherited);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(super.hashCode(), originalType, inherited);
+        return Objects.hash(super.hashCode(), originalTypes, inherited);
     }
 }

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

@@ -327,12 +327,13 @@ public abstract class FieldExtractorTestCase extends ESRestTestCase {
     }
 
     public void testFlattenedUnsupported() throws IOException {
+        assumeOriginalTypesReported();
         new Test("flattened").createIndex("test", "flattened");
         index("test", """
             {"flattened": {"a": "foo"}}""");
         Map<String, Object> result = runEsql("FROM test* | LIMIT 2");
 
-        assertResultMap(result, List.of(columnInfo("flattened", "unsupported")), List.of(matchesList().item(null)));
+        assertResultMap(result, List.of(unsupportedColumnInfo("flattened", "flattened")), List.of(matchesList().item(null)));
     }
 
     public void testEmptyMapping() throws IOException {
@@ -689,6 +690,7 @@ public abstract class FieldExtractorTestCase extends ESRestTestCase {
      * </pre>.
      */
     public void testIncompatibleTypes() throws IOException {
+        assumeOriginalTypesReported();
         keywordTest().createIndex("test1", "f");
         index("test1", """
             {"f": "f1"}""");
@@ -697,7 +699,11 @@ public abstract class FieldExtractorTestCase extends ESRestTestCase {
             {"f": 1}""");
 
         Map<String, Object> result = runEsql("FROM test*");
-        assertResultMap(result, List.of(columnInfo("f", "unsupported")), List.of(matchesList().item(null), matchesList().item(null)));
+        assertResultMap(
+            result,
+            List.of(unsupportedColumnInfo("f", "keyword", "long")),
+            List.of(matchesList().item(null), matchesList().item(null))
+        );
         ResponseException e = expectThrows(ResponseException.class, () -> runEsql("FROM test* | SORT f | LIMIT 3"));
         String err = EntityUtils.toString(e.getResponse().getEntity());
         assertThat(
@@ -758,10 +764,7 @@ public abstract class FieldExtractorTestCase extends ESRestTestCase {
      * </pre>.
      */
     public void testMergeKeywordAndObject() throws IOException {
-        assumeTrue(
-            "order of fields in error message inconsistent before 8.14",
-            getCachedNodesVersions().stream().allMatch(v -> Version.fromString(v).onOrAfter(Version.V_8_14_0))
-        );
+        assumeOriginalTypesReported();
         keywordTest().createIndex("test1", "file");
         index("test1", """
             {"file": "f1"}""");
@@ -797,7 +800,7 @@ public abstract class FieldExtractorTestCase extends ESRestTestCase {
         Map<String, Object> result = runEsql("FROM test* | SORT file.raw | LIMIT 2");
         assertResultMap(
             result,
-            List.of(columnInfo("file", "unsupported"), columnInfo("file.raw", "keyword")),
+            List.of(unsupportedColumnInfo("file", "keyword", "object"), columnInfo("file.raw", "keyword")),
             List.of(matchesList().item(null).item("o2"), matchesList().item(null).item(null))
         );
     }
@@ -817,6 +820,7 @@ public abstract class FieldExtractorTestCase extends ESRestTestCase {
      * </pre>.
      */
     public void testPropagateUnsupportedToSubFields() throws IOException {
+        assumeOriginalTypesReported();
         createIndex("test", index -> {
             index.startObject("properties");
             index.startObject("f");
@@ -842,7 +846,7 @@ public abstract class FieldExtractorTestCase extends ESRestTestCase {
         Map<String, Object> result = runEsql("FROM test* | LIMIT 2");
         assertResultMap(
             result,
-            List.of(columnInfo("f", "unsupported"), columnInfo("f.raw", "unsupported")),
+            List.of(unsupportedColumnInfo("f", "ip_range"), unsupportedColumnInfo("f.raw", "ip_range")),
             List.of(matchesList().item(null).item(null))
         );
     }
@@ -867,10 +871,7 @@ public abstract class FieldExtractorTestCase extends ESRestTestCase {
      * </pre>.
      */
     public void testMergeUnsupportedAndObject() throws IOException {
-        assumeTrue(
-            "order of fields in error message inconsistent before 8.14",
-            getCachedNodesVersions().stream().allMatch(v -> Version.fromString(v).onOrAfter(Version.V_8_14_0))
-        );
+        assumeOriginalTypesReported();
         createIndex("test1", index -> {
             index.startObject("properties");
             index.startObject("f").field("type", "ip_range").endObject();
@@ -905,7 +906,7 @@ public abstract class FieldExtractorTestCase extends ESRestTestCase {
         Map<String, Object> result = runEsql("FROM test* | LIMIT 2");
         assertResultMap(
             result,
-            List.of(columnInfo("f", "unsupported"), columnInfo("f.raw", "unsupported")),
+            List.of(unsupportedColumnInfo("f", "ip_range"), unsupportedColumnInfo("f.raw", "ip_range")),
             List.of(matchesList().item(null).item(null), matchesList().item(null).item(null))
         );
     }
@@ -958,10 +959,7 @@ public abstract class FieldExtractorTestCase extends ESRestTestCase {
      * In an ideal world we'd promote the {@code integer} to an {@code long} and just go.
      */
     public void testLongIntegerConflict() throws IOException {
-        assumeTrue(
-            "order of fields in error message inconsistent before 8.14",
-            getCachedNodesVersions().stream().allMatch(v -> Version.fromString(v).onOrAfter(Version.V_8_14_0))
-        );
+        assumeOriginalTypesReported();
         longTest().sourceMode(SourceMode.DEFAULT).createIndex("test1", "emp_no");
         index("test1", """
             {"emp_no": 1}""");
@@ -980,7 +978,11 @@ public abstract class FieldExtractorTestCase extends ESRestTestCase {
         );
 
         Map<String, Object> result = runEsql("FROM test* | LIMIT 2");
-        assertResultMap(result, List.of(columnInfo("emp_no", "unsupported")), List.of(matchesList().item(null), matchesList().item(null)));
+        assertResultMap(
+            result,
+            List.of(unsupportedColumnInfo("emp_no", "integer", "long")),
+            List.of(matchesList().item(null), matchesList().item(null))
+        );
     }
 
     /**
@@ -1000,10 +1002,7 @@ public abstract class FieldExtractorTestCase extends ESRestTestCase {
      * In an ideal world we'd promote the {@code short} to an {@code integer} and just go.
      */
     public void testIntegerShortConflict() throws IOException {
-        assumeTrue(
-            "order of fields in error message inconsistent before 8.14",
-            getCachedNodesVersions().stream().allMatch(v -> Version.fromString(v).onOrAfter(Version.V_8_14_0))
-        );
+        assumeOriginalTypesReported();
         intTest().sourceMode(SourceMode.DEFAULT).createIndex("test1", "emp_no");
         index("test1", """
             {"emp_no": 1}""");
@@ -1022,7 +1021,11 @@ public abstract class FieldExtractorTestCase extends ESRestTestCase {
         );
 
         Map<String, Object> result = runEsql("FROM test* | LIMIT 2");
-        assertResultMap(result, List.of(columnInfo("emp_no", "unsupported")), List.of(matchesList().item(null), matchesList().item(null)));
+        assertResultMap(
+            result,
+            List.of(unsupportedColumnInfo("emp_no", "integer", "short")),
+            List.of(matchesList().item(null), matchesList().item(null))
+        );
     }
 
     /**
@@ -1048,10 +1051,7 @@ public abstract class FieldExtractorTestCase extends ESRestTestCase {
      * </pre>.
      */
     public void testTypeConflictInObject() throws IOException {
-        assumeTrue(
-            "order of fields in error message inconsistent before 8.14",
-            getCachedNodesVersions().stream().allMatch(v -> Version.fromString(v).onOrAfter(Version.V_8_14_0))
-        );
+        assumeOriginalTypesReported();
         createIndex("test1", empNoInObject("integer"));
         index("test1", """
             {"foo": {"emp_no": 1}}""");
@@ -1060,7 +1060,10 @@ public abstract class FieldExtractorTestCase extends ESRestTestCase {
             {"foo": {"emp_no": "cat"}}""");
 
         Map<String, Object> result = runEsql("FROM test* | LIMIT 3");
-        assertMap(result, getResultMatcher(result).entry("columns", List.of(columnInfo("foo.emp_no", "unsupported"))).extraOk());
+        assertMap(
+            result,
+            getResultMatcher(result).entry("columns", List.of(unsupportedColumnInfo("foo.emp_no", "integer", "keyword"))).extraOk()
+        );
 
         ResponseException e = expectThrows(ResponseException.class, () -> runEsql("FROM test* | SORT foo.emp_no | LIMIT 3"));
         String err = EntityUtils.toString(e.getResponse().getEntity());
@@ -1370,6 +1373,12 @@ public abstract class FieldExtractorTestCase extends ESRestTestCase {
         );
     }
 
+    private void assumeOriginalTypesReported() throws IOException {
+        var capsName = EsqlCapabilities.Cap.REPORT_ORIGINAL_TYPES.name().toLowerCase(Locale.ROOT);
+        boolean requiredClusterCapability = clusterHasCapability("POST", "/_query", List.of(), List.of(capsName)).orElse(false);
+        assumeTrue("This test makes sense for versions that report original types", requiredClusterCapability);
+    }
+
     private CheckedConsumer<XContentBuilder, IOException> empNoInObject(String empNoType) {
         return index -> {
             index.startObject("properties");
@@ -1705,6 +1714,10 @@ public abstract class FieldExtractorTestCase extends ESRestTestCase {
         return Map.of("name", name, "type", type);
     }
 
+    private static Map<String, Object> unsupportedColumnInfo(String name, String... originalTypes) {
+        return Map.of("name", name, "type", "unsupported", "original_types", List.of(originalTypes));
+    }
+
     private static void index(String name, String... docs) throws IOException {
         Request request = new Request("POST", "/" + name + "/_bulk");
         request.addParameter("refresh", "true");

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

@@ -24,6 +24,7 @@ import java.io.IOException;
 import java.io.InputStream;
 import java.util.Collections;
 import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Map;
 
 import static java.util.Collections.emptyMap;
@@ -110,7 +111,7 @@ public class LoadMapping {
                 field = DateEsField.dateEsField(name, properties, docValues);
             } else if (esDataType == UNSUPPORTED) {
                 String type = content.get("type").toString();
-                field = new UnsupportedEsField(name, type, null, properties);
+                field = new UnsupportedEsField(name, List.of(type), null, properties);
                 propagateUnsupportedType(name, type, properties);
             } else {
                 field = new EsField(name, esDataType, properties, docValues);
@@ -165,9 +166,9 @@ public class LoadMapping {
                 UnsupportedEsField u;
                 if (field instanceof UnsupportedEsField) {
                     u = (UnsupportedEsField) field;
-                    u = new UnsupportedEsField(u.getName(), originalType, inherited, u.getProperties());
+                    u = new UnsupportedEsField(u.getName(), List.of(originalType), inherited, u.getProperties());
                 } else {
-                    u = new UnsupportedEsField(field.getName(), originalType, inherited, field.getProperties());
+                    u = new UnsupportedEsField(field.getName(), List.of(originalType), inherited, field.getProperties());
                 }
                 entry.setValue(u);
                 propagateUnsupportedType(inherited, originalType, u.getProperties());

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/AsyncEsqlQueryActionIT.java

@@ -94,7 +94,7 @@ public class AsyncEsqlQueryActionIT extends AbstractPausableIntegTestCase {
             try (var finalResponse = future.get()) {
                 assertThat(finalResponse, notNullValue());
                 assertThat(finalResponse.isRunning(), is(false));
-                assertThat(finalResponse.columns(), equalTo(List.of(new ColumnInfoImpl("sum(pause_me)", "long"))));
+                assertThat(finalResponse.columns(), equalTo(List.of(new ColumnInfoImpl("sum(pause_me)", "long", null))));
                 assertThat(getValuesList(finalResponse).size(), equalTo(1));
             }
 
@@ -103,7 +103,7 @@ public class AsyncEsqlQueryActionIT extends AbstractPausableIntegTestCase {
             try (var finalResponse = again.get()) {
                 assertThat(finalResponse, notNullValue());
                 assertThat(finalResponse.isRunning(), is(false));
-                assertThat(finalResponse.columns(), equalTo(List.of(new ColumnInfoImpl("sum(pause_me)", "long"))));
+                assertThat(finalResponse.columns(), equalTo(List.of(new ColumnInfoImpl("sum(pause_me)", "long", null))));
                 assertThat(getValuesList(finalResponse).size(), equalTo(1));
             }
 
@@ -231,7 +231,7 @@ public class AsyncEsqlQueryActionIT extends AbstractPausableIntegTestCase {
 
         try (var response = request.execute().actionGet(60, TimeUnit.SECONDS)) {
             assertThat(response.isRunning(), is(false));
-            assertThat(response.columns(), equalTo(List.of(new ColumnInfoImpl("sum(pause_me)", "long"))));
+            assertThat(response.columns(), equalTo(List.of(new ColumnInfoImpl("sum(pause_me)", "long", null))));
             assertThat(getValuesList(response).size(), equalTo(1));
 
             if (keepOnCompletion) {
@@ -244,7 +244,7 @@ public class AsyncEsqlQueryActionIT extends AbstractPausableIntegTestCase {
                 try (var resp = future.actionGet(60, TimeUnit.SECONDS)) {
                     assertThat(resp.asyncExecutionId().get(), equalTo(id));
                     assertThat(resp.isRunning(), is(false));
-                    assertThat(resp.columns(), equalTo(List.of(new ColumnInfoImpl("sum(pause_me)", "long"))));
+                    assertThat(resp.columns(), equalTo(List.of(new ColumnInfoImpl("sum(pause_me)", "long", null))));
                     assertThat(getValuesList(resp).size(), equalTo(1));
                 }
             } else {

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClusterAsyncQueryIT.java

@@ -31,6 +31,7 @@ import static org.elasticsearch.xpack.esql.action.EsqlAsyncTestUtils.startAsyncQ
 import static org.elasticsearch.xpack.esql.action.EsqlAsyncTestUtils.waitForCluster;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.hasItem;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.lessThanOrEqualTo;
 import static org.hamcrest.Matchers.not;
@@ -151,10 +152,10 @@ public class CrossClusterAsyncQueryIT extends AbstractCrossClusterTestCase {
 
             } else {
                 assertThat(resp.columns().size(), equalTo(4));
-                assertThat(resp.columns().contains(new ColumnInfoImpl("const", "long")), is(true));
-                assertThat(resp.columns().contains(new ColumnInfoImpl("id", "keyword")), is(true));
-                assertThat(resp.columns().contains(new ColumnInfoImpl("tag", "keyword")), is(true));
-                assertThat(resp.columns().contains(new ColumnInfoImpl("v", "long")), is(true));
+                assertThat(resp.columns(), hasItem(new ColumnInfoImpl("const", "long", null)));
+                assertThat(resp.columns(), hasItem(new ColumnInfoImpl("id", "keyword", null)));
+                assertThat(resp.columns(), hasItem(new ColumnInfoImpl("tag", "keyword", null)));
+                assertThat(resp.columns(), hasItem(new ColumnInfoImpl("v", "long", null)));
                 assertThat(resp.values().hasNext(), is(false));  // values should be empty list
 
                 assertNotNull(executionInfo);

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/EsqlActionIT.java

@@ -128,7 +128,7 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
 
     public void testProjectConstant() {
         try (EsqlQueryResponse results = run("from test | eval x = 1 | keep x")) {
-            assertThat(results.columns(), equalTo(List.of(new ColumnInfoImpl("x", "integer"))));
+            assertThat(results.columns(), equalTo(List.of(new ColumnInfoImpl("x", "integer", null))));
             assertThat(getValuesList(results).size(), equalTo(40));
             assertThat(getValuesList(results).get(0).get(0), equalTo(1));
         }
@@ -136,7 +136,7 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
 
     public void testStatsOverConstant() {
         try (EsqlQueryResponse results = run("from test | eval x = 1 | stats x = count(x)")) {
-            assertThat(results.columns(), equalTo(List.of(new ColumnInfoImpl("x", "long"))));
+            assertThat(results.columns(), equalTo(List.of(new ColumnInfoImpl("x", "long", null))));
             assertThat(getValuesList(results).size(), equalTo(1));
             assertThat(getValuesList(results).get(0).get(0), equalTo(40L));
         }
@@ -564,10 +564,14 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
             logger.info(results);
             assertEquals(1, getValuesList(results).size());
             assertEquals(2, getValuesList(results).get(0).size());
-            assertEquals(50, (double) getValuesList(results).get(0).get(results.columns().indexOf(new ColumnInfoImpl("x", "double"))), 1d);
+            assertEquals(
+                50,
+                (double) getValuesList(results).get(0).get(results.columns().indexOf(new ColumnInfoImpl("x", "double", null))),
+                1d
+            );
             assertEquals(
                 43,
-                (double) getValuesList(results).get(0).get(results.columns().indexOf(new ColumnInfoImpl("avg_count", "double"))),
+                (double) getValuesList(results).get(0).get(results.columns().indexOf(new ColumnInfoImpl("avg_count", "double", null))),
                 1d
             );
         }
@@ -577,7 +581,7 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
         try (EsqlQueryResponse results = run("from test | where count > 40")) {
             logger.info(results);
             assertEquals(30, getValuesList(results).size());
-            var countIndex = results.columns().indexOf(new ColumnInfoImpl("count", "long"));
+            var countIndex = results.columns().indexOf(new ColumnInfoImpl("count", "long", null));
             for (List<Object> values : getValuesList(results)) {
                 assertThat((Long) values.get(countIndex), greaterThan(40L));
             }
@@ -588,7 +592,7 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
         try (EsqlQueryResponse results = run("from test | keep count | where count > 40")) {
             logger.info(results);
             assertEquals(30, getValuesList(results).size());
-            int countIndex = results.columns().indexOf(new ColumnInfoImpl("count", "long"));
+            int countIndex = results.columns().indexOf(new ColumnInfoImpl("count", "long", null));
             for (List<Object> values : getValuesList(results)) {
                 assertThat((Long) values.get(countIndex), greaterThan(40L));
             }
@@ -599,7 +603,7 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
         try (EsqlQueryResponse results = run("from test | eval x = count / 2 | where x > 20")) {
             logger.info(results);
             assertEquals(30, getValuesList(results).size());
-            int countIndex = results.columns().indexOf(new ColumnInfoImpl("x", "long"));
+            int countIndex = results.columns().indexOf(new ColumnInfoImpl("x", "long", null));
             for (List<Object> values : getValuesList(results)) {
                 assertThat((Long) values.get(countIndex), greaterThan(20L));
             }
@@ -617,7 +621,7 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
         try (EsqlQueryResponse results = run("row a = null | sort a")) {
             logger.info(results);
             assertEquals(1, getValuesList(results).size());
-            int countIndex = results.columns().indexOf(new ColumnInfoImpl("a", "null"));
+            int countIndex = results.columns().indexOf(new ColumnInfoImpl("a", "null", null));
             assertThat(results.columns().stream().map(ColumnInfo::name).toList(), contains("a"));
             assertThat(results.columns().stream().map(ColumnInfoImpl::type).toList(), contains(DataType.NULL));
             assertNull(getValuesList(results).getFirst().get(countIndex));
@@ -628,7 +632,7 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
         try (EsqlQueryResponse results = run("row a = null | stats by a")) {
             logger.info(results);
             assertEquals(1, getValuesList(results).size());
-            int countIndex = results.columns().indexOf(new ColumnInfoImpl("a", "null"));
+            int countIndex = results.columns().indexOf(new ColumnInfoImpl("a", "null", null));
             assertThat(results.columns().stream().map(ColumnInfo::name).toList(), contains("a"));
             assertThat(results.columns().stream().map(ColumnInfoImpl::type).toList(), contains(DataType.NULL));
             assertNull(getValuesList(results).getFirst().get(countIndex));
@@ -639,7 +643,7 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
         try (EsqlQueryResponse results = run("from test | eval l = length(color)")) {
             logger.info(results);
             assertThat(getValuesList(results), hasSize(40));
-            int countIndex = results.columns().indexOf(new ColumnInfoImpl("l", "integer"));
+            int countIndex = results.columns().indexOf(new ColumnInfoImpl("l", "integer", null));
             for (List<Object> values : getValuesList(results)) {
                 assertThat((Integer) values.get(countIndex), greaterThanOrEqualTo(3));
             }
@@ -658,11 +662,11 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
         try (EsqlQueryResponse results = run("from test | eval newCount = count + 1 | where newCount > 1")) {
             logger.info(results);
             assertEquals(40, getValuesList(results).size());
-            assertThat(results.columns(), hasItem(equalTo(new ColumnInfoImpl("count", "long"))));
-            assertThat(results.columns(), hasItem(equalTo(new ColumnInfoImpl("count_d", "double"))));
-            assertThat(results.columns(), hasItem(equalTo(new ColumnInfoImpl("data", "long"))));
-            assertThat(results.columns(), hasItem(equalTo(new ColumnInfoImpl("data_d", "double"))));
-            assertThat(results.columns(), hasItem(equalTo(new ColumnInfoImpl("time", "long"))));
+            assertThat(results.columns(), hasItem(equalTo(new ColumnInfoImpl("count", "long", null))));
+            assertThat(results.columns(), hasItem(equalTo(new ColumnInfoImpl("count_d", "double", null))));
+            assertThat(results.columns(), hasItem(equalTo(new ColumnInfoImpl("data", "long", null))));
+            assertThat(results.columns(), hasItem(equalTo(new ColumnInfoImpl("data_d", "double", null))));
+            assertThat(results.columns(), hasItem(equalTo(new ColumnInfoImpl("time", "long", null))));
         }
     }
 
@@ -696,7 +700,7 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
             assertEquals(40, getValuesList(results).size());
             assertEquals(1, results.columns().stream().filter(c -> c.name().equals("count")).count());
             int countIndex = results.columns().size() - 1;
-            assertEquals(new ColumnInfoImpl("count", "long"), results.columns().get(countIndex));
+            assertEquals(new ColumnInfoImpl("count", "long", null), results.columns().get(countIndex));
             for (List<Object> values : getValuesList(results)) {
                 assertThat((Long) values.get(countIndex), greaterThanOrEqualTo(42L));
             }
@@ -707,7 +711,7 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
         try (var results = run("from test | eval y = count | rename count as x | keep x, y")) {
             logger.info(results);
             assertEquals(40, getValuesList(results).size());
-            assertThat(results.columns(), contains(new ColumnInfoImpl("x", "long"), new ColumnInfoImpl("y", "long")));
+            assertThat(results.columns(), contains(new ColumnInfoImpl("x", "long", null), new ColumnInfoImpl("y", "long", null)));
             for (List<Object> values : getValuesList(results)) {
                 assertThat((Long) values.get(0), greaterThanOrEqualTo(40L));
                 assertThat(values.get(1), is(values.get(0)));
@@ -722,10 +726,10 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
             assertThat(
                 results.columns(),
                 contains(
-                    new ColumnInfoImpl("x", "long"),
-                    new ColumnInfoImpl("y", "long"),
-                    new ColumnInfoImpl("x2", "long"),
-                    new ColumnInfoImpl("y2", "long")
+                    new ColumnInfoImpl("x", "long", null),
+                    new ColumnInfoImpl("y", "long", null),
+                    new ColumnInfoImpl("x2", "long", null),
+                    new ColumnInfoImpl("y2", "long", null)
                 )
             );
             for (List<Object> values : getValuesList(results)) {
@@ -743,7 +747,11 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
             assertEquals(40, getValuesList(results).size());
             assertThat(
                 results.columns(),
-                contains(new ColumnInfoImpl("x", "long"), new ColumnInfoImpl("y", "long"), new ColumnInfoImpl("z", "long"))
+                contains(
+                    new ColumnInfoImpl("x", "long", null),
+                    new ColumnInfoImpl("y", "long", null),
+                    new ColumnInfoImpl("z", "long", null)
+                )
             );
             for (List<Object> values : getValuesList(results)) {
                 assertThat((Long) values.get(0), greaterThanOrEqualTo(40L));
@@ -757,7 +765,7 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
         try (var results = run("from test | eval cnt = count | rename count as data | keep cnt, data")) {
             logger.info(results);
             assertEquals(40, getValuesList(results).size());
-            assertThat(results.columns(), contains(new ColumnInfoImpl("cnt", "long"), new ColumnInfoImpl("data", "long")));
+            assertThat(results.columns(), contains(new ColumnInfoImpl("cnt", "long", null), new ColumnInfoImpl("data", "long", null)));
             for (List<Object> values : getValuesList(results)) {
                 assertThat(values.get(1), is(values.get(0)));
             }
@@ -927,7 +935,7 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
     public void testFromStatsLimit() {
         try (EsqlQueryResponse results = run("from test | stats ac = avg(count) by data | limit 1")) {
             logger.info(results);
-            assertThat(results.columns(), contains(new ColumnInfoImpl("ac", "double"), new ColumnInfoImpl("data", "long")));
+            assertThat(results.columns(), contains(new ColumnInfoImpl("ac", "double", null), new ColumnInfoImpl("data", "long", null)));
             assertThat(getValuesList(results), contains(anyOf(contains(42.0, 1L), contains(44.0, 2L))));
         }
     }
@@ -935,7 +943,7 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
     public void testFromLimit() {
         try (EsqlQueryResponse results = run("from test | keep data | limit 2")) {
             logger.info(results);
-            assertThat(results.columns(), contains(new ColumnInfoImpl("data", "long")));
+            assertThat(results.columns(), contains(new ColumnInfoImpl("data", "long", null)));
             assertThat(getValuesList(results), contains(anyOf(contains(1L), contains(2L)), anyOf(contains(1L), contains(2L))));
         }
     }
@@ -944,7 +952,7 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
         try (EsqlQueryResponse results = run("from test | keep data | drop data | eval a = 1")) {
             logger.info(results);
             assertThat(results.columns(), hasSize(1));
-            assertThat(results.columns(), contains(new ColumnInfoImpl("a", "integer")));
+            assertThat(results.columns(), contains(new ColumnInfoImpl("a", "integer", null)));
             assertThat(getValuesList(results), is(empty()));
         }
     }
@@ -1233,7 +1241,10 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
     public void testEmptyIndex() {
         assertAcked(client().admin().indices().prepareCreate("test_empty").setMapping("k", "type=keyword", "v", "type=long").get());
         try (EsqlQueryResponse results = run("from test_empty")) {
-            assertThat(results.columns(), equalTo(List.of(new ColumnInfoImpl("k", "keyword"), new ColumnInfoImpl("v", "long"))));
+            assertThat(
+                results.columns(),
+                equalTo(List.of(new ColumnInfoImpl("k", "keyword", null), new ColumnInfoImpl("v", "long", null)))
+            );
             assertThat(getValuesList(results), empty());
         }
     }
@@ -1244,9 +1255,9 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
                 results.columns(),
                 equalTo(
                     List.of(
-                        new ColumnInfoImpl("version", "keyword"),
-                        new ColumnInfoImpl("date", "keyword"),
-                        new ColumnInfoImpl("hash", "keyword")
+                        new ColumnInfoImpl("version", "keyword", null),
+                        new ColumnInfoImpl("date", "keyword", null),
+                        new ColumnInfoImpl("hash", "keyword", null)
                     )
                 )
             );
@@ -1259,7 +1270,7 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
 
     public void testInWithNullValue() {
         try (EsqlQueryResponse results = run("from test | where null in (data, 2) | keep data")) {
-            assertThat(results.columns(), equalTo(List.of(new ColumnInfoImpl("data", "long"))));
+            assertThat(results.columns(), equalTo(List.of(new ColumnInfoImpl("data", "long", null))));
             assertThat(getValuesList(results).size(), equalTo(0));
         }
     }
@@ -1420,7 +1431,7 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
 
     public void testLoadId() {
         try (EsqlQueryResponse results = run("from test metadata _id | keep _id | sort _id ")) {
-            assertThat(results.columns(), equalTo(List.of(new ColumnInfoImpl("_id", "keyword"))));
+            assertThat(results.columns(), equalTo(List.of(new ColumnInfoImpl("_id", "keyword", null))));
             ListMatcher values = matchesList();
             for (int i = 10; i < 50; i++) {
                 values = values.item(List.of(Integer.toString(i)));
@@ -1629,12 +1640,15 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
 
         try (EsqlQueryResponse resp = run(from + "METADATA _source | EVAL x = 123")) {
             assertFalse(resp.values().hasNext());
-            assertThat(resp.columns(), equalTo(List.of(new ColumnInfoImpl("_source", "_source"), new ColumnInfoImpl("x", "integer"))));
+            assertThat(
+                resp.columns(),
+                equalTo(List.of(new ColumnInfoImpl("_source", "_source", null), new ColumnInfoImpl("x", "integer", null)))
+            );
         }
 
         try (EsqlQueryResponse resp = run(from)) {
             assertFalse(resp.values().hasNext());
-            assertThat(resp.columns(), equalTo(List.of(new ColumnInfoImpl("<no-fields>", "null"))));
+            assertThat(resp.columns(), equalTo(List.of(new ColumnInfoImpl("<no-fields>", "null", null))));
         }
     }
 
@@ -1664,9 +1678,9 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
                 resp.columns(),
                 equalTo(
                     List.of(
-                        new ColumnInfoImpl("name", "text"),
-                        new ColumnInfoImpl("_source", "_source"),
-                        new ColumnInfoImpl("x", "integer")
+                        new ColumnInfoImpl("name", "text", null),
+                        new ColumnInfoImpl("_source", "_source", null),
+                        new ColumnInfoImpl("x", "integer", null)
                     )
                 )
             );
@@ -1674,26 +1688,29 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
 
         try (EsqlQueryResponse resp = run(from)) {
             assertFalse(resp.values().hasNext());
-            assertThat(resp.columns(), equalTo(List.of(new ColumnInfoImpl("name", "text"))));
+            assertThat(resp.columns(), equalTo(List.of(new ColumnInfoImpl("name", "text", null))));
         }
     }
 
     private void assertEmptyIndexQueries(String from) {
         try (EsqlQueryResponse resp = run(from + "METADATA _source | KEEP _source | LIMIT 1")) {
             assertFalse(resp.values().hasNext());
-            assertThat(resp.columns(), equalTo(List.of(new ColumnInfoImpl("_source", "_source"))));
+            assertThat(resp.columns(), equalTo(List.of(new ColumnInfoImpl("_source", "_source", null))));
         }
 
         try (EsqlQueryResponse resp = run(from + "| EVAL y = 1 | KEEP y | LIMIT 1 | EVAL x = 1")) {
             assertFalse(resp.values().hasNext());
-            assertThat(resp.columns(), equalTo(List.of(new ColumnInfoImpl("y", "integer"), new ColumnInfoImpl("x", "integer"))));
+            assertThat(
+                resp.columns(),
+                equalTo(List.of(new ColumnInfoImpl("y", "integer", null), new ColumnInfoImpl("x", "integer", null)))
+            );
         }
 
         try (EsqlQueryResponse resp = run(from + "| STATS c = count()")) {
             assertTrue(resp.values().hasNext());
             Iterator<Object> row = resp.values().next();
             assertThat(row.next(), equalTo((long) 0));
-            assertThat(resp.columns(), equalTo(List.of(new ColumnInfoImpl("c", "long"))));
+            assertThat(resp.columns(), equalTo(List.of(new ColumnInfoImpl("c", "long", null))));
         }
 
         try (EsqlQueryResponse resp = run(from + "| STATS c = count() | EVAL x = 123")) {
@@ -1702,7 +1719,7 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
             assertThat(row.next(), equalTo((long) 0));
             assertThat(row.next(), equalTo(123));
             assertFalse(row.hasNext());
-            assertThat(resp.columns(), equalTo(List.of(new ColumnInfoImpl("c", "long"), new ColumnInfoImpl("x", "integer"))));
+            assertThat(resp.columns(), equalTo(List.of(new ColumnInfoImpl("c", "long", null), new ColumnInfoImpl("x", "integer", null))));
         }
     }
 
@@ -1779,7 +1796,7 @@ public class EsqlActionIT extends AbstractEsqlIntegTestCase {
 
     private void assertNoNestedDocuments(String query, int docsCount, long minValue, long maxValue) {
         try (EsqlQueryResponse results = run(query)) {
-            assertThat(results.columns(), contains(new ColumnInfoImpl("data", "long")));
+            assertThat(results.columns(), contains(new ColumnInfoImpl("data", "long", null)));
             assertThat(results.columns().size(), is(1));
             assertThat(getValuesList(results).size(), is(docsCount));
             for (List<Object> row : getValuesList(results)) {

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/TimeSeriesIT.java

@@ -196,7 +196,7 @@ public class TimeSeriesIT extends AbstractEsqlIntegTestCase {
             }
         }
         try (var resp = run("METRICS hosts | STATS sum(rate(request_count, 1second))")) {
-            assertThat(resp.columns(), equalTo(List.of(new ColumnInfoImpl("sum(rate(request_count, 1second))", "double"))));
+            assertThat(resp.columns(), equalTo(List.of(new ColumnInfoImpl("sum(rate(request_count, 1second))", "double", null))));
             List<List<Object>> values = EsqlTestUtils.getValuesList(resp);
             assertThat(values, hasSize(1));
             assertThat(values.get(0), hasSize(1));
@@ -207,8 +207,8 @@ public class TimeSeriesIT extends AbstractEsqlIntegTestCase {
                 resp.columns(),
                 equalTo(
                     List.of(
-                        new ColumnInfoImpl("max(rate(request_count))", "double"),
-                        new ColumnInfoImpl("min(rate(request_count))", "double")
+                        new ColumnInfoImpl("max(rate(request_count))", "double", null),
+                        new ColumnInfoImpl("min(rate(request_count))", "double", null)
                     )
                 )
             );
@@ -225,9 +225,9 @@ public class TimeSeriesIT extends AbstractEsqlIntegTestCase {
                 resp.columns(),
                 equalTo(
                     List.of(
-                        new ColumnInfoImpl("max(rate(request_count))", "double"),
-                        new ColumnInfoImpl("avg(rate(request_count))", "double"),
-                        new ColumnInfoImpl("max(rate(request_count, 1minute))", "double")
+                        new ColumnInfoImpl("max(rate(request_count))", "double", null),
+                        new ColumnInfoImpl("avg(rate(request_count))", "double", null),
+                        new ColumnInfoImpl("max(rate(request_count, 1minute))", "double", null)
                     )
                 )
             );
@@ -244,8 +244,8 @@ public class TimeSeriesIT extends AbstractEsqlIntegTestCase {
                 resp.columns(),
                 equalTo(
                     List.of(
-                        new ColumnInfoImpl("avg(rate(request_count))", "double"),
-                        new ColumnInfoImpl("avg(rate(request_count, 1second))", "double")
+                        new ColumnInfoImpl("avg(rate(request_count))", "double", null),
+                        new ColumnInfoImpl("avg(rate(request_count, 1second))", "double", null)
                     )
                 )
             );
@@ -261,10 +261,10 @@ public class TimeSeriesIT extends AbstractEsqlIntegTestCase {
                 resp.columns(),
                 equalTo(
                     List.of(
-                        new ColumnInfoImpl("max(rate(request_count))", "double"),
-                        new ColumnInfoImpl("min(rate(request_count))", "double"),
-                        new ColumnInfoImpl("min(cpu)", "double"),
-                        new ColumnInfoImpl("max(cpu)", "double")
+                        new ColumnInfoImpl("max(rate(request_count))", "double", null),
+                        new ColumnInfoImpl("min(rate(request_count))", "double", null),
+                        new ColumnInfoImpl("min(cpu)", "double", null),
+                        new ColumnInfoImpl("max(cpu)", "double", null)
                     )
                 )
             );
@@ -299,7 +299,9 @@ public class TimeSeriesIT extends AbstractEsqlIntegTestCase {
         try (var resp = run("METRICS hosts | STATS sum(rate(request_count)) BY cluster | SORT cluster")) {
             assertThat(
                 resp.columns(),
-                equalTo(List.of(new ColumnInfoImpl("sum(rate(request_count))", "double"), new ColumnInfoImpl("cluster", "keyword")))
+                equalTo(
+                    List.of(new ColumnInfoImpl("sum(rate(request_count))", "double", null), new ColumnInfoImpl("cluster", "keyword", null))
+                )
             );
             List<List<Object>> values = EsqlTestUtils.getValuesList(resp);
             assertThat(values, hasSize(bucketToRates.size()));
@@ -314,7 +316,9 @@ public class TimeSeriesIT extends AbstractEsqlIntegTestCase {
         try (var resp = run("METRICS hosts | STATS avg(rate(request_count)) BY cluster | SORT cluster")) {
             assertThat(
                 resp.columns(),
-                equalTo(List.of(new ColumnInfoImpl("avg(rate(request_count))", "double"), new ColumnInfoImpl("cluster", "keyword")))
+                equalTo(
+                    List.of(new ColumnInfoImpl("avg(rate(request_count))", "double", null), new ColumnInfoImpl("cluster", "keyword", null))
+                )
             );
             List<List<Object>> values = EsqlTestUtils.getValuesList(resp);
             assertThat(values, hasSize(bucketToRates.size()));
@@ -339,9 +343,9 @@ public class TimeSeriesIT extends AbstractEsqlIntegTestCase {
                 resp.columns(),
                 equalTo(
                     List.of(
-                        new ColumnInfoImpl("avg(rate(request_count, 1minute))", "double"),
-                        new ColumnInfoImpl("avg(rate(request_count))", "double"),
-                        new ColumnInfoImpl("cluster", "keyword")
+                        new ColumnInfoImpl("avg(rate(request_count, 1minute))", "double", null),
+                        new ColumnInfoImpl("avg(rate(request_count))", "double", null),
+                        new ColumnInfoImpl("cluster", "keyword", null)
                     )
                 )
             );
@@ -385,7 +389,7 @@ public class TimeSeriesIT extends AbstractEsqlIntegTestCase {
         try (var resp = run("METRICS hosts | STATS sum(rate(request_count)) BY ts=bucket(@timestamp, 1 minute) | SORT ts | LIMIT 5")) {
             assertThat(
                 resp.columns(),
-                equalTo(List.of(new ColumnInfoImpl("sum(rate(request_count))", "double"), new ColumnInfoImpl("ts", "date")))
+                equalTo(List.of(new ColumnInfoImpl("sum(rate(request_count))", "double", null), new ColumnInfoImpl("ts", "date", null)))
             );
             List<List<Object>> values = EsqlTestUtils.getValuesList(resp);
             assertThat(values, hasSize(sortedKeys.size()));
@@ -405,7 +409,7 @@ public class TimeSeriesIT extends AbstractEsqlIntegTestCase {
         try (var resp = run("METRICS hosts | STATS avg(rate(request_count)) BY ts=bucket(@timestamp, 1minute) | SORT ts | LIMIT 5")) {
             assertThat(
                 resp.columns(),
-                equalTo(List.of(new ColumnInfoImpl("avg(rate(request_count))", "double"), new ColumnInfoImpl("ts", "date")))
+                equalTo(List.of(new ColumnInfoImpl("avg(rate(request_count))", "double", null), new ColumnInfoImpl("ts", "date", null)))
             );
             List<List<Object>> values = EsqlTestUtils.getValuesList(resp);
             assertThat(values, hasSize(sortedKeys.size()));
@@ -433,9 +437,9 @@ public class TimeSeriesIT extends AbstractEsqlIntegTestCase {
                 resp.columns(),
                 equalTo(
                     List.of(
-                        new ColumnInfoImpl("avg(rate(request_count, 1minute))", "double"),
-                        new ColumnInfoImpl("avg(rate(request_count))", "double"),
-                        new ColumnInfoImpl("ts", "date")
+                        new ColumnInfoImpl("avg(rate(request_count, 1minute))", "double", null),
+                        new ColumnInfoImpl("avg(rate(request_count))", "double", null),
+                        new ColumnInfoImpl("ts", "date", null)
                     )
                 )
             );
@@ -496,9 +500,9 @@ public class TimeSeriesIT extends AbstractEsqlIntegTestCase {
                 resp.columns(),
                 equalTo(
                     List.of(
-                        new ColumnInfoImpl("sum(rate(request_count))", "double"),
-                        new ColumnInfoImpl("ts", "date"),
-                        new ColumnInfoImpl("cluster", "keyword")
+                        new ColumnInfoImpl("sum(rate(request_count))", "double", null),
+                        new ColumnInfoImpl("ts", "date", null),
+                        new ColumnInfoImpl("cluster", "keyword", null)
                     )
                 )
             );
@@ -527,9 +531,9 @@ public class TimeSeriesIT extends AbstractEsqlIntegTestCase {
                 resp.columns(),
                 equalTo(
                     List.of(
-                        new ColumnInfoImpl("avg(rate(request_count))", "double"),
-                        new ColumnInfoImpl("ts", "date"),
-                        new ColumnInfoImpl("cluster", "keyword")
+                        new ColumnInfoImpl("avg(rate(request_count))", "double", null),
+                        new ColumnInfoImpl("ts", "date", null),
+                        new ColumnInfoImpl("cluster", "keyword", null)
                     )
                 )
             );
@@ -559,10 +563,10 @@ public class TimeSeriesIT extends AbstractEsqlIntegTestCase {
                 resp.columns(),
                 equalTo(
                     List.of(
-                        new ColumnInfoImpl("avg(rate(request_count, 1minute))", "double"),
-                        new ColumnInfoImpl("avg(rate(request_count))", "double"),
-                        new ColumnInfoImpl("ts", "date"),
-                        new ColumnInfoImpl("cluster", "keyword")
+                        new ColumnInfoImpl("avg(rate(request_count, 1minute))", "double", null),
+                        new ColumnInfoImpl("avg(rate(request_count))", "double", null),
+                        new ColumnInfoImpl("ts", "date", null),
+                        new ColumnInfoImpl("cluster", "keyword", null)
                     )
                 )
             );
@@ -602,11 +606,11 @@ public class TimeSeriesIT extends AbstractEsqlIntegTestCase {
                 resp.columns(),
                 equalTo(
                     List.of(
-                        new ColumnInfoImpl("avg_rate", "double"),
-                        new ColumnInfoImpl("max(rate(request_count))", "double"),
-                        new ColumnInfoImpl("avg(rate(request_count))", "double"),
-                        new ColumnInfoImpl("ts", "date"),
-                        new ColumnInfoImpl("cluster", "keyword")
+                        new ColumnInfoImpl("avg_rate", "double", null),
+                        new ColumnInfoImpl("max(rate(request_count))", "double", null),
+                        new ColumnInfoImpl("avg(rate(request_count))", "double", null),
+                        new ColumnInfoImpl("ts", "date", null),
+                        new ColumnInfoImpl("cluster", "keyword", null)
                     )
                 )
             );
@@ -640,10 +644,10 @@ public class TimeSeriesIT extends AbstractEsqlIntegTestCase {
                 resp.columns(),
                 equalTo(
                     List.of(
-                        new ColumnInfoImpl("sum(rate(request_count))", "double"),
-                        new ColumnInfoImpl("max(cpu)", "double"),
-                        new ColumnInfoImpl("ts", "date"),
-                        new ColumnInfoImpl("cluster", "keyword")
+                        new ColumnInfoImpl("sum(rate(request_count))", "double", null),
+                        new ColumnInfoImpl("max(cpu)", "double", null),
+                        new ColumnInfoImpl("ts", "date", null),
+                        new ColumnInfoImpl("cluster", "keyword", null)
                     )
                 )
             );
@@ -678,10 +682,10 @@ public class TimeSeriesIT extends AbstractEsqlIntegTestCase {
                 resp.columns(),
                 equalTo(
                     List.of(
-                        new ColumnInfoImpl("sum(rate(request_count))", "double"),
-                        new ColumnInfoImpl("avg(cpu)", "double"),
-                        new ColumnInfoImpl("ts", "date"),
-                        new ColumnInfoImpl("cluster", "keyword")
+                        new ColumnInfoImpl("sum(rate(request_count))", "double", null),
+                        new ColumnInfoImpl("avg(cpu)", "double", null),
+                        new ColumnInfoImpl("ts", "date", null),
+                        new ColumnInfoImpl("cluster", "keyword", null)
                     )
                 )
             );
@@ -727,14 +731,14 @@ public class TimeSeriesIT extends AbstractEsqlIntegTestCase {
             }
         }
         try (var resp = run("METRICS hosts | STATS sum(abs(rate(request_count, 1second)))")) {
-            assertThat(resp.columns(), equalTo(List.of(new ColumnInfoImpl("sum(abs(rate(request_count, 1second)))", "double"))));
+            assertThat(resp.columns(), equalTo(List.of(new ColumnInfoImpl("sum(abs(rate(request_count, 1second)))", "double", null))));
             List<List<Object>> values = EsqlTestUtils.getValuesList(resp);
             assertThat(values, hasSize(1));
             assertThat(values.get(0), hasSize(1));
             assertThat((double) values.get(0).get(0), closeTo(rates.stream().mapToDouble(d -> d).sum(), 0.1));
         }
         try (var resp = run("METRICS hosts | STATS sum(10.0 * rate(request_count, 1second))")) {
-            assertThat(resp.columns(), equalTo(List.of(new ColumnInfoImpl("sum(10.0 * rate(request_count, 1second))", "double"))));
+            assertThat(resp.columns(), equalTo(List.of(new ColumnInfoImpl("sum(10.0 * rate(request_count, 1second))", "double", null))));
             List<List<Object>> values = EsqlTestUtils.getValuesList(resp);
             assertThat(values, hasSize(1));
             assertThat(values.get(0), hasSize(1));
@@ -745,7 +749,11 @@ public class TimeSeriesIT extends AbstractEsqlIntegTestCase {
                 resp.columns(),
                 equalTo(
                     List.of(
-                        new ColumnInfoImpl("sum(20 * rate(request_count, 1second) + 10 * floor(rate(request_count, 1second)))", "double")
+                        new ColumnInfoImpl(
+                            "sum(20 * rate(request_count, 1second) + 10 * floor(rate(request_count, 1second)))",
+                            "double",
+                            null
+                        )
                     )
                 )
             );
@@ -764,8 +772,8 @@ public class TimeSeriesIT extends AbstractEsqlIntegTestCase {
         }
         refresh("events");
         List<ColumnInfoImpl> columns = List.of(
-            new ColumnInfoImpl("_index", DataType.KEYWORD),
-            new ColumnInfoImpl("_index_mode", DataType.KEYWORD)
+            new ColumnInfoImpl("_index", DataType.KEYWORD, null),
+            new ColumnInfoImpl("_index_mode", DataType.KEYWORD, null)
         );
         try (EsqlQueryResponse resp = run("""
             FROM events,hosts METADATA _index_mode, _index

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/ColumnInfoImpl.java

@@ -7,8 +7,10 @@
 
 package org.elasticsearch.xpack.esql.action;
 
+import org.elasticsearch.TransportVersions;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.core.Nullable;
 import org.elasticsearch.xcontent.InstantiatingObjectParser;
 import org.elasticsearch.xcontent.ParseField;
 import org.elasticsearch.xcontent.ParserConstructor;
@@ -19,9 +21,11 @@ import org.elasticsearch.xpack.core.esql.action.ColumnInfo;
 import org.elasticsearch.xpack.esql.core.type.DataType;
 
 import java.io.IOException;
+import java.util.List;
 import java.util.Objects;
 
 import static org.elasticsearch.xcontent.ConstructingObjectParser.constructorArg;
+import static org.elasticsearch.xcontent.ConstructingObjectParser.optionalConstructorArg;
 
 public class ColumnInfoImpl implements ColumnInfo {
 
@@ -34,6 +38,7 @@ public class ColumnInfoImpl implements ColumnInfo {
         );
         parser.declareString(constructorArg(), new ParseField("name"));
         parser.declareString(constructorArg(), new ParseField("type"));
+        parser.declareStringArray(optionalConstructorArg(), new ParseField("original_types"));
         PARSER = parser.build();
     }
 
@@ -43,41 +48,58 @@ public class ColumnInfoImpl implements ColumnInfo {
             return true;
         }
         if ((o instanceof ColumnInfoImpl that)) {
-            return Objects.equals(name, that.name) && Objects.equals(type, that.type);
+            return Objects.equals(name, that.name) && Objects.equals(type, that.type) && Objects.equals(originalTypes, that.originalTypes);
         }
         return false;
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(name, type);
+        return Objects.hash(name, type, originalTypes);
     }
 
     public static ColumnInfo fromXContent(XContentParser parser) {
         return PARSER.apply(parser, null);
     }
 
-    private String name;
-    private DataType type;
+    private final String name;
+    private final DataType type;
+    /**
+     * If this field is unsupported this contains the underlying ES types. If there
+     * is a type conflict this will have many elements, some or all of which may
+     * be actually supported types.
+     */
+    @Nullable
+    private final List<String> originalTypes;
 
     @ParserConstructor
-    public ColumnInfoImpl(String name, String type) {
-        this(name, DataType.fromEs(type));
+    public ColumnInfoImpl(String name, String type, @Nullable List<String> originalTypes) {
+        this(name, DataType.fromEs(type), originalTypes);
     }
 
-    public ColumnInfoImpl(String name, DataType type) {
+    public ColumnInfoImpl(String name, DataType type, @Nullable List<String> originalTypes) {
         this.name = name;
         this.type = type;
+        this.originalTypes = originalTypes;
     }
 
     public ColumnInfoImpl(StreamInput in) throws IOException {
-        this(in.readString(), in.readString());
+        this.name = in.readString();
+        this.type = DataType.fromEs(in.readString());
+        if (in.getTransportVersion().onOrAfter(TransportVersions.ESQL_REPORT_ORIGINAL_TYPES)) {
+            this.originalTypes = in.readOptionalStringCollectionAsList();
+        } else {
+            this.originalTypes = null;
+        }
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         out.writeString(name);
         out.writeString(type.outputType());
+        if (out.getTransportVersion().onOrAfter(TransportVersions.ESQL_REPORT_ORIGINAL_TYPES)) {
+            out.writeOptionalStringCollection(originalTypes);
+        }
     }
 
     @Override
@@ -85,6 +107,9 @@ public class ColumnInfoImpl implements ColumnInfo {
         builder.startObject();
         builder.field("name", name);
         builder.field("type", type.outputType());
+        if (originalTypes != null) {
+            builder.field("original_types", originalTypes);
+        }
         builder.endObject();
         return builder;
     }
@@ -102,4 +127,9 @@ public class ColumnInfoImpl implements ColumnInfo {
     public DataType type() {
         return type;
     }
+
+    @Nullable
+    public List<String> originalTypes() {
+        return originalTypes;
+    }
 }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java

@@ -919,6 +919,11 @@ public class EsqlCapabilities {
          */
         AGGREGATE_METRIC_DOUBLE_CONVERT_TO(AGGREGATE_METRIC_DOUBLE_FEATURE_FLAG),
 
+        /**
+         * The {@code _query} API now reports the original types.
+         */
+        REPORT_ORIGINAL_TYPES,
+
         /**
          * The metrics command
          */

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/analysis/Analyzer.java

@@ -1660,7 +1660,7 @@ public class Analyzer extends ParameterizedRuleExecutor<LogicalPlan, AnalyzerCon
         static Attribute checkUnresolved(FieldAttribute fa) {
             if (fa.field() instanceof InvalidMappedField imf) {
                 String unresolvedMessage = "Cannot use field [" + fa.name() + "] due to ambiguities being " + imf.errorMessage();
-                String types = imf.getTypesToIndices().keySet().stream().collect(Collectors.joining(","));
+                List<String> types = imf.getTypesToIndices().keySet().stream().toList();
                 return new UnsupportedAttribute(
                     fa.source(),
                     fa.name(),

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/UnsupportedAttribute.java

@@ -28,6 +28,7 @@ import org.elasticsearch.xpack.esql.core.util.PlanStreamOutput;
 import org.elasticsearch.xpack.esql.io.stream.PlanStreamInput;
 
 import java.io.IOException;
+import java.util.List;
 import java.util.Objects;
 
 import static org.elasticsearch.xpack.esql.core.util.PlanStreamInput.readCachedStringWithVersionCheck;
@@ -60,7 +61,7 @@ public final class UnsupportedAttribute extends FieldAttribute implements Unreso
     private final boolean hasCustomMessage; // TODO remove me and just use message != null?
 
     private static String errorMessage(String name, UnsupportedEsField field) {
-        return "Cannot use field [" + name + "] with unsupported type [" + field.getOriginalType() + "]";
+        return "Cannot use field [" + name + "] with unsupported type [" + String.join(",", field.getOriginalTypes()) + "]";
     }
 
     public UnsupportedAttribute(Source source, String name, UnsupportedEsField field) {
@@ -174,4 +175,9 @@ public final class UnsupportedAttribute extends FieldAttribute implements Unreso
         }
         return false;
     }
+
+    @Override
+    public List<String> originalTypes() {
+        return field().getOriginalTypes();
+    }
 }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/TransportEsqlQueryAction.java

@@ -55,6 +55,8 @@ import org.elasticsearch.xpack.esql.session.Result;
 
 import java.io.IOException;
 import java.time.ZoneOffset;
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -314,7 +316,17 @@ public class TransportEsqlQueryAction extends HandledTransportAction<EsqlQueryRe
     }
 
     private EsqlQueryResponse toResponse(Task task, EsqlQueryRequest request, Configuration configuration, Result result) {
-        List<ColumnInfoImpl> columns = result.schema().stream().map(c -> new ColumnInfoImpl(c.name(), c.dataType().outputType())).toList();
+        List<ColumnInfoImpl> columns = result.schema().stream().map(c -> {
+            List<String> originalTypes;
+            if (c.originalTypes() == null) {
+                originalTypes = null;
+            } else {
+                // Sort the original types so they are easier to test against and prettier.
+                originalTypes = new ArrayList<>(c.originalTypes());
+                Collections.sort(originalTypes);
+            }
+            return new ColumnInfoImpl(c.name(), c.dataType().outputType(), originalTypes);
+        }).toList();
         EsqlQueryResponse.Profile profile = configuration.profile() ? new EsqlQueryResponse.Profile(result.profiles()) : null;
         threadPool.getThreadContext().addResponseHeader(AsyncExecutionId.ASYNC_EXECUTION_IS_RUNNING_HEADER, "?0");
         if (task instanceof EsqlQueryTask asyncTask && request.keepOnCompletion()) {

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/IndexResolver.java

@@ -137,7 +137,7 @@ public class IndexResolver {
                 ? createField(fieldCapsResponse, name, fullName, fcs, isAlias)
                 : new UnsupportedEsField(
                     fullName,
-                    firstUnsupportedParent.getOriginalType(),
+                    firstUnsupportedParent.getOriginalTypes(),
                     firstUnsupportedParent.getName(),
                     new HashMap<>()
                 );
@@ -243,7 +243,7 @@ public class IndexResolver {
 
     private static UnsupportedEsField unsupported(String name, IndexFieldCapabilities fc) {
         String originalType = fc.metricType() == TimeSeriesParams.MetricType.COUNTER ? "counter" : fc.type();
-        return new UnsupportedEsField(name, originalType);
+        return new UnsupportedEsField(name, List.of(originalType));
     }
 
     private static EsField conflictingTypes(String name, String fullName, FieldCapabilitiesResponse fieldCapsResponse) {

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/action/EsqlQueryResponseTests.java

@@ -53,6 +53,7 @@ import org.elasticsearch.xcontent.XContentType;
 import org.elasticsearch.xcontent.json.JsonXContent;
 import org.elasticsearch.xpack.esql.core.type.DataType;
 import org.elasticsearch.xpack.esql.planner.PlannerUtils;
+import org.elasticsearch.xpack.esql.type.UnsupportedEsFieldTests;
 import org.elasticsearch.xpack.versionfield.Version;
 import org.junit.After;
 import org.junit.Before;
@@ -176,7 +177,12 @@ public class EsqlQueryResponseTests extends AbstractChunkedSerializingTestCase<E
                 || t == DataType.AGGREGATE_METRIC_DOUBLE,
             () -> randomFrom(DataType.types())
         ).widenSmallNumeric();
-        return new ColumnInfoImpl(randomAlphaOfLength(10), type.esType());
+        return new ColumnInfoImpl(randomAlphaOfLength(10), type.esType(), randomOriginalTypes());
+    }
+
+    @Nullable
+    public static List<String> randomOriginalTypes() {
+        return randomBoolean() ? null : UnsupportedEsFieldTests.randomOriginalTypes();
     }
 
     private EsqlQueryResponse.Profile randomProfile() {
@@ -254,7 +260,10 @@ public class EsqlQueryResponseTests extends AbstractChunkedSerializingTestCase<E
                 int mutCol = between(0, instance.columns().size() - 1);
                 List<ColumnInfoImpl> cols = new ArrayList<>(instance.columns());
                 // keep the type the same so the values are still valid but change the name
-                cols.set(mutCol, new ColumnInfoImpl(cols.get(mutCol).name() + "mut", cols.get(mutCol).type()));
+                cols.set(
+                    mutCol,
+                    new ColumnInfoImpl(cols.get(mutCol).name() + "mut", cols.get(mutCol).type(), cols.get(mutCol).originalTypes())
+                );
                 yield new EsqlQueryResponse(
                     cols,
                     deepCopyOfPages(instance),
@@ -627,7 +636,7 @@ public class EsqlQueryResponseTests extends AbstractChunkedSerializingTestCase<E
     public void testBasicXContentIdAndRunning() {
         try (
             EsqlQueryResponse response = new EsqlQueryResponse(
-                List.of(new ColumnInfoImpl("foo", "integer")),
+                List.of(new ColumnInfoImpl("foo", "integer", null)),
                 List.of(new Page(blockFactory.newIntArrayVector(new int[] { 40, 80 }, 2).asBlock())),
                 null,
                 false,
@@ -642,10 +651,28 @@ public class EsqlQueryResponseTests extends AbstractChunkedSerializingTestCase<E
         }
     }
 
+    public void testXContentOriginalTypes() {
+        try (
+            EsqlQueryResponse response = new EsqlQueryResponse(
+                List.of(new ColumnInfoImpl("foo", "unsupported", List.of("foo", "bar"))),
+                List.of(new Page(blockFactory.newConstantNullBlock(2))),
+                null,
+                false,
+                null,
+                false,
+                false,
+                null
+            )
+        ) {
+            assertThat(Strings.toString(response), equalTo("""
+                {"columns":[{"name":"foo","type":"unsupported","original_types":["foo","bar"]}],"values":[[null],[null]]}"""));
+        }
+    }
+
     public void testNullColumnsXContentDropNulls() {
         try (
             EsqlQueryResponse response = new EsqlQueryResponse(
-                List.of(new ColumnInfoImpl("foo", "integer"), new ColumnInfoImpl("all_null", "integer")),
+                List.of(new ColumnInfoImpl("foo", "integer", null), new ColumnInfoImpl("all_null", "integer", null)),
                 List.of(new Page(blockFactory.newIntArrayVector(new int[] { 40, 80 }, 2).asBlock(), blockFactory.newConstantNullBlock(2))),
                 null,
                 false,
@@ -675,7 +702,7 @@ public class EsqlQueryResponseTests extends AbstractChunkedSerializingTestCase<E
             b.appendNull();
             try (
                 EsqlQueryResponse response = new EsqlQueryResponse(
-                    List.of(new ColumnInfoImpl("foo", "integer"), new ColumnInfoImpl("all_null", "integer")),
+                    List.of(new ColumnInfoImpl("foo", "integer", null), new ColumnInfoImpl("all_null", "integer", null)),
                     List.of(new Page(blockFactory.newIntArrayVector(new int[] { 40, 80 }, 2).asBlock(), b.build())),
                     null,
                     false,
@@ -702,7 +729,7 @@ public class EsqlQueryResponseTests extends AbstractChunkedSerializingTestCase<E
 
     private EsqlQueryResponse simple(boolean columnar, boolean async) {
         return new EsqlQueryResponse(
-            List.of(new ColumnInfoImpl("foo", "integer")),
+            List.of(new ColumnInfoImpl("foo", "integer", null)),
             List.of(new Page(blockFactory.newIntArrayVector(new int[] { 40, 80 }, 2).asBlock())),
             null,
             columnar,
@@ -714,7 +741,7 @@ public class EsqlQueryResponseTests extends AbstractChunkedSerializingTestCase<E
     public void testProfileXContent() {
         try (
             EsqlQueryResponse response = new EsqlQueryResponse(
-                List.of(new ColumnInfoImpl("foo", "integer")),
+                List.of(new ColumnInfoImpl("foo", "integer", null)),
                 List.of(new Page(blockFactory.newIntArrayVector(new int[] { 40, 80 }, 2).asBlock())),
                 new EsqlQueryResponse.Profile(
                     List.of(
@@ -798,7 +825,7 @@ public class EsqlQueryResponseTests extends AbstractChunkedSerializingTestCase<E
         var intBlk2 = blockFactory.newIntArrayVector(new int[] { 30, 40, 50 }, 3).asBlock();
         var longBlk1 = blockFactory.newLongArrayVector(new long[] { 100L, 200L }, 2).asBlock();
         var longBlk2 = blockFactory.newLongArrayVector(new long[] { 300L, 400L, 500L }, 3).asBlock();
-        var columnInfo = List.of(new ColumnInfoImpl("foo", "integer"), new ColumnInfoImpl("bar", "long"));
+        var columnInfo = List.of(new ColumnInfoImpl("foo", "integer", null), new ColumnInfoImpl("bar", "long", null));
         var pages = List.of(new Page(intBlk1, longBlk1), new Page(intBlk2, longBlk2));
         try (var response = new EsqlQueryResponse(columnInfo, pages, null, false, null, false, false, null)) {
             assertThat(columnValues(response.column(0)), contains(10, 20, 30, 40, 50));
@@ -810,7 +837,7 @@ public class EsqlQueryResponseTests extends AbstractChunkedSerializingTestCase<E
 
     public void testColumnsIllegalArg() {
         var intBlk1 = blockFactory.newIntArrayVector(new int[] { 10 }, 1).asBlock();
-        var columnInfo = List.of(new ColumnInfoImpl("foo", "integer"));
+        var columnInfo = List.of(new ColumnInfoImpl("foo", "integer", null));
         var pages = List.of(new Page(intBlk1));
         try (var response = new EsqlQueryResponse(columnInfo, pages, null, false, null, false, false, null)) {
             expectThrows(IllegalArgumentException.class, () -> response.column(-1));
@@ -829,7 +856,7 @@ public class EsqlQueryResponseTests extends AbstractChunkedSerializingTestCase<E
             blk2 = bb2.appendInt(30).appendNull().appendNull().appendInt(60).build();
             blk3 = bb3.appendNull().appendInt(80).appendInt(90).appendNull().build();
         }
-        var columnInfo = List.of(new ColumnInfoImpl("foo", "integer"));
+        var columnInfo = List.of(new ColumnInfoImpl("foo", "integer", null));
         var pages = List.of(new Page(blk1), new Page(blk2), new Page(blk3));
         try (var response = new EsqlQueryResponse(columnInfo, pages, null, false, null, false, false, null)) {
             assertThat(columnValues(response.column(0)), contains(10, null, 30, null, null, 60, null, 80, 90, null));
@@ -849,7 +876,7 @@ public class EsqlQueryResponseTests extends AbstractChunkedSerializingTestCase<E
             blk2 = bb2.beginPositionEntry().appendInt(40).appendInt(50).endPositionEntry().build();
             blk3 = bb3.appendNull().appendInt(70).appendInt(80).appendNull().build();
         }
-        var columnInfo = List.of(new ColumnInfoImpl("foo", "integer"));
+        var columnInfo = List.of(new ColumnInfoImpl("foo", "integer", null));
         var pages = List.of(new Page(blk1), new Page(blk2), new Page(blk3));
         try (var response = new EsqlQueryResponse(columnInfo, pages, null, false, null, false, false, null)) {
             assertThat(columnValues(response.column(0)), contains(List.of(10, 20), null, List.of(40, 50), null, 70, 80, null));

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/analysis/VerifierTests.java

@@ -85,7 +85,7 @@ public class VerifierTests extends ESTestCase {
         final String unsupported = "unsupported";
         final String multiTyped = "multi_typed";
 
-        EsField unsupportedField = new UnsupportedEsField(unsupported, "flattened");
+        EsField unsupportedField = new UnsupportedEsField(unsupported, List.of("flattened"));
         // Use linked maps/sets to fix the order in the error message.
         LinkedHashSet<String> ipIndices = new LinkedHashSet<>();
         ipIndices.add("test1");

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/formatter/TextFormatTests.java

@@ -269,18 +269,18 @@ public class TextFormatTests extends ESTestCase {
     }
 
     private static EsqlQueryResponse emptyData() {
-        return new EsqlQueryResponse(singletonList(new ColumnInfoImpl("name", "keyword")), emptyList(), null, false, false, null);
+        return new EsqlQueryResponse(singletonList(new ColumnInfoImpl("name", "keyword", null)), emptyList(), null, false, false, null);
     }
 
     private static EsqlQueryResponse regularData() {
         BlockFactory blockFactory = TestBlockFactory.getNonBreakingInstance();
         // headers
         List<ColumnInfoImpl> headers = asList(
-            new ColumnInfoImpl("string", "keyword"),
-            new ColumnInfoImpl("number", "integer"),
-            new ColumnInfoImpl("location", "geo_point"),
-            new ColumnInfoImpl("location2", "cartesian_point"),
-            new ColumnInfoImpl("null_field", "keyword")
+            new ColumnInfoImpl("string", "keyword", null),
+            new ColumnInfoImpl("number", "integer", null),
+            new ColumnInfoImpl("location", "geo_point", null),
+            new ColumnInfoImpl("location2", "cartesian_point", null),
+            new ColumnInfoImpl("null_field", "keyword", null)
         );
 
         BytesRefArray geoPoints = new BytesRefArray(2, BigArrays.NON_RECYCLING_INSTANCE);
@@ -308,7 +308,10 @@ public class TextFormatTests extends ESTestCase {
 
     private static EsqlQueryResponse escapedData() {
         // headers
-        List<ColumnInfoImpl> headers = asList(new ColumnInfoImpl("first", "keyword"), new ColumnInfoImpl("\"special\"", "keyword"));
+        List<ColumnInfoImpl> headers = asList(
+            new ColumnInfoImpl("first", "keyword", null),
+            new ColumnInfoImpl("\"special\"", "keyword", null)
+        );
 
         // values
         List<Page> values = List.of(

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/formatter/TextFormatterTests.java

@@ -34,16 +34,16 @@ public class TextFormatterTests extends ESTestCase {
     static BlockFactory blockFactory = TestBlockFactory.getNonBreakingInstance();
 
     private final List<ColumnInfoImpl> columns = Arrays.asList(
-        new ColumnInfoImpl("foo", "keyword"),
-        new ColumnInfoImpl("bar", "long"),
-        new ColumnInfoImpl("15charwidename!", "double"),
-        new ColumnInfoImpl("null_field1", "integer"),
-        new ColumnInfoImpl("superduperwidename!!!", "double"),
-        new ColumnInfoImpl("baz", "keyword"),
-        new ColumnInfoImpl("date", "date"),
-        new ColumnInfoImpl("location", "geo_point"),
-        new ColumnInfoImpl("location2", "cartesian_point"),
-        new ColumnInfoImpl("null_field2", "keyword")
+        new ColumnInfoImpl("foo", "keyword", null),
+        new ColumnInfoImpl("bar", "long", null),
+        new ColumnInfoImpl("15charwidename!", "double", null),
+        new ColumnInfoImpl("null_field1", "integer", null),
+        new ColumnInfoImpl("superduperwidename!!!", "double", null),
+        new ColumnInfoImpl("baz", "keyword", null),
+        new ColumnInfoImpl("date", "date", null),
+        new ColumnInfoImpl("location", "geo_point", null),
+        new ColumnInfoImpl("location2", "cartesian_point", null),
+        new ColumnInfoImpl("null_field2", "keyword", null)
     );
 
     private static final BytesRefArray geoPoints = new BytesRefArray(2, BigArrays.NON_RECYCLING_INSTANCE);
@@ -213,7 +213,7 @@ public class TextFormatterTests extends ESTestCase {
             getTextBodyContent(
                 new TextFormatter(
                     new EsqlQueryResponse(
-                        List.of(new ColumnInfoImpl("foo", "keyword")),
+                        List.of(new ColumnInfoImpl("foo", "keyword", null)),
                         List.of(
                             new Page(
                                 blockFactory.newBytesRefBlockBuilder(2)

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/type/UnsupportedEsFieldTests.java

@@ -10,15 +10,20 @@ package org.elasticsearch.xpack.esql.type;
 import org.elasticsearch.xpack.esql.core.type.EsField;
 import org.elasticsearch.xpack.esql.core.type.UnsupportedEsField;
 
+import java.util.List;
 import java.util.Map;
 
 public class UnsupportedEsFieldTests extends AbstractEsFieldTypeTests<UnsupportedEsField> {
     public static UnsupportedEsField randomUnsupportedEsField(int maxPropertiesDepth) {
         String name = randomAlphaOfLength(4);
-        String originalType = randomAlphaOfLength(5);
+        List<String> originalTypes = randomOriginalTypes();
         String inherited = randomBoolean() ? null : randomAlphaOfLength(5);
         Map<String, EsField> properties = randomProperties(maxPropertiesDepth);
-        return new UnsupportedEsField(name, originalType, inherited, properties);
+        return new UnsupportedEsField(name, originalTypes, inherited, properties);
+    }
+
+    public static List<String> randomOriginalTypes() {
+        return randomBoolean() ? List.of(randomAlphaOfLength(5)) : randomList(4, 4, () -> randomAlphaOfLength(5));
     }
 
     @Override
@@ -29,16 +34,16 @@ public class UnsupportedEsFieldTests extends AbstractEsFieldTypeTests<Unsupporte
     @Override
     protected UnsupportedEsField mutate(UnsupportedEsField instance) {
         String name = instance.getName();
-        String originalType = randomAlphaOfLength(5);
-        String inherited = randomBoolean() ? null : randomAlphaOfLength(5);
+        List<String> originalTypes = instance.getOriginalTypes();
+        String inherited = instance.getInherited();
         Map<String, EsField> properties = instance.getProperties();
         switch (between(0, 3)) {
             case 0 -> name = randomAlphaOfLength(name.length() + 1);
-            case 1 -> originalType = randomValueOtherThan(originalType, () -> randomAlphaOfLength(4));
+            case 1 -> originalTypes = randomValueOtherThan(originalTypes, UnsupportedEsFieldTests::randomOriginalTypes);
             case 2 -> inherited = randomValueOtherThan(inherited, () -> randomBoolean() ? null : randomAlphaOfLength(4));
             case 3 -> properties = randomValueOtherThan(properties, () -> randomProperties(4));
             default -> throw new IllegalArgumentException();
         }
-        return new UnsupportedEsField(name, originalType, inherited, properties);
+        return new UnsupportedEsField(name, originalTypes, inherited, properties);
     }
 }

x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/esql/160_union_types.yml

@@ -263,6 +263,15 @@ load single index ip_long stats invalid grouping:
 
 ---
 load two indices, showing unsupported type and null value for event_duration:
+  - requires:
+      test_runner_features: [capabilities]
+      capabilities:
+        - method: POST
+          path: /_query
+          parameters: []
+          capabilities: [REPORT_ORIGINAL_TYPES]
+      reason: "Uses original_types"
+
   - do:
       allowed_warnings_regex:
         - "No limit defined, adding default limit of \\[.*\\]"
@@ -276,8 +285,10 @@ load two indices, showing unsupported type and null value for event_duration:
   - match: { columns.0.type: "date" }
   - match: { columns.1.name: "client_ip" }
   - match: { columns.1.type: "ip" }
+  - is_false: columns.1.original_types
   - match: { columns.2.name: "event_duration" }
   - match: { columns.2.type: "unsupported" }
+  - match: { columns.2.original_types: ["keyword", "long"] }
   - match: { columns.3.name: "message" }
   - match: { columns.3.type: "keyword" }
   - match: { columns.4.name: "_index" }
@@ -302,8 +313,8 @@ load two indices with no conversion function, but needs TO_LONG conversion:
         - method: POST
           path: /_query
           parameters: []
-          capabilities: [union_types_fix_rename_resolution]
-      reason: "Union type resolution fix for rename also allows direct usage of unsupported fields in KEEP"
+          capabilities: [REPORT_ORIGINAL_TYPES]
+      reason: "Uses original_types"
 
   - do:
       allowed_warnings_regex:
@@ -318,7 +329,9 @@ load two indices with no conversion function, but needs TO_LONG conversion:
   - match: { columns.1.type: "date" }
   - match: { columns.2.name: "client_ip" }
   - match: { columns.2.type: "ip" }
+  - is_false: columns.2.original_types
   - match: { columns.3.name: "event_duration" }
+  - match: { columns.3.original_types: ["keyword", "long"] }
   - match: { columns.3.type: "unsupported" }
   - match: { columns.4.name: "message" }
   - match: { columns.4.type: "keyword" }

x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/esql/40_unsupported_types.yml

@@ -110,17 +110,13 @@ setup:
 ---
 unsupported:
   - requires:
-      cluster_features: ["gte_v8.13.0"]
-      reason: "Latest types supported in ESQL as of 8.13.0"
-
-  - requires:
-      test_runner_features: [ capabilities ]
+      test_runner_features: [capabilities, contains]
       capabilities:
         - method: POST
           path: /_query
-          parameters: [ ]
-          capabilities: [ aggregate_metric_double_rendering ]
-      reason: "support for rendering aggregate_metric_double type"
+          parameters: []
+          capabilities: [REPORT_ORIGINAL_TYPES]
+      reason: "uses original_type"
 
   - do:
       allowed_warnings_regex:
@@ -132,10 +128,13 @@ unsupported:
 
   - match: { columns.0.name: aggregate_metric_double }
   - match: { columns.0.type: aggregate_metric_double }
+  - is_false: columns.0.original_types
   - match: { columns.1.name: binary }
   - match: { columns.1.type: unsupported }
+  - match: { columns.1.original_types: [binary] }
   - match: { columns.2.name: completion }
   - match: { columns.2.type: unsupported }
+  - match: { columns.2.original_types: [completion] }
   - match: { columns.3.name: date_nanos }
   - match: { columns.3.type: date_nanos }
   - match: { columns.4.name: date_range }
@@ -228,10 +227,13 @@ unsupported:
           query: 'from test | limit 0'
   - match: { columns.0.name: aggregate_metric_double }
   - match: { columns.0.type: aggregate_metric_double }
+  - is_false: columns.0.original_types
   - match: { columns.1.name: binary }
   - match: { columns.1.type: unsupported }
+  - match: { columns.1.original_types: [binary] }
   - match: { columns.2.name: completion }
   - match: { columns.2.type: unsupported }
+  - match: { columns.2.original_types: [completion] }
   - match: { columns.3.name: date_nanos }
   - match: { columns.3.type: date_nanos }
   - match: { columns.4.name: date_range }