diff --git a/docs/changelog/111475.yaml b/docs/changelog/111475.yaml
new file mode 100644
index 0000000000000..264c975444868
--- /dev/null
+++ b/docs/changelog/111475.yaml
@@ -0,0 +1,6 @@
+pr: 111475
+summary: "ESQL: Fix for overzealous validation in case of invalid mapped fields"
+area: ES|QL
+type: bug
+issues:
+ - 111452
diff --git a/docs/reference/esql/esql-multi-index.asciidoc b/docs/reference/esql/esql-multi-index.asciidoc
index cf98cbe959237..25874a132d93d 100644
--- a/docs/reference/esql/esql-multi-index.asciidoc
+++ b/docs/reference/esql/esql-multi-index.asciidoc
@@ -97,8 +97,7 @@ In addition, if the query refers to this unsupported field directly, the query f
 [source.merge.styled,esql]
 ----
 FROM events_*
-| KEEP @timestamp, client_ip, event_duration, message
-| SORT @timestamp DESC
+| SORT client_ip DESC
 ----
 
 [source,bash]
@@ -118,9 +117,8 @@ experimental::[]
 {esql} has a way to handle <<esql-multi-index-invalid-mapping, field type mismatches>>. When the same field is mapped to multiple types in multiple indices,
 the type of the field is understood to be a _union_ of the various types in the index mappings.
 As seen in the preceding examples, this _union type_ cannot be used in the results,
-and cannot be referred to by the query
--- except when it's passed to a type conversion function that accepts all the types in the _union_ and converts the field
-to a single type. {esql} offers a suite of <<esql-type-conversion-functions,type conversion functions>> to achieve this.
+and cannot be referred to by the query -- except in `KEEP`, `DROP` or when it's passed to a type conversion function that accepts all the types in
+the _union_ and converts the field to a single type. {esql} offers a suite of <<esql-type-conversion-functions,type conversion functions>> to achieve this.
 
 In the above examples, the query can use a command like `EVAL client_ip = TO_IP(client_ip)` to resolve
 the union of `ip` and `keyword` to just `ip`.
diff --git a/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/Alias.java b/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/Alias.java
index 01cc716a20547..c68e77d47ed0c 100644
--- a/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/Alias.java
+++ b/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/Alias.java
@@ -115,6 +115,11 @@ public Nullability nullable() {
         return child.nullable();
     }
 
+    @Override
+    protected TypeResolution resolveType() {
+        return child.resolveType();
+    }
+
     @Override
     public DataType dataType() {
         return child.dataType();
diff --git a/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/NamedExpression.java b/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/NamedExpression.java
index e3e9a60180da7..266ad8e2bb051 100644
--- a/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/NamedExpression.java
+++ b/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/NamedExpression.java
@@ -55,6 +55,10 @@ public boolean synthetic() {
         return synthetic;
     }
 
+    /**
+     * Try to return either {@code this} if it is an {@link Attribute}, or a {@link ReferenceAttribute} to it otherwise.
+     * Return an {@link UnresolvedAttribute} if this is unresolved.
+     */
     public abstract Attribute toAttribute();
 
     @Override
diff --git a/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/TypeResolutions.java b/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/TypeResolutions.java
index c3593e91c537e..ab05a71b0e1c6 100644
--- a/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/TypeResolutions.java
+++ b/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/TypeResolutions.java
@@ -9,6 +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 java.util.Locale;
 import java.util.StringJoiner;
@@ -176,19 +177,60 @@ public static TypeResolution isType(
         ParamOrdinal paramOrd,
         String... acceptedTypes
     ) {
-        return predicate.test(e.dataType()) || e.dataType() == NULL
-            ? TypeResolution.TYPE_RESOLVED
-            : new TypeResolution(
-                format(
-                    null,
-                    "{}argument of [{}] must be [{}], found value [{}] type [{}]",
-                    paramOrd == null || paramOrd == DEFAULT ? "" : paramOrd.name().toLowerCase(Locale.ROOT) + " ",
-                    operationName,
-                    acceptedTypesForErrorMsg(acceptedTypes),
-                    name(e),
-                    e.dataType().typeName()
-                )
-            );
+        return isType(e, predicate, operationName, paramOrd, false, acceptedTypes);
+    }
+
+    public static TypeResolution isTypeOrUnionType(
+        Expression e,
+        Predicate<DataType> predicate,
+        String operationName,
+        ParamOrdinal paramOrd,
+        String... acceptedTypes
+    ) {
+        return isType(e, predicate, operationName, paramOrd, true, acceptedTypes);
+    }
+
+    public static TypeResolution isType(
+        Expression e,
+        Predicate<DataType> predicate,
+        String operationName,
+        ParamOrdinal paramOrd,
+        boolean allowUnionTypes,
+        String... acceptedTypes
+    ) {
+        if (predicate.test(e.dataType()) || e.dataType() == NULL) {
+            return TypeResolution.TYPE_RESOLVED;
+        }
+
+        // TODO: Shouldn't we perform widening of small numerical types here?
+        if (allowUnionTypes
+            && e instanceof FieldAttribute fa
+            && fa.field() instanceof InvalidMappedField imf
+            && imf.types().stream().allMatch(predicate)) {
+            return TypeResolution.TYPE_RESOLVED;
+        }
+
+        return new TypeResolution(
+            errorStringIncompatibleTypes(operationName, paramOrd, name(e), e.dataType(), acceptedTypesForErrorMsg(acceptedTypes))
+        );
+    }
+
+    private static String errorStringIncompatibleTypes(
+        String operationName,
+        ParamOrdinal paramOrd,
+        String argumentName,
+        DataType foundType,
+        String... acceptedTypes
+    ) {
+        return format(
+            null,
+            "{}argument of [{}] must be [{}], found value [{}] type [{}]",
+            paramOrd == null || paramOrd == DEFAULT ? "" : paramOrd.name().toLowerCase(Locale.ROOT) + " ",
+            operationName,
+            acceptedTypesForErrorMsg(acceptedTypes),
+            argumentName,
+            foundType.typeName()
+        );
     }
 
     private static String acceptedTypesForErrorMsg(String... acceptedTypes) {
diff --git a/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/UnresolvedAttribute.java b/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/UnresolvedAttribute.java
index 87ef37cb84d1f..fe04df8e7e1d8 100644
--- a/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/UnresolvedAttribute.java
+++ b/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/UnresolvedAttribute.java
@@ -96,6 +96,11 @@ public UnresolvedAttribute withUnresolvedMessage(String unresolvedMessage) {
         return new UnresolvedAttribute(source(), name(), qualifier(), id(), unresolvedMessage, resolutionMetadata());
     }
 
+    @Override
+    protected TypeResolution resolveType() {
+        return new TypeResolution("unresolved attribute [" + name() + "]");
+    }
+
     @Override
     public DataType dataType() {
         throw new UnresolvedException("dataType", this);
diff --git a/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/type/InvalidMappedField.java b/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/type/InvalidMappedField.java
index 9b088cfb19f6c..9b3d7950c2a01 100644
--- a/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/type/InvalidMappedField.java
+++ b/x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/type/InvalidMappedField.java
@@ -17,6 +17,7 @@
 import java.util.Objects;
 import java.util.Set;
 import java.util.TreeMap;
+import java.util.stream.Collectors;
 
 /**
  * Representation of field mapped differently across indices.
@@ -64,6 +65,10 @@ private InvalidMappedField(StreamInput in) throws IOException {
         this(in.readString(), in.readString(), in.readImmutableMap(StreamInput::readString, i -> i.readNamedWriteable(EsField.class)));
     }
 
+    public Set<DataType> types() {
+        return typesToIndices.keySet().stream().map(DataType::fromTypeName).collect(Collectors.toSet());
+    }
+
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         out.writeString(getName());
diff --git a/x-pack/plugin/esql/qa/testFixtures/src/main/resources/stats.csv-spec b/x-pack/plugin/esql/qa/testFixtures/src/main/resources/stats.csv-spec
index be4342b95c6b8..923b3b8dacff1 100644
--- a/x-pack/plugin/esql/qa/testFixtures/src/main/resources/stats.csv-spec
+++ b/x-pack/plugin/esql/qa/testFixtures/src/main/resources/stats.csv-spec
@@ -1900,6 +1900,19 @@ M
 null
 ;
 
+shadowingInternalWithGroup2#[skip:-8.14.1,reason:implemented in 8.14]
+FROM employees
+| STATS x = MAX(emp_no), y = count(x) BY x = emp_no, x = gender
+| SORT x ASC
+;
+
+y:long | x:keyword
+    33 | F
+    57 | M
+     0 | null
+;
+
+
 shadowingTheGroup
 FROM employees
 | STATS gender = MAX(emp_no), gender = MIN(emp_no) BY gender
diff --git a/x-pack/plugin/esql/qa/testFixtures/src/main/resources/union_types.csv-spec b/x-pack/plugin/esql/qa/testFixtures/src/main/resources/union_types.csv-spec
index eaf27dca83b3e..bf7e01f2eb823 100644
--- a/x-pack/plugin/esql/qa/testFixtures/src/main/resources/union_types.csv-spec
+++ b/x-pack/plugin/esql/qa/testFixtures/src/main/resources/union_types.csv-spec
@@ -954,3 +954,240 @@ event_duration:long | _index:keyword      | ts:date                   | ts_str:k
 8268153             | sample_data_str     | 2023-10-23T13:52:55.015Z  | 2023-10-23T13:52:55.015Z  | 1698069175015  |  172.21.3.15   |  172.21.3.15
 8268153             | sample_data_ts_long | 2023-10-23T13:52:55.015Z  | 1698069175015             | 1698069175015  |  172.21.3.15   |  172.21.3.15
 ;
+
+
+multiIndexIndirectUseOfUnionTypesInSort
+// TODO: `union_types` is required only because this makes the test skip in the csv tests; better solution:
+// make the csv tests work with multiple indices.
+required_capability: union_types
+FROM sample_data, sample_data_ts_long
+| SORT client_ip ASC
+| LIMIT 1
+;
+
+@timestamp:null | client_ip:ip | event_duration:long | message:keyword
+           null | 172.21.0.5   |             1232382 | Disconnected
+;
+
+multiIndexIndirectUseOfUnionTypesInEval
+// TODO: `union_types` is required only because this makes the test skip in the csv tests; better solution:
+// make the csv tests work with multiple indices.
+required_capability: union_types
+FROM sample_data, sample_data_ts_long
+| EVAL foo = event_duration > 1232381
+| SORT client_ip ASC
+| LIMIT 1
+;
+
+@timestamp:null | client_ip:ip | event_duration:long | message:keyword | foo:boolean
+           null | 172.21.0.5   |             1232382 | Disconnected    | true
+;
+
+multiIndexIndirectUseOfUnionTypesInRename
+// TODO: `union_types` is required only because this makes the test skip in the csv tests; better solution:
+// make the csv tests work with multiple indices.
+required_capability: union_types
+required_capability: union_types_fix_rename_resolution
+FROM sample_data, sample_data_ts_long
+| RENAME message AS event_message
+| SORT client_ip ASC
+| LIMIT 1
+;
+
+@timestamp:null | client_ip:ip | event_duration:long | event_message:keyword
+           null | 172.21.0.5   |             1232382 | Disconnected
+;
+
+multiIndexIndirectUseOfUnionTypesInKeep
+// TODO: `union_types` is required only because this makes the test skip in the csv tests; better solution:
+// make the csv tests work with multiple indices.
+required_capability: union_types
+FROM sample_data, sample_data_ts_long
+| KEEP client_ip, event_duration, message
+| SORT client_ip ASC
+| LIMIT 1
+;
+
+client_ip:ip | event_duration:long | message:keyword
+172.21.0.5   |             1232382 | Disconnected
+;
+
+multiIndexIndirectUseOfUnionTypesInWildcardKeep
+// TODO: `union_types` is required only because this makes the test skip in the csv tests; better solution:
+// make the csv tests work with multiple indices.
+required_capability: union_types
+required_capability: union_types_fix_rename_resolution
+FROM sample_data, sample_data_ts_long
+| KEEP *
+| SORT client_ip ASC
+| LIMIT 1
+;
+
+@timestamp:null | client_ip:ip | event_duration:long | message:keyword
+           null | 172.21.0.5   |             1232382 | Disconnected
+;
+
+multiIndexIndirectUseOfUnionTypesInWildcardKeep2
+// TODO: `union_types` is required only because this makes the test skip in the csv tests; better solution:
+// make the csv tests work with multiple indices.
+required_capability: union_types
+required_capability: union_types_fix_rename_resolution
+FROM sample_data, sample_data_ts_long
+| KEEP *e*
+| SORT client_ip ASC
+| LIMIT 1
+;
+
+@timestamp:null | client_ip:ip | event_duration:long | message:keyword
+           null | 172.21.0.5   |             1232382 | Disconnected
+;
+
+
+multiIndexUseOfUnionTypesInKeep
+// TODO: `union_types` is required only because this makes the test skip in the csv tests; better solution:
+// make the csv tests work with multiple indices.
+required_capability: union_types
+required_capability: union_types_fix_rename_resolution
+FROM sample_data, sample_data_ts_long
+| KEEP @timestamp
+| LIMIT 1
+;
+
+@timestamp:null
+null
+;
+
+multiIndexUseOfUnionTypesInDrop
+// TODO: `union_types` is required only because this makes the test skip in the csv tests; better solution:
+// make the csv tests work with multiple indices.
+required_capability: union_types
+required_capability: union_types_fix_rename_resolution
+FROM sample_data, sample_data_ts_long
+| DROP @timestamp
+| SORT client_ip ASC
+| LIMIT 1
+;
+
+client_ip:ip | event_duration:long | message:keyword
+172.21.0.5   |             1232382 | Disconnected
+;
+
+multiIndexIndirectUseOfUnionTypesInWildcardDrop
+// TODO: `union_types` is required only because this makes the test skip in the csv tests; better solution:
+// make the csv tests work with multiple indices.
+required_capability: union_types
+required_capability: union_types_fix_rename_resolution
+FROM sample_data, sample_data_ts_long
+| DROP *time*
+| SORT client_ip ASC
+| LIMIT 1
+;
+
+client_ip:ip | event_duration:long | message:keyword
+172.21.0.5   |             1232382 | Disconnected
+;
+
+multiIndexIndirectUseOfUnionTypesInWhere
+// TODO: `union_types` is required only because this makes the test skip in the csv tests; better solution:
+// make the csv tests work with multiple indices.
+required_capability: union_types
+FROM sample_data, sample_data_ts_long
+| WHERE message == "Disconnected"
+;
+
+@timestamp:null | client_ip:ip | event_duration:long | message:keyword
+           null | 172.21.0.5   |             1232382 | Disconnected
+           null | 172.21.0.5   |             1232382 | Disconnected
+;
+
+multiIndexIndirectUseOfUnionTypesInDissect
+// TODO: `union_types` is required only because this makes the test skip in the csv tests; better solution:
+// make the csv tests work with multiple indices.
+required_capability: union_types
+FROM sample_data, sample_data_ts_long
+| DISSECT message "%{foo}"
+| SORT client_ip ASC
+| LIMIT 1
+;
+
+@timestamp:null | client_ip:ip | event_duration:long | message:keyword | foo:keyword
+           null | 172.21.0.5   |             1232382 | Disconnected    | Disconnected
+;
+
+multiIndexIndirectUseOfUnionTypesInGrok
+// TODO: `union_types` is required only because this makes the test skip in the csv tests; better solution:
+// make the csv tests work with multiple indices.
+required_capability: union_types
+FROM sample_data, sample_data_ts_long
+| GROK message "%{WORD:foo}"
+| SORT client_ip ASC
+| LIMIT 1
+;
+
+@timestamp:null | client_ip:ip | event_duration:long | message:keyword | foo:keyword
+           null | 172.21.0.5   |             1232382 | Disconnected    | Disconnected
+;
+
+multiIndexIndirectUseOfUnionTypesInEnrich
+// TODO: `union_types` is required only because this makes the test skip in the csv tests; better solution:
+// make the csv tests work with multiple indices.
+required_capability: union_types
+required_capability: enrich_load
+FROM sample_data, sample_data_ts_long
+| EVAL client_ip = client_ip::keyword
+| ENRICH clientip_policy ON client_ip WITH env
+| SORT client_ip ASC
+| LIMIT 1
+;
+
+@timestamp:null | event_duration:long | message:keyword | client_ip:keyword | env:keyword
+           null |             1232382 | Disconnected    | 172.21.0.5        | Development
+;
+
+multiIndexIndirectUseOfUnionTypesInStats
+// TODO: `union_types` is required only because this makes the test skip in the csv tests; better solution:
+// make the csv tests work with multiple indices.
+required_capability: union_types
+FROM sample_data, sample_data_ts_long
+| STATS foo = max(event_duration) BY client_ip
+| SORT client_ip ASC
+;
+
+foo:long | client_ip:ip
+ 1232382 | 172.21.0.5
+ 2764889 | 172.21.2.113
+ 3450233 | 172.21.2.162
+ 8268153 | 172.21.3.15
+;
+
+multiIndexIndirectUseOfUnionTypesInLookup
+// TODO: `union_types` is required only because this makes the test skip in the csv tests; better solution:
+// make the csv tests work with multiple indices.
+required_capability: union_types
+required_capability: lookup_v4
+FROM sample_data, sample_data_ts_long
+| SORT client_ip ASC
+| LIMIT 1
+| EVAL int = (event_duration - 1232380)::integer
+| LOOKUP int_number_names ON int
+;
+
+@timestamp:null | client_ip:ip | event_duration:long | message:keyword | int:integer | name:keyword
+           null | 172.21.0.5   |             1232382 | Disconnected    |           2 | two
+;
+
+multiIndexIndirectUseOfUnionTypesInMvExpand
+// TODO: `union_types` is required only because this makes the test skip in the csv tests; better solution:
+// make the csv tests work with multiple indices.
+required_capability: union_types
+FROM sample_data, sample_data_ts_long
+| EVAL foo = MV_APPEND(message, message)
+| SORT client_ip ASC
+| LIMIT 1
+| MV_EXPAND foo
+;
+
+@timestamp:null | client_ip:ip | event_duration:long | message:keyword | foo:keyword
+           null | 172.21.0.5   |             1232382 | Disconnected    | Disconnected
+           null | 172.21.0.5   |             1232382 | Disconnected    | Disconnected
+;
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java
index 918e9614f9070..98ed8bcb9910c 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java
@@ -123,6 +123,12 @@ public enum Cap {
          */
         UNION_TYPES_REMOVE_FIELDS,
 
+        /**
+         * Fix for union-types when renaming unrelated columns.
+         * https://github.com/elastic/elasticsearch/issues/111452
+         */
+        UNION_TYPES_FIX_RENAME_RESOLUTION,
+
         /**
          * Fix a parsing issue where numbers below Long.MIN_VALUE threw an exception instead of parsing as doubles.
          * see <a href="https://github.com/elastic/elasticsearch/issues/104323"> Parsing large numbers is inconsistent #104323 </a>
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/analysis/Analyzer.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/analysis/Analyzer.java
index 37c8cceb3f605..ded3e75174ae1 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/analysis/Analyzer.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/analysis/Analyzer.java
@@ -22,7 +22,6 @@
 import org.elasticsearch.xpack.esql.core.common.Failure;
 import org.elasticsearch.xpack.esql.core.expression.Alias;
 import org.elasticsearch.xpack.esql.core.expression.Attribute;
-import org.elasticsearch.xpack.esql.core.expression.AttributeMap;
 import org.elasticsearch.xpack.esql.core.expression.EmptyAttribute;
 import org.elasticsearch.xpack.esql.core.expression.Expression;
 import org.elasticsearch.xpack.esql.core.expression.Expressions;
@@ -437,6 +436,7 @@ private LogicalPlan resolveAggregate(Aggregate a, List<Attribute> childrenOutput
             // e.g. STATS a ... GROUP BY a = x + 1
             Holder<Boolean> changed = new Holder<>(false);
             List<Expression> groupings = a.groupings();
+            List<? extends NamedExpression> aggregates = a.aggregates();
             // first resolve groupings since the aggs might refer to them
             // trying to globally resolve unresolved attributes will lead to some being marked as unresolvable
             if (Resolvables.resolved(groupings) == false) {
@@ -456,16 +456,16 @@ private LogicalPlan resolveAggregate(Aggregate a, List<Attribute> childrenOutput
             }
 
             if (a.expressionsResolved() == false) {
-                AttributeMap<Expression> resolved = new AttributeMap<>();
+                ArrayList<Attribute> resolved = new ArrayList<>();
                 for (Expression e : groupings) {
                     Attribute attr = Expressions.attribute(e);
                     if (attr != null && attr.resolved()) {
-                        resolved.put(attr, attr);
+                        resolved.add(attr);
                     }
                 }
-                List<Attribute> resolvedList = NamedExpressions.mergeOutputAttributes(new ArrayList<>(resolved.keySet()), childrenOutput);
-                List<NamedExpression> newAggregates = new ArrayList<>();
+                List<Attribute> resolvedList = NamedExpressions.mergeOutputAttributes(resolved, childrenOutput);
 
+                List<NamedExpression> newAggregates = new ArrayList<>();
                 for (NamedExpression aggregate : a.aggregates()) {
                     var agg = (NamedExpression) aggregate.transformUp(UnresolvedAttribute.class, ua -> {
                         Expression ne = ua;
@@ -809,9 +809,18 @@ private LogicalPlan resolveEnrich(Enrich enrich, List<Attribute> childrenOutput)
                     String matchType = enrich.policy().getType();
                     DataType[] allowed = allowedEnrichTypes(matchType);
                     if (Arrays.asList(allowed).contains(dataType) == false) {
-                        String suffix = "only " + Arrays.toString(allowed) + " allowed for type [" + matchType + "]";
+                        String suffix = "only ["
+                            + Arrays.stream(allowed).map(DataType::typeName).collect(Collectors.joining(", "))
+                            + "] allowed for type ["
+                            + matchType
+                            + "]";
                         resolved = ua.withUnresolvedMessage(
-                            "Unsupported type [" + resolved.dataType() + "] for enrich matching field [" + ua.name() + "]; " + suffix
+                            "Unsupported type ["
+                                + resolved.dataType().typeName()
+                                + "] for enrich matching field ["
+                                + ua.name()
+                                + "]; "
+                                + suffix
                         );
                     }
                 }
@@ -1064,24 +1073,19 @@ public static Expression castStringLiteral(Expression from, DataType target) {
                     target,
                     e.getMessage()
                 );
-                return new UnsupportedAttribute(
-                    from.source(),
-                    String.valueOf(from.fold()),
-                    new UnsupportedEsField(String.valueOf(from.fold()), from.dataType().typeName()),
-                    message
-                );
+                return new UnresolvedAttribute(from.source(), String.valueOf(from.fold()), null, message);
             }
         }
     }
 
     /**
      * The EsqlIndexResolver will create InvalidMappedField instances for fields that are ambiguous (i.e. have multiple mappings).
-     * During ResolveRefs we do not convert these to UnresolvedAttribute instances, as we want to first determine if they can
+     * During {@link ResolveRefs} we do not convert these to UnresolvedAttribute instances, as we want to first determine if they can
      * instead be handled by conversion functions within the query. This rule looks for matching conversion functions and converts
      * those fields into MultiTypeEsField, which encapsulates the knowledge of how to convert these into a single type.
      * This knowledge will be used later in generating the FieldExtractExec with built-in type conversion.
      * Any fields which could not be resolved by conversion functions will be converted to UnresolvedAttribute instances in a later rule
-     * (See UnresolveUnionTypes below).
+     * (See {@link UnionTypesCleanup} below).
      */
     private static class ResolveUnionTypes extends Rule<LogicalPlan, LogicalPlan> {
 
@@ -1101,7 +1105,7 @@ public LogicalPlan apply(LogicalPlan plan) {
                 }
             });
 
-            return plan.transformUp(LogicalPlan.class, p -> p.resolved() || p.childrenResolved() == false ? p : doRule(p));
+            return plan.transformUp(LogicalPlan.class, p -> p.childrenResolved() == false ? p : doRule(p));
         }
 
         private LogicalPlan doRule(LogicalPlan plan) {
@@ -1117,23 +1121,6 @@ private LogicalPlan doRule(LogicalPlan plan) {
                 return plan;
             }
 
-            // In ResolveRefs the aggregates are resolved from the groupings, which might have an unresolved MultiTypeEsField.
-            // Now that we have resolved those, we need to re-resolve the aggregates.
-            if (plan instanceof EsqlAggregate agg) {
-                // If the union-types resolution occurred in a child of the aggregate, we need to check the groupings
-                plan = agg.transformExpressionsOnly(FieldAttribute.class, UnionTypesCleanup::checkUnresolved);
-
-                // Aggregates where the grouping key comes from a union-type field need to be resolved against the grouping key
-                Map<Attribute, Expression> resolved = new HashMap<>();
-                for (Expression e : agg.groupings()) {
-                    Attribute attr = Expressions.attribute(e);
-                    if (attr != null && attr.resolved()) {
-                        resolved.put(attr, e);
-                    }
-                }
-                plan = plan.transformExpressionsOnly(UnresolvedAttribute.class, ua -> resolveAttribute(ua, resolved));
-            }
-
             // And add generated fields to EsRelation, so these new attributes will appear in the OutputExec of the Fragment
             // and thereby get used in FieldExtractExec
             plan = plan.transformDown(EsRelation.class, esr -> {
@@ -1155,21 +1142,12 @@ private LogicalPlan doRule(LogicalPlan plan) {
             return plan;
         }
 
-        private Expression resolveAttribute(UnresolvedAttribute ua, Map<Attribute, Expression> resolved) {
-            var named = resolveAgainstList(ua, resolved.keySet());
-            return switch (named.size()) {
-                case 0 -> ua;
-                case 1 -> named.get(0).equals(ua) ? ua : resolved.get(named.get(0));
-                default -> ua.withUnresolvedMessage("Resolved [" + ua + "] unexpectedly to multiple attributes " + named);
-            };
-        }
-
         private Expression resolveConvertFunction(AbstractConvertFunction convert, List<FieldAttribute> unionFieldAttributes) {
             if (convert.field() instanceof FieldAttribute fa && fa.field() instanceof InvalidMappedField imf) {
                 HashMap<TypeResolutionKey, Expression> typeResolutions = new HashMap<>();
                 Set<DataType> supportedTypes = convert.supportedTypes();
-                imf.getTypesToIndices().keySet().forEach(typeName -> {
-                    DataType type = DataType.fromTypeName(typeName);
+                imf.types().forEach(type -> {
+                    // TODO: Shouldn't we perform widening of small numerical types here?
                     if (supportedTypes.contains(type)) {
                         TypeResolutionKey key = new TypeResolutionKey(fa.name(), type);
                         var concreteConvert = typeSpecificConvert(convert, fa.source(), type, imf);
@@ -1242,13 +1220,10 @@ private Expression typeSpecificConvert(AbstractConvertFunction convert, Source s
      */
     private static class UnionTypesCleanup extends Rule<LogicalPlan, LogicalPlan> {
         public LogicalPlan apply(LogicalPlan plan) {
-            LogicalPlan planWithCheckedUnionTypes = plan.transformUp(LogicalPlan.class, p -> {
-                if (p instanceof EsRelation esRelation) {
-                    // Leave esRelation as InvalidMappedField so that UNSUPPORTED fields can still pass through
-                    return esRelation;
-                }
-                return p.transformExpressionsOnly(FieldAttribute.class, UnionTypesCleanup::checkUnresolved);
-            });
+            LogicalPlan planWithCheckedUnionTypes = plan.transformUp(
+                LogicalPlan.class,
+                p -> p.transformExpressionsOnly(FieldAttribute.class, UnionTypesCleanup::checkUnresolved)
+            );
 
             // To drop synthetic attributes at the end, we need to compute the plan's output.
             // This is only legal to do if the plan is resolved.
@@ -1260,7 +1235,14 @@ public LogicalPlan apply(LogicalPlan plan) {
         static Attribute checkUnresolved(FieldAttribute fa) {
             if (fa.field() instanceof InvalidMappedField imf) {
                 String unresolvedMessage = "Cannot use field [" + fa.name() + "] due to ambiguities being " + imf.errorMessage();
-                return new UnresolvedAttribute(fa.source(), fa.name(), fa.qualifier(), fa.id(), unresolvedMessage, null);
+                String types = imf.getTypesToIndices().keySet().stream().collect(Collectors.joining(","));
+                return new UnsupportedAttribute(
+                    fa.source(),
+                    fa.name(),
+                    new UnsupportedEsField(imf.getName(), types),
+                    unresolvedMessage,
+                    fa.id()
+                );
             }
             return fa;
         }
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/analysis/Verifier.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/analysis/Verifier.java
index 514a53b0933e9..dc94353712c6d 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/analysis/Verifier.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/analysis/Verifier.java
@@ -100,16 +100,23 @@ else if (p.resolved()) {
                 }
 
                 e.forEachUp(ae -> {
-                    // we're only interested in the children
+                    // Special handling for Project and unsupported/union types: disallow renaming them but pass them through otherwise.
+                    if (p instanceof Project) {
+                        if (ae instanceof Alias as && as.child() instanceof UnsupportedAttribute ua) {
+                            failures.add(fail(ae, ua.unresolvedMessage()));
+                        }
+                        if (ae instanceof UnsupportedAttribute) {
+                            return;
+                        }
+                    }
+
+                    // Do not fail multiple times in case the children are already unresolved.
                     if (ae.childrenResolved() == false) {
                         return;
                     }
 
                     if (ae instanceof Unresolvable u) {
-                        // special handling for Project and unsupported types
-                        if (p instanceof Project == false || u instanceof UnsupportedAttribute == false) {
-                            failures.add(fail(ae, u.unresolvedMessage()));
-                        }
+                        failures.add(fail(ae, u.unresolvedMessage()));
                     }
                     if (ae.typeResolved().unresolved()) {
                         failures.add(fail(ae, ae.typeResolved().message()));
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/UnresolvedFunction.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/UnresolvedFunction.java
index ab3475635ddbd..5c898631b28fb 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/UnresolvedFunction.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/UnresolvedFunction.java
@@ -132,6 +132,11 @@ public boolean analyzed() {
         return analyzed;
     }
 
+    @Override
+    protected TypeResolution resolveType() {
+        return new TypeResolution("unresolved function [" + name + "]");
+    }
+
     @Override
     public DataType dataType() {
         throw new UnresolvedException("dataType", this);
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/convert/AbstractConvertFunction.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/convert/AbstractConvertFunction.java
index 0fed02f89fd92..2e83d048a2f68 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/convert/AbstractConvertFunction.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/convert/AbstractConvertFunction.java
@@ -37,7 +37,7 @@
 import java.util.Set;
 import java.util.function.Function;
 
-import static org.elasticsearch.xpack.esql.core.expression.TypeResolutions.isType;
+import static org.elasticsearch.xpack.esql.core.expression.TypeResolutions.isTypeOrUnionType;
 
 /**
  * Base class for functions that converts a field into a function-specific type.
@@ -77,14 +77,14 @@ protected final TypeResolution resolveType() {
         if (childrenResolved() == false) {
             return new TypeResolution("Unresolved children");
         }
-        return isType(field(), factories()::containsKey, sourceText(), null, supportedTypesNames(supportedTypes()));
+        return isTypeOrUnionType(field(), factories()::containsKey, sourceText(), null, supportedTypesNames(supportedTypes()));
     }
 
     public Set<DataType> supportedTypes() {
         return factories().keySet();
     }
 
-    public static String supportedTypesNames(Set<DataType> types) {
+    private static String supportedTypesNames(Set<DataType> types) {
         List<String> supportedTypesNames = new ArrayList<>(types.size());
         HashSet<DataType> supportTypes = new HashSet<>(types);
         if (supportTypes.containsAll(NUMERIC_TYPES)) {
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/analysis/AnalyzerTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/analysis/AnalyzerTests.java
index 82a9699412a34..edf05a3d1f281 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/analysis/AnalyzerTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/analysis/AnalyzerTests.java
@@ -631,30 +631,6 @@ public void testRenameReuseAlias() {
             """, "_meta_field", "e", "gender", "job", "job.raw", "languages", "last_name", "long_noidx", "salary");
     }
 
-    public void testRenameUnsupportedField() {
-        assertProjectionWithMapping("""
-            from test
-            | rename unsupported as u
-            | keep int, u, float
-            """, "mapping-multi-field-variation.json", "int", "u", "float");
-    }
-
-    public void testRenameUnsupportedFieldChained() {
-        assertProjectionWithMapping("""
-            from test
-            | rename unsupported as u1, u1 as u2
-            | keep int, u2, float
-            """, "mapping-multi-field-variation.json", "int", "u2", "float");
-    }
-
-    public void testRenameUnsupportedAndResolved() {
-        assertProjectionWithMapping("""
-            from test
-            | rename unsupported as u, float as f
-            | keep int, u, f
-            """, "mapping-multi-field-variation.json", "int", "u", "f");
-    }
-
     public void testRenameUnsupportedSubFieldAndResolved() {
         assertProjectionWithMapping("""
             from test
@@ -1539,7 +1515,7 @@ public void testEnrichWrongMatchFieldType() {
             | enrich languages on x
             | keep first_name, language_name, id
             """));
-        assertThat(e.getMessage(), containsString("Unsupported type [BOOLEAN] for enrich matching field [x]; only [KEYWORD,"));
+        assertThat(e.getMessage(), containsString("Unsupported type [boolean] for enrich matching field [x]; only [keyword, "));
 
         e = expectThrows(VerificationException.class, () -> analyze("""
             FROM airports
@@ -1547,7 +1523,7 @@ public void testEnrichWrongMatchFieldType() {
             | ENRICH city_boundaries ON x
             | KEEP abbrev, airport, region
             """, "airports", "mapping-airports.json"));
-        assertThat(e.getMessage(), containsString("Unsupported type [KEYWORD] for enrich matching field [x]; only [GEO_POINT,"));
+        assertThat(e.getMessage(), containsString("Unsupported type [keyword] for enrich matching field [x]; only [geo_point, "));
     }
 
     public void testValidEnrich() {
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/analysis/VerifierTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/analysis/VerifierTests.java
index ad08130c5b0d9..dc99c39331341 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/analysis/VerifierTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/analysis/VerifierTests.java
@@ -10,14 +10,23 @@
 import org.elasticsearch.Build;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xpack.esql.VerificationException;
+import org.elasticsearch.xpack.esql.core.index.EsIndex;
+import org.elasticsearch.xpack.esql.core.index.IndexResolution;
 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.UnsupportedEsField;
 import org.elasticsearch.xpack.esql.parser.EsqlParser;
 import org.elasticsearch.xpack.esql.parser.QueryParam;
 import org.elasticsearch.xpack.esql.parser.QueryParams;
 import org.elasticsearch.xpack.esql.type.EsqlDataTypes;
 
 import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
 
 import static org.elasticsearch.xpack.esql.EsqlTestUtils.withDefaultLimitWarning;
 import static org.elasticsearch.xpack.esql.analysis.AnalyzerTestUtils.loadMapping;
@@ -46,6 +55,164 @@ public void testIncompatibleTypesInMathOperation() {
         );
     }
 
+    public void testUnsupportedAndMultiTypedFields() {
+        final String unsupported = "unsupported";
+        final String multiTyped = "multi_typed";
+
+        EsField unsupportedField = new UnsupportedEsField(unsupported, "flattened");
+        // Use linked maps/sets to fix the order in the error message.
+        LinkedHashSet<String> ipIndices = new LinkedHashSet<>();
+        ipIndices.add("test1");
+        ipIndices.add("test2");
+        LinkedHashMap<String, Set<String>> typesToIndices = new LinkedHashMap<>();
+        typesToIndices.put("ip", ipIndices);
+        typesToIndices.put("keyword", Set.of("test3"));
+        EsField multiTypedField = new InvalidMappedField(multiTyped, typesToIndices);
+
+        // Also add an unsupported/multityped field under the names `int` and `double` so we can use `LOOKUP int_number_names ...` and
+        // `LOOKUP double_number_names` without renaming the fields first.
+        IndexResolution indexWithUnsupportedAndMultiTypedField = IndexResolution.valid(
+            new EsIndex(
+                "test*",
+                Map.of(unsupported, unsupportedField, multiTyped, multiTypedField, "int", unsupportedField, "double", multiTypedField)
+            )
+        );
+        Analyzer analyzer = AnalyzerTestUtils.analyzer(indexWithUnsupportedAndMultiTypedField);
+
+        assertEquals(
+            "1:22: Cannot use field [unsupported] with unsupported type [flattened]",
+            error("from test* | dissect unsupported \"%{foo}\"", analyzer)
+        );
+        assertEquals(
+            "1:22: Cannot use field [multi_typed] due to ambiguities being mapped as [2] incompatible types:"
+                + " [ip] in [test1, test2], [keyword] in [test3]",
+            error("from test* | dissect multi_typed \"%{foo}\"", analyzer)
+        );
+
+        assertEquals(
+            "1:19: Cannot use field [unsupported] with unsupported type [flattened]",
+            error("from test* | grok unsupported \"%{WORD:foo}\"", analyzer)
+        );
+        assertEquals(
+            "1:19: Cannot use field [multi_typed] due to ambiguities being mapped as [2] incompatible types:"
+                + " [ip] in [test1, test2], [keyword] in [test3]",
+            error("from test* | grok multi_typed \"%{WORD:foo}\"", analyzer)
+        );
+
+        assertEquals(
+            "1:36: Cannot use field [unsupported] with unsupported type [flattened]",
+            error("from test* | enrich client_cidr on unsupported", analyzer)
+        );
+        assertEquals(
+            "1:36: Unsupported type [unsupported] for enrich matching field [multi_typed];"
+                + " only [keyword, text, ip, long, integer, float, double, datetime] allowed for type [range]",
+            error("from test* | enrich client_cidr on multi_typed", analyzer)
+        );
+
+        assertEquals(
+            "1:23: Cannot use field [unsupported] with unsupported type [flattened]",
+            error("from test* | eval x = unsupported", analyzer)
+        );
+        assertEquals(
+            "1:23: Cannot use field [multi_typed] due to ambiguities being mapped as [2] incompatible types:"
+                + " [ip] in [test1, test2], [keyword] in [test3]",
+            error("from test* | eval x = multi_typed", analyzer)
+        );
+
+        assertEquals(
+            "1:32: Cannot use field [unsupported] with unsupported type [flattened]",
+            error("from test* | eval x = to_lower(unsupported)", analyzer)
+        );
+        assertEquals(
+            "1:32: Cannot use field [multi_typed] due to ambiguities being mapped as [2] incompatible types:"
+                + " [ip] in [test1, test2], [keyword] in [test3]",
+            error("from test* | eval x = to_lower(multi_typed)", analyzer)
+        );
+
+        assertEquals(
+            "1:32: Cannot use field [unsupported] with unsupported type [flattened]",
+            error("from test* | stats count(1) by unsupported", analyzer)
+        );
+        assertEquals(
+            "1:32: Cannot use field [multi_typed] due to ambiguities being mapped as [2] incompatible types:"
+                + " [ip] in [test1, test2], [keyword] in [test3]",
+            error("from test* | stats count(1) by multi_typed", analyzer)
+        );
+        if (Build.current().isSnapshot()) {
+            assertEquals(
+                "1:38: Cannot use field [unsupported] with unsupported type [flattened]",
+                error("from test* | inlinestats count(1) by unsupported", analyzer)
+            );
+            assertEquals(
+                "1:38: Cannot use field [multi_typed] due to ambiguities being mapped as [2] incompatible types:"
+                    + " [ip] in [test1, test2], [keyword] in [test3]",
+                error("from test* | inlinestats count(1) by multi_typed", analyzer)
+            );
+        }
+
+        assertEquals(
+            "1:27: Cannot use field [unsupported] with unsupported type [flattened]",
+            error("from test* | stats values(unsupported)", analyzer)
+        );
+        assertEquals(
+            "1:27: Cannot use field [multi_typed] due to ambiguities being mapped as [2] incompatible types:"
+                + " [ip] in [test1, test2], [keyword] in [test3]",
+            error("from test* | stats values(multi_typed)", analyzer)
+        );
+        if (Build.current().isSnapshot()) {
+            assertEquals(
+                "1:33: Cannot use field [unsupported] with unsupported type [flattened]",
+                error("from test* | inlinestats values(unsupported)", analyzer)
+            );
+            assertEquals(
+                "1:33: Cannot use field [multi_typed] due to ambiguities being mapped as [2] incompatible types:"
+                    + " [ip] in [test1, test2], [keyword] in [test3]",
+                error("from test* | inlinestats values(multi_typed)", analyzer)
+            );
+        }
+
+        assertEquals(
+            "1:27: Cannot use field [unsupported] with unsupported type [flattened]",
+            error("from test* | stats values(unsupported)", analyzer)
+        );
+        assertEquals(
+            "1:27: Cannot use field [multi_typed] due to ambiguities being mapped as [2] incompatible types:"
+                + " [ip] in [test1, test2], [keyword] in [test3]",
+            error("from test* | stats values(multi_typed)", analyzer)
+        );
+
+        // LOOKUP with unsupported type
+        assertEquals(
+            "1:41: column type mismatch, table column was [integer] and original column was [unsupported]",
+            error("from test* | lookup int_number_names on int", analyzer)
+        );
+        // LOOKUP with multi-typed field
+        assertEquals(
+            "1:44: column type mismatch, table column was [double] and original column was [unsupported]",
+            error("from test* | lookup double_number_names on double", analyzer)
+        );
+
+        assertEquals(
+            "1:24: Cannot use field [unsupported] with unsupported type [flattened]",
+            error("from test* | mv_expand unsupported", analyzer)
+        );
+        assertEquals(
+            "1:24: Cannot use field [multi_typed] due to ambiguities being mapped as [2] incompatible types:"
+                + " [ip] in [test1, test2], [keyword] in [test3]",
+            error("from test* | mv_expand multi_typed", analyzer)
+        );
+
+        assertEquals(
+            "1:21: Cannot use field [unsupported] with unsupported type [flattened]",
+            error("from test* | rename unsupported as x", analyzer)
+        );
+        assertEquals(
+            "1:21: Cannot use field [multi_typed] due to ambiguities being mapped as [2] incompatible types:"
+                + " [ip] in [test1, test2], [keyword] in [test3]",
+            error("from test* | rename multi_typed as x", analyzer)
+        );
+    }
+
     public void testRoundFunctionInvalidInputs() {
         assertEquals(
             "1:31: first argument of [round(b, 3)] must be [numeric], found value [b] type [keyword]",
diff --git a/x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/esql/160_union_types.yml b/x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/esql/160_union_types.yml
index 003b1d0651d11..92b3f4d1b084d 100644
--- a/x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/esql/160_union_types.yml
+++ b/x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/esql/160_union_types.yml
@@ -296,12 +296,44 @@ load two indices, showing unsupported type and null value for event_duration:
 
 ---
 load two indices with no conversion function, but needs TO_LONG conversion:
+  - requires:
+      test_runner_features: [capabilities]
+      capabilities:
+        - 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"
+
   - do:
-      catch: '/Cannot use field \[event_duration\] due to ambiguities being mapped as \[2\] incompatible types: \[keyword\] in \[events_ip_keyword\], \[long\] in \[events_ip_long\]/'
+      allowed_warnings_regex:
+        - "No limit defined, adding default limit of \\[.*\\]"
       esql.query:
         body:
           query: 'FROM events_ip_* METADATA _index | KEEP _index, @timestamp, client_ip, event_duration, message | SORT _index ASC, @timestamp DESC'
 
+  - match: { columns.0.name: "_index" }
+  - match: { columns.0.type: "keyword" }
+  - match: { columns.1.name: "@timestamp" }
+  - match: { columns.1.type: "date" }
+  - match: { columns.2.name: "client_ip" }
+  - match: { columns.2.type: "ip" }
+  - match: { columns.3.name: "event_duration" }
+  - match: { columns.3.type: "unsupported" }
+  - match: { columns.4.name: "message" }
+  - match: { columns.4.type: "keyword" }
+  - length: { values: 14 }
+  - match: { values.0.0: "events_ip_keyword" }
+  - match: { values.0.1: "2023-10-23T13:55:01.543Z" }
+  - match: { values.0.2: "172.21.3.15" }
+  - match: { values.0.3: null }
+  - match: { values.0.4: "Connected to 10.1.0.1" }
+  - match: { values.7.0: "events_ip_long" }
+  - match: { values.7.1: "2023-10-23T13:55:01.543Z" }
+  - match: { values.7.2: "172.21.3.15" }
+  - match: { values.7.3: null }
+  - match: { values.7.4: "Connected to 10.1.0.1" }
+
 ---
 load two indices with incorrect conversion function, TO_IP instead of TO_LONG:
   - do:
@@ -450,12 +482,44 @@ load two indices, showing unsupported type and null value for client_ip:
 
 ---
 load two indices with no conversion function, but needs TO_IP conversion:
+  - requires:
+      test_runner_features: [capabilities]
+      capabilities:
+        - 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"
+
   - do:
-      catch: '/Cannot use field \[client_ip\] due to ambiguities being mapped as \[2\] incompatible types: \[ip\] in \[events_ip_long\], \[keyword\] in \[events_keyword_long\]/'
+      allowed_warnings_regex:
+        - "No limit defined, adding default limit of \\[.*\\]"
       esql.query:
         body:
           query: 'FROM events_*_long METADATA _index | KEEP _index, @timestamp, client_ip, event_duration, message | SORT _index ASC, @timestamp DESC'
 
+  - match: { columns.0.name: "_index" }
+  - match: { columns.0.type: "keyword" }
+  - match: { columns.1.name: "@timestamp" }
+  - match: { columns.1.type: "date" }
+  - match: { columns.2.name: "client_ip" }
+  - match: { columns.2.type: "unsupported" }
+  - match: { columns.3.name: "event_duration" }
+  - match: { columns.3.type: "long" }
+  - match: { columns.4.name: "message" }
+  - match: { columns.4.type: "keyword" }
+  - length: { values: 14 }
+  - match: { values.0.0: "events_ip_long" }
+  - match: { values.0.1: "2023-10-23T13:55:01.543Z" }
+  - match: { values.0.2: null }
+  - match: { values.0.3: 1756467 }
+  - match: { values.0.4: "Connected to 10.1.0.1" }
+  - match: { values.7.0: "events_keyword_long" }
+  - match: { values.7.1: "2023-10-23T13:55:01.543Z" }
+  - match: { values.7.2: null }
+  - match: { values.7.3: 1756467 }
+  - match: { values.7.4: "Connected to 10.1.0.1" }
+
 ---
 load two indices with incorrect conversion function, TO_LONG instead of TO_IP:
   - do:
@@ -629,20 +693,103 @@ load two indexes, convert client_ip and group by something invalid:
 
 ---
 load four indices with single conversion function TO_LONG:
+  - requires:
+      test_runner_features: [capabilities]
+      capabilities:
+        - 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"
+
   - do:
-      catch: '/Cannot use field \[client_ip\] due to ambiguities being mapped as \[2\] incompatible types: \[ip\] in \[events_ip_keyword, events_ip_long\], \[keyword\] in \[events_keyword_keyword, events_keyword_long\]/'
+      allowed_warnings_regex:
+        - "No limit defined, adding default limit of \\[.*\\]"
       esql.query:
         body:
           query: 'FROM events_* METADATA _index | EVAL event_duration = TO_LONG(event_duration) | KEEP _index, @timestamp, client_ip, event_duration, message | SORT _index ASC, @timestamp DESC'
 
+  - match: { columns.0.name: "_index" }
+  - match: { columns.0.type: "keyword" }
+  - match: { columns.1.name: "@timestamp" }
+  - match: { columns.1.type: "date" }
+  - match: { columns.2.name: "client_ip" }
+  - match: { columns.2.type: "unsupported" }
+  - match: { columns.3.name: "event_duration" }
+  - match: { columns.3.type: "long" }
+  - match: { columns.4.name: "message" }
+  - match: { columns.4.type: "keyword" }
+  - length: { values: 28 }
+  - match: { values.0.0: "events_ip_keyword" }
+  - match: { values.0.1: "2023-10-23T13:55:01.543Z" }
+  - match: { values.0.2: null }
+  - match: { values.0.3: 1756467 }
+  - match: { values.0.4: "Connected to 10.1.0.1" }
+  - match: { values.7.0: "events_ip_long" }
+  - match: { values.7.1: "2023-10-23T13:55:01.543Z" }
+  - match: { values.7.2: null }
+  - match: { values.7.3: 1756467 }
+  - match: { values.7.4: "Connected to 10.1.0.1" }
+  - match: { values.14.0: "events_keyword_keyword" }
+  - match: { values.14.1: "2023-10-23T13:55:01.543Z" }
+  - match: { values.14.2: null }
+  - match: { values.14.3: 1756467 }
+  - match: { values.14.4: "Connected to 10.1.0.1" }
+  - match: { values.21.0: "events_keyword_long" }
+  - match: { values.21.1: "2023-10-23T13:55:01.543Z" }
+  - match: { values.21.2: null }
+  - match: { values.21.3: 1756467 }
+  - match: { values.21.4: "Connected to 10.1.0.1" }
+
 ---
 load four indices with single conversion function TO_IP:
+  - requires:
+      test_runner_features: [capabilities]
+      capabilities:
+        - 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"
+
   - do:
-      catch: '/Cannot use field \[event_duration\] due to ambiguities being mapped as \[2\] incompatible types: \[keyword\] in \[events_ip_keyword, events_keyword_keyword\], \[long\] in \[events_ip_long, events_keyword_long\]/'
+      allowed_warnings_regex:
+        - "No limit defined, adding default limit of \\[.*\\]"
       esql.query:
         body:
           query: 'FROM events_* METADATA _index | EVAL client_ip = TO_IP(client_ip) | KEEP _index, @timestamp, client_ip, event_duration, message | SORT _index ASC, @timestamp DESC'
 
+  - match: { columns.0.name: "_index" }
+  - match: { columns.0.type: "keyword" }
+  - match: { columns.1.name: "@timestamp" }
+  - match: { columns.1.type: "date" }
+  - match: { columns.2.name: "client_ip" }
+  - match: { columns.2.type: "ip" }
+  - match: { columns.3.name: "event_duration" }
+  - match: { columns.3.type: "unsupported" }
+  - match: { columns.4.name: "message" }
+  - match: { columns.4.type: "keyword" }
+  - length: { values: 28 }
+  - match: { values.0.0: "events_ip_keyword" }
+  - match: { values.0.1: "2023-10-23T13:55:01.543Z" }
+  - match: { values.0.2: "172.21.3.15" }
+  - match: { values.0.3: null }
+  - match: { values.0.4: "Connected to 10.1.0.1" }
+  - match: { values.7.0: "events_ip_long" }
+  - match: { values.7.1: "2023-10-23T13:55:01.543Z" }
+  - match: { values.7.2: "172.21.3.15" }
+  - match: { values.7.3: null }
+  - match: { values.7.4: "Connected to 10.1.0.1" }
+  - match: { values.14.0: "events_keyword_keyword" }
+  - match: { values.14.1: "2023-10-23T13:55:01.543Z" }
+  - match: { values.14.2: "172.21.3.15" }
+  - match: { values.14.3: null }
+  - match: { values.14.4: "Connected to 10.1.0.1" }
+  - match: { values.21.0: "events_keyword_long" }
+  - match: { values.21.1: "2023-10-23T13:55:01.543Z" }
+  - match: { values.21.2: "172.21.3.15" }
+  - match: { values.21.3: null }
+  - match: { values.21.4: "Connected to 10.1.0.1" }
 ---
 load four indices with multiple conversion functions TO_LONG and TO_IP:
   - do:
diff --git a/x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/esql/40_tsdb.yml b/x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/esql/40_tsdb.yml
index 8f291600acbf6..642407ac6d45b 100644
--- a/x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/esql/40_tsdb.yml
+++ b/x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/esql/40_tsdb.yml
@@ -253,12 +253,24 @@ from index pattern unsupported counter:
 
 ---
 from index pattern explicit counter use:
+  - requires:
+      test_runner_features: [capabilities]
+      capabilities:
+        - 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"
+
   - do:
-      catch: '/Cannot use field \[k8s.pod.network.tx\] due to ambiguities being mapped as different metric types in indices: \[test, test2\]/'
+      allowed_warnings_regex:
+        - "No limit defined, adding default limit of \\[.*\\]"
       esql.query:
         body:
           query: 'FROM test* | keep *.tx'
-
+  - match: {columns.0.name: "k8s.pod.network.tx"}
+  - match: {columns.0.type: "unsupported"}
+  - length: {values: 10}
 
 ---
 _source: