[ESQL] Consolidate EsqlProject into Project

x-pack/plugin/esql/compute/gen/build.gradle

@@ -2,6 +2,7 @@ apply plugin: 'elasticsearch.build'
 
 dependencies {
   api project(':x-pack:plugin:esql:compute:ann')
+  implementation project(':libs:core')
   api 'com.squareup:javapoet:1.13.0'
 }
 

x-pack/plugin/esql/compute/gen/src/main/java/module-info.java

@@ -13,6 +13,7 @@
     requires com.squareup.javapoet;
     requires org.elasticsearch.compute.ann;
     requires java.compiler;
+    requires org.elasticsearch.base;
 
     exports org.elasticsearch.compute.gen;
     exports org.elasticsearch.compute.gen.argument;

x-pack/plugin/esql/compute/gen/src/main/java/org/elasticsearch/compute/gen/MarkerAnnotationProcessor.java

@@ -0,0 +1,67 @@
+/*
+ * 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.compute.gen;
+
+import org.elasticsearch.core.UpdateForV10;
+
+import java.util.Set;
+
+import javax.annotation.processing.Completion;
+import javax.annotation.processing.ProcessingEnvironment;
+import javax.annotation.processing.Processor;
+import javax.annotation.processing.RoundEnvironment;
+import javax.lang.model.SourceVersion;
+import javax.lang.model.element.AnnotationMirror;
+import javax.lang.model.element.Element;
+import javax.lang.model.element.ExecutableElement;
+import javax.lang.model.element.TypeElement;
+
+/**
+ * A no-op annotation processor that claims marker annotations like {@code @UpdateForV10}.
+ * <p>
+ * These annotations are documentation-only markers (with {@code @Retention(SOURCE)}) used to
+ * track code that needs cleanup in future versions. Since the ESQL module uses annotation
+ * processors for code generation, the compiler warns about unclaimed annotations. This
+ * processor claims them to suppress those warnings.
+ */
+public class MarkerAnnotationProcessor implements Processor {
+
+    @Override
+    public Set<String> getSupportedOptions() {
+        return Set.of();
+    }
+
+    @Override
+    public Set<String> getSupportedAnnotationTypes() {
+        // Marker annotations that are documentation-only and don't require processing.
+        return Set.of(UpdateForV10.class.getCanonicalName());
+    }
+
+    @Override
+    public SourceVersion getSupportedSourceVersion() {
+        return SourceVersion.RELEASE_21;
+    }
+
+    @Override
+    public void init(ProcessingEnvironment processingEnvironment) {}
+
+    @Override
+    public Iterable<? extends Completion> getCompletions(
+        Element element,
+        AnnotationMirror annotationMirror,
+        ExecutableElement executableElement,
+        String s
+    ) {
+        return Set.of();
+    }
+
+    @Override
+    public boolean process(Set<? extends TypeElement> annotations, RoundEnvironment roundEnvironment) {
+        return true;
+    }
+}

x-pack/plugin/esql/compute/gen/src/main/resources/META-INF/services/javax.annotation.processing.Processor

@@ -1,3 +1,4 @@
 org.elasticsearch.compute.gen.AggregatorProcessor
 org.elasticsearch.compute.gen.ConsumeProcessor
 org.elasticsearch.compute.gen.EvaluatorProcessor
+org.elasticsearch.compute.gen.MarkerAnnotationProcessor

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

@@ -143,7 +143,6 @@
 import org.elasticsearch.xpack.esql.plan.logical.join.JoinType;
 import org.elasticsearch.xpack.esql.plan.logical.join.JoinTypes;
 import org.elasticsearch.xpack.esql.plan.logical.join.LookupJoin;
-import org.elasticsearch.xpack.esql.plan.logical.local.EsqlProject;
 import org.elasticsearch.xpack.esql.plan.logical.local.LocalRelation;
 import org.elasticsearch.xpack.esql.plan.logical.local.LocalSupplier;
 import org.elasticsearch.xpack.esql.plan.logical.promql.PromqlCommand;
@@ -908,7 +907,7 @@ private LogicalPlan resolveFork(Fork fork, AnalyzerContext context) {
                 }
 
                 List<String> subPlanColumns = logicalPlan.output().stream().map(Attribute::name).toList();
-                // We need to add an explicit EsqlProject to align the outputs.
+                // We need to add an explicit Project to align the outputs.
                 if (logicalPlan instanceof Project == false || subPlanColumns.equals(forkColumns) == false) {
                     changed = true;
                     List<Attribute> newOutput = new ArrayList<>();
@@ -1264,7 +1263,7 @@ private LogicalPlan resolveKeep(Project p, List<Attribute> childOutput) {
                 resolvedProjections = new ArrayList<>(priorities.keySet());
             }
 
-            return new EsqlProject(p.source(), p.child(), resolvedProjections);
+            return new Project(p.source(), p.child(), resolvedProjections);
         }
 
         private LogicalPlan resolveDrop(Drop drop, List<Attribute> childOutput) {
@@ -1294,13 +1293,13 @@ private LogicalPlan resolveDrop(Drop drop, List<Attribute> childOutput) {
                 });
             }
 
-            return new EsqlProject(drop.source(), drop.child(), resolvedProjections);
+            return new Project(drop.source(), drop.child(), resolvedProjections);
         }
 
         private LogicalPlan resolveRename(Rename rename, List<Attribute> childrenOutput) {
             List<NamedExpression> projections = projectionsForRename(rename, childrenOutput, log);
 
-            return new EsqlProject(rename.source(), rename.child(), projections);
+            return new Project(rename.source(), rename.child(), projections);
         }
 
         /**
@@ -2591,12 +2590,12 @@ private static Map<String, Set<AbstractConvertFunction>> collectConvertFunctions
          * Push down the conversion functions into the child plan by adding an Eval with the new aliases on top of the child plan.
          */
         private static LogicalPlan maybePushDownConvertFunctionsToChild(LogicalPlan child, List<Alias> aliases, List<Attribute> output) {
-            // Fork/UnionAll adds an EsqlProject on top of each child plan during resolveFork, check this pattern before pushing down
+            // Fork/UnionAll adds an Project on top of each child plan during resolveFork, check this pattern before pushing down
             // If the pattern doesn't match, something unexpected happened, just return the child as is
-            if (aliases.isEmpty() == false && child instanceof EsqlProject esqlProject) {
-                LogicalPlan childOfProject = esqlProject.child();
+            if (aliases.isEmpty() == false && child instanceof Project project) {
+                LogicalPlan childOfProject = project.child();
                 Eval eval = new Eval(childOfProject.source(), childOfProject, aliases);
-                return new EsqlProject(esqlProject.source(), eval, output);
+                return new Project(project.source(), eval, output);
             }
             return child;
         }
@@ -2736,7 +2735,7 @@ private static LogicalPlan implicitCastingUnionAllOutput(
                         outputChanged = true;
                     }
                 }
-                // create a new eval for the casting expressions, and push it down under the EsqlProject
+                // create a new eval for the casting expressions, and push it down under the Project
                 newChildren.add(maybePushDownConvertFunctionsToChild(child, newAliases, newChildOutput));
             }
 

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

@@ -114,6 +114,7 @@ Collection<Failure> verify(LogicalPlan plan, BitSet partialMetrics) {
 
             checkOperationsOnUnsignedLong(p, failures);
             checkBinaryComparison(p, failures);
+            checkUnsupportedAttributeRenaming(p, failures);
             checkInsist(p, failures);
             checkLimitBeforeInlineStats(p, failures);
         });
@@ -152,14 +153,10 @@ else if (p.resolved()) {
                 }
 
                 e.forEachUp(ae -> {
-                    // Special handling for Project and unsupported/union types: disallow renaming them but pass them through otherwise.
-                    if (p instanceof Project || p instanceof Insist) {
-                        if (ae instanceof Alias as && as.child() instanceof UnsupportedAttribute ua) {
-                            failures.add(fail(ae, ua.unresolvedMessage()));
-                        }
-                        if (ae instanceof UnsupportedAttribute) {
-                            return;
-                        }
+                    // UnsupportedAttribute can pass through Project/Insist unchanged.
+                    // Renaming is checked separately in #checkUnsupportedAttributeRenaming.
+                    if ((p instanceof Project || p instanceof Insist) && ae instanceof UnsupportedAttribute) {
+                        return;
                     }
 
                     // Do not fail multiple times in case the children are already unresolved.
@@ -275,6 +272,22 @@ private static void checkInsist(LogicalPlan p, Failures failures) {
         }
     }
 
+    /**
+     * Check that UnsupportedAttribute is not renamed via Alias in Project or Insist.
+     * UnsupportedAttribute can pass through these plans unchanged, but renaming is not allowed.
+     * This check runs unconditionally (not gated by {@link LogicalPlan#resolved()}) because
+     * {@link Project#expressionsResolved()} treats UnsupportedAttribute as resolved to allow pass-through.
+     */
+    private static void checkUnsupportedAttributeRenaming(LogicalPlan p, Failures failures) {
+        if (p instanceof Project || p instanceof Insist) {
+            p.forEachExpression(Alias.class, alias -> {
+                if (alias.child() instanceof UnsupportedAttribute ua) {
+                    failures.add(fail(alias, ua.unresolvedMessage()));
+                }
+            });
+        }
+    }
+
     /*
      * This is a rudimentary check to prevent INLINE STATS after LIMIT. A LIMIT command can be added by other commands by default,
      * the best example being FORK. A more robust solution would be to track the commands that add LIMIT and prevent them from doing so

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/PushDownFilterAndLimitIntoUnionAll.java

@@ -40,23 +40,23 @@
  * to {@code Limit} optimization can be applied.
  *
  * This rule applies for certain patterns of {@code UnionAll} branches. The branches of a {@code UnionAll}/{@code Fork} plan has a similar
- * pattern, as {@code Fork} adds {@code EsqlProject}, an optional {@code Eval} and {@code Limit} on top of its actual children. In case
+ * pattern, as {@code Fork} adds {@code Project}, an optional {@code Eval} and {@code Limit} on top of its actual children. In case
  * there is mismatched data types on the same field across different {@code UnionAll} branches, a {@code ConvertFunction} could also be
  * added in the optional {@code Eval}.
  *
  * If the patterns of the {@code UnionAll} branches do not match the following expected patterns, the rule is not applied.
  *
- *   EsqlProject
+ *   Project
  *     Eval (optional) - added when the output of each UnionAll branch are not exactly the same
  *       Limit
  *         EsRelation
  * or
- *   EsqlProject
+ *   Project
  *     Eval (optional)
  *       Limit
  *         Subquery
  * or
- *   Limit   - CombineProjections may remove the EsqlProject on top of the limit
+ *   Limit   - CombineProjections may remove the Project on top of the limit
  *     Subquery
  */
 public final class PushDownFilterAndLimitIntoUnionAll extends Rule<LogicalPlan, LogicalPlan> {

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/local/PushExpressionsToFieldLoad.java

@@ -26,7 +26,6 @@
 import org.elasticsearch.xpack.esql.plan.logical.Project;
 import org.elasticsearch.xpack.esql.plan.logical.Row;
 import org.elasticsearch.xpack.esql.plan.logical.join.StubRelation;
-import org.elasticsearch.xpack.esql.plan.logical.local.EsqlProject;
 import org.elasticsearch.xpack.esql.rule.ParameterizedRule;
 
 import java.util.ArrayList;
@@ -203,7 +202,7 @@ private LogicalPlan transformPotentialInvocation(LogicalPlan plan) {
                 return plan;
             }
             // Found a new pushable attribute, discard it *after* use so we don't modify the output.
-            return new EsqlProject(Source.EMPTY, transformedPlan, transformedPlan.output());
+            return new Project(Source.EMPTY, transformedPlan, transformedPlan.output());
         }
 
         private Expression transformExpression(LogicalPlan nodeWithExpression, Expression e, BlockLoaderExpression ble) {

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

@@ -31,7 +31,6 @@
 import org.elasticsearch.xpack.esql.plan.logical.join.Join;
 import org.elasticsearch.xpack.esql.plan.logical.local.CopyingLocalSupplier;
 import org.elasticsearch.xpack.esql.plan.logical.local.EmptyLocalSupplier;
-import org.elasticsearch.xpack.esql.plan.logical.local.EsqlProject;
 import org.elasticsearch.xpack.esql.plan.logical.local.ImmediateLocalSupplier;
 import org.elasticsearch.xpack.esql.plan.logical.local.LocalRelation;
 import org.elasticsearch.xpack.esql.plan.physical.AggregateExec;
@@ -79,7 +78,6 @@ public static List<NamedWriteableRegistry.Entry> logical() {
             Dissect.ENTRY,
             Enrich.ENTRY,
             EsRelation.ENTRY,
-            EsqlProject.ENTRY,
             Eval.ENTRY,
             Filter.ENTRY,
             Grok.ENTRY,
@@ -92,6 +90,7 @@ public static List<NamedWriteableRegistry.Entry> logical() {
             MvExpand.ENTRY,
             OrderBy.ENTRY,
             Project.ENTRY,
+            Project.V9_ENTRY, // Backward compatibility for reading old "EsqlProject" type
             Rerank.ENTRY,
             Sample.ENTRY,
             Subquery.ENTRY,

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/Insist.java

@@ -46,7 +46,7 @@ public Insist replaceChild(LogicalPlan newChild) {
 
     @Override
     public boolean expressionsResolved() {
-        // Like EsqlProject, we allow unsupported attributes to flow through the engine.
+        // Like Project, we allow unsupported attributes to flow through the engine.
         return insistedAttributes().stream().allMatch(a -> a.resolved() || a instanceof UnsupportedAttribute);
     }
 

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/Project.java

@@ -9,7 +9,7 @@
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
-import org.elasticsearch.xpack.esql.core.capabilities.Resolvables;
+import org.elasticsearch.core.UpdateForV10;
 import org.elasticsearch.xpack.esql.core.expression.Alias;
 import org.elasticsearch.xpack.esql.core.expression.Attribute;
 import org.elasticsearch.xpack.esql.core.expression.Expressions;
@@ -18,6 +18,7 @@
 import org.elasticsearch.xpack.esql.core.tree.NodeInfo;
 import org.elasticsearch.xpack.esql.core.tree.Source;
 import org.elasticsearch.xpack.esql.expression.function.Functions;
+import org.elasticsearch.xpack.esql.expression.function.UnsupportedAttribute;
 import org.elasticsearch.xpack.esql.io.stream.PlanStreamInput;
 
 import java.io.IOException;
@@ -30,11 +31,42 @@
 public class Project extends UnaryPlan implements Streaming, SortAgnostic, SortPreserving {
     public static final NamedWriteableRegistry.Entry ENTRY = new NamedWriteableRegistry.Entry(LogicalPlan.class, "Project", Project::new);
 
+    /**
+     * Backward compatibility entry + name for reading the consolidated `EsqlProject` plans from pre-9.4.0 nodes.
+     */
+    private static final String LEGACY_PROJECT_NAME = "EsqlProject";
+
+    @UpdateForV10(owner = UpdateForV10.Owner.SEARCH_ANALYTICS)
+    public static final NamedWriteableRegistry.Entry V9_ENTRY = new NamedWriteableRegistry.Entry(
+        LogicalPlan.class,
+        LEGACY_PROJECT_NAME,
+        Project::readLegacyEsqlProject
+    );
+
+    private static Project readLegacyEsqlProject(StreamInput in) throws IOException {
+        return new Project(
+            Source.readFrom((PlanStreamInput) in),
+            in.readNamedWriteable(LogicalPlan.class),
+            in.readNamedWriteableCollectionAsList(NamedExpression.class),
+            V9_ENTRY.name
+        );
+    }
+
     private final List<? extends NamedExpression> projections;
+    private final String writeableName;
 
     public Project(Source source, LogicalPlan child, List<? extends NamedExpression> projections) {
+        this(source, child, projections, ENTRY.name);
+    }
+
+    /**
+     * Constructor that allows specifying a custom writeable name for backward compatibility.
+     * Used when deserializing legacy "EsqlProject" plans from older cluster versions.
+     */
+    private Project(Source source, LogicalPlan child, List<? extends NamedExpression> projections, String writeableName) {
         super(source, child);
         this.projections = projections;
+        this.writeableName = writeableName;
         assert validateProjections(projections);
     }
 
@@ -55,7 +87,8 @@ private Project(StreamInput in) throws IOException {
         this(
             Source.readFrom((PlanStreamInput) in),
             in.readNamedWriteable(LogicalPlan.class),
-            in.readNamedWriteableCollectionAsList(NamedExpression.class)
+            in.readNamedWriteableCollectionAsList(NamedExpression.class),
+            ENTRY.name
         );
     }
 
@@ -68,7 +101,7 @@ public void writeTo(StreamOutput out) throws IOException {
 
     @Override
     public String getWriteableName() {
-        return ENTRY.name;
+        return writeableName;
     }
 
     @Override
@@ -96,7 +129,13 @@ public boolean resolved() {
 
     @Override
     public boolean expressionsResolved() {
-        return Resolvables.resolved(projections);
+        for (NamedExpression projection : projections) {
+            // don't call dataType() - it will fail on UnresolvedAttribute
+            if (projection.resolved() == false && projection instanceof UnsupportedAttribute == false) {
+                return false;
+            }
+        }
+        return true;
     }
 
     @Override