[ES|QL] Verify aggregation filter's type is boolean to avoid class_cast_exception (#116274) (#116360)

* validate agg filter's type is boolean

(cherry picked from commit 0e044d7)

# Conflicts:
#	x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/analysis/Verifier.java

Author: fang-xing-esql

docs/changelog/116274.yaml

@@ -0,0 +1,5 @@
+pr: 116274
+summary: "[ES|QL] Verify aggregation filter's type is boolean to avoid `class_cast_exception`"
+area: ES|QL
+type: bug
+issues: []

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

@@ -69,6 +69,7 @@
 import static org.elasticsearch.xpack.esql.common.Failure.fail;
 import static org.elasticsearch.xpack.esql.core.expression.TypeResolutions.ParamOrdinal.FIRST;
 import static org.elasticsearch.xpack.esql.core.type.DataType.BOOLEAN;
+import static org.elasticsearch.xpack.esql.core.type.DataType.NULL;
 import static org.elasticsearch.xpack.esql.optimizer.rules.physical.local.PushFiltersToSource.canPushToSource;
 
 /**
@@ -317,6 +318,10 @@ private static void checkInvalidNamedExpressionUsage(
                 Expression filter = fe.filter();
                 failures.add(fail(filter, "WHERE clause allowed only for aggregate functions, none found in [{}]", fe.sourceText()));
             }
+            Expression f = fe.filter(); // check the filter has to be a boolean term, similar as checkFilterConditionType
+            if (f.dataType() != NULL && f.dataType() != BOOLEAN) {
+                failures.add(fail(f, "Condition expression needs to be boolean, found [{}]", f.dataType()));
+            }
             // but that the filter doesn't use grouping or aggregate functions
             fe.filter().forEachDown(c -> {
                 if (c instanceof AggregateFunction af) {

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

@@ -394,6 +394,19 @@ public void testAggFilterOnBucketingOrAggFunctions() {
         assertEquals("1:60: Unknown column [m]", error("from test | stats m = max(languages), min(languages) WHERE m + 2 > 1 by emp_no"));
     }
 
+    public void testAggWithNonBooleanFilter() {
+        for (String filter : List.of("\"true\"", "1", "1 + 0", "concat(\"a\", \"b\")")) {
+            String type = (filter.equals("1") || filter.equals("1 + 0")) ? "INTEGER" : "KEYWORD";
+            assertEquals("1:19: Condition expression needs to be boolean, found [" + type + "]", error("from test | where " + filter));
+            for (String by : List.of("", " by languages", " by bucket(salary, 10)")) {
+                assertEquals(
+                    "1:34: Condition expression needs to be boolean, found [" + type + "]",
+                    error("from test | stats count(*) where " + filter + by)
+                );
+            }
+        }
+    }
+
     public void testGroupingInsideAggsAsAgg() {
         assertEquals(
             "1:18: can only use grouping function [bucket(emp_no, 5.)] part of the BY clause",