Merge branch 'main' into 2022-12-22-ccr-chunking

x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/AbstractRollupFieldProducer.java

@@ -7,31 +7,23 @@
 
 package org.elasticsearch.xpack.downsample;
 
-import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.index.fielddata.FormattedDocValues;
 
 import java.io.IOException;
 
 /**
- * Base class for classes that read metric and label fields.
+ * Base class that reads fields from the source index and produces their downsampled values
  */
-abstract class AbstractRollupFieldProducer<T> {
+abstract class AbstractRollupFieldProducer implements RollupFieldSerializer {
 
-    protected final String name;
+    private final String name;
     protected boolean isEmpty;
 
     AbstractRollupFieldProducer(String name) {
         this.name = name;
         this.isEmpty = true;
     }
 
-    /**
-     * Collect a value for the field applying the specific subclass collection strategy.
-     *
-     * @param field the name of the field to collect
-     * @param value the value to collect.
-     */
-    public abstract void collect(String field, T value);
-
     /**
      * @return the name of the field.
      */
@@ -44,15 +36,12 @@ public String name() {
      */
     public abstract void reset();
 
-    /**
-     * Serialize the downsampled value of the field.
-     */
-    public abstract void write(XContentBuilder builder) throws IOException;
-
     /**
      * @return true if the field has not collected any value.
      */
     public boolean isEmpty() {
         return isEmpty;
     }
+
+    public abstract void collect(FormattedDocValues docValues, int docId) throws IOException;
 }

x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/AggregateMetricFieldSerializer.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.downsample;
+
+import org.elasticsearch.xcontent.XContentBuilder;
+
+import java.io.IOException;
+import java.util.Collection;
+
+public class AggregateMetricFieldSerializer implements RollupFieldSerializer {
+    private final Collection<AbstractRollupFieldProducer> producers;
+    private final String name;
+
+    /**
+     * @param name the name of the aggregate_metric_double field as it will be serialized
+     *             in the downsampled index
+     * @param producers a collection of {@link AbstractRollupFieldProducer} instances with the subfields
+     *                  of the aggregate_metric_double field.
+     */
+    public AggregateMetricFieldSerializer(String name, Collection<AbstractRollupFieldProducer> producers) {
+        this.name = name;
+        this.producers = producers;
+    }
+
+    @Override
+    public void write(XContentBuilder builder) throws IOException {
+        if (isEmpty()) {
+            return;
+        }
+
+        builder.startObject(name);
+        for (AbstractRollupFieldProducer rollupFieldProducer : producers) {
+            assert name.equals(rollupFieldProducer.name()) : "producer has a different name";
+            if (rollupFieldProducer.isEmpty() == false) {
+                if (rollupFieldProducer instanceof MetricFieldProducer metricFieldProducer) {
+                    for (MetricFieldProducer.Metric metric : metricFieldProducer.metrics()) {
+                        if (metric.get() != null) {
+                            builder.field(metric.name(), metric.get());
+                        }
+                    }
+                } else if (rollupFieldProducer instanceof LabelFieldProducer labelFieldProducer) {
+                    LabelFieldProducer.Label label = labelFieldProducer.label();
+                    if (label.get() != null) {
+                        builder.field(label.name(), label.get());
+                    }
+                }
+            }
+        }
+        builder.endObject();
+    }
+
+    private boolean isEmpty() {
+        for (AbstractRollupFieldProducer p : producers) {
+            if (p.isEmpty() == false) {
+                return false;
+            }
+        }
+        return true;
+    }
+}

x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/AggregateMetricFieldValueFetcher.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.downsample;
+
+import org.elasticsearch.index.fielddata.IndexFieldData;
+import org.elasticsearch.index.mapper.MappedFieldType;
+import org.elasticsearch.index.mapper.NumberFieldMapper;
+import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper;
+import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.AggregateDoubleMetricFieldType;
+
+import java.util.List;
+
+public class AggregateMetricFieldValueFetcher extends FieldValueFetcher {
+
+    private AggregateDoubleMetricFieldType aggMetricFieldType;
+
+    private final AbstractRollupFieldProducer rollupFieldProducer;
+
+    protected AggregateMetricFieldValueFetcher(
+        MappedFieldType fieldType,
+        AggregateDoubleMetricFieldType aggMetricFieldType,
+        IndexFieldData<?> fieldData
+    ) {
+        super(fieldType, fieldData);
+        this.aggMetricFieldType = aggMetricFieldType;
+        this.rollupFieldProducer = createRollupFieldProducer();
+    }
+
+    public AbstractRollupFieldProducer rollupFieldProducer() {
+        return rollupFieldProducer;
+    }
+
+    private AbstractRollupFieldProducer createRollupFieldProducer() {
+        AggregateDoubleMetricFieldMapper.Metric metric = null;
+        for (var e : aggMetricFieldType.getMetricFields().entrySet()) {
+            NumberFieldMapper.NumberFieldType metricSubField = e.getValue();
+            if (metricSubField.name().equals(name())) {
+                metric = e.getKey();
+                break;
+            }
+        }
+        assert metric != null : "Cannot resolve metric type for field " + name();
+
+        if (aggMetricFieldType.getMetricType() != null) {
+            // If the field is an aggregate_metric_double field, we should use the correct subfields
+            // for each aggregation. This is a rollup-of-rollup case
+            MetricFieldProducer.Metric metricOperation = switch (metric) {
+                case max -> new MetricFieldProducer.Max();
+                case min -> new MetricFieldProducer.Min();
+                case sum -> new MetricFieldProducer.Sum();
+                // To compute value_count summary, we must sum all field values
+                case value_count -> new MetricFieldProducer.Sum(AggregateDoubleMetricFieldMapper.Metric.value_count.name());
+            };
+            return new MetricFieldProducer.GaugeMetricFieldProducer(aggMetricFieldType.name(), List.of(metricOperation));
+        } else {
+            // If field is not a metric, we downsample it as a label
+            return new LabelFieldProducer.AggregateMetricFieldProducer.AggregateMetricFieldProducer(aggMetricFieldType.name(), metric);
+        }
+    }
+}

x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/FieldValueFetcher.java

@@ -16,51 +16,59 @@
 import org.elasticsearch.search.DocValueFormat;
 import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper;
 
+import java.util.ArrayList;
 import java.util.Collections;
-import java.util.LinkedHashMap;
-import java.util.Map;
+import java.util.List;
 
 /**
  * Utility class used for fetching field values by reading field data
  */
 class FieldValueFetcher {
-    private final String name;
+
     private final MappedFieldType fieldType;
-    private final DocValueFormat format;
     private final IndexFieldData<?> fieldData;
+    private final AbstractRollupFieldProducer rollupFieldProducer;
 
-    protected FieldValueFetcher(String name, MappedFieldType fieldType, IndexFieldData<?> fieldData) {
-        this.name = name;
+    protected FieldValueFetcher(MappedFieldType fieldType, IndexFieldData<?> fieldData) {
         this.fieldType = fieldType;
-        this.format = fieldType.docValueFormat(null, null);
         this.fieldData = fieldData;
+        this.rollupFieldProducer = createRollupFieldProducer();
     }
 
     public String name() {
-        return name;
+        return fieldType().name();
     }
 
     public MappedFieldType fieldType() {
         return fieldType;
     }
 
-    public DocValueFormat format() {
-        return format;
+    public FormattedDocValues getLeaf(LeafReaderContext context) {
+        DocValueFormat format = fieldType.docValueFormat(null, null);
+        return fieldData.load(context).getFormattedValues(format);
     }
 
-    public IndexFieldData<?> fieldData() {
-        return fieldData;
+    public AbstractRollupFieldProducer rollupFieldProducer() {
+        return rollupFieldProducer;
     }
 
-    FormattedDocValues getLeaf(LeafReaderContext context) {
-        return fieldData.load(context).getFormattedValues(format);
+    private AbstractRollupFieldProducer createRollupFieldProducer() {
+        if (fieldType.getMetricType() != null) {
+            return switch (fieldType.getMetricType()) {
+                case gauge -> new MetricFieldProducer.GaugeMetricFieldProducer(name());
+                case counter -> new MetricFieldProducer.CounterMetricFieldProducer(name());
+            };
+        } else {
+            // If field is not a metric, we downsample it as a label
+            return new LabelFieldProducer.LabelLastValueFieldProducer(name());
+        }
     }
 
     /**
      * Retrieve field value fetchers for a list of fields.
      */
-    static Map<String, FieldValueFetcher> create(SearchExecutionContext context, String[] fields) {
-        Map<String, FieldValueFetcher> fetchers = new LinkedHashMap<>();
+    static List<FieldValueFetcher> create(SearchExecutionContext context, String[] fields) {
+        List<FieldValueFetcher> fetchers = new ArrayList<>();
         for (String field : fields) {
             MappedFieldType fieldType = context.getFieldType(field);
             assert fieldType != null : "Unknown field type for field: [" + field + "]";
@@ -71,24 +79,16 @@ static Map<String, FieldValueFetcher> create(SearchExecutionContext context, Str
                 for (NumberFieldMapper.NumberFieldType metricSubField : aggMetricFieldType.getMetricFields().values()) {
                     if (context.fieldExistsInIndex(metricSubField.name())) {
                         IndexFieldData<?> fieldData = context.getForField(metricSubField, MappedFieldType.FielddataOperation.SEARCH);
-                        fetchers.put(metricSubField.name(), new FieldValueFetcher(metricSubField.name(), fieldType, fieldData));
+                        fetchers.add(new AggregateMetricFieldValueFetcher(metricSubField, aggMetricFieldType, fieldData));
                     }
                 }
             } else {
                 if (context.fieldExistsInIndex(field)) {
                     IndexFieldData<?> fieldData = context.getForField(fieldType, MappedFieldType.FielddataOperation.SEARCH);
-                    fetchers.put(field, new FieldValueFetcher(field, fieldType, fieldData));
+                    fetchers.add(new FieldValueFetcher(fieldType, fieldData));
                 }
             }
         }
-        return Collections.unmodifiableMap(fetchers);
-    }
-
-    static Map<String, FormattedDocValues> docValuesFetchers(LeafReaderContext ctx, Map<String, FieldValueFetcher> fieldValueFetchers) {
-        final Map<String, FormattedDocValues> docValuesFetchers = new LinkedHashMap<>(fieldValueFetchers.size());
-        for (FieldValueFetcher fetcher : fieldValueFetchers.values()) {
-            docValuesFetchers.put(fetcher.name(), fetcher.getLeaf(ctx));
-        }
-        return Collections.unmodifiableMap(docValuesFetchers);
+        return Collections.unmodifiableList(fetchers);
     }
 }