ES|QL: Fix KQL/QSTR with unmapped fields in NULLIFY mode

docs/changelog/143399.yaml

@@ -0,0 +1,7 @@
+area: ES|QL
+issues:
+ - 142968
+ - 142959
+pr: 143399
+summary: Fix KQL/QSTR with unmapped fields in NULLIFY mode
+type: bug

server/src/main/resources/transport/definitions/referable/esql_missing_es_field.csv

@@ -0,0 +1 @@
+9305000

server/src/main/resources/transport/upper_bounds/9.4.csv

@@ -1 +1 @@
-inference_azure_openai_task_settings_headers,9304000
+esql_missing_es_field,9305000

x-pack/plugin/esql/qa/testFixtures/src/main/resources/unmapped-nullify.csv-spec

@@ -482,3 +482,47 @@ ROW x = 1
 x:integer      |does_not_exist:null |y:keyword | language_name:keyword
 1              |null                |null      |null
 ;
+
+kqlWithUnmappedField
+required_capability: optional_fields_nullify_tech_preview
+required_capability: kql_function
+required_capability: fix_unmapped_fields_in_esrelation
+
+// Reproducer for https://github.com/elastic/elasticsearch/issues/142968
+// KQL function should work when unmapped fields are referenced later in the query
+SET unmapped_fields="nullify"\;
+FROM books
+| WHERE KQL("author: Faulkner")
+| EVAL x = does_not_exist_field + 1
+| KEEP book_no, author, does_not_exist_field, x
+| SORT book_no
+| LIMIT 3
+;
+
+book_no:keyword | author:text                                        | does_not_exist_field:null | x:integer
+2378            | [Carol Faulkner, Holly Byers Ochoa, Lucretia Mott] | null                      | null
+2713            | William Faulkner                                   | null                      | null
+2847            | Colleen Faulkner                                   | null                      | null
+;
+
+qstrWithUnmappedField
+required_capability: optional_fields_nullify_tech_preview
+required_capability: qstr_function
+required_capability: fix_unmapped_fields_in_esrelation
+
+// Reproducer for https://github.com/elastic/elasticsearch/issues/142959
+// QSTR function should work when unmapped fields are referenced later in the query
+SET unmapped_fields="nullify"\;
+FROM books
+| WHERE QSTR("author: Faulkner")
+| EVAL x = does_not_exist_field + 1
+| KEEP book_no, author, does_not_exist_field, x
+| SORT book_no
+| LIMIT 3
+;
+
+book_no:keyword | author:text                                        | does_not_exist_field:null | x:integer
+2378            | [Carol Faulkner, Holly Byers Ochoa, Lucretia Mott] | null                      | null
+2713            | William Faulkner                                   | null                      | null
+2847            | Colleen Faulkner                                   | null                      | null
+;

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java

@@ -2263,6 +2263,13 @@ public enum Cap {
          */
         FIX_AGG_FIRST_LAST_FOLDABLES_IN_SORT_FIELD,
 
+        /**
+         * Fix for KQL/QSTR functions failing when used with unmapped fields in NULLIFY mode.
+         * Unmapped fields are now added directly to EsRelation output with NULL type instead of using Eval nodes.
+         * https://github.com/elastic/elasticsearch/issues/142968
+         */
+        FIX_UNMAPPED_FIELDS_IN_ESRELATION,
+
         // Last capability should still have a comma for fewer merge conflicts when adding new ones :)
         // This comment prevents the semicolon from being on the previous capability when Spotless formats the file.
         ;

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

@@ -24,6 +24,9 @@
 import org.elasticsearch.xpack.esql.core.expression.UnresolvedAttribute;
 import org.elasticsearch.xpack.esql.core.expression.UnresolvedPattern;
 import org.elasticsearch.xpack.esql.core.expression.UnresolvedTimestamp;
+import org.elasticsearch.xpack.esql.core.type.DataType;
+import org.elasticsearch.xpack.esql.core.type.EsField;
+import org.elasticsearch.xpack.esql.core.type.MissingEsField;
 import org.elasticsearch.xpack.esql.core.type.PotentiallyUnmappedKeywordEsField;
 import org.elasticsearch.xpack.esql.core.util.Holder;
 import org.elasticsearch.xpack.esql.plan.logical.EsRelation;
@@ -97,19 +100,53 @@ private static LogicalPlan resolve(LogicalPlan plan, boolean load) {
     }
 
     /**
-     * The method introduces {@code EVAL missing_field = NULL}-equivalent into the plan, on top of the source, for every attribute in
-     * {@code unresolved}. It also "patches" the introduced attributes through the plan, where needed (like through Fork/UntionAll).
+     * This method introduces null-typed field attributes for every attribute in {@code unresolved}, within the {@link EsRelation}s
+     * in the plan. The fields are added with {@link DataType#NULL} type, which causes {@code ReplaceFieldWithConstantOrNull}
+     * (in the local logical optimizer) to replace them with {@code Literal.NULL}.
+     * <p>
+     * For non-EsRelation sources (Row, LocalRelation), it falls back to inserting Eval nodes with null assignments.
+     * <p>
+     * It also "patches" the introduced attributes through the plan, where needed (like through Fork/UnionAll).
      */
     private static LogicalPlan nullify(LogicalPlan plan, Set<UnresolvedAttribute> unresolved) {
-        // insert an Eval on top of every LeafPlan, if there's a UnaryPlan atop it
-        var transformed = plan.transformUp(
-            n -> n instanceof UnaryPlan unary && unary.child() instanceof LeafPlan,
+        // For EsRelation sources: add null-typed fields to the relation's output
+        var transformed = plan.transformUp(EsRelation.class, esr -> {
+            if (esr.indexMode() == IndexMode.LOOKUP) {
+                return esr;
+            }
+            List<FieldAttribute> fieldsToNullify = fieldsToNullify(unresolved, Expressions.names(esr.output()));
+            return fieldsToNullify.isEmpty() ? esr : esr.withAttributes(combine(esr.output(), fieldsToNullify));
+        });
+
+        // For non-EsRelation sources (Row, LocalRelation): insert Eval nodes with null assignments
+        // This handles cases like: ROW x = 1 | EVAL y = unmapped_field
+        transformed = transformed.transformUp(
+            n -> n instanceof UnaryPlan unary && unary.child() instanceof LeafPlan leaf && (leaf instanceof EsRelation == false),
             p -> evalUnresolvedAtopUnary((UnaryPlan) p, nullAliases(unresolved))
         );
-        // insert an Eval on top of those LeafPlan that are children of n-ary plans (could happen with UnionAll)
         return transformed.transformUp(
             n -> n instanceof UnaryPlan == false && n instanceof LeafPlan == false,
-            nAry -> evalUnresolvedAtopNary(nAry, nullAliases(unresolved))
+            nAry -> evalUnresolvedAtopNaryNonEsRelation(nAry, nullAliases(unresolved))
+        );
+    }
+
+    private static List<FieldAttribute> fieldsToNullify(Set<UnresolvedAttribute> unresolved, List<String> exclude) {
+        List<FieldAttribute> nullified = new ArrayList<>(unresolved.size());
+        for (var ua : unresolved) {
+            if (exclude.contains(ua.name()) == false) {
+                nullified.add(nullifyField(ua));
+            }
+        }
+        return nullified;
+    }
+
+    private static FieldAttribute nullifyField(Attribute attribute) {
+        return new FieldAttribute(
+            attribute.source(),
+            null,
+            attribute.qualifier(),
+            attribute.name(),
+            new MissingEsField(attribute.name(), DataType.NULL, Map.of(), false, EsField.TimeSeriesFieldType.NONE)
         );
     }
 
@@ -222,6 +259,24 @@ private static LogicalPlan evalUnresolvedAtopNary(LogicalPlan nAry, List<Alias>
         return changed ? nAry.replaceChildren(newChildren) : nAry;
     }
 
+    /**
+     * Inserts an Eval atop each child of the given {@code nAry}, if the child is a non-EsRelation LeafPlan.
+     * EsRelation sources are handled separately by adding fields to their output.
+     */
+    private static LogicalPlan evalUnresolvedAtopNaryNonEsRelation(LogicalPlan nAry, List<Alias> nullAliases) {
+        List<LogicalPlan> newChildren = new ArrayList<>(nAry.children().size());
+        boolean changed = false;
+        for (var child : nAry.children()) {
+            if (child instanceof LeafPlan source && (source instanceof EsRelation == false)) {
+                assertSourceType(source);
+                child = new Eval(source.source(), source, nullAliases);
+                changed = true;
+            }
+            newChildren.add(child);
+        }
+        return changed ? nAry.replaceChildren(newChildren) : nAry;
+    }
+
     /**
      * Inserts an Eval atop the given {@code unaryAtopSource}, if this isn't an Eval already. Otherwise it merges the nullAliases into it.
      */

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/core/type/DateEsField.java

@@ -6,6 +6,7 @@
  */
 package org.elasticsearch.xpack.esql.core.type;
 
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.xpack.esql.io.stream.PlanStreamInput;
@@ -51,7 +52,7 @@ public void writeContent(StreamOutput out) throws IOException {
         writeTimeSeriesFieldType(out);
     }
 
-    public String getWriteableName() {
+    public String getWriteableName(TransportVersion transportVersion) {
         return "DateEsField";
     }
 

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/core/type/EsField.java

@@ -105,6 +105,7 @@ public static TimeSeriesFieldType fromIndexFieldCapabilities(IndexFieldCapabilit
         Map.entry("EsField", EsField::new),
         Map.entry("InvalidMappedField", InvalidMappedField::new),
         Map.entry("KeywordEsField", KeywordEsField::new),
+        Map.entry("MissingEsField", MissingEsField::new),
         Map.entry("MultiTypeEsField", MultiTypeEsField::new),
         Map.entry("PotentiallyUnmappedKeywordEsField", PotentiallyUnmappedKeywordEsField::new),
         Map.entry("TextEsField", TextEsField::new),
@@ -167,7 +168,7 @@ public static <A extends EsField> A readFrom(StreamInput in) throws IOException
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
-        if (((PlanStreamOutput) out).writeEsFieldCacheHeader(this)) {
+        if (((PlanStreamOutput) out).writeEsFieldCacheHeader(this, out.getTransportVersion())) {
             writeContent(out);
         }
     }
@@ -201,7 +202,7 @@ protected static TimeSeriesFieldType readTimeSeriesFieldType(StreamInput in) thr
     /**
      * This needs to be overridden by subclasses for specific serialization
      */
-    public String getWriteableName() {
+    public String getWriteableName(TransportVersion transportVersion) {
         return "EsField";
     }
 

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/core/type/InvalidMappedField.java

@@ -7,6 +7,7 @@
 
 package org.elasticsearch.xpack.esql.core.type;
 
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.xpack.esql.core.QlIllegalArgumentException;
@@ -81,7 +82,7 @@ public void writeContent(StreamOutput out) throws IOException {
         writeTimeSeriesFieldType(out);
     }
 
-    public String getWriteableName() {
+    public String getWriteableName(TransportVersion transportVersion) {
         return "InvalidMappedField";
     }
 

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/core/type/KeywordEsField.java

@@ -6,6 +6,7 @@
  */
 package org.elasticsearch.xpack.esql.core.type;
 
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.xpack.esql.io.stream.PlanStreamInput;
@@ -76,7 +77,7 @@ public void writeContent(StreamOutput out) throws IOException {
         writeTimeSeriesFieldType(out);
     }
 
-    public String getWriteableName() {
+    public String getWriteableName(TransportVersion transportVersion) {
         return "KeywordEsField";
     }
 

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/core/type/MissingEsField.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.esql.core.type;
+
+import org.elasticsearch.TransportVersion;
+import org.elasticsearch.common.io.stream.StreamInput;
+
+import java.io.IOException;
+import java.util.Map;
+
+public class MissingEsField extends EsField {
+
+    private static final TransportVersion ESQL_MISSING_ES_FIELD = TransportVersion.fromName("esql_missing_es_field");
+
+    public MissingEsField(
+        String name,
+        DataType esDataType,
+        Map<String, EsField> properties,
+        boolean aggregatable,
+        TimeSeriesFieldType timeSeriesFieldType
+    ) {
+        super(name, esDataType, properties, aggregatable, timeSeriesFieldType);
+    }
+
+    public MissingEsField(
+        String name,
+        DataType esDataType,
+        Map<String, EsField> properties,
+        boolean aggregatable,
+        boolean isAlias,
+        TimeSeriesFieldType timeSeriesFieldType
+    ) {
+        super(name, esDataType, properties, aggregatable, isAlias, timeSeriesFieldType);
+    }
+
+    public MissingEsField(StreamInput in) throws IOException {
+        super(in);
+    }
+
+    public String getWriteableName(TransportVersion transportVersion) {
+        if (transportVersion.supports(ESQL_MISSING_ES_FIELD)) {
+            return "MissingEsField";
+        }
+        // This was introduced because of https://github.com/elastic/elasticsearch/issues/142968
+        // Things should still work fine with nodes that don't have MissingEsField,
+        // so for BWC we just fall back to EsField, which is the closest thing to MissingEsField that older nodes will understand.
+        return "EsField";
+    }
+}

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/core/type/MultiTypeEsField.java

@@ -7,6 +7,7 @@
 
 package org.elasticsearch.xpack.esql.core.type;
 
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.xpack.esql.core.expression.Expression;
@@ -62,7 +63,7 @@ public void writeContent(StreamOutput out) throws IOException {
         writeTimeSeriesFieldType(out);
     }
 
-    public String getWriteableName() {
+    public String getWriteableName(TransportVersion transportVersion) {
         return "MultiTypeEsField";
     }
 

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/core/type/PotentiallyUnmappedKeywordEsField.java

@@ -6,6 +6,7 @@
  */
 package org.elasticsearch.xpack.esql.core.type;
 
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.common.io.stream.StreamInput;
 
 import java.io.IOException;
@@ -25,7 +26,7 @@ public PotentiallyUnmappedKeywordEsField(StreamInput in) throws IOException {
         super(in);
     }
 
-    public String getWriteableName() {
+    public String getWriteableName(TransportVersion transportVersion) {
         return "PotentiallyUnmappedKeywordEsField";
     }
 }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/core/type/TextEsField.java

@@ -6,6 +6,7 @@
  */
 package org.elasticsearch.xpack.esql.core.type;
 
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.core.Tuple;
@@ -54,7 +55,7 @@ public void writeContent(StreamOutput out) throws IOException {
         writeTimeSeriesFieldType(out);
     }
 
-    public String getWriteableName() {
+    public String getWriteableName(TransportVersion transportVersion) {
         return "TextEsField";
     }
 

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/core/type/UnsupportedEsField.java

@@ -80,7 +80,7 @@ public void writeContent(StreamOutput out) throws IOException {
         writeTimeSeriesFieldType(out);
     }
 
-    public String getWriteableName() {
+    public String getWriteableName(TransportVersion transportVersion) {
         return "UnsupportedEsField";
     }
 

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

@@ -193,9 +193,11 @@ private int cacheAttribute(Attribute attr) {
      * Writes a cache header for an {@link org.elasticsearch.xpack.esql.core.type.EsField} and caches it if it is not already in the cache.
      * In that case, the field will have to serialize itself into this stream immediately after this method call.
      * @param field The EsField to serialize
+     * @param transportVersion The transport version to use for serialization,
+     *                         needed to get the correct field name in case of versioned fields.
      * @return true if the attribute needs to serialize itself, false otherwise (ie. if already cached)
      */
-    public boolean writeEsFieldCacheHeader(EsField field) throws IOException {
+    public boolean writeEsFieldCacheHeader(EsField field, TransportVersion transportVersion) throws IOException {
         Integer cacheId = esFieldIdFromCache(field);
         if (cacheId != null) {
             writeZLong(cacheId);
@@ -204,7 +206,7 @@ public boolean writeEsFieldCacheHeader(EsField field) throws IOException {
 
         cacheId = cacheEsField(field);
         writeZLong(-1 - cacheId);
-        writeCachedString(field.getWriteableName());
+        writeCachedString(field.getWriteableName(transportVersion));
         return true;
     }