[FEATURE][ML] Skip rows that have missing data (#36067)

This also prepares for allowing the `DataFrameDataExtractor`
to be reused while joining the results with the raw documents.

Author: dimitris-athanasiou

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/analytics/DataFrameDataExtractor.java

@@ -15,6 +15,7 @@
 import org.elasticsearch.action.search.SearchScrollAction;
 import org.elasticsearch.action.search.SearchScrollRequestBuilder;
 import org.elasticsearch.client.Client;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.search.sort.SortOrder;
@@ -68,18 +69,18 @@ public void cancel() {
         isCancelled = true;
     }
 
-    public Optional<List<String[]>> next() throws IOException {
+    public Optional<List<Row>> next() throws IOException {
         if (!hasNext()) {
             throw new NoSuchElementException();
         }
-        Optional<List<String[]>> records = scrollId == null ? Optional.ofNullable(initScroll()) : Optional.ofNullable(continueScroll());
-        if (!records.isPresent()) {
+        Optional<List<Row>> hits = scrollId == null ? Optional.ofNullable(initScroll()) : Optional.ofNullable(continueScroll());
+        if (!hits.isPresent()) {
             hasNext = false;
         }
-        return records;
+        return hits;
     }
 
-    protected List<String[]> initScroll() throws IOException {
+    protected List<Row> initScroll() throws IOException {
         LOGGER.debug("[{}] Initializing scroll", "analytics");
         SearchResponse searchResponse = executeSearchRequest(buildSearchRequest());
         LOGGER.debug("[{}] Search response was obtained", context.jobId);
@@ -106,7 +107,7 @@ private SearchRequestBuilder buildSearchRequest() {
         return searchRequestBuilder;
     }
 
-    private List<String[]> processSearchResponse(SearchResponse searchResponse) throws IOException {
+    private List<Row> processSearchResponse(SearchResponse searchResponse) throws IOException {
 
         if (searchResponse.getFailedShards() > 0 && searchHasShardFailure == false) {
             LOGGER.debug("[{}] Resetting scroll search after shard failure", context.jobId);
@@ -123,29 +124,35 @@ private List<String[]> processSearchResponse(SearchResponse searchResponse) thro
         }
 
         SearchHit[] hits = searchResponse.getHits().getHits();
-        List<String[]> records = new ArrayList<>(hits.length);
+        List<Row> rows = new ArrayList<>(hits.length);
         for (SearchHit hit : hits) {
             if (isCancelled) {
                 hasNext = false;
                 clearScroll(scrollId);
                 break;
             }
-            records.add(toStringArray(hit));
+            rows.add(createRow(hit));
         }
-        return records;
+        return rows;
+
     }
 
-    private String[] toStringArray(SearchHit hit) {
-        String[] result = new String[context.extractedFields.getAllFields().size()];
-        for (int i = 0; i < result.length; ++i) {
+    private Row createRow(SearchHit hit) {
+        String[] extractedValues = new String[context.extractedFields.getAllFields().size()];
+        for (int i = 0; i < extractedValues.length; ++i) {
             ExtractedField field = context.extractedFields.getAllFields().get(i);
             Object[] values = field.value(hit);
-            result[i] = (values.length == 1) ? Objects.toString(values[0]) : "";
+            if (values.length == 1 && values[0] instanceof Number) {
+                extractedValues[i] = Objects.toString(values[0]);
+            } else {
+                extractedValues = null;
+                break;
+            }
         }
-        return result;
+        return new Row(extractedValues);
     }
 
-    private List<String[]> continueScroll() throws IOException {
+    private List<Row> continueScroll() throws IOException {
         LOGGER.debug("[{}] Continuing scroll with id [{}]", context.jobId, scrollId);
         SearchResponse searchResponse = executeSearchScrollRequest(scrollId);
         LOGGER.debug("[{}] Search response was obtained", context.jobId);
@@ -210,4 +217,19 @@ public DataSummary(long rows, long cols) {
             this.cols = cols;
         }
     }
+
+    public static class Row {
+
+        @Nullable
+        private String[] values;
+
+        private Row(String[] values) {
+            this.values = values;
+        }
+
+        @Nullable
+        public String[] getValues() {
+            return values;
+        }
+    }
 }

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/analytics/process/AnalyticsProcessManager.java

@@ -48,10 +48,13 @@ public void processData(String jobId, DataFrameDataExtractor dataExtractor) {
                 process.writeRecord(dataExtractor.getFieldNamesArray());
 
                 while (dataExtractor.hasNext()) {
-                    Optional<List<String[]>> records = dataExtractor.next();
-                    if (records.isPresent()) {
-                        for (String[] record : records.get()) {
-                            process.writeRecord(record);
+                    Optional<List<DataFrameDataExtractor.Row>> rows = dataExtractor.next();
+                    if (rows.isPresent()) {
+                        for (DataFrameDataExtractor.Row row : rows.get()) {
+                            String[] rowValues = row.getValues();
+                            if (rowValues != null) {
+                                process.writeRecord(rowValues);
+                            }
                         }
                     }
                 }