Refactor/efficient union type memory

benchmarks/src/main/java/org/elasticsearch/benchmark/_nightly/esql/QueryPlanningBenchmark.java

@@ -107,7 +107,7 @@ public void setup() {
             mapping.put("field" + i, new EsField("field-" + i, TEXT, emptyMap(), true, EsField.TimeSeriesFieldType.NONE));
         }
 
-        var esIndex = new EsIndex("test", mapping, Map.of("test", IndexMode.STANDARD), Map.of(), Map.of(), Map.of());
+        var esIndex = new EsIndex("test", mapping, Map.of("test", IndexMode.STANDARD), Map.of(), Map.of());
 
         var functionRegistry = new EsqlFunctionRegistry();
         parser = new EsqlParser(new EsqlConfig(functionRegistry));

benchmarks/src/main/java/org/elasticsearch/benchmark/esql/ViewResolutionBenchmarkBase.java

@@ -168,7 +168,7 @@ public void setup() {
             String name = "col" + i;
             mapping.put(name, new EsField(name, KEYWORD, emptyMap(), true, EsField.TimeSeriesFieldType.NONE));
         }
-        EsIndex esIndex = new EsIndex("test", mapping, Map.of("test", IndexMode.STANDARD), Map.of(), Map.of(), Map.of());
+        EsIndex esIndex = new EsIndex("test", mapping, Map.of("test", IndexMode.STANDARD), Map.of(), Map.of());
 
         Configuration config = new Configuration(
             DateUtils.UTC,

server/src/main/resources/transport/definitions/referable/compact_multi_type_es_field.csv

@@ -0,0 +1 @@
+9366000

server/src/main/resources/transport/upper_bounds/9.5.csv

@@ -1 +1 @@
-inference_api_audio_video_pdf_support,9365000
+compact_multi_type_es_field,9366000

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

@@ -154,8 +154,7 @@ public TestAnalyzer addNoFieldsIndex() {
             Map.of(),
             Map.of(noFieldsIndexName, IndexMode.STANDARD),
             Map.of("", List.of(noFieldsIndexName)),
-            Map.of("", List.of(noFieldsIndexName)),
-            Map.of()
+            Map.of("", List.of(noFieldsIndexName))
         );
         addIndex(noFieldsIndexName, IndexResolution.valid(noFieldsIndex));
         return this;
@@ -781,7 +780,7 @@ public AnalyzerContext buildContext() {
      */
     public static IndexResolution loadMapping(String resource, String indexName, IndexMode indexMode) {
         return IndexResolution.valid(
-            new EsIndex(indexName, EsqlTestUtils.loadMapping(resource), Map.of(indexName, indexMode), Map.of(), Map.of(), Map.of())
+            new EsIndex(indexName, EsqlTestUtils.loadMapping(resource), Map.of(indexName, indexMode), Map.of(), Map.of())
         );
     }
 

x-pack/plugin/esql/qa/testFixtures/src/main/resources/unmapped-load.csv-spec

@@ -537,6 +537,22 @@ sample_data            | Connected to 10.1.0.2
 sample_data            | Connected to 10.1.0.1
 ;
 
+partiallyUnmappedSmallNumericFieldIsWidened
+required_capability: optional_fields_v5
+
+SET unmapped_fields="load"\;
+FROM apps_short, partial_mapping_sample_data
+| KEEP id, name
+| SORT name DESC NULLS LAST
+| LIMIT 3
+;
+
+id:integer | name:keyword
+14         | mmmmm
+13         | lllll
+11         | kkkkk
+;
+
 fieldIsPartiallyUnmappedAndRenamedMultiIndex
 required_capability: optional_fields_v5
 

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

@@ -7,6 +7,7 @@
 
 package org.elasticsearch.xpack.esql.analysis;
 
+import org.elasticsearch.core.Nullable;
 import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.index.mapper.flattened.FlattenedFieldMapper;
 import org.elasticsearch.license.XPackLicenseState;
@@ -31,8 +32,10 @@
 import org.elasticsearch.xpack.esql.core.expression.predicate.operator.comparison.BinaryComparison;
 import org.elasticsearch.xpack.esql.core.tree.Node;
 import org.elasticsearch.xpack.esql.core.type.DataType;
-import org.elasticsearch.xpack.esql.core.type.MultiTypeEsField;
+import org.elasticsearch.xpack.esql.core.type.EsField;
 import org.elasticsearch.xpack.esql.core.type.PotentiallyUnmappedKeywordEsField;
+import org.elasticsearch.xpack.esql.core.type.TypeConflictField;
+import org.elasticsearch.xpack.esql.core.type.UnionTypeEsField;
 import org.elasticsearch.xpack.esql.core.type.UnsupportedEsField;
 import org.elasticsearch.xpack.esql.core.util.Holder;
 import org.elasticsearch.xpack.esql.expression.function.TimestampAware;
@@ -44,7 +47,6 @@
 import org.elasticsearch.xpack.esql.expression.predicate.operator.comparison.Equals;
 import org.elasticsearch.xpack.esql.expression.predicate.operator.comparison.EsqlBinaryComparison;
 import org.elasticsearch.xpack.esql.expression.predicate.operator.comparison.NotEquals;
-import org.elasticsearch.xpack.esql.index.EsIndex;
 import org.elasticsearch.xpack.esql.index.IndexResolution;
 import org.elasticsearch.xpack.esql.plan.IndexPattern;
 import org.elasticsearch.xpack.esql.plan.logical.Aggregate;
@@ -613,13 +615,12 @@ private static AttributeSet partiallyUnmappedNonKeywords(LogicalPlan plan, Map<I
         plan.forEachUp(EsRelation.class, relation -> {
             IndexResolution indexResolution = indexResolutions.get(new IndexPattern(relation.source(), relation.indexPattern()));
             if (indexResolution != null && indexResolution.isValid()) {
-                EsIndex index = indexResolution.get();
+                Set<String> punkFieldNames = collectPotentiallyUnmappedNonKeywords(indexResolution.get().mapping());
                 for (Attribute attr : relation.output()) {
-                    if (attr instanceof FieldAttribute fa
-                        && index.isPartiallyUnmappedField(fa.fieldName().string())
-                        && fa.dataType() != DataType.KEYWORD
                     // punk_field::long is fine; in this case, the FieldAttribute contains a MultiTypeEsField with the conversions.
-                        && fa.field() instanceof MultiTypeEsField == false) {
+                    if (attr instanceof FieldAttribute fa
+                        && punkFieldNames.contains(fa.fieldName().string())
+                        && fa.field() instanceof UnionTypeEsField == false) {
                         punks.add(fa);
                     }
                 }
@@ -629,6 +630,29 @@ private static AttributeSet partiallyUnmappedNonKeywords(LogicalPlan plan, Map<I
         return punks.build();
     }
 
+    private static Set<String> collectPotentiallyUnmappedNonKeywords(Map<String, EsField> mapping) {
+        HashSet<String> result = new HashSet<>();
+        collectPotentiallyUnmappedNonKeywords(mapping, null, result);
+        return result;
+    }
+
+    private static void collectPotentiallyUnmappedNonKeywords(
+        Map<String, EsField> mapping,
+        @Nullable String prefix,
+        Set<String> aggregator
+    ) {
+        for (Map.Entry<String, EsField> entry : mapping.entrySet()) {
+            String name = prefix == null ? entry.getKey() : prefix + "." + entry.getKey();
+            EsField field = entry.getValue();
+            if (field instanceof TypeConflictField imf && imf.isPotentiallyUnmapped()) {
+                aggregator.add(name);
+            }
+            if (field.getProperties() != null && field.getProperties().isEmpty() == false) {
+                collectPotentiallyUnmappedNonKeywords(field.getProperties(), name, aggregator);
+            }
+        }
+    }
+
     private void licenseCheck(LogicalPlan plan, Failures failures) {
         Consumer<Node<?>> licenseCheck = n -> {
             if (n instanceof LicenseAware la && la.licenseCheck(licenseState) == false) {

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

@@ -8,7 +8,7 @@
 
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.xpack.esql.core.type.DataType;
-import org.elasticsearch.xpack.esql.core.type.InvalidMappedField;
+import org.elasticsearch.xpack.esql.core.type.TypeConflictField;
 
 import java.util.ArrayList;
 import java.util.Collection;
@@ -40,7 +40,7 @@ public static List<Attribute> asAttributes(List<? extends NamedExpression> named
      * Converts named expressions to {@link ReferenceAttribute}s, preserving {@link NameId}s for attributes whose name
      * matches one in {@code existingOutput}. Genuinely new attributes get fresh NameIds.
      * <p>
-     * Exception: a {@link FieldAttribute} backed by an {@link InvalidMappedField} (ambiguous type across indices) is instead
+     * Exception: a {@link FieldAttribute} backed by a {@link TypeConflictField} (ambiguous type across indices) is instead
      * converted to an {@link UnsupportedAttribute} via
      * {@link FieldAttribute#flagTypeConflicts()}, so the analyzer can surface a clear user-facing error.
      */
@@ -60,7 +60,7 @@ public static List<Attribute> toReferenceAttributesPreservingIds(
             Attribute existing = existingByName.get(exp.name());
             NameId id = existing != null ? existing.id() : new NameId();
             Attribute refAttr = switch (exp) {
-                case FieldAttribute fa when fa.field() instanceof InvalidMappedField -> fa.flagTypeConflicts();
+                case FieldAttribute fa when fa.field() instanceof TypeConflictField -> fa.flagTypeConflicts();
                 case ReferenceAttribute ra -> ra.withId(id);
                 default -> new ReferenceAttribute(exp.source(), null, exp.name(), exp.dataType(), exp.nullable(), id, exp.synthetic());
             };

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

@@ -16,7 +16,7 @@
 import org.elasticsearch.xpack.esql.core.tree.Source;
 import org.elasticsearch.xpack.esql.core.type.DataType;
 import org.elasticsearch.xpack.esql.core.type.EsField;
-import org.elasticsearch.xpack.esql.core.type.InvalidMappedField;
+import org.elasticsearch.xpack.esql.core.type.TypeConflictField;
 import org.elasticsearch.xpack.esql.core.type.UnsupportedEsField;
 import org.elasticsearch.xpack.esql.io.stream.PlanStreamInput;
 import org.elasticsearch.xpack.esql.io.stream.PlanStreamOutput;
@@ -243,12 +243,12 @@ public FieldName fieldName() {
     }
 
     /**
-     * If the underlying field is an {@link InvalidMappedField} (ambiguous type across indices),
+     * If the underlying field is a {@link TypeConflictField} (ambiguous type across indices),
      * converts this attribute into an {@link UnsupportedAttribute} with a descriptive error message
      * so the analyzer can surface a clear user-facing error.
      */
     public Attribute flagTypeConflicts() {
-        if (field instanceof InvalidMappedField imf) {
+        if (field instanceof TypeConflictField imf) {
             // Field has conflicting types across indices — build a user-facing error message.
             String unresolvedMessage = "Cannot use field [" + name() + "] due to ambiguities being " + imf.errorMessage();
             List<String> types = imf.getTypesToIndices().keySet().stream().toList();

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

@@ -9,7 +9,7 @@
 import org.elasticsearch.xpack.esql.core.expression.Expression.TypeResolution;
 import org.elasticsearch.xpack.esql.core.type.DataType;
 import org.elasticsearch.xpack.esql.core.type.EsField;
-import org.elasticsearch.xpack.esql.core.type.InvalidMappedField;
+import org.elasticsearch.xpack.esql.core.type.TypeConflictField;
 
 import java.util.Locale;
 import java.util.StringJoiner;
@@ -250,7 +250,7 @@ public static TypeResolution isType(
         // TODO: Shouldn't we perform widening of small numerical types here?
         if (allowUnionTypes
             && e instanceof FieldAttribute fa
-            && fa.field() instanceof InvalidMappedField imf
+            && fa.field() instanceof TypeConflictField imf
             && imf.types().stream().allMatch(predicate)) {
             return TypeResolution.TYPE_RESOLVED;
         }

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

@@ -0,0 +1,136 @@
+/*
+ * 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.checkerframework.checker.nullness.qual.NonNull;
+import org.elasticsearch.TransportVersion;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.xpack.esql.core.QlIllegalArgumentException;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.TreeMap;
+
+/**
+ * Memory-frugal counterpart to {@link InvalidMappedField}: stores at most {@value #MAX_INDICES_PER_TYPE} concrete index names per source
+ * type instead of the full per-type index list.
+ */
+public final class CompactInvalidMappedField extends EsField implements TypeConflictField {
+    private static final int MAX_INDICES_PER_TYPE = 3;
+
+    private final String errorMessage;
+    private final Map<DataType, Set<String>> typesToIndices;
+    private final boolean isPotentiallyUnmapped;
+
+    public CompactInvalidMappedField(String name, Map<DataType, Set<String>> typesToIndices) {
+        this(name, makeErrorMessage(typesToIndices, false), truncate(typesToIndices), false);
+    }
+
+    public static CompactInvalidMappedField potentiallyUnmapped(String name, Map<DataType, Set<String>> typesToIndices) {
+        return new CompactInvalidMappedField(name, makeErrorMessage(typesToIndices, true), truncate(typesToIndices), true);
+    }
+
+    private CompactInvalidMappedField(
+        String name,
+        String errorMessage,
+        Map<DataType, Set<String>> typesToIndices,
+        boolean isPotentiallyUnmapped
+    ) {
+        super(name, DataType.UNSUPPORTED, new TreeMap<>(), false, TimeSeriesFieldType.UNKNOWN);
+        this.errorMessage = errorMessage;
+        this.typesToIndices = typesToIndices;
+        this.isPotentiallyUnmapped = isPotentiallyUnmapped;
+    }
+
+    @Override
+    public void writeContent(StreamOutput out) throws IOException {
+        throw new UnsupportedOperationException("CompactInvalidMappedField shouldn't be transported");
+    }
+
+    @Override
+    public String getWriteableName(TransportVersion transportVersion) {
+        return "InvalidMappedField";
+    }
+
+    @Override
+    public String errorMessage() {
+        return errorMessage;
+    }
+
+    @Override
+    public Map<String, Set<String>> getTypesToIndices() {
+        Map<String, Set<String>> result = new TreeMap<>();
+        typesToIndices.forEach((k, v) -> result.put(k.typeName(), v));
+        return result;
+    }
+
+    @Override
+    public boolean isPotentiallyUnmapped() {
+        return isPotentiallyUnmapped;
+    }
+
+    @Override
+    public Set<DataType> types() {
+        return typesToIndices.keySet();
+    }
+
+    @Override
+    public EsField getExactField() {
+        throw new QlIllegalArgumentException("Field [" + getName() + "] is invalid, cannot access it");
+    }
+
+    @Override
+    public Exact getExactInfo() {
+        return new Exact(false, "Field [" + getName() + "] is invalid, cannot access it");
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(super.hashCode(), errorMessage);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (super.equals(obj) == false) {
+            return false;
+        }
+        CompactInvalidMappedField other = (CompactInvalidMappedField) obj;
+        return Objects.equals(errorMessage, other.errorMessage);
+    }
+
+    /** Cap each per-type index set at {@value #MAX_INDICES_PER_TYPE} entries. */
+    private static Map<DataType, Set<String>> truncate(Map<DataType, Set<String>> typesToIndices) {
+        Map<DataType, Set<String>> result = new TreeMap<>();
+        for (Map.Entry<DataType, Set<String>> entry : typesToIndices.entrySet()) {
+            Set<String> indices = entry.getValue();
+            result.put(entry.getKey(), indices.size() <= MAX_INDICES_PER_TYPE ? Set.copyOf(indices) : truncate(indices));
+        }
+        return result;
+    }
+
+    private static @NonNull Set<String> truncate(Set<String> indices) {
+        Set<String> truncated = new LinkedHashSet<>(MAX_INDICES_PER_TYPE + 1);
+        indices.stream().sorted().limit(MAX_INDICES_PER_TYPE).forEach(truncated::add);
+        truncated.add("...");
+        return Collections.unmodifiableSet(truncated);
+    }
+
+    /**
+     * Adapter onto {@link TypeConflictField#makeErrorMessage(Map, boolean)} since that one is shared with {@link InvalidMappedField} and
+     * therefore takes string keys. The string-keyed view is built ad-hoc here, used to render the message, and discarded.
+     */
+    private static String makeErrorMessage(Map<DataType, Set<String>> typesToIndices, boolean includeInsistKeyword) {
+        Map<String, Set<String>> stringKeyed = new TreeMap<>();
+        typesToIndices.forEach((k, v) -> stringKeyed.put(k.typeName(), v));
+        return TypeConflictField.makeErrorMessage(stringKeyed, includeInsistKeyword);
+    }
+}