Fix attribute serialization behavior

Author: alex-spies

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/Attribute.java

@@ -6,14 +6,19 @@
  */
 package org.elasticsearch.xpack.esql.core.expression;
 
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.xpack.esql.core.tree.Source;
 import org.elasticsearch.xpack.esql.core.type.DataType;
+import org.elasticsearch.xpack.esql.core.util.PlanStreamInput;
+import org.elasticsearch.xpack.esql.core.util.PlanStreamOutput;
 
+import java.io.IOException;
 import java.util.List;
 import java.util.Objects;
 
 import static java.util.Collections.emptyList;
+import static org.elasticsearch.TransportVersions.ESQL_QUALIFIERS_IN_ATTRIBUTES;
 
 /**
  * {@link Expression}s that can be materialized and describe properties of the derived table.
@@ -197,4 +202,22 @@ public static boolean dataTypeEquals(List<Attribute> left, List<Attribute> right
      * @return true if the attribute represents a TSDB dimension type
      */
     public abstract boolean isDimension();
+
+    protected void checkAndSerializeQualifier(PlanStreamOutput out, TransportVersion version) throws IOException {
+        if (version.onOrAfter(ESQL_QUALIFIERS_IN_ATTRIBUTES)) {
+            out.writeOptionalCachedString(qualifier());
+        } else if (qualifier() != null) {
+            // Non-null qualifier means the query specifically defined one. Old nodes don't know what to do with it and just writing
+            // null would lose information and lead to undefined, likely invalid queries.
+            // IllegalArgumentException returns a 400 to the user, which is what we want here.
+            throw new IllegalArgumentException("Trying to serialize an Attribute with a qualifier to an old node");
+        }
+    }
+
+    protected static String readQualifier(PlanStreamInput in, TransportVersion version) throws IOException {
+        if (version.onOrAfter(ESQL_QUALIFIERS_IN_ATTRIBUTES)) {
+            return in.readOptionalCachedString();
+        }
+        return null;
+    }
 }

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/FieldAttribute.java

@@ -22,7 +22,6 @@
 import java.util.Objects;
 
 import static org.elasticsearch.TransportVersions.ESQL_FIELD_ATTRIBUTE_DROP_TYPE;
-import static org.elasticsearch.TransportVersions.ESQL_QUALIFIERS_IN_ATTRIBUTES;
 import static org.elasticsearch.xpack.esql.core.util.PlanStreamInput.readCachedStringWithVersionCheck;
 import static org.elasticsearch.xpack.esql.core.util.PlanStreamOutput.writeCachedStringWithVersionCheck;
 
@@ -105,10 +104,7 @@ private static FieldAttribute innerReadFrom(StreamInput in) throws IOException {
          */
         Source source = Source.readFrom((StreamInput & PlanStreamInput) in);
         String parentName = ((PlanStreamInput) in).readOptionalCachedString();
-        String qualifier = null;
-        if (in.getTransportVersion().onOrAfter(ESQL_QUALIFIERS_IN_ATTRIBUTES)) {
-            qualifier = ((PlanStreamInput) in).readOptionalCachedString();
-        }
+        String qualifier = readQualifier((PlanStreamInput) in, in.getTransportVersion());
         String name = readCachedStringWithVersionCheck(in);
         if (in.getTransportVersion().before(ESQL_FIELD_ATTRIBUTE_DROP_TYPE)) {
             DataType.readFrom(in);
@@ -128,16 +124,14 @@ public void writeTo(StreamOutput out) throws IOException {
         if (((PlanStreamOutput) out).writeAttributeCacheHeader(this)) {
             Source.EMPTY.writeTo(out);
             ((PlanStreamOutput) out).writeOptionalCachedString(parentName);
-            if (out.getTransportVersion().onOrAfter(ESQL_QUALIFIERS_IN_ATTRIBUTES)) {
-                ((PlanStreamOutput) out).writeOptionalCachedString(qualifier());
-            }
+            checkAndSerializeQualifier((PlanStreamOutput) out, out.getTransportVersion());
             writeCachedStringWithVersionCheck(out, name());
             if (out.getTransportVersion().before(ESQL_FIELD_ATTRIBUTE_DROP_TYPE)) {
                 dataType().writeTo(out);
             }
             field.writeTo(out);
             if (out.getTransportVersion().before(ESQL_FIELD_ATTRIBUTE_DROP_TYPE)) {
-                // We used to write the qualifier here. We can still do if needed in the future.
+                // We used to write the qualifier here, even though it was always null.
                 out.writeOptionalString(null);
             }
             out.writeEnum(nullable());

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/ReferenceAttribute.java

@@ -18,6 +18,8 @@
 
 import java.io.IOException;
 
+import static org.elasticsearch.TransportVersions.ESQL_QUALIFIERS_IN_ATTRIBUTES;
+
 /**
  * Attribute based on a reference to an expression.
  */
@@ -58,7 +60,11 @@ public void writeTo(StreamOutput out) throws IOException {
             Source.EMPTY.writeTo(out);
             out.writeString(name());
             dataType().writeTo(out);
-            out.writeOptionalString(qualifier());
+            checkAndSerializeQualifier((PlanStreamOutput) out, out.getTransportVersion());
+            if (out.getTransportVersion().before(ESQL_QUALIFIERS_IN_ATTRIBUTES)) {
+                // We used to always serialize a null qualifier here, so do the same for bwc.
+                out.writeOptionalString(null);
+            }
             out.writeEnum(nullable());
             id().writeTo(out);
             out.writeBoolean(synthetic());
@@ -82,8 +88,10 @@ private static ReferenceAttribute innerReadFrom(StreamInput in) throws IOExcepti
         // We could cache this if we wanted to.
         String name = in.readString();
         DataType dataType = DataType.readFrom(in);
-        // Also cacheable
-        String qualifier = in.readOptionalString();
+        String qualifier = readQualifier((PlanStreamInput) in, in.getTransportVersion());
+        if (in.getTransportVersion().before(ESQL_QUALIFIERS_IN_ATTRIBUTES)) {
+            in.readOptionalString();
+        }
         Nullability nullability = in.readEnum(Nullability.class);
         NameId id = NameId.readFrom((StreamInput & PlanStreamInput) in);
         boolean synthetic = in.readBoolean();

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/UnsupportedAttribute.java

@@ -88,20 +88,24 @@ public UnsupportedAttribute(
         this.message = customMessage == null ? errorMessage(name(), field) : customMessage;
     }
 
-    private UnsupportedAttribute(StreamInput in) throws IOException {
-        this(
-            Source.readFrom((PlanStreamInput) in),
-            readCachedStringWithVersionCheck(in),
-            in.getTransportVersion().onOrAfter(TransportVersions.V_8_15_2) ? EsField.readFrom(in) : new UnsupportedEsField(in),
-            in.readOptionalString(),
-            NameId.readFrom((PlanStreamInput) in)
-        );
+    private static UnsupportedAttribute innerReadFrom(StreamInput in) throws IOException {
+        Source source = Source.readFrom((PlanStreamInput) in);
+        String qualifier = readQualifier((PlanStreamInput) in, in.getTransportVersion());
+        String name = readCachedStringWithVersionCheck(in);
+        UnsupportedEsField field = in.getTransportVersion().onOrAfter(TransportVersions.V_8_15_2)
+            ? EsField.readFrom(in)
+            : new UnsupportedEsField(in);
+        String message = in.readOptionalString();
+        NameId id = NameId.readFrom((PlanStreamInput) in);
+
+        return new UnsupportedAttribute(source, qualifier, name, field, message, id);
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         if (((PlanStreamOutput) out).writeAttributeCacheHeader(this)) {
             Source.EMPTY.writeTo(out);
+            checkAndSerializeQualifier((PlanStreamOutput) out, out.getTransportVersion());
             writeCachedStringWithVersionCheck(out, name());
             if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_15_2)) {
                 field().writeTo(out);
@@ -114,7 +118,7 @@ public void writeTo(StreamOutput out) throws IOException {
     }
 
     public static UnsupportedAttribute readFrom(StreamInput in) throws IOException {
-        return ((PlanStreamInput) in).readAttributeWithCache(UnsupportedAttribute::new);
+        return ((PlanStreamInput) in).readAttributeWithCache(UnsupportedAttribute::innerReadFrom);
     }
 
     @Override

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/parser/ExpressionBuilder.java

@@ -941,7 +941,7 @@ private Alias visitRerankField(EsqlBaseParser.RerankFieldContext ctx, Source sou
 
         var boolExprCtx = ctx.booleanExpression();
         Expression value = boolExprCtx == null ? id : expression(boolExprCtx);
-        return new Alias(source, source(ctx.qualifiedName()).text(), value);
+        return new Alias(source, id.qualifier() != null ? id.qualifiedName() : id.name(), value);
     }
 
     @Override

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/UnsupportedAttributeTests.java

@@ -19,25 +19,28 @@ protected UnsupportedAttribute create() {
     }
 
     public static UnsupportedAttribute randomUnsupportedAttribute() {
+        String qualifier = randomBoolean() ? null : randomAlphaOfLength(3);
         String name = randomAlphaOfLength(5);
         UnsupportedEsField field = UnsupportedEsFieldTests.randomUnsupportedEsField(4);
         String customMessage = randomBoolean() ? null : randomAlphaOfLength(9);
         NameId id = new NameId();
-        return new UnsupportedAttribute(Source.EMPTY, name, field, customMessage, id);
+        return new UnsupportedAttribute(Source.EMPTY, qualifier, name, field, customMessage, id);
     }
 
     @Override
     protected UnsupportedAttribute mutate(UnsupportedAttribute instance) {
         Source source = instance.source();
+        String qualifier = instance.qualifier();
         String name = instance.name();
         UnsupportedEsField field = instance.field();
         String customMessage = instance.hasCustomMessage() ? instance.unresolvedMessage() : null;
-        switch (between(0, 2)) {
-            case 0 -> name = randomAlphaOfLength(name.length() + 1);
-            case 1 -> field = randomValueOtherThan(field, () -> UnsupportedEsFieldTests.randomUnsupportedEsField(4));
-            case 2 -> customMessage = randomValueOtherThan(customMessage, () -> randomBoolean() ? null : randomAlphaOfLength(9));
+        switch (between(0, 3)) {
+            case 0 -> qualifier = randomAlphaOfLength(qualifier == null ? 3 : qualifier.length() + 1);
+            case 1 -> name = randomAlphaOfLength(name.length() + 1);
+            case 2 -> field = randomValueOtherThan(field, () -> UnsupportedEsFieldTests.randomUnsupportedEsField(4));
+            case 3 -> customMessage = randomValueOtherThan(customMessage, () -> randomBoolean() ? null : randomAlphaOfLength(9));
             default -> throw new IllegalArgumentException();
         }
-        return new UnsupportedAttribute(source, name, field, customMessage, new NameId());
+        return new UnsupportedAttribute(source, qualifier, name, field, customMessage, new NameId());
     }
 }