Verify that the field is aggregatable before attempting cardinality aggregation (#53874)

Author: przemekwitek

x-pack/plugin/ml/qa/native-multi-node-tests/src/test/java/org/elasticsearch/xpack/ml/integration/ClassificationIT.java

@@ -56,6 +56,7 @@ public class ClassificationIT extends MlNativeDataFrameAnalyticsIntegTestCase {
     private static final String BOOLEAN_FIELD = "boolean-field";
     private static final String NUMERICAL_FIELD = "numerical-field";
     private static final String DISCRETE_NUMERICAL_FIELD = "discrete-numerical-field";
+    private static final String TEXT_FIELD = "text-field";
     private static final String KEYWORD_FIELD = "keyword-field";
     private static final String NESTED_FIELD = "outer-field.inner-field";
     private static final String ALIAS_TO_KEYWORD_FIELD = "alias-to-keyword-field";
@@ -260,6 +261,14 @@ public void testWithOnlyTrainingRowsAndTrainingPercentIsFifty_DependentVariableI
         assertThat(e.getMessage(), startsWith("invalid types [double] for required field [numerical-field];"));
     }
 
+    public void testWithOnlyTrainingRowsAndTrainingPercentIsFifty_DependentVariableIsText() throws Exception {
+        ElasticsearchStatusException e = expectThrows(
+            ElasticsearchStatusException.class,
+            () -> testWithOnlyTrainingRowsAndTrainingPercentIsFifty(
+                "classification_training_percent_is_50_text", TEXT_FIELD, KEYWORD_FIELD_VALUES, null));
+        assertThat(e.getMessage(), startsWith("field [text-field] of type [text] is non-aggregatable"));
+    }
+
     public void testWithOnlyTrainingRowsAndTrainingPercentIsFifty_DependentVariableIsBoolean() throws Exception {
         testWithOnlyTrainingRowsAndTrainingPercentIsFifty(
             "classification_training_percent_is_50_boolean", BOOLEAN_FIELD, BOOLEAN_FIELD_VALUES, "boolean");
@@ -517,6 +526,7 @@ private static void createIndex(String index) {
                 BOOLEAN_FIELD, "type=boolean",
                 NUMERICAL_FIELD, "type=double",
                 DISCRETE_NUMERICAL_FIELD, "type=integer",
+                TEXT_FIELD, "type=text",
                 KEYWORD_FIELD, "type=keyword",
                 NESTED_FIELD, "type=keyword",
                 ALIAS_TO_KEYWORD_FIELD, "type=alias,path=" + KEYWORD_FIELD,
@@ -532,6 +542,7 @@ private static void indexData(String sourceIndex, int numTrainingRows, int numNo
                 BOOLEAN_FIELD, BOOLEAN_FIELD_VALUES.get(i % BOOLEAN_FIELD_VALUES.size()),
                 NUMERICAL_FIELD, NUMERICAL_FIELD_VALUES.get(i % NUMERICAL_FIELD_VALUES.size()),
                 DISCRETE_NUMERICAL_FIELD, DISCRETE_NUMERICAL_FIELD_VALUES.get(i % DISCRETE_NUMERICAL_FIELD_VALUES.size()),
+                TEXT_FIELD, KEYWORD_FIELD_VALUES.get(i % KEYWORD_FIELD_VALUES.size()),
                 KEYWORD_FIELD, KEYWORD_FIELD_VALUES.get(i % KEYWORD_FIELD_VALUES.size()),
                 NESTED_FIELD, KEYWORD_FIELD_VALUES.get(i % KEYWORD_FIELD_VALUES.size()));
             IndexRequest indexRequest = new IndexRequest(sourceIndex).source(source.toArray());
@@ -549,6 +560,9 @@ private static void indexData(String sourceIndex, int numTrainingRows, int numNo
                 source.addAll(
                     List.of(DISCRETE_NUMERICAL_FIELD, DISCRETE_NUMERICAL_FIELD_VALUES.get(i % DISCRETE_NUMERICAL_FIELD_VALUES.size())));
             }
+            if (TEXT_FIELD.equals(dependentVariable) == false) {
+                source.addAll(List.of(TEXT_FIELD, KEYWORD_FIELD_VALUES.get(i % KEYWORD_FIELD_VALUES.size())));
+            }
             if (KEYWORD_FIELD.equals(dependentVariable) == false) {
                 source.addAll(List.of(KEYWORD_FIELD, KEYWORD_FIELD_VALUES.get(i % KEYWORD_FIELD_VALUES.size())));
             }

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/dataframe/extractor/ExtractedFieldsDetectorFactory.java

@@ -10,6 +10,7 @@
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.admin.indices.settings.get.GetSettingsRequest;
 import org.elasticsearch.action.admin.indices.settings.get.GetSettingsResponse;
+import org.elasticsearch.action.fieldcaps.FieldCapabilities;
 import org.elasticsearch.action.fieldcaps.FieldCapabilitiesAction;
 import org.elasticsearch.action.fieldcaps.FieldCapabilitiesRequest;
 import org.elasticsearch.action.fieldcaps.FieldCapabilitiesResponse;
@@ -78,7 +79,7 @@ private void create(String[] index, DataFrameAnalyticsConfig config, ActionListe
         ActionListener<FieldCapabilitiesResponse> fieldCapabilitiesHandler = ActionListener.wrap(
             fieldCapabilitiesResponse -> {
                 fieldCapsResponseHolder.set(fieldCapabilitiesResponse);
-                getCardinalitiesForFieldsWithConstraints(index, config, fieldCardinalitiesHandler);
+                getCardinalitiesForFieldsWithConstraints(index, config, fieldCapabilitiesResponse, fieldCardinalitiesHandler);
             },
             listener::onFailure
         );
@@ -96,7 +97,9 @@ private void create(String[] index, DataFrameAnalyticsConfig config, ActionListe
         getDocValueFieldsLimit(index, docValueFieldsLimitListener);
     }
 
-    private void getCardinalitiesForFieldsWithConstraints(String[] index, DataFrameAnalyticsConfig config,
+    private void getCardinalitiesForFieldsWithConstraints(String[] index,
+                                                          DataFrameAnalyticsConfig config,
+                                                          FieldCapabilitiesResponse fieldCapabilitiesResponse,
                                                           ActionListener<Map<String, Long>> listener) {
         List<FieldCardinalityConstraint> fieldCardinalityConstraints = config.getAnalysis().getFieldCardinalityConstraints();
         if (fieldCardinalityConstraints.isEmpty()) {
@@ -111,6 +114,12 @@ private void getCardinalitiesForFieldsWithConstraints(String[] index, DataFrameA
 
         SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder().size(0).query(config.getSource().getParsedQuery());
         for (FieldCardinalityConstraint constraint : fieldCardinalityConstraints) {
+            for (FieldCapabilities fieldCaps : fieldCapabilitiesResponse.getField(constraint.getField()).values()) {
+                if (fieldCaps.isAggregatable() == false) {
+                    throw ExceptionsHelper.badRequestException("field [{}] of type [{}] is non-aggregatable",
+                        fieldCaps.getName(), fieldCaps.getType());
+                }
+            }
             searchSourceBuilder.aggregation(
                 AggregationBuilders.cardinality(constraint.getField())
                     .field(constraint.getField())