I was hitting this exception with `from * | limit 10` with the security rally track.

```
java.lang.IllegalArgumentException: Unknown writeable [org.elasticsearch.xpack.ql.type.EsField][InvalidMappedField]
	at org.elasticsearch.xpack.esql.io.stream.PlanNameRegistry.throwOnUnknownWritable(PlanNameRegistry.java:252) ~[?:?]
	at org.elasticsearch.xpack.esql.io.stream.PlanNameRegistry.getWriter(PlanNameRegistry.java:217) ~[?:?]
	at org.elasticsearch.xpack.esql.io.stream.PlanNameRegistry.getWriter(PlanNameRegistry.java:232) ~[?:?]
	at org.elasticsearch.xpack.esql.io.stream.PlanStreamOutput.writeNamed(PlanStreamOutput.java:76) ~[?:?]
	at org.elasticsearch.xpack.esql.io.stream.PlanNamedTypes.writeFieldAttribute(PlanNamedTypes.java:793) ~[?:?]
	at org.elasticsearch.xpack.esql.io.stream.PlanStreamOutput.writeNamed(PlanStreamOutput.java:78) ~[?:?]
	at org.elasticsearch.xpack.esql.io.stream.PlanStreamOutput.writeAttribute(PlanStreamOutput.java:61) ~[?:?]
	at org.elasticsearch.xpack.esql.io.stream.PlanNameRegistry$PlanWriter.write(PlanNameRegistry.java:46) ~[?:?]
	at org.elasticsearch.common.io.stream.StreamOutput.writeCollection(StreamOutput.java:1060) ~[elasticsearch-8.11.0-SNAPSHOT.jar:?]
	at org.elasticsearch.xpack.esql.io.stream.PlanNamedTypes.writeAttributes(PlanNamedTypes.java:764) ~[?:?]
	at org.elasticsearch.xpack.esql.io.stream.PlanNamedTypes.writeExchangeSinkExec(PlanNamedTypes.java:485) ~[?:?]
	at org.elasticsearch.xpack.esql.io.stream.PlanStreamOutput.writeNamed(PlanStreamOutput.java:78) ~[?:?]
	at org.elasticsearch.xpack.esql.io.stream.PlanStreamOutput.writePhysicalPlanNode(PlanStreamOutput.java:49) ~[?:?]
	at org.elasticsearch.xpack.esql.plugin.DataNodeRequest.writeTo(DataNodeRequest.java:72) ~[?:?]
```
x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/io/stream/PlanNamedTypes.java
out.writeExpression(bucket.buckets());
        out.writeExpression(bucket.from());
        out.writeExpression(bucket.to());
    }
    static final Map<String, TriFunction<Source, Expression, List<Expression>, ScalarFunction>> VARARG_CTORS = Map.ofEntries(
        entry(name(Case.class), Case::new),
        entry(name(Coalesce.class), Coalesce::new),
        entry(name(Concat.class), Concat::new),
        entry(name(Greatest.class), Greatest::new),
        entry(name(Least.class), Least::new)
    );
    static ScalarFunction readVarag(PlanStreamInput in, String name) throws IOException {
        return VARARG_CTORS.get(name)
            .apply(Source.EMPTY, in.readExpression(), in.readList(readerFromPlanReader(PlanStreamInput::readExpression)));
    }
    static void writeVararg(PlanStreamOutput out, ScalarFunction vararg) throws IOException {
        out.writeExpression(vararg.children().get(0));
        out.writeCollection(
            vararg.children().subList(1, vararg.children().size()),
            writerFromPlanWriter(PlanStreamOutput::writeExpression)
        );
    }
    static CountDistinct readCountDistinct(PlanStreamInput in) throws IOException {
        return new CountDistinct(Source.EMPTY, in.readExpression(), in.readOptionalNamed(Expression.class));
    }
    static void writeCountDistinct(PlanStreamOutput out, CountDistinct countDistinct) throws IOException {
        List<Expression> fields = countDistinct.children();
        assert fields.size() == 1 || fields.size() == 2;
        out.writeExpression(fields.get(0));
        out.writeOptionalWriteable(fields.size() == 2 ? o -> out.writeExpression(fields.get(1)) : null);