ES|QL: Check for unsupported fields in FUSE (elastic#135530)

Author: ioanatia

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

@@ -1054,7 +1054,13 @@ private LogicalPlan resolveFuse(Fuse fuse, List<Attribute> childrenOutput) {
                 if (attr.name().equals(score.name())) {
                     continue;
                 }
-                aggregates.add(new Alias(source, attr.name(), new Values(source, attr, aggFilter)));
+                var valuesAgg = new Values(source, attr, aggFilter);
+                // Use VALUES only on supported fields.
+                // FuseScoreEval will check that the input contains only columns with supported data types
+                // and will fail with an appropriate error message if it doesn't.
+                if (valuesAgg.resolved()) {
+                    aggregates.add(new Alias(source, attr.name(), valuesAgg));
+                }
             }
 
             return resolveAggregate(new Aggregate(source, scoreEval, new ArrayList<>(keys), aggregates), childrenOutput);

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/fuse/FuseScoreEval.java

@@ -27,6 +27,7 @@
 import org.elasticsearch.xpack.esql.core.tree.Source;
 import org.elasticsearch.xpack.esql.core.type.DataType;
 import org.elasticsearch.xpack.esql.core.util.Holder;
+import org.elasticsearch.xpack.esql.expression.function.aggregate.Values;
 import org.elasticsearch.xpack.esql.plan.logical.ExecutesOn;
 import org.elasticsearch.xpack.esql.plan.logical.LogicalPlan;
 import org.elasticsearch.xpack.esql.plan.logical.PipelineBreaker;
@@ -130,7 +131,8 @@ public boolean licenseCheck(XPackLicenseState state) {
 
     @Override
     public void postAnalysisVerification(Failures failures) {
-        validateLimitedInput(failures);
+        validateInput(failures);
+        validatePipelineBreakerBeforeFuse(failures);
         if (options == null) {
             return;
         }
@@ -141,7 +143,27 @@ public void postAnalysisVerification(Failures failures) {
         }
     }
 
-    private void validateLimitedInput(Failures failures) {
+    private void validateInput(Failures failures) {
+        // Since we use STATS BY to merge rows together, we need to make sure that all columns can be used in STATS BY.
+        // When the input of FUSE contains unsupported columns, we don't want to fail with a STATS BY validation error,
+        // but with an error specific to FUSE.
+        Expression aggFilter = new Literal(source(), true, DataType.BOOLEAN);
+
+        for (Attribute attr : child().output()) {
+            var valuesAgg = new Values(source(), attr, aggFilter);
+
+            if (valuesAgg.resolved() == false) {
+                failures.add(
+                    new Failure(
+                        this,
+                        "cannot use [" + attr.name() + "] as an input of FUSE. Consider using [DROP " + attr.name() + "] before FUSE."
+                    )
+                );
+            }
+        }
+    }
+
+    private void validatePipelineBreakerBeforeFuse(Failures failures) {
         var myself = this;
         Holder<Boolean> hasLimitedInput = new Holder<>(false);
         this.forEachUp(LogicalPlan.class, plan -> {

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/analysis/VerifierTests.java

@@ -303,6 +303,13 @@ public void testUnsupportedAndMultiTypedFields() {
                 )
             );
         }
+
+        if (EsqlCapabilities.Cap.FUSE_V6.isEnabled()) {
+            assertEquals(
+                "1:76: cannot use [double] as an input of FUSE. Consider using [DROP double] before FUSE.",
+                error("from test* METADATA _id, _index, _score | FORK (where true) (where true) | FUSE", analyzer)
+            );
+        }
     }
 
     public void testRoundFunctionInvalidInputs() {