ESQL: Fix synthetic attribute pruning

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

@@ -195,7 +195,13 @@ static TransportVersion def(int id) {
     public static final TransportVersion ESQL_PROFILE_SLEEPS = def(8_725_00_0);
     public static final TransportVersion ZDT_NANOS_SUPPORT = def(8_726_00_0);
     public static final TransportVersion LTR_SERVERLESS_RELEASE = def(8_727_00_0);
-    public static final TransportVersion ESQL_AGGREGATE_EXEC_TRACKS_INTERMEDIATE_ATTRS = def(8_728_00_0);
+    public static final TransportVersion ALLOW_PARTIAL_SEARCH_RESULTS_IN_PIT = def(8_728_00_0);
+    public static final TransportVersion RANK_DOCS_RETRIEVER = def(8_729_00_0);
+    public static final TransportVersion ESQL_ES_FIELD_CACHED_SERIALIZATION = def(8_730_00_0);
+    public static final TransportVersion ADD_MANAGE_ROLES_PRIVILEGE = def(8_731_00_0);
+    public static final TransportVersion REPOSITORIES_TELEMETRY = def(8_732_00_0);
+    public static final TransportVersion ML_INFERENCE_ALIBABACLOUD_SEARCH_ADDED = def(8_733_00_0);
+    public static final TransportVersion ESQL_AGGREGATE_EXEC_TRACKS_INTERMEDIATE_ATTRS = def(8_734_00_0);
 
     /*
      * STOP! READ THIS FIRST! No, really,

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

@@ -65,9 +65,11 @@ public FieldAttribute(
         this(source, parent, name, field.getDataType(), field, nullability, id, synthetic);
     }
 
-    // TODO: Should this become private? An explicitly set type will be discarded the moment this is `clone`d, so using this constructor is
-    // deceptive.
-    public FieldAttribute(
+    /**
+     * Used only for testing. Do not use this otherwise, as an explicitly set type will be ignored the next time this FieldAttribute is
+     * {@link FieldAttribute#clone}d.
+     */
+    FieldAttribute(
         Source source,
         FieldAttribute parent,
         String name,
@@ -197,6 +199,11 @@ protected Attribute clone(Source source, String name, DataType type, Nullability
         return new FieldAttribute(source, parent, name, field, nullability, id, synthetic);
     }
 
+    @Override
+    public Attribute withDataType(DataType type) {
+        throw new UnsupportedOperationException("FieldAttribute obtains its type from the contained EsField.");
+    }
+
     @Override
     public int hashCode() {
         return Objects.hash(super.hashCode(), path, field);

x-pack/plugin/esql-core/src/test/java/org/elasticsearch/xpack/esql/core/expression/FieldAttributeTestUtils.java

@@ -0,0 +1,27 @@
+/*
+ * 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.expression;
+
+import org.elasticsearch.xpack.esql.core.tree.Source;
+import org.elasticsearch.xpack.esql.core.type.DataType;
+import org.elasticsearch.xpack.esql.core.type.EsField;
+
+public class FieldAttributeTestUtils {
+    public static final FieldAttribute newFieldAttributeWithType(
+        Source source,
+        FieldAttribute parent,
+        String name,
+        DataType type,
+        EsField field,
+        Nullability nullability,
+        NameId id,
+        boolean synthetic
+    ) {
+        return new FieldAttribute(source, parent, name, type, field, nullability, id, synthetic);
+    }
+}

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/physical/AggregateExec.java

@@ -36,10 +36,9 @@ public class AggregateExec extends UnaryExec implements EstimatesRowSize {
     private final List<? extends Expression> groupings;
     private final List<? extends NamedExpression> aggregates;
     /**
-     * The output attributes of {@link AggregatorMode#INITIAL} aggregations, resp.
-     * the input attributes of {@link AggregatorMode#FINAL} aggregations.
+     * The output attributes of {@link AggregatorMode#INITIAL} and {@link AggregatorMode#INTERMEDIATE} aggregations, resp.
+     * the input attributes of {@link AggregatorMode#FINAL} and {@link AggregatorMode#INTERMEDIATE} aggregations.
      */
-    // TODO: For INTERMEDIATE, should the input attributes be the same as the output attributes? Currently, they are.
     private final List<Attribute> intermediateAttributes;
 
     private final AggregatorMode mode;

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

@@ -14,7 +14,6 @@
 import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.compute.operator.DriverProfile;
 import org.elasticsearch.core.Releasables;
-import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.logging.LogManager;
 import org.elasticsearch.logging.Logger;
@@ -54,7 +53,6 @@
 import org.elasticsearch.xpack.esql.plan.logical.Phased;
 import org.elasticsearch.xpack.esql.plan.logical.Project;
 import org.elasticsearch.xpack.esql.plan.logical.RegexExtract;
-import org.elasticsearch.xpack.esql.plan.logical.UnresolvedRelation;
 import org.elasticsearch.xpack.esql.plan.physical.EstimatesRowSize;
 import org.elasticsearch.xpack.esql.plan.physical.FragmentExec;
 import org.elasticsearch.xpack.esql.plan.physical.PhysicalPlan;
@@ -302,9 +300,9 @@ static Set<String> fieldNames(LogicalPlan parsed, Set<String> enrichPolicyMatchF
                 references.addAll(enrichRefs);
             } else {
                 references.addAll(p.references());
-                // TODO: Is this still needed? This branch used to live in UnresolvedRelation.references().
-                if (p instanceof UnresolvedRelation ur && ur.indexMode() == IndexMode.TIME_SERIES) {
-                    references.add(new UnresolvedAttribute(ur.source(), MetadataAttribute.TIMESTAMP_FIELD));
+                if (p instanceof Aggregate aggregate && aggregate.aggregateType() == Aggregate.AggregateType.METRICS) {
+                    // The METRICS command relies on @timestamp without the user having to mention it.
+                    references.add(new UnresolvedAttribute(aggregate.source(), MetadataAttribute.TIMESTAMP_FIELD));
                 }
                 // special handling for UnresolvedPattern (which is not an UnresolvedAttribute)
                 p.forEachExpression(UnresolvedNamePattern.class, up -> {

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/FieldAttributeTests.java

@@ -15,6 +15,8 @@
 import org.elasticsearch.xpack.esql.core.type.EsField;
 import org.elasticsearch.xpack.esql.type.AbstractEsFieldTypeTests;
 
+import static org.elasticsearch.xpack.esql.core.expression.FieldAttributeTestUtils.newFieldAttributeWithType;
+
 public class FieldAttributeTests extends AbstractAttributeTestCase<FieldAttribute> {
     public static FieldAttribute createFieldAttribute(int maxDepth, boolean onlyRepresentable) {
         Source source = Source.EMPTY;
@@ -26,7 +28,7 @@ public static FieldAttribute createFieldAttribute(int maxDepth, boolean onlyRepr
         EsField field = AbstractEsFieldTypeTests.randomAnyEsField(maxDepth);
         Nullability nullability = randomFrom(Nullability.values());
         boolean synthetic = randomBoolean();
-        return new FieldAttribute(source, parent, name, type, field, nullability, new NameId(), synthetic);
+        return newFieldAttributeWithType(source, parent, name, type, field, nullability, new NameId(), synthetic);
     }
 
     @Override
@@ -51,6 +53,6 @@ protected FieldAttribute mutate(FieldAttribute instance) {
             case 4 -> nullability = randomValueOtherThan(nullability, () -> randomFrom(Nullability.values()));
             case 5 -> synthetic = false == synthetic;
         }
-        return new FieldAttribute(source, parent, name, type, field, nullability, new NameId(), synthetic);
+        return newFieldAttributeWithType(source, parent, name, type, field, nullability, new NameId(), synthetic);
     }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/io/stream/PlanNamedTypesTests.java

@@ -70,6 +70,7 @@
 
 import static org.elasticsearch.test.ListMatcher.matchesList;
 import static org.elasticsearch.test.MapMatcher.assertMap;
+import static org.elasticsearch.xpack.esql.core.expression.FieldAttributeTestUtils.newFieldAttributeWithType;
 import static org.hamcrest.Matchers.equalTo;
 
 public class PlanNamedTypesTests extends ESTestCase {
@@ -151,7 +152,7 @@ static FieldAttribute randomFieldAttributeOrNull() {
     }
 
     static FieldAttribute randomFieldAttribute() {
-        return new FieldAttribute(
+        return newFieldAttributeWithType(
             Source.EMPTY,
             randomFieldAttributeOrNull(), // parent
             randomAlphaOfLength(randomIntBetween(1, 25)), // name

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/session/IndexResolverFieldNamesTests.java

@@ -1227,6 +1227,26 @@ public void testEnrichOnDefaultField() {
         assertThat(fieldNames, equalTo(ALL_FIELDS));
     }
 
+    public void testMetrics() {
+        Set<String> fieldNames = EsqlSession.fieldNames(parser.createStatement("""
+            METRICS k8s bytes=sum(rate(network.total_bytes_in)), sum(rate(network.total_cost)) BY cluster"""), Set.of());
+        assertThat(
+            fieldNames,
+            equalTo(
+                Set.of(
+                    "@timestamp",
+                    "@timestamp.*",
+                    "network.total_bytes_in",
+                    "network.total_bytes_in.*",
+                    "network.total_cost",
+                    "network.total_cost.*",
+                    "cluster",
+                    "cluster.*"
+                )
+            )
+        );
+    }
+
     private void assertFieldNames(String query, Set<String> expected) {
         Set<String> fieldNames = EsqlSession.fieldNames(parser.createStatement(query), Collections.emptySet());
         assertThat(fieldNames, equalTo(expected));