Adds the ability to specify a format on date_histogram source

This commit adds the ability to specify a date format on the `date_histogram` composite source.
If the format is defined, the key for the source is returned as a formatted date.

Closes elastic#27923

docs/reference/aggregations/bucket/composite-aggregation.asciidoc

@@ -225,7 +225,41 @@ Note that fractional time values are not supported, but you can address this by
 time unit (e.g., `1.5h` could instead be specified as `90m`).
 
 [float]
-===== Time Zone
+====== Format
+
+Internally, a date is represented as a 64 bit number representing a timestamp in milliseconds-since-the-epoch.
+These timestamps are returned as the bucket keys. It is possible to return a formatted date string instead using
+the format specified with the format parameter:
+
+[source,js]
+--------------------------------------------------
+GET /_search
+{
+    "aggs" : {
+        "my_buckets": {
+            "composite" : {
+                "sources" : [
+                    {
+                        "date": {
+                            "date_histogram" : {
+                                "field": "timestamp",
+                                "interval": "1d",
+                                "format": "yyyy-MM-dd" <1>
+                            }
+                        }
+                    }
+                ]
+            }
+        }
+    }
+}
+--------------------------------------------------
+// CONSOLE
+
+<1> Supports expressive date <<date-format-pattern,format pattern>>
+
+[float]
+====== Time Zone
 
 Date-times are stored in Elasticsearch in UTC.  By default, all bucketing and
 rounding is also done in UTC. The `time_zone` parameter can be used to indicate

rest-api-spec/src/main/resources/rest-api-spec/test/search.aggregation/230_composite.yml

@@ -7,6 +7,8 @@ setup:
               mappings:
                 doc:
                   properties:
+                    date:
+                      type: date
                     keyword:
                       type: keyword
                     long:
@@ -40,6 +42,20 @@ setup:
         id:    4
         body:  { "keyword": "bar", "long": [1000, 0] }
 
+  - do:
+      index:
+        index: test
+        type:  doc
+        id:    5
+        body:  { "date": "2017-10-20T03:08:45" }
+
+  - do:
+      index:
+        index: test
+        type:  doc
+        id:    6
+        body:  { "date": "2017-10-21T07:00:00" }
+
   - do:
       indices.refresh:
         index: [test]
@@ -66,7 +82,7 @@ setup:
                   }
                 ]
 
-  - match: {hits.total: 4}
+  - match: {hits.total: 6}
   - length: { aggregations.test.buckets: 2 }
   - match: { aggregations.test.buckets.0.key.kw: "bar" }
   - match: { aggregations.test.buckets.0.doc_count: 3 }
@@ -104,7 +120,7 @@ setup:
                   }
                 ]
 
-  - match: {hits.total: 4}
+  - match: {hits.total: 6}
   - length: { aggregations.test.buckets: 5 }
   - match: { aggregations.test.buckets.0.key.long: 0}
   - match: { aggregations.test.buckets.0.key.kw: "bar" }
@@ -154,7 +170,7 @@ setup:
                 ]
                 after: { "long": 20, "kw": "foo" }
 
-  - match: {hits.total: 4}
+  - match: {hits.total: 6}
   - length: { aggregations.test.buckets: 2 }
   - match: { aggregations.test.buckets.0.key.long: 100 }
   - match: { aggregations.test.buckets.0.key.kw: "bar" }
@@ -188,7 +204,7 @@ setup:
                 ]
                 after: { "kw": "delta" }
 
-  - match: {hits.total: 4}
+  - match: {hits.total: 6}
   - length: { aggregations.test.buckets: 1 }
   - match: { aggregations.test.buckets.0.key.kw: "foo" }
   - match: { aggregations.test.buckets.0.doc_count: 2 }
@@ -220,3 +236,35 @@ setup:
                         }
                       }
                     ]
+
+---
+"Composite aggregation with format":
+  - skip:
+      version: " - 6.99.99"
+      reason:  this uses a new option (format) added 7.0.0
+
+  - do:
+      search:
+        index: test
+        body:
+          aggregations:
+            test:
+              composite:
+                sources: [
+                  {
+                    "date": {
+                      "date_histogram": {
+                        "field": "date",
+                        "interval": "1d",
+                        "format": "yyyy-MM-dd"
+                      }
+                    }
+                  }
+                ]
+
+  - match: {hits.total: 6}
+  - length: { aggregations.test.buckets: 2 }
+  - match: { aggregations.test.buckets.0.key.date: "2017-10-20" }
+  - match: { aggregations.test.buckets.0.doc_count: 1 }
+  - match: { aggregations.test.buckets.1.key.date: "2017-10-21" }
+  - match: { aggregations.test.buckets.1.doc_count: 1 }

server/src/main/java/org/elasticsearch/search/aggregations/bucket/composite/CompositeAggregationBuilder.java

@@ -147,17 +147,15 @@ protected AggregatorFactory<?> doBuild(SearchContext context, AggregatorFactory<
             Sort sort = indexSortConfig.buildIndexSort(shardContext::fieldMapper, shardContext::getForField);
             System.arraycopy(sort.getSort(), 0, sortFields, 0, sortFields.length);
         }
-        List<String> sourceNames = new ArrayList<>();
         for (int i = 0; i < configs.length; i++) {
             configs[i] = sources.get(i).build(context, i, configs.length, sortFields[i]);
-            sourceNames.add(sources.get(i).name());
             if (configs[i].valuesSource().needsScores()) {
                 throw new IllegalArgumentException("[sources] cannot access _score");
             }
         }
         final CompositeKey afterKey;
         if (after != null) {
-            if (after.size() != sources.size()) {
+            if (after.size() != configs.length) {
                 throw new IllegalArgumentException("[after] has " + after.size() +
                     " value(s) but [sources] has " + sources.size());
             }
@@ -179,7 +177,7 @@ protected AggregatorFactory<?> doBuild(SearchContext context, AggregatorFactory<
         } else {
             afterKey = null;
         }
-        return new CompositeAggregationFactory(name, context, parent, subfactoriesBuilder, metaData, size, configs, sourceNames, afterKey);
+        return new CompositeAggregationFactory(name, context, parent, subfactoriesBuilder, metaData, size, configs, afterKey);
     }
 
 

server/src/main/java/org/elasticsearch/search/aggregations/bucket/composite/CompositeAggregationFactory.java

@@ -32,24 +32,21 @@
 class CompositeAggregationFactory extends AggregatorFactory<CompositeAggregationFactory> {
     private final int size;
     private final CompositeValuesSourceConfig[] sources;
-    private final List<String> sourceNames;
     private final CompositeKey afterKey;
 
     CompositeAggregationFactory(String name, SearchContext context, AggregatorFactory<?> parent,
                                 AggregatorFactories.Builder subFactoriesBuilder, Map<String, Object> metaData,
-                                int size, CompositeValuesSourceConfig[] sources,
-                                List<String> sourceNames, CompositeKey afterKey) throws IOException {
+                                int size, CompositeValuesSourceConfig[] sources, CompositeKey afterKey) throws IOException {
         super(name, context, parent, subFactoriesBuilder, metaData);
         this.size = size;
         this.sources = sources;
-        this.sourceNames = sourceNames;
         this.afterKey = afterKey;
     }
 
     @Override
     protected Aggregator createInternal(Aggregator parent, boolean collectsFromSingleBucket,
                                         List<PipelineAggregator> pipelineAggregators, Map<String, Object> metaData) throws IOException {
         return new CompositeAggregator(name, factories, context, parent, pipelineAggregators, metaData,
-            size, sources, sourceNames, afterKey);
+            size, sources, afterKey);
     }
 }

server/src/main/java/org/elasticsearch/search/aggregations/bucket/composite/CompositeAggregator.java

@@ -27,6 +27,7 @@
 import org.apache.lucene.search.Scorer;
 import org.apache.lucene.search.Weight;
 import org.apache.lucene.util.RoaringDocIdSet;
+import org.elasticsearch.search.DocValueFormat;
 import org.elasticsearch.search.aggregations.Aggregator;
 import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.InternalAggregation;
@@ -43,11 +44,13 @@
 import java.util.List;
 import java.util.Map;
 import java.util.TreeMap;
+import java.util.stream.Collectors;
 
 final class CompositeAggregator extends BucketsAggregator {
     private final int size;
     private final CompositeValuesSourceConfig[] sources;
     private final List<String> sourceNames;
+    private final List<DocValueFormat> formats;
     private final boolean canEarlyTerminate;
 
     private final TreeMap<Integer, Integer> keys;
@@ -59,12 +62,12 @@ final class CompositeAggregator extends BucketsAggregator {
 
     CompositeAggregator(String name, AggregatorFactories factories, SearchContext context, Aggregator parent,
                             List<PipelineAggregator> pipelineAggregators, Map<String, Object> metaData,
-                            int size, CompositeValuesSourceConfig[] sources, List<String> sourceNames,
-                            CompositeKey rawAfterKey) throws IOException {
+                            int size, CompositeValuesSourceConfig[] sources, CompositeKey rawAfterKey) throws IOException {
         super(name, factories, context, parent, pipelineAggregators, metaData);
         this.size = size;
         this.sources = sources;
-        this.sourceNames = sourceNames;
+        this.sourceNames = Arrays.stream(sources).map(CompositeValuesSourceConfig::name).collect(Collectors.toList());
+        this.formats = Arrays.stream(sources).map(CompositeValuesSourceConfig::format).collect(Collectors.toList());
         // we use slot 0 to fill the current document (size+1).
         this.array = new CompositeValuesComparator(context.searcher().getIndexReader(), sources, size+1);
         if (rawAfterKey != null) {
@@ -131,15 +134,17 @@ public InternalAggregation buildAggregation(long zeroBucket) throws IOException
             CompositeKey key = array.toCompositeKey(slot);
             InternalAggregations aggs = bucketAggregations(slot);
             int docCount = bucketDocCount(slot);
-            buckets[pos++] = new InternalComposite.InternalBucket(sourceNames, key, reverseMuls, docCount, aggs);
+            buckets[pos++] = new InternalComposite.InternalBucket(sourceNames, formats, key, reverseMuls, docCount, aggs);
         }
-        return new InternalComposite(name, size, sourceNames, Arrays.asList(buckets), reverseMuls, pipelineAggregators(), metaData());
+        return new InternalComposite(name, size, sourceNames, formats, Arrays.asList(buckets), reverseMuls,
+            pipelineAggregators(), metaData());
     }
 
     @Override
     public InternalAggregation buildEmptyAggregation() {
         final int[] reverseMuls = getReverseMuls();
-        return new InternalComposite(name, size, sourceNames, Collections.emptyList(), reverseMuls, pipelineAggregators(), metaData());
+        return new InternalComposite(name, size, sourceNames, formats, Collections.emptyList(), reverseMuls,
+            pipelineAggregators(), metaData());
     }
 
     @Override

server/src/main/java/org/elasticsearch/search/aggregations/bucket/composite/CompositeValuesSourceBuilder.java

@@ -25,6 +25,7 @@
 import org.apache.lucene.index.SortedNumericDocValues;
 import org.apache.lucene.index.SortedSetDocValues;
 import org.apache.lucene.search.SortField;
+import org.elasticsearch.Version;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
@@ -51,6 +52,7 @@ public abstract class CompositeValuesSourceBuilder<AB extends CompositeValuesSou
     private ValueType valueType = null;
     private Object missing = null;
     private SortOrder order = SortOrder.ASC;
+    private String format = null;
 
     CompositeValuesSourceBuilder(String name) {
         this(name, null);
@@ -72,6 +74,11 @@ public abstract class CompositeValuesSourceBuilder<AB extends CompositeValuesSou
         }
         this.missing = in.readGenericValue();
         this.order = SortOrder.readFromStream(in);
+        if (in.getVersion().onOrAfter(Version.V_7_0_0_alpha1)) {
+            this.format = in.readOptionalString();
+        } else {
+            this.format = null;
+        }
     }
 
     @Override
@@ -90,6 +97,9 @@ public final void writeTo(StreamOutput out) throws IOException {
         }
         out.writeGenericValue(missing);
         order.writeTo(out);
+        if (out.getVersion().onOrAfter(Version.V_7_0_0_alpha1)) {
+            out.writeOptionalString(format);
+        }
         innerWriteTo(out);
     }
 
@@ -112,6 +122,9 @@ public final XContentBuilder toXContent(XContentBuilder builder, Params params)
         if (valueType != null) {
             builder.field("value_type", valueType.getPreferredName());
         }
+        if (format != null) {
+            builder.field("format", format);
+        }
         builder.field("order", order);
         doXContentBody(builder, params);
         builder.endObject();
@@ -120,7 +133,7 @@ public final XContentBuilder toXContent(XContentBuilder builder, Params params)
 
     @Override
     public final int hashCode() {
-        return Objects.hash(field, missing, script, valueType, order, innerHashCode());
+        return Objects.hash(field, missing, script, valueType, order, format, innerHashCode());
     }
 
     protected abstract int innerHashCode();
@@ -137,6 +150,7 @@ public boolean equals(Object o) {
             Objects.equals(valueType, that.valueType()) &&
             Objects.equals(missing, that.missing()) &&
             Objects.equals(order, that.order()) &&
+            Objects.equals(format, that.format()) &&
             innerEquals(that);
     }
 
@@ -254,6 +268,24 @@ public SortOrder order() {
         return order;
     }
 
+    /**
+     * Sets the format to use for the output of the aggregation.
+     */
+    public AB format(String format) {
+        if (format == null) {
+            throw new IllegalArgumentException("[format] must not be null: [" + name + "]");
+        }
+        this.format = format;
+        return (AB) this;
+    }
+
+    /**
+     * Gets the format to use for the output of the aggregation.
+     */
+    public String format() {
+        return format;
+    }
+
     /**
      * Creates a {@link CompositeValuesSourceConfig} for this source.
      *
@@ -271,7 +303,7 @@ protected abstract CompositeValuesSourceConfig innerBuild(SearchContext context,
 
     public final CompositeValuesSourceConfig build(SearchContext context, int pos, int numPos, SortField sortField) throws IOException {
         ValuesSourceConfig<?> config = ValuesSourceConfig.resolve(context.getQueryShardContext(),
-            valueType, field, script, missing, null, null);
+            valueType, field, script, missing, null, format);
         return innerBuild(context, config, pos, numPos, sortField);
     }