EsSourceExec

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().name(), plan.indexMode(), plan.index().indexNameWithModes(), attributes, plan.query());
+        return new EsQueryExec(plan.source(), plan.indexName(), plan.indexMode(), plan.indexNameWithModes(), attributes, plan.query());
     }
 }

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

@@ -22,45 +22,64 @@
 
 import java.io.IOException;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 
+import static org.elasticsearch.TransportVersions.ESQ_SKIP_ES_INDEX_SERIALIZATION;
+
 public class EsSourceExec extends LeafExec {
     public static final NamedWriteableRegistry.Entry ENTRY = new NamedWriteableRegistry.Entry(
         PhysicalPlan.class,
         "EsSourceExec",
         EsSourceExec::readFrom
     );
 
-    private final EsIndex index;
-    private final List<Attribute> attributes;
-    private final QueryBuilder query;
+    private final String indexName;
     private final IndexMode indexMode;
+    private final Map<String, IndexMode> indexNameWithModes;
+    private final QueryBuilder query;
+    private final List<Attribute> attributes;
 
     public EsSourceExec(EsRelation relation) {
-        this(relation.source(), relation.index(), relation.output(), null, relation.indexMode());
+        this(relation.source(), relation.index().name(), relation.indexMode(), relation.index().indexNameWithModes(), null, relation.output());
     }
 
-    public EsSourceExec(Source source, EsIndex index, List<Attribute> attributes, QueryBuilder query, IndexMode indexMode) {
+    public EsSourceExec(Source source, String indexName, IndexMode indexMode, Map<String, IndexMode> indexNameWithModes, QueryBuilder query, List<Attribute> attributes) {
         super(source);
-        this.index = index;
-        this.attributes = attributes;
-        this.query = query;
+        this.indexName = indexName;
         this.indexMode = indexMode;
+        this.indexNameWithModes = indexNameWithModes;
+        this.query = query;
+        this.attributes = attributes;
     }
 
     private static EsSourceExec readFrom(StreamInput in) throws IOException {
         var source = Source.readFrom((PlanStreamInput) in);
-        var index = new EsIndex(in);
+        String indexName;
+        Map<String, IndexMode> indexNameWithModes;
+        if (in.getTransportVersion().onOrAfter(ESQ_SKIP_ES_INDEX_SERIALIZATION)) {
+            indexName = in.readString();
+            indexNameWithModes = in.readMap(IndexMode::readFrom);
+        } else {
+            var index = new EsIndex(in);
+            indexName = index.name();
+            indexNameWithModes = index.indexNameWithModes();
+        }
         var attributes = in.readNamedWriteableCollectionAsList(Attribute.class);
         var query = in.readOptionalNamedWriteable(QueryBuilder.class);
         var indexMode = EsRelation.readIndexMode(in);
-        return new EsSourceExec(source, index, attributes, query, indexMode);
+        return new EsSourceExec(source, indexName, indexMode, indexNameWithModes, query, attributes);
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         Source.EMPTY.writeTo(out);
-        index().writeTo(out);
+        if (out.getTransportVersion().onOrAfter(ESQ_SKIP_ES_INDEX_SERIALIZATION)) {
+            out.writeString(indexName);
+            out.writeMap(indexNameWithModes, (o, v) -> IndexMode.writeTo(v, out));
+        } else {
+            new EsIndex(indexName, Map.of(), indexNameWithModes).writeTo(out);
+        }
         out.writeNamedWriteableCollection(output());
         out.writeOptionalNamedWriteable(query());
         EsRelation.writeIndexMode(out, indexMode());
@@ -71,31 +90,35 @@ public String getWriteableName() {
         return ENTRY.name;
     }
 
-    public EsIndex index() {
-        return index;
-    }
-
-    public QueryBuilder query() {
-        return query;
+    public String indexName() {
+        return indexName;
     }
 
     public IndexMode indexMode() {
         return indexMode;
     }
 
+    public Map<String, IndexMode> indexNameWithModes() {
+        return indexNameWithModes;
+    }
+
+    public QueryBuilder query() {
+        return query;
+    }
+
     @Override
     public List<Attribute> output() {
         return attributes;
     }
 
     @Override
     protected NodeInfo<? extends PhysicalPlan> info() {
-        return NodeInfo.create(this, EsSourceExec::new, index, attributes, query, indexMode);
+        return NodeInfo.create(this, EsSourceExec::new, indexName, indexMode, indexNameWithModes, query, attributes);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(index, attributes, query, indexMode);
+        return Objects.hash(indexName, indexMode, indexNameWithModes, query, attributes);
     }
 
     @Override
@@ -109,14 +132,15 @@ public boolean equals(Object obj) {
         }
 
         EsSourceExec other = (EsSourceExec) obj;
-        return Objects.equals(index, other.index)
-            && Objects.equals(attributes, other.attributes)
+        return Objects.equals(indexName, other.indexName)
+            && Objects.equals(indexMode, other.indexMode)
+            && Objects.equals(indexNameWithModes, other.indexNameWithModes)
             && Objects.equals(query, other.query)
-            && Objects.equals(indexMode, other.indexMode);
+            && Objects.equals(attributes, other.attributes);
     }
 
     @Override
     public String nodeString() {
-        return nodeName() + "[" + index + "]" + NodeUtils.limitedToString(attributes);
+        return nodeName() + "[" + indexName + "]" + NodeUtils.limitedToString(attributes);
     }
 }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/PlannerUtils.java

@@ -186,7 +186,7 @@ public static PhysicalPlan localPlan(
             if (filter != null) {
                 physicalFragment = physicalFragment.transformUp(
                     EsSourceExec.class,
-                    query -> new EsSourceExec(Source.EMPTY, query.index(), query.output(), filter, query.indexMode())
+                    query -> new EsSourceExec(Source.EMPTY, query.indexName(), query.indexMode(), query.indexNameWithModes(), filter, query.output())
                 );
             }
             var localOptimized = physicalOptimizer.localOptimize(physicalFragment);

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/PhysicalPlanOptimizerTests.java

@@ -2576,10 +2576,9 @@ public void testFieldExtractWithoutSourceAttributes() {
             | where round(emp_no) > 10
             """));
         // Transform the verified plan so that it is invalid (i.e. no source attributes)
-        List<Attribute> emptyAttrList = List.of();
         var badPlan = verifiedPlan.transformDown(
             EsQueryExec.class,
-            node -> new EsSourceExec(node.source(), null /*TODO 112998*/, emptyAttrList, node.query(), IndexMode.STANDARD)
+            node -> new EsSourceExec(node.source(), node.indexName(), IndexMode.STANDARD, node.indexNameWithModes(), node.query(), List.of())
         );
 
         var e = expectThrows(VerificationException.class, () -> physicalPlanOptimizer.verify(badPlan));

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plan/physical/EsSourceExecSerializationTests.java

@@ -10,24 +10,26 @@
 import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.index.query.TermQueryBuilder;
+import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xpack.esql.core.expression.Attribute;
-import org.elasticsearch.xpack.esql.core.tree.Source;
-import org.elasticsearch.xpack.esql.index.EsIndex;
 import org.elasticsearch.xpack.esql.index.EsIndexSerializationTests;
 
 import java.io.IOException;
 import java.util.List;
+import java.util.Map;
 
-import static org.elasticsearch.xpack.esql.plan.logical.AbstractLogicalPlanSerializationTests.randomFieldAttributes;
+import static org.elasticsearch.xpack.esql.index.EsIndexSerializationTests.randomIndexNameWithModes;
 
 public class EsSourceExecSerializationTests extends AbstractPhysicalPlanSerializationTests<EsSourceExec> {
     public static EsSourceExec randomEsSourceExec() {
-        Source source = randomSource();
-        EsIndex index = EsIndexSerializationTests.randomEsIndex();
-        List<Attribute> attributes = randomFieldAttributes(1, 10, false);
-        QueryBuilder query = new TermQueryBuilder(randomAlphaOfLength(5), randomAlphaOfLength(5));
-        IndexMode indexMode = randomFrom(IndexMode.values());
-        return new EsSourceExec(source, index, attributes, query, indexMode);
+        return new EsSourceExec(
+            randomSource(),
+            randomIdentifier(),
+            randomFrom(IndexMode.values()),
+            randomIndexNameWithModes(),
+            new TermQueryBuilder(randomAlphaOfLength(5), randomAlphaOfLength(5)),
+            randomFieldAttributes(1, 10, false)
+        );
     }
 
     @Override
@@ -37,18 +39,20 @@ protected EsSourceExec createTestInstance() {
 
     @Override
     protected EsSourceExec mutateInstance(EsSourceExec instance) throws IOException {
-        EsIndex index = instance.index();
-        List<Attribute> attributes = instance.output();
-        QueryBuilder query = instance.query();
+        String indexName = instance.indexName();
         IndexMode indexMode = instance.indexMode();
-        switch (between(0, 3)) {
-            case 0 -> index = randomValueOtherThan(index, EsIndexSerializationTests::randomEsIndex);
-            case 1 -> attributes = randomValueOtherThan(attributes, () -> randomFieldAttributes(1, 10, false));
-            case 2 -> query = randomValueOtherThan(query, () -> new TermQueryBuilder(randomAlphaOfLength(5), randomAlphaOfLength(5)));
-            case 3 -> indexMode = randomValueOtherThan(indexMode, () -> randomFrom(IndexMode.values()));
+        Map<String, IndexMode> indexNameWithModes = instance.indexNameWithModes();
+        QueryBuilder query = instance.query();
+        List<Attribute> attributes = instance.output();
+        switch (between(0, 4)) {
+            case 0 -> indexName = randomValueOtherThan(indexName, ESTestCase::randomIdentifier);
+            case 1 -> indexMode = randomValueOtherThan(indexMode, () -> randomFrom(IndexMode.values()));
+            case 2 -> indexNameWithModes = randomValueOtherThan(indexNameWithModes, EsIndexSerializationTests::randomIndexNameWithModes);
+            case 3 -> query = randomValueOtherThan(query, () -> new TermQueryBuilder(randomAlphaOfLength(5), randomAlphaOfLength(5)));
+            case 4 -> attributes = randomValueOtherThan(attributes, () -> randomFieldAttributes(1, 10, false));
             default -> throw new IllegalStateException();
         }
-        return new EsSourceExec(instance.source(), index, attributes, query, indexMode);
+        return new EsSourceExec(instance.source(), indexName, indexMode, indexNameWithModes, query, attributes);
     }
 
     @Override