ES|QL: Implement serialization of InvalidMappedField (#98972)

Fixes https://github.com/elastic/elasticsearch/issues/98851

As from the subject, implement the serialization/deserialization of
InvalidMappedField objects. These objects are instantiated in the
execution plan when a field mapping cannot be defined, eg. if two fields
with the same name and different types appear in two indexes for the
same index pattern.

docs/changelog/98972.yaml

@@ -0,0 +1,6 @@
+pr: 98972
+summary: "ES|QL: Implement serialization of `InvalidMappedField`"
+area: ES|QL
+type: bug
+issues:
+ - 98851

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/io/stream/PlanNamedTypes.java

@@ -168,6 +168,7 @@ import org.elasticsearch.xpack.ql.plan.logical.Project;
 import org.elasticsearch.xpack.ql.tree.Source;
 import org.elasticsearch.xpack.ql.type.DateEsField;
 import org.elasticsearch.xpack.ql.type.EsField;
+import org.elasticsearch.xpack.ql.type.InvalidMappedField;
 import org.elasticsearch.xpack.ql.type.KeywordEsField;
 import org.elasticsearch.xpack.ql.type.TextEsField;
 import org.elasticsearch.xpack.ql.type.UnsupportedEsField;
@@ -266,6 +267,7 @@ public final class PlanNamedTypes {
             // EsFields
             of(EsField.class, EsField.class, PlanNamedTypes::writeEsField, PlanNamedTypes::readEsField),
             of(EsField.class, DateEsField.class, PlanNamedTypes::writeDateEsField, PlanNamedTypes::readDateEsField),
+            of(EsField.class, InvalidMappedField.class, PlanNamedTypes::writeInvalidMappedField, PlanNamedTypes::readInvalidMappedField),
             of(EsField.class, KeywordEsField.class, PlanNamedTypes::writeKeywordEsField, PlanNamedTypes::readKeywordEsField),
             of(EsField.class, TextEsField.class, PlanNamedTypes::writeTextEsField, PlanNamedTypes::readTextEsField),
             of(EsField.class, UnsupportedEsField.class, PlanNamedTypes::writeUnsupportedEsField, PlanNamedTypes::readUnsupportedEsField),
@@ -904,6 +906,15 @@ public final class PlanNamedTypes {
         out.writeBoolean(dateEsField.isAggregatable());
     }
 
+    static InvalidMappedField readInvalidMappedField(PlanStreamInput in) throws IOException {
+        return new InvalidMappedField(in.readString(), in.readString());
+    }
+
+    static void writeInvalidMappedField(PlanStreamOutput out, InvalidMappedField field) throws IOException {
+        out.writeString(field.getName());
+        out.writeString(field.errorMessage());
+    }
+
     static KeywordEsField readKeywordEsField(PlanStreamInput in) throws IOException {
         return new KeywordEsField(
             in.readString(),

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

@@ -82,6 +82,7 @@ import org.elasticsearch.xpack.ql.type.DataType;
 import org.elasticsearch.xpack.ql.type.DataTypes;
 import org.elasticsearch.xpack.ql.type.DateEsField;
 import org.elasticsearch.xpack.ql.type.EsField;
+import org.elasticsearch.xpack.ql.type.InvalidMappedField;
 import org.elasticsearch.xpack.ql.type.KeywordEsField;
 import org.elasticsearch.xpack.ql.type.TextEsField;
 import org.elasticsearch.xpack.ql.type.UnsupportedEsField;
@@ -246,6 +247,19 @@ public class PlanNamedTypesTests extends ESTestCase {
         Stream.generate(PlanNamedTypesTests::randomTextEsField).limit(100).forEach(PlanNamedTypesTests::assertNamedEsField);
     }
 
+    public void testInvalidMappedFieldSimple() throws IOException {
+        var orig = new InvalidMappedField("foo", "bar");
+        BytesStreamOutput bso = new BytesStreamOutput();
+        PlanStreamOutput out = new PlanStreamOutput(bso, planNameRegistry);
+        PlanNamedTypes.writeInvalidMappedField(out, orig);
+        var deser = PlanNamedTypes.readInvalidMappedField(planStreamInput(bso));
+        EqualsHashCodeTestUtils.checkEqualsAndHashCode(orig, unused -> deser);
+    }
+
+    public void testInvalidMappedField() {
+        Stream.generate(PlanNamedTypesTests::randomInvalidMappedField).limit(100).forEach(PlanNamedTypesTests::assertNamedEsField);
+    }
+
     public void testEsDateFieldSimple() throws IOException {
         var orig = DateEsField.dateEsField("birth_date", Map.of(), false);
         BytesStreamOutput bso = new BytesStreamOutput();
@@ -454,6 +468,13 @@ public class PlanNamedTypesTests extends ESTestCase {
         );
     }
 
+    static InvalidMappedField randomInvalidMappedField() {
+        return new InvalidMappedField(
+            randomAlphaOfLength(randomIntBetween(1, 25)), // name
+            randomAlphaOfLength(randomIntBetween(1, 25)) // error message
+        );
+    }
+
     static BinaryComparison randomBinaryComparison() {
         int v = randomIntBetween(0, 6);
         var left = field(randomName(), randomDataType());