ES|QL: Configurable score, key and group by columns for FUSE (#135079)

Author: ioanatia

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

@@ -204,7 +204,7 @@ book_no:keyword | title:text                                            | author
 
 reranker after FUSE
 required_capability: fork_v9
-required_capability: fuse_v3
+required_capability: fuse_v4
 required_capability: match_operator_colon
 required_capability: rerank
 

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

@@ -28,7 +28,7 @@ protected Collection<Class<? extends Plugin>> nodePlugins() {
 
     @Before
     public void setupIndex() {
-        assumeTrue("requires FUSE capability", EsqlCapabilities.Cap.FUSE_V3.isEnabled());
+        assumeTrue("requires FUSE capability", EsqlCapabilities.Cap.FUSE_V4.isEnabled());
         createAndPopulateIndex();
     }
 

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/FuseIT.java

@@ -31,7 +31,7 @@ protected Collection<Class<? extends Plugin>> nodePlugins() {
 
     @Before
     public void setupIndex() {
-        assumeTrue("requires FUSE capability", EsqlCapabilities.Cap.FUSE_V3.isEnabled());
+        assumeTrue("requires FUSE capability", EsqlCapabilities.Cap.FUSE_V4.isEnabled());
         var indexName = "test";
         var client = client().admin().indices();
         var CreateRequest = client.prepareCreate(indexName)

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/FuseWithInvalidLicenseIT.java

@@ -338,7 +338,14 @@ insistCommand
     ;
 
 fuseCommand
-    : DEV_FUSE (fuseType=identifier)? fuseOptions=commandNamedParameters
+    : DEV_FUSE (fuseType=identifier)? (fuseConfiguration)*
+    ;
+
+fuseConfiguration
+    : SCORE BY score=qualifiedName
+    | KEY BY key=fields
+    | GROUP BY group=qualifiedName
+    | WITH options=mapExpression
     ;
 
 setCommand

x-pack/plugin/esql/src/main/antlr/EsqlBaseLexer.tokens

@@ -14,8 +14,13 @@ FUSE_PIPE : PIPE -> type(PIPE), popMode;
 // explicit popMode of RP to allow FUSE in FORK branches
 FUSE_RP : RP -> type(RP), popMode, popMode;
 
+GROUP: 'group';
+SCORE: 'score';
+KEY : 'key';
+
 FUSE_WITH: WITH -> type(WITH), popMode, pushMode(EXPRESSION_MODE);
 FUSE_COMMA: COMMA -> type(COMMA);
+FUSE_BY: BY -> type(BY);
 FUSE_QUOTED_IDENTIFIER: QUOTED_IDENTIFIER -> type(QUOTED_IDENTIFIER);
 FUSE_UNQUOTED_IDENTIFIER: UNQUOTED_IDENTIFIER -> type(UNQUOTED_IDENTIFIER);
 FUSE_LINE_COMMENT: LINE_COMMENT -> channel(HIDDEN);

x-pack/plugin/esql/src/main/antlr/EsqlBaseParser.g4

@@ -1357,7 +1357,7 @@ public enum Cap {
         /**
          * FUSE command
          */
-        FUSE_V3(Build.current().isSnapshot()),
+        FUSE_V4(Build.current().isSnapshot()),
 
         /**
          * Support improved behavior for LIKE operator when used with index fields.

x-pack/plugin/esql/src/main/antlr/EsqlBaseParser.tokens

@@ -1026,14 +1026,19 @@ private LogicalPlan resolveFuse(Fuse fuse, List<Attribute> childrenOutput) {
                 discriminator = maybeResolveAttribute((UnresolvedAttribute) discriminator, childrenOutput);
             }
 
-            List<NamedExpression> groupings = fuse.groupings()
+            List<NamedExpression> keys = fuse.keys()
                 .stream()
                 .map(attr -> attr instanceof UnresolvedAttribute ? maybeResolveAttribute((UnresolvedAttribute) attr, childrenOutput) : attr)
                 .toList();
 
-            // some attributes were unresolved - we return Fuse here so that the Verifier can raise an error message
-            if (score instanceof UnresolvedAttribute || discriminator instanceof UnresolvedAttribute) {
-                return new Fuse(fuse.source(), fuse.child(), score, discriminator, groupings, fuse.fuseType(), fuse.options());
+            // some attributes were unresolved or the wrong type
+            // we return Fuse here so that the Verifier can raise an error message
+            if (score instanceof UnresolvedAttribute
+                || (score.resolved() && score.dataType() != DOUBLE)
+                || discriminator instanceof UnresolvedAttribute
+                || (discriminator.resolved() && DataType.isString(discriminator.dataType()) == false)
+                || keys.stream().allMatch(attr -> attr.resolved() && DataType.isString(attr.dataType())) == false) {
+                return new Fuse(fuse.source(), fuse.child(), score, discriminator, keys, fuse.fuseType(), fuse.options());
             }
 
             LogicalPlan scoreEval = new FuseScoreEval(source, fuse.child(), score, discriminator, fuse.fuseType(), fuse.options());
@@ -1051,7 +1056,7 @@ private LogicalPlan resolveFuse(Fuse fuse, List<Attribute> childrenOutput) {
                 aggregates.add(new Alias(source, attr.name(), new Values(source, attr, aggFilter)));
             }
 
-            return resolveAggregate(new Aggregate(source, scoreEval, new ArrayList<>(groupings), aggregates), childrenOutput);
+            return resolveAggregate(new Aggregate(source, scoreEval, new ArrayList<>(keys), aggregates), childrenOutput);
         }
 
         private Attribute maybeResolveAttribute(UnresolvedAttribute ua, List<Attribute> childrenOutput) {