Do not serialize EsIndex in plan (#119580)

Certain plan classes (such as EsRelation, EsSourceExec, EsQueryExec) contain and serialize the entire EsIndex instance.
This instance might contain huge mapping that is never used in plan. This change replaces EsIndex usage with indexPattern and indexNameWithModes to minimize the size of the serialized plan.

Author: idegtiarenko

docs/changelog/119580.yaml

@@ -0,0 +1,5 @@
+pr: 119580
+summary: Do not serialize `EsIndex` in plan
+area: ES|QL
+type: enhancement
+issues: []

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

@@ -158,6 +158,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion ESQL_PROFILE_ROWS_PROCESSED = def(8_824_00_0);
     public static final TransportVersion BYTE_SIZE_VALUE_ALWAYS_USES_BYTES_1 = def(8_825_00_0);
     public static final TransportVersion REVERT_BYTE_SIZE_VALUE_ALWAYS_USES_BYTES_1 = def(8_826_00_0);
+    public static final TransportVersion ESQL_SKIP_ES_INDEX_SERIALIZATION = def(8_827_00_0);
 
     /*
      * STOP! READ THIS FIRST! No, really,

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

@@ -215,7 +215,7 @@ public static Range rangeOf(Expression value, Expression lower, boolean includeL
     }
 
     public static EsRelation relation() {
-        return new EsRelation(EMPTY, new EsIndex(randomAlphaOfLength(8), emptyMap()), IndexMode.STANDARD, randomBoolean());
+        return new EsRelation(EMPTY, new EsIndex(randomAlphaOfLength(8), emptyMap()), IndexMode.STANDARD);
     }
 
     /**

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

@@ -269,7 +269,13 @@ private LogicalPlan resolveIndex(UnresolvedRelation plan, IndexResolution indexR
             }
             var attributes = mappingAsAttributes(plan.source(), esIndex.mapping());
             attributes.addAll(plan.metadataFields());
-            return new EsRelation(plan.source(), esIndex, attributes.isEmpty() ? NO_FIELDS : attributes, plan.indexMode());
+            return new EsRelation(
+                plan.source(),
+                esIndex.name(),
+                plan.indexMode(),
+                esIndex.indexNameWithModes(),
+                attributes.isEmpty() ? NO_FIELDS : attributes
+            );
         }
     }
 
@@ -1371,9 +1377,13 @@ private LogicalPlan doRule(LogicalPlan plan) {
                 }
 
                 if (missing.isEmpty() == false) {
-                    List<Attribute> newOutput = new ArrayList<>(esr.output());
-                    newOutput.addAll(missing);
-                    return new EsRelation(esr.source(), esr.index(), newOutput, esr.indexMode(), esr.frozen());
+                    return new EsRelation(
+                        esr.source(),
+                        esr.indexPattern(),
+                        esr.indexMode(),
+                        esr.indexNameWithModes(),
+                        CollectionUtils.combine(esr.output(), missing)
+                    );
                 }
                 return esr;
             });

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

@@ -102,13 +102,13 @@ public LogicalPlan apply(LogicalPlan plan) {
                             p = new Eval(eval.source(), eval.child(), remaining);
                         }
                     }
-                } else if (p instanceof EsRelation esRelation && esRelation.indexMode() == IndexMode.LOOKUP) {
+                } else if (p instanceof EsRelation esr && esr.indexMode() == IndexMode.LOOKUP) {
                     // Normally, pruning EsRelation has no effect because InsertFieldExtraction only extracts the required fields, anyway.
                     // However, InsertFieldExtraction can't be currently used in LOOKUP JOIN right index,
                     // it works differently as we extract all fields (other than the join key) that the EsRelation has.
-                    var remaining = removeUnused(esRelation.output(), used);
+                    var remaining = removeUnused(esr.output(), used);
                     if (remaining != null) {
-                        p = new EsRelation(esRelation.source(), esRelation.index(), remaining, esRelation.indexMode(), esRelation.frozen());
+                        p = new EsRelation(esr.source(), esr.indexPattern(), esr.indexMode(), esr.indexNameWithModes(), remaining);
                     }
                 }
             } while (recheck);

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

@@ -16,6 +16,6 @@ public final class SkipQueryOnEmptyMappings extends OptimizerRules.OptimizerRule
 
     @Override
     protected LogicalPlan rule(EsRelation plan) {
-        return plan.index().concreteIndices().isEmpty() ? new LocalRelation(plan.source(), plan.output(), LocalSupplier.EMPTY) : plan;
+        return plan.concreteIndices().isEmpty() ? new LocalRelation(plan.source(), plan.output(), LocalSupplier.EMPTY) : plan;
     }
 }

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

@@ -220,7 +220,7 @@ private static Aggregate toStandardAggregate(Aggregate metrics) {
             if (attributes.stream().noneMatch(a -> a.name().equals(MetadataAttribute.TIMESTAMP_FIELD))) {
                 attributes.removeIf(a -> a.name().equals(MetadataAttribute.TIMESTAMP_FIELD));
             }
-            return new EsRelation(r.source(), r.index(), new ArrayList<>(attributes), IndexMode.STANDARD);
+            return new EsRelation(r.source(), r.indexPattern(), IndexMode.STANDARD, r.indexNameWithModes(), new ArrayList<>(attributes));
         });
         return new Aggregate(metrics.source(), child, Aggregate.AggregateType.STANDARD, metrics.groupings(), metrics.aggregates());
     }

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

@@ -104,8 +104,9 @@ private static PhysicalPlan rewrite(
             var query = Queries.combine(Queries.Clause.FILTER, asList(queryExec.query(), planQuery));
             queryExec = new EsQueryExec(
                 queryExec.source(),
-                queryExec.index(),
+                queryExec.indexPattern(),
                 queryExec.indexMode(),
+                queryExec.indexNameWithModes(),
                 queryExec.output(),
                 query,
                 queryExec.limit(),

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

@@ -59,7 +59,7 @@ protected PhysicalPlan rule(AggregateExec aggregateExec, LocalPhysicalOptimizerC
             if (tuple.v2().size() == aggregateExec.aggregates().size()) {
                 plan = new EsStatsQueryExec(
                     aggregateExec.source(),
-                    queryExec.index(),
+                    queryExec.indexPattern(),
                     queryExec.query(),
                     queryExec.limit(),
                     tuple.v1(),

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

@@ -53,6 +53,6 @@ protected PhysicalPlan rule(EsSourceExec plan) {
                 attributes.add(ma);
             }
         });
-        return new EsQueryExec(plan.source(), plan.index(), plan.indexMode(), attributes, plan.query());
+        return new EsQueryExec(plan.source(), plan.indexPattern(), plan.indexMode(), plan.indexNameWithModes(), attributes, plan.query());
     }
 }

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

@@ -27,6 +27,9 @@
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Objects;
+import java.util.Set;
+
+import static org.elasticsearch.TransportVersions.ESQL_SKIP_ES_INDEX_SERIALIZATION;
 
 public class EsRelation extends LeafPlan {
     public static final NamedWriteableRegistry.Entry ENTRY = new NamedWriteableRegistry.Entry(
@@ -35,30 +38,41 @@ public class EsRelation extends LeafPlan {
         EsRelation::readFrom
     );
 
-    private final EsIndex index;
-    private final List<Attribute> attrs;
-    private final boolean frozen;
+    private final String indexPattern;
     private final IndexMode indexMode;
+    private final Map<String, IndexMode> indexNameWithModes;
+    private final List<Attribute> attrs;
 
-    public EsRelation(Source source, EsIndex index, IndexMode indexMode, boolean frozen) {
-        this(source, index, flatten(source, index.mapping()), indexMode, frozen);
-    }
-
-    public EsRelation(Source source, EsIndex index, List<Attribute> attributes, IndexMode indexMode) {
-        this(source, index, attributes, indexMode, false);
+    public EsRelation(Source source, EsIndex index, IndexMode indexMode) {
+        this(source, index.name(), indexMode, index.indexNameWithModes(), flatten(source, index.mapping()));
     }
 
-    public EsRelation(Source source, EsIndex index, List<Attribute> attributes, IndexMode indexMode, boolean frozen) {
+    public EsRelation(
+        Source source,
+        String indexPattern,
+        IndexMode indexMode,
+        Map<String, IndexMode> indexNameWithModes,
+        List<Attribute> attributes
+    ) {
         super(source);
-        this.index = index;
-        this.attrs = attributes;
+        this.indexPattern = indexPattern;
         this.indexMode = indexMode;
-        this.frozen = frozen;
+        this.indexNameWithModes = indexNameWithModes;
+        this.attrs = attributes;
     }
 
     private static EsRelation readFrom(StreamInput in) throws IOException {
         Source source = Source.readFrom((PlanStreamInput) in);
-        EsIndex esIndex = EsIndex.readFrom(in);
+        String indexPattern;
+        Map<String, IndexMode> indexNameWithModes;
+        if (in.getTransportVersion().onOrAfter(ESQL_SKIP_ES_INDEX_SERIALIZATION)) {
+            indexPattern = in.readString();
+            indexNameWithModes = in.readMap(IndexMode::readFrom);
+        } else {
+            var index = EsIndex.readFrom(in);
+            indexPattern = index.name();
+            indexNameWithModes = index.indexNameWithModes();
+        }
         List<Attribute> attributes = in.readNamedWriteableCollectionAsList(Attribute.class);
         if (supportingEsSourceOptions(in.getTransportVersion())) {
             // We don't do anything with these strings
@@ -67,23 +81,32 @@ private static EsRelation readFrom(StreamInput in) throws IOException {
             in.readOptionalString();
         }
         IndexMode indexMode = readIndexMode(in);
-        boolean frozen = in.readBoolean();
-        return new EsRelation(source, esIndex, attributes, indexMode, frozen);
+        if (in.getTransportVersion().before(ESQL_SKIP_ES_INDEX_SERIALIZATION)) {
+            in.readBoolean();
+        }
+        return new EsRelation(source, indexPattern, indexMode, indexNameWithModes, attributes);
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         Source.EMPTY.writeTo(out);
-        index().writeTo(out);
-        out.writeNamedWriteableCollection(output());
+        if (out.getTransportVersion().onOrAfter(ESQL_SKIP_ES_INDEX_SERIALIZATION)) {
+            out.writeString(indexPattern);
+            out.writeMap(indexNameWithModes, (o, v) -> IndexMode.writeTo(v, out));
+        } else {
+            new EsIndex(indexPattern, Map.of(), indexNameWithModes).writeTo(out);
+        }
+        out.writeNamedWriteableCollection(attrs);
         if (supportingEsSourceOptions(out.getTransportVersion())) {
             // write (null) string fillers expected by remote
             out.writeOptionalString(null);
             out.writeOptionalString(null);
             out.writeOptionalString(null);
         }
-        writeIndexMode(out, indexMode());
-        out.writeBoolean(frozen());
+        writeIndexMode(out, indexMode);
+        if (out.getTransportVersion().before(ESQL_SKIP_ES_INDEX_SERIALIZATION)) {
+            out.writeBoolean(false);
+        }
     }
 
     private static boolean supportingEsSourceOptions(TransportVersion version) {
@@ -97,7 +120,7 @@ public String getWriteableName() {
 
     @Override
     protected NodeInfo<EsRelation> info() {
-        return NodeInfo.create(this, EsRelation::new, index, attrs, indexMode, frozen);
+        return NodeInfo.create(this, EsRelation::new, indexPattern, indexMode, indexNameWithModes, attrs);
     }
 
     private static List<Attribute> flatten(Source source, Map<String, EsField> mapping) {
@@ -128,23 +151,27 @@ private static List<Attribute> flatten(Source source, Map<String, EsField> mappi
         return list;
     }
 
-    public EsIndex index() {
-        return index;
-    }
-
-    public boolean frozen() {
-        return frozen;
+    public String indexPattern() {
+        return indexPattern;
     }
 
     public IndexMode indexMode() {
         return indexMode;
     }
 
+    public Map<String, IndexMode> indexNameWithModes() {
+        return indexNameWithModes;
+    }
+
     @Override
     public List<Attribute> output() {
         return attrs;
     }
 
+    public Set<String> concreteIndices() {
+        return indexNameWithModes.keySet();
+    }
+
     @Override
     public String commandName() {
         return "FROM";
@@ -159,7 +186,7 @@ public boolean expressionsResolved() {
 
     @Override
     public int hashCode() {
-        return Objects.hash(index, indexMode, frozen, attrs);
+        return Objects.hash(indexPattern, indexMode, indexNameWithModes, attrs);
     }
 
     @Override
@@ -173,17 +200,17 @@ public boolean equals(Object obj) {
         }
 
         EsRelation other = (EsRelation) obj;
-        return Objects.equals(index, other.index)
-            && indexMode == other.indexMode()
-            && frozen == other.frozen
+        return Objects.equals(indexPattern, other.indexPattern)
+            && Objects.equals(indexMode, other.indexMode)
+            && Objects.equals(indexNameWithModes, other.indexNameWithModes)
             && Objects.equals(attrs, other.attrs);
     }
 
     @Override
     public String nodeString() {
         return nodeName()
             + "["
-            + index
+            + indexPattern
             + "]"
             + (indexMode != IndexMode.STANDARD ? "[" + indexMode.name() + "]" : "")
             + NodeUtils.limitedToString(attrs);