Forbid usage of _score aggregations on STATS when there is a WHERE clause

Author: carlosdelest

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

@@ -551,9 +551,10 @@ required_capability: match_function
 required_capability: full_text_functions_in_stats_where
 
 from books metadata _score 
-| stats avg_score = avg(_score) where match(title, "Lord Rings", {"operator": "AND"})
+| where match(title, "Lord Rings", {"operator": "AND"})
+| stats avg_score = avg(_score), max_score = max(_score), min_score = min(_score)
 ;
 
-avg_score:double
-3.869828939437866 
+avg_score:double   | max_score:double   | min_score:double
+3.869828939437866  | 5.123856544494629  | 3.0124807357788086
 ;

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

@@ -13,11 +13,13 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.xpack.esql.VerificationException;
 import org.elasticsearch.xpack.esql.action.AbstractEsqlIntegTestCase;
+import org.hamcrest.Matchers;
 import org.junit.Before;
 
 import java.util.List;
 
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+import static org.elasticsearch.xpack.esql.EsqlTestUtils.getValuesList;
 import static org.hamcrest.CoreMatchers.containsString;
 
 //@TestLogging(value = "org.elasticsearch.xpack.esql:TRACE,org.elasticsearch.compute:TRACE", reason = "debug")
@@ -265,6 +267,21 @@ public void testMatchWithStats() {
             assertColumnTypes(resp.columns(), List.of("long", "long"));
             assertValues(resp.values(), List.of(List.of(2L, 4L)));
         }
+
+        query = """
+            FROM test METADATA _score
+            | WHERE match(content, "fox")
+            | STATS m = max(_score), n = min(_score)
+            """;
+
+        try (var resp = run(query)) {
+            assertColumnNames(resp.columns(), List.of("m", "n"));
+            assertColumnTypes(resp.columns(), List.of("double", "double"));
+            List<List<Object>> valuesList = getValuesList(resp.values());
+            assertEquals(1, valuesList.size());
+            assertThat((double) valuesList.get(0).get(0), Matchers.greaterThan(1.0));
+            assertThat((double) valuesList.get(0).get(1), Matchers.greaterThan(0.0));
+        }
     }
 
     public void testMatchWithinEval() {

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

@@ -22,13 +22,16 @@
 import org.elasticsearch.xpack.esql.core.expression.Expression;
 import org.elasticsearch.xpack.esql.core.expression.Expressions;
 import org.elasticsearch.xpack.esql.core.expression.FieldAttribute;
+import org.elasticsearch.xpack.esql.core.expression.MetadataAttribute;
 import org.elasticsearch.xpack.esql.core.expression.NamedExpression;
 import org.elasticsearch.xpack.esql.core.tree.NodeInfo;
 import org.elasticsearch.xpack.esql.core.tree.Source;
+import org.elasticsearch.xpack.esql.core.util.Holder;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.AggregateFunction;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.FilteredExpression;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.Rate;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.TimeSeriesAggregateFunction;
+import org.elasticsearch.xpack.esql.expression.function.fulltext.FullTextFunction;
 import org.elasticsearch.xpack.esql.expression.function.grouping.Categorize;
 import org.elasticsearch.xpack.esql.expression.function.grouping.GroupingFunction;
 import org.elasticsearch.xpack.esql.io.stream.PlanStreamInput;
@@ -229,7 +232,22 @@ public void postAnalysisVerification(Failures failures) {
             );
         }
         checkCategorizeGrouping(failures);
+        checkMultipleScoreAggregations(failures);
+    }
 
+    private void checkMultipleScoreAggregations(Failures failures) {
+        Holder<Boolean> hasScoringAggs = new Holder<>();
+        forEachExpression(FilteredExpression.class, fe -> {
+            if (fe.delegate() instanceof AggregateFunction aggregateFunction) {
+                if (aggregateFunction.field() instanceof MetadataAttribute metadataAttribute) {
+                    if (MetadataAttribute.SCORE.equals(metadataAttribute.name())) {
+                        if (fe.filter().anyMatch(e -> e instanceof FullTextFunction)) {
+                            failures.add(fail(fe, "cannot use _score aggregations with a WHERE filter in a STATS command"));
+                        }
+                    }
+                }
+            }
+        });
     }
 
     /**

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

@@ -2493,6 +2493,27 @@ public void testInsistNotOnTopOfFrom() {
         );
     }
 
+    public void testFullTextFunctionsInStats() {
+        checkFullTextFunctionsInStats("match(last_name, \"Smith\")");
+        checkFullTextFunctionsInStats("multi_match(\"Smith\", first_name, last_name)");
+        checkFullTextFunctionsInStats("last_name : \"Smith\"");
+        checkFullTextFunctionsInStats("qstr(\"last_name: Smith\")");
+        checkFullTextFunctionsInStats("kql(\"last_name: Smith\")");
+    }
+
+    private void checkFullTextFunctionsInStats(String functionInvocation) {
+
+        query("from test | stats c = max(salary) where " + functionInvocation);
+        query("from test | stats c = max(salary) where " + functionInvocation + " or length(first_name) > 10");
+        query("from test metadata _score |  where " + functionInvocation + " | stats c = max(_score)");
+        query("from test metadata _score |  where " + functionInvocation + " or length(first_name) > 10 | stats c = max(_score)");
+
+        assertThat(
+            error("from test metadata _score | stats c = max(_score) where " + functionInvocation),
+            containsString("cannot use _score aggregations with a WHERE filter in a STATS command")
+        );
+    }
+
     private void query(String query) {
         query(query, defaultAnalyzer);
     }