Make MappedFieldType responsible for providing a parser/formatter. #17546

Aggregations need to perform instanceof calls on MappedFieldType instances in
order to know how they should be parsed or formatted. Instead, we should let
the field types provide a formatter/parser that can can be used.

6 files changed, 89 insertions, 99 deletions

diff --git a/core/src/main/java/org/elasticsearch/search/aggregations/bucket/range/InternalRange.java b/core/src/main/java/org/elasticsearch/search/aggregations/bucket/range/InternalRange.java
index c38334a80f..ce4e0ba60f 100644
--- a/core/src/main/java/org/elasticsearch/search/aggregations/bucket/range/InternalRange.java
+++ b/core/src/main/java/org/elasticsearch/search/aggregations/bucket/range/InternalRange.java
@@ -21,6 +21,7 @@ package org.elasticsearch.search.aggregations.bucket.range;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.search.DocValueFormat;
 import org.elasticsearch.search.aggregations.AggregationStreams;
 import org.elasticsearch.search.aggregations.Aggregations;
 import org.elasticsearch.search.aggregations.InternalAggregation;
@@ -31,8 +32,6 @@ import org.elasticsearch.search.aggregations.bucket.BucketStreams;
 import org.elasticsearch.search.aggregations.pipeline.PipelineAggregator;
 import org.elasticsearch.search.aggregations.support.ValueType;
 import org.elasticsearch.search.aggregations.support.ValuesSourceType;
-import org.elasticsearch.search.aggregations.support.format.ValueFormatter;
-import org.elasticsearch.search.aggregations.support.format.ValueFormatterStreams;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -61,7 +60,7 @@ public class InternalRange<B extends InternalRange.Bucket, R extends InternalRan
     private final static BucketStreams.Stream<Bucket> BUCKET_STREAM = new BucketStreams.Stream<Bucket>() {
         @Override
         public Bucket readResult(StreamInput in, BucketStreamContext context) throws IOException {
-            Bucket buckets = new Bucket(context.keyed(), context.formatter());
+            Bucket buckets = new Bucket(context.keyed(), context.format());
             buckets.readFrom(in);
             return buckets;
         }
@@ -69,7 +68,7 @@ public class InternalRange<B extends InternalRange.Bucket, R extends InternalRan
         @Override
         public BucketStreamContext getBucketStreamContext(Bucket bucket) {
             BucketStreamContext context = new BucketStreamContext();
-            context.formatter(bucket.formatter);
+            context.format(bucket.format);
             context.keyed(bucket.keyed);
             return context;
         }
@@ -83,20 +82,20 @@ public class InternalRange<B extends InternalRange.Bucket, R extends InternalRan
     public static class Bucket extends InternalMultiBucketAggregation.InternalBucket implements Range.Bucket {
 
         protected transient final boolean keyed;
-        protected transient final ValueFormatter formatter;
+        protected transient final DocValueFormat format;
         protected double from;
         protected double to;
         private long docCount;
         InternalAggregations aggregations;
         private String key;
 
-        public Bucket(boolean keyed, ValueFormatter formatter) {
+        public Bucket(boolean keyed, DocValueFormat formatter) {
             this.keyed = keyed;
-            this.formatter = formatter;
+            this.format = formatter;
         }
 
         public Bucket(String key, double from, double to, long docCount, InternalAggregations aggregations, boolean keyed,
-                ValueFormatter formatter) {
+                DocValueFormat formatter) {
             this(keyed, formatter);
             this.key = key != null ? key : generateKey(from, to, formatter);
             this.from = from;
@@ -129,8 +128,8 @@ public class InternalRange<B extends InternalRange.Bucket, R extends InternalRan
             return keyed;
         }
 
-        public ValueFormatter getFormatter() {
-            return formatter;
+        public DocValueFormat getFormat() {
+            return format;
         }
 
         @Override
@@ -138,7 +137,7 @@ public class InternalRange<B extends InternalRange.Bucket, R extends InternalRan
             if (Double.isInfinite(from)) {
                 return null;
             } else {
-                return formatter.format(from);
+                return format.format(from);
             }
         }
 
@@ -147,7 +146,7 @@ public class InternalRange<B extends InternalRange.Bucket, R extends InternalRan
             if (Double.isInfinite(to)) {
                 return null;
             } else {
-                return formatter.format(to);
+                return format.format(to);
             }
         }
 
@@ -173,7 +172,7 @@ public class InternalRange<B extends InternalRange.Bucket, R extends InternalRan
                 aggregationsList.add(range.aggregations);
             }
             final InternalAggregations aggs = InternalAggregations.reduce(aggregationsList, context);
-            return getFactory().createBucket(key, from, to, docCount, aggs, keyed, formatter);
+            return getFactory().createBucket(key, from, to, docCount, aggs, keyed, format);
         }
 
         @Override
@@ -186,14 +185,14 @@ public class InternalRange<B extends InternalRange.Bucket, R extends InternalRan
             }
             if (!Double.isInfinite(from)) {
                 builder.field(CommonFields.FROM, from);
-                if (formatter != null) {
-                    builder.field(CommonFields.FROM_AS_STRING, formatter.format(from));
+                if (format != DocValueFormat.RAW) {
+                    builder.field(CommonFields.FROM_AS_STRING, format.format(from));
                 }
             }
             if (!Double.isInfinite(to)) {
                 builder.field(CommonFields.TO, to);
-                if (formatter != null) {
-                    builder.field(CommonFields.TO_AS_STRING, formatter.format(to));
+                if (format != DocValueFormat.RAW) {
+                    builder.field(CommonFields.TO_AS_STRING, format.format(to));
                 }
             }
             builder.field(CommonFields.DOC_COUNT, docCount);
@@ -202,7 +201,7 @@ public class InternalRange<B extends InternalRange.Bucket, R extends InternalRan
             return builder;
         }
 
-        protected String generateKey(double from, double to, ValueFormatter formatter) {
+        protected String generateKey(double from, double to, DocValueFormat formatter) {
             StringBuilder sb = new StringBuilder();
             sb.append(Double.isInfinite(from) ? "*" : formatter.format(from));
             sb.append("-");
@@ -236,43 +235,43 @@ public class InternalRange<B extends InternalRange.Bucket, R extends InternalRan
         }
 
         @SuppressWarnings("unchecked")
-        public R create(String name, List<B> ranges, ValueFormatter formatter, boolean keyed, List<PipelineAggregator> pipelineAggregators,
+        public R create(String name, List<B> ranges, DocValueFormat formatter, boolean keyed, List<PipelineAggregator> pipelineAggregators,
                 Map<String, Object> metaData) {
             return (R) new InternalRange<>(name, ranges, formatter, keyed, pipelineAggregators, metaData);
         }
 
         @SuppressWarnings("unchecked")
         public B createBucket(String key, double from, double to, long docCount, InternalAggregations aggregations, boolean keyed,
-                ValueFormatter formatter) {
+                DocValueFormat formatter) {
             return (B) new Bucket(key, from, to, docCount, aggregations, keyed, formatter);
         }
 
         @SuppressWarnings("unchecked")
         public R create(List<B> ranges, R prototype) {
-            return (R) new InternalRange<>(prototype.name, ranges, prototype.formatter, prototype.keyed, prototype.pipelineAggregators(),
+            return (R) new InternalRange<>(prototype.name, ranges, prototype.format, prototype.keyed, prototype.pipelineAggregators(),
                     prototype.metaData);
         }
 
         @SuppressWarnings("unchecked")
         public B createBucket(InternalAggregations aggregations, B prototype) {
             return (B) new Bucket(prototype.getKey(), prototype.from, prototype.to, prototype.getDocCount(), aggregations, prototype.keyed,
-                    prototype.formatter);
+                    prototype.format);
         }
     }
 
     private List<B> ranges;
     private Map<String, B> rangeMap;
-    protected ValueFormatter formatter;
+    protected DocValueFormat format;
     protected boolean keyed;
 
     public InternalRange() {} // for serialization
 
-    public InternalRange(String name, List<B> ranges, ValueFormatter formatter, boolean keyed,
+    public InternalRange(String name, List<B> ranges, DocValueFormat format, boolean keyed,
             List<PipelineAggregator> pipelineAggregators,
             Map<String, Object> metaData) {
         super(name, pipelineAggregators, metaData);
         this.ranges = ranges;
-        this.formatter = formatter;
+        this.format = format;
         this.keyed = keyed;
     }
 
@@ -320,18 +319,18 @@ public class InternalRange<B extends InternalRange.Bucket, R extends InternalRan
         for (int i = 0; i < this.ranges.size(); ++i) {
             ranges.add((B) rangeList[i].get(0).reduce(rangeList[i], reduceContext));
         }
-        return getFactory().create(name, ranges, formatter, keyed, pipelineAggregators(), getMetaData());
+        return getFactory().create(name, ranges, format, keyed, pipelineAggregators(), getMetaData());
     }
 
     @Override
     protected void doReadFrom(StreamInput in) throws IOException {
-        formatter = ValueFormatterStreams.readOptional(in);
+        format = in.readValueFormat();
         keyed = in.readBoolean();
         int size = in.readVInt();
         List<B> ranges = new ArrayList<>(size);
         for (int i = 0; i < size; i++) {
             String key = in.readOptionalString();
-            ranges.add(getFactory().createBucket(key, in.readDouble(), in.readDouble(), in.readVLong(), InternalAggregations.readAggregations(in), keyed, formatter));
+            ranges.add(getFactory().createBucket(key, in.readDouble(), in.readDouble(), in.readVLong(), InternalAggregations.readAggregations(in), keyed, format));
         }
         this.ranges = ranges;
         this.rangeMap = null;
@@ -339,7 +338,7 @@ public class InternalRange<B extends InternalRange.Bucket, R extends InternalRan
 
     @Override
     protected void doWriteTo(StreamOutput out) throws IOException {
-        ValueFormatterStreams.writeOptional(formatter, out);
+        out.writeValueFormat(format);
         out.writeBoolean(keyed);
         out.writeVInt(ranges.size());
         for (B bucket : ranges) {
diff --git a/core/src/main/java/org/elasticsearch/search/aggregations/bucket/range/RangeAggregator.java b/core/src/main/java/org/elasticsearch/search/aggregations/bucket/range/RangeAggregator.java
index 392e3c1422..45bfefff2c 100644
--- a/core/src/main/java/org/elasticsearch/search/aggregations/bucket/range/RangeAggregator.java
+++ b/core/src/main/java/org/elasticsearch/search/aggregations/bucket/range/RangeAggregator.java
@@ -29,6 +29,7 @@ import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.index.fielddata.SortedNumericDoubleValues;
+import org.elasticsearch.search.DocValueFormat;
 import org.elasticsearch.search.aggregations.Aggregator;
 import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.InternalAggregation;
@@ -40,9 +41,6 @@ import org.elasticsearch.search.aggregations.bucket.BucketsAggregator;
 import org.elasticsearch.search.aggregations.pipeline.PipelineAggregator;
 import org.elasticsearch.search.aggregations.support.AggregationContext;
 import org.elasticsearch.search.aggregations.support.ValuesSource;
-import org.elasticsearch.search.aggregations.support.format.ValueFormat;
-import org.elasticsearch.search.aggregations.support.format.ValueFormatter;
-import org.elasticsearch.search.aggregations.support.format.ValueParser;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -97,15 +95,15 @@ public class RangeAggregator extends BucketsAggregator {
             return "[" + from + " to " + to + ")";
         }
 
-        public Range process(ValueParser parser, SearchContext context) {
+        public Range process(DocValueFormat parser, SearchContext context) {
             assert parser != null;
             Double from = this.from;
             Double to = this.to;
             if (fromAsStr != null) {
-                from = parser.parseDouble(fromAsStr, context);
+                from = parser.parseDouble(fromAsStr, false, context.nowCallable());
             }
             if (toAsStr != null) {
-                to = parser.parseDouble(toAsStr, context);
+                to = parser.parseDouble(toAsStr, false, context.nowCallable());
             }
             return new Range(key, from, fromAsStr, to, toAsStr);
         }
@@ -205,28 +203,27 @@ public class RangeAggregator extends BucketsAggregator {
     }
 
     final ValuesSource.Numeric valuesSource;
-    final ValueFormatter formatter;
+    final DocValueFormat format;
     final Range[] ranges;
     final boolean keyed;
     final InternalRange.Factory rangeFactory;
 
     final double[] maxTo;
 
-    public RangeAggregator(String name, AggregatorFactories factories, ValuesSource.Numeric valuesSource, ValueFormat format,
+    public RangeAggregator(String name, AggregatorFactories factories, ValuesSource.Numeric valuesSource, DocValueFormat format,
             InternalRange.Factory rangeFactory, List<? extends Range> ranges, boolean keyed, AggregationContext aggregationContext,
             Aggregator parent, List<PipelineAggregator> pipelineAggregators, Map<String, Object> metaData) throws IOException {
 
         super(name, factories, aggregationContext, parent, pipelineAggregators, metaData);
         assert valuesSource != null;
         this.valuesSource = valuesSource;
-        this.formatter = format.formatter();
+        this.format = format;
         this.keyed = keyed;
         this.rangeFactory = rangeFactory;
 
         this.ranges = new Range[ranges.size()];
-        ValueParser parser = format != null ? format.parser() : ValueParser.RAW;
         for (int i = 0; i < this.ranges.length; i++) {
-            this.ranges[i] = ranges.get(i).process(parser, context.searchContext());
+            this.ranges[i] = ranges.get(i).process(format, context.searchContext());
         }
         sortRanges(this.ranges);
 
@@ -320,11 +317,11 @@ public class RangeAggregator extends BucketsAggregator {
             Range range = ranges[i];
             final long bucketOrd = subBucketOrdinal(owningBucketOrdinal, i);
             org.elasticsearch.search.aggregations.bucket.range.Range.Bucket bucket =
-                    rangeFactory.createBucket(range.key, range.from, range.to, bucketDocCount(bucketOrd), bucketAggregations(bucketOrd), keyed, formatter);
+                    rangeFactory.createBucket(range.key, range.from, range.to, bucketDocCount(bucketOrd), bucketAggregations(bucketOrd), keyed, format);
             buckets.add(bucket);
         }
         // value source can be null in the case of unmapped fields
-        return rangeFactory.create(name, buckets, formatter, keyed, pipelineAggregators(), metaData());
+        return rangeFactory.create(name, buckets, format, keyed, pipelineAggregators(), metaData());
     }
 
     @Override
@@ -334,11 +331,11 @@ public class RangeAggregator extends BucketsAggregator {
         for (int i = 0; i < ranges.length; i++) {
             Range range = ranges[i];
             org.elasticsearch.search.aggregations.bucket.range.Range.Bucket bucket =
-                    rangeFactory.createBucket(range.key, range.from, range.to, 0, subAggs, keyed, formatter);
+                    rangeFactory.createBucket(range.key, range.from, range.to, 0, subAggs, keyed, format);
             buckets.add(bucket);
         }
         // value source can be null in the case of unmapped fields
-        return rangeFactory.create(name, buckets, formatter, keyed, pipelineAggregators(), metaData());
+        return rangeFactory.create(name, buckets, format, keyed, pipelineAggregators(), metaData());
     }
 
     private static final void sortRanges(final Range[] ranges) {
@@ -367,22 +364,21 @@ public class RangeAggregator extends BucketsAggregator {
         private final List<R> ranges;
         private final boolean keyed;
         private final InternalRange.Factory factory;
-        private final ValueFormatter formatter;
+        private final DocValueFormat format;
 
         @SuppressWarnings("unchecked")
-        public Unmapped(String name, List<R> ranges, boolean keyed, ValueFormat format,
+        public Unmapped(String name, List<R> ranges, boolean keyed, DocValueFormat format,
                 AggregationContext context,
                 Aggregator parent, InternalRange.Factory factory, List<PipelineAggregator> pipelineAggregators, Map<String, Object> metaData)
                 throws IOException {
 
             super(name, context, parent, pipelineAggregators, metaData);
             this.ranges = new ArrayList<>();
-            ValueParser parser = format != null ? format.parser() : ValueParser.RAW;
             for (R range : ranges) {
-                this.ranges.add((R) range.process(parser, context.searchContext()));
+                this.ranges.add((R) range.process(format, context.searchContext()));
             }
             this.keyed = keyed;
-            this.formatter = format.formatter();
+            this.format = format;
             this.factory = factory;
         }
 
@@ -391,9 +387,9 @@ public class RangeAggregator extends BucketsAggregator {
             InternalAggregations subAggs = buildEmptySubAggregations();
             List<org.elasticsearch.search.aggregations.bucket.range.Range.Bucket> buckets = new ArrayList<>(ranges.size());
             for (RangeAggregator.Range range : ranges) {
-                buckets.add(factory.createBucket(range.key, range.from, range.to, 0, subAggs, keyed, formatter));
+                buckets.add(factory.createBucket(range.key, range.from, range.to, 0, subAggs, keyed, format));
             }
-            return factory.create(name, buckets, formatter, keyed, pipelineAggregators(), metaData());
+            return factory.create(name, buckets, format, keyed, pipelineAggregators(), metaData());
         }
     }
 
diff --git a/core/src/main/java/org/elasticsearch/search/aggregations/bucket/range/date/InternalDateRange.java b/core/src/main/java/org/elasticsearch/search/aggregations/bucket/range/date/InternalDateRange.java
index 88568bcd00..80411c5904 100644
--- a/core/src/main/java/org/elasticsearch/search/aggregations/bucket/range/date/InternalDateRange.java
+++ b/core/src/main/java/org/elasticsearch/search/aggregations/bucket/range/date/InternalDateRange.java
@@ -19,6 +19,7 @@
 package org.elasticsearch.search.aggregations.bucket.range.date;
 
 import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.search.DocValueFormat;
 import org.elasticsearch.search.aggregations.AggregationStreams;
 import org.elasticsearch.search.aggregations.InternalAggregation;
 import org.elasticsearch.search.aggregations.InternalAggregations;
@@ -27,7 +28,6 @@ import org.elasticsearch.search.aggregations.bucket.BucketStreams;
 import org.elasticsearch.search.aggregations.bucket.range.InternalRange;
 import org.elasticsearch.search.aggregations.pipeline.PipelineAggregator;
 import org.elasticsearch.search.aggregations.support.ValueType;
-import org.elasticsearch.search.aggregations.support.format.ValueFormatter;
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
 
@@ -54,7 +54,7 @@ public class InternalDateRange extends InternalRange<InternalDateRange.Bucket, I
     private final static BucketStreams.Stream<Bucket> BUCKET_STREAM = new BucketStreams.Stream<Bucket>() {
         @Override
         public Bucket readResult(StreamInput in, BucketStreamContext context) throws IOException {
-            Bucket buckets = new Bucket(context.keyed(), context.formatter());
+            Bucket buckets = new Bucket(context.keyed(), context.format());
             buckets.readFrom(in);
             return buckets;
         }
@@ -62,7 +62,7 @@ public class InternalDateRange extends InternalRange<InternalDateRange.Bucket, I
         @Override
         public BucketStreamContext getBucketStreamContext(Bucket bucket) {
             BucketStreamContext context = new BucketStreamContext();
-            context.formatter(bucket.formatter());
+            context.format(bucket.format());
             context.keyed(bucket.keyed());
             return context;
         }
@@ -77,15 +77,15 @@ public class InternalDateRange extends InternalRange<InternalDateRange.Bucket, I
 
     public static class Bucket extends InternalRange.Bucket {
 
-        public Bucket(boolean keyed, ValueFormatter formatter) {
+        public Bucket(boolean keyed, DocValueFormat formatter) {
             super(keyed, formatter);
         }
 
-        public Bucket(String key, double from, double to, long docCount, List<InternalAggregation> aggregations, boolean keyed, ValueFormatter formatter) {
+        public Bucket(String key, double from, double to, long docCount, List<InternalAggregation> aggregations, boolean keyed, DocValueFormat formatter) {
             super(key, from, to, docCount, new InternalAggregations(aggregations), keyed, formatter);
         }
 
-        public Bucket(String key, double from, double to, long docCount, InternalAggregations aggregations, boolean keyed, ValueFormatter formatter) {
+        public Bucket(String key, double from, double to, long docCount, InternalAggregations aggregations, boolean keyed, DocValueFormat formatter) {
             super(key, from, to, docCount, aggregations, keyed, formatter);
         }
 
@@ -108,8 +108,8 @@ public class InternalDateRange extends InternalRange<InternalDateRange.Bucket, I
             return keyed;
         }
 
-        ValueFormatter formatter() {
-            return formatter;
+        DocValueFormat format() {
+            return format;
         }
     }
 
@@ -126,32 +126,32 @@ public class InternalDateRange extends InternalRange<InternalDateRange.Bucket, I
         }
 
         @Override
-        public InternalDateRange create(String name, List<InternalDateRange.Bucket> ranges, ValueFormatter formatter, boolean keyed,
+        public InternalDateRange create(String name, List<InternalDateRange.Bucket> ranges, DocValueFormat formatter, boolean keyed,
                 List<PipelineAggregator> pipelineAggregators, Map<String, Object> metaData) {
             return new InternalDateRange(name, ranges, formatter, keyed, pipelineAggregators, metaData);
         }
 
         @Override
         public InternalDateRange create(List<Bucket> ranges, InternalDateRange prototype) {
-            return new InternalDateRange(prototype.name, ranges, prototype.formatter, prototype.keyed, prototype.pipelineAggregators(),
+            return new InternalDateRange(prototype.name, ranges, prototype.format, prototype.keyed, prototype.pipelineAggregators(),
                     prototype.metaData);
         }
 
         @Override
-        public Bucket createBucket(String key, double from, double to, long docCount, InternalAggregations aggregations, boolean keyed, ValueFormatter formatter) {
+        public Bucket createBucket(String key, double from, double to, long docCount, InternalAggregations aggregations, boolean keyed, DocValueFormat formatter) {
             return new Bucket(key, from, to, docCount, aggregations, keyed, formatter);
         }
 
         @Override
         public Bucket createBucket(InternalAggregations aggregations, Bucket prototype) {
             return new Bucket(prototype.getKey(), ((Number) prototype.getFrom()).doubleValue(), ((Number) prototype.getTo()).doubleValue(),
-                    prototype.getDocCount(), aggregations, prototype.getKeyed(), prototype.getFormatter());
+                    prototype.getDocCount(), aggregations, prototype.getKeyed(), prototype.getFormat());
         }
     }
 
     InternalDateRange() {} // for serialization
 
-    InternalDateRange(String name, List<InternalDateRange.Bucket> ranges, ValueFormatter formatter, boolean keyed,
+    InternalDateRange(String name, List<InternalDateRange.Bucket> ranges, DocValueFormat formatter, boolean keyed,
             List<PipelineAggregator> pipelineAggregators, Map<String, Object> metaData) {
         super(name, ranges, formatter, keyed, pipelineAggregators, metaData);
     }
diff --git a/core/src/main/java/org/elasticsearch/search/aggregations/bucket/range/geodistance/InternalGeoDistance.java b/core/src/main/java/org/elasticsearch/search/aggregations/bucket/range/geodistance/InternalGeoDistance.java
index 2c16d93ea0..13ed3a44fb 100644
--- a/core/src/main/java/org/elasticsearch/search/aggregations/bucket/range/geodistance/InternalGeoDistance.java
+++ b/core/src/main/java/org/elasticsearch/search/aggregations/bucket/range/geodistance/InternalGeoDistance.java
@@ -19,6 +19,7 @@
 package org.elasticsearch.search.aggregations.bucket.range.geodistance;
 
 import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.search.DocValueFormat;
 import org.elasticsearch.search.aggregations.AggregationStreams;
 import org.elasticsearch.search.aggregations.InternalAggregation;
 import org.elasticsearch.search.aggregations.InternalAggregations;
@@ -28,7 +29,6 @@ import org.elasticsearch.search.aggregations.bucket.range.InternalRange;
 import org.elasticsearch.search.aggregations.pipeline.PipelineAggregator;
 import org.elasticsearch.search.aggregations.support.ValueType;
 import org.elasticsearch.search.aggregations.support.ValuesSourceType;
-import org.elasticsearch.search.aggregations.support.format.ValueFormatter;
 
 import java.io.IOException;
 import java.util.List;
@@ -53,7 +53,7 @@ public class InternalGeoDistance extends InternalRange<InternalGeoDistance.Bucke
     private final static BucketStreams.Stream<Bucket> BUCKET_STREAM = new BucketStreams.Stream<Bucket>() {
         @Override
         public Bucket readResult(StreamInput in, BucketStreamContext context) throws IOException {
-            Bucket buckets = new Bucket(context.keyed(), context.formatter());
+            Bucket buckets = new Bucket(context.keyed());
             buckets.readFrom(in);
             return buckets;
         }
@@ -61,7 +61,7 @@ public class InternalGeoDistance extends InternalRange<InternalGeoDistance.Bucke
         @Override
         public BucketStreamContext getBucketStreamContext(Bucket bucket) {
             BucketStreamContext context = new BucketStreamContext();
-            context.formatter(bucket.formatter());
+            context.format(DocValueFormat.RAW);
             context.keyed(bucket.keyed());
             return context;
         }
@@ -76,17 +76,16 @@ public class InternalGeoDistance extends InternalRange<InternalGeoDistance.Bucke
 
     static class Bucket extends InternalRange.Bucket {
 
-        Bucket(boolean keyed, ValueFormatter formatter) {
-            super(keyed, formatter);
+        Bucket(boolean keyed) {
+            super(keyed, DocValueFormat.RAW);
         }
 
-        Bucket(String key, double from, double to, long docCount, List<InternalAggregation> aggregations, boolean keyed,
-                ValueFormatter formatter) {
-            this(key, from, to, docCount, new InternalAggregations(aggregations), keyed, formatter);
+        Bucket(String key, double from, double to, long docCount, List<InternalAggregation> aggregations, boolean keyed) {
+            this(key, from, to, docCount, new InternalAggregations(aggregations), keyed);
         }
 
-        Bucket(String key, double from, double to, long docCount, InternalAggregations aggregations, boolean keyed, ValueFormatter formatter) {
-            super(key, from, to, docCount, aggregations, keyed, formatter);
+        Bucket(String key, double from, double to, long docCount, InternalAggregations aggregations, boolean keyed) {
+            super(key, from, to, docCount, aggregations, keyed, DocValueFormat.RAW);
         }
 
         @Override
@@ -97,10 +96,6 @@ public class InternalGeoDistance extends InternalRange<InternalGeoDistance.Bucke
         boolean keyed() {
             return keyed;
         }
-
-        ValueFormatter formatter() {
-            return formatter;
-        }
     }
 
     public static class Factory extends InternalRange.Factory<InternalGeoDistance.Bucket, InternalGeoDistance> {
@@ -121,36 +116,36 @@ public class InternalGeoDistance extends InternalRange<InternalGeoDistance.Bucke
         }
 
         @Override
-        public InternalGeoDistance create(String name, List<Bucket> ranges, ValueFormatter formatter, boolean keyed,
+        public InternalGeoDistance create(String name, List<Bucket> ranges, DocValueFormat format, boolean keyed,
                 List<PipelineAggregator> pipelineAggregators, Map<String, Object> metaData) {
-            return new InternalGeoDistance(name, ranges, formatter, keyed, pipelineAggregators, metaData);
+            return new InternalGeoDistance(name, ranges, keyed, pipelineAggregators, metaData);
         }
 
         @Override
         public InternalGeoDistance create(List<Bucket> ranges, InternalGeoDistance prototype) {
-            return new InternalGeoDistance(prototype.name, ranges, prototype.formatter, prototype.keyed, prototype.pipelineAggregators(),
+            return new InternalGeoDistance(prototype.name, ranges, prototype.keyed, prototype.pipelineAggregators(),
                     prototype.metaData);
         }
 
         @Override
         public Bucket createBucket(String key, double from, double to, long docCount, InternalAggregations aggregations, boolean keyed,
-                ValueFormatter formatter) {
-            return new Bucket(key, from, to, docCount, aggregations, keyed, formatter);
+                DocValueFormat format) {
+            return new Bucket(key, from, to, docCount, aggregations, keyed);
         }
 
         @Override
         public Bucket createBucket(InternalAggregations aggregations, Bucket prototype) {
             return new Bucket(prototype.getKey(), ((Number) prototype.getFrom()).doubleValue(), ((Number) prototype.getTo()).doubleValue(),
-                    prototype.getDocCount(), aggregations, prototype.getKeyed(), prototype.getFormatter());
+                    prototype.getDocCount(), aggregations, prototype.getKeyed());
         }
     }
 
     InternalGeoDistance() {} // for serialization
 
-    public InternalGeoDistance(String name, List<Bucket> ranges, ValueFormatter formatter, boolean keyed,
+    public InternalGeoDistance(String name, List<Bucket> ranges, boolean keyed,
             List<PipelineAggregator> pipelineAggregators,
             Map<String, Object> metaData) {
-        super(name, ranges, formatter, keyed, pipelineAggregators, metaData);
+        super(name, ranges, DocValueFormat.RAW, keyed, pipelineAggregators, metaData);
     }
 
     @Override
diff --git a/core/src/main/java/org/elasticsearch/search/aggregations/bucket/range/ipv4/IPv4RangeAggregatorBuilder.java b/core/src/main/java/org/elasticsearch/search/aggregations/bucket/range/ipv4/IPv4RangeAggregatorBuilder.java
index a8c1a69126..7c867d56fa 100644
--- a/core/src/main/java/org/elasticsearch/search/aggregations/bucket/range/ipv4/IPv4RangeAggregatorBuilder.java
+++ b/core/src/main/java/org/elasticsearch/search/aggregations/bucket/range/ipv4/IPv4RangeAggregatorBuilder.java
@@ -27,13 +27,13 @@ import org.elasticsearch.common.network.Cidrs;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.search.aggregations.AggregatorFactory;
+import org.elasticsearch.search.DocValueFormat;
 import org.elasticsearch.search.aggregations.AggregatorFactories.Builder;
 import org.elasticsearch.search.aggregations.bucket.range.AbstractRangeBuilder;
 import org.elasticsearch.search.aggregations.bucket.range.RangeAggregator;
 import org.elasticsearch.search.aggregations.support.AggregationContext;
 import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.aggregations.support.ValuesSource.Numeric;
-import org.elasticsearch.search.aggregations.support.format.ValueParser;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -171,16 +171,16 @@ public class IPv4RangeAggregatorBuilder extends AbstractRangeBuilder<IPv4RangeAg
         }
 
         @Override
-        public Range process(ValueParser parser, SearchContext context) {
+        public Range process(DocValueFormat parser, SearchContext context) {
             assert parser != null;
             Double from = this.from;
             Double to = this.to;
             String key = this.key;
             if (fromAsStr != null) {
-                from = parser.parseDouble(fromAsStr, context);
+                from = parser.parseDouble(fromAsStr, false, context.nowCallable());
             }
             if (toAsStr != null) {
-                to = parser.parseDouble(toAsStr, context);
+                to = parser.parseDouble(toAsStr, false, context.nowCallable());
             }
             if (cidr != null) {
                 long[] fromTo = Cidrs.cidrMaskToMinMax(cidr);
diff --git a/core/src/main/java/org/elasticsearch/search/aggregations/bucket/range/ipv4/InternalIPv4Range.java b/core/src/main/java/org/elasticsearch/search/aggregations/bucket/range/ipv4/InternalIPv4Range.java
index a6c3ed38b1..f58df79c16 100644
--- a/core/src/main/java/org/elasticsearch/search/aggregations/bucket/range/ipv4/InternalIPv4Range.java
+++ b/core/src/main/java/org/elasticsearch/search/aggregations/bucket/range/ipv4/InternalIPv4Range.java
@@ -19,6 +19,7 @@
 package org.elasticsearch.search.aggregations.bucket.range.ipv4;
 
 import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.search.DocValueFormat;
 import org.elasticsearch.search.aggregations.AggregationStreams;
 import org.elasticsearch.search.aggregations.InternalAggregation;
 import org.elasticsearch.search.aggregations.InternalAggregations;
@@ -27,7 +28,6 @@ import org.elasticsearch.search.aggregations.bucket.BucketStreams;
 import org.elasticsearch.search.aggregations.bucket.range.InternalRange;
 import org.elasticsearch.search.aggregations.pipeline.PipelineAggregator;
 import org.elasticsearch.search.aggregations.support.ValueType;
-import org.elasticsearch.search.aggregations.support.format.ValueFormatter;
 
 import java.io.IOException;
 import java.util.List;
@@ -76,27 +76,27 @@ public class InternalIPv4Range extends InternalRange<InternalIPv4Range.Bucket, I
     public static class Bucket extends InternalRange.Bucket {
 
         public Bucket(boolean keyed) {
-            super(keyed, ValueFormatter.IPv4);
+            super(keyed, DocValueFormat.IP);
         }
 
         public Bucket(String key, double from, double to, long docCount, List<InternalAggregation> aggregations, boolean keyed) {
-            super(key, from, to, docCount, new InternalAggregations(aggregations), keyed, ValueFormatter.IPv4);
+            super(key, from, to, docCount, new InternalAggregations(aggregations), keyed, DocValueFormat.IP);
         }
 
         public Bucket(String key, double from, double to, long docCount, InternalAggregations aggregations, boolean keyed) {
-            super(key, from, to, docCount, aggregations, keyed, ValueFormatter.IPv4);
+            super(key, from, to, docCount, aggregations, keyed, DocValueFormat.IP);
         }
 
         @Override
         public String getFromAsString() {
             double from = ((Number) this.from).doubleValue();
-            return Double.isInfinite(from) ? null : from == 0 ? null : ValueFormatter.IPv4.format(from);
+            return Double.isInfinite(from) ? null : from == 0 ? null : DocValueFormat.IP.format(from);
         }
 
         @Override
         public String getToAsString() {
             double to = ((Number) this.to).doubleValue();
-            return Double.isInfinite(to) ? null : MAX_IP == to ? null : ValueFormatter.IPv4.format(to);
+            return Double.isInfinite(to) ? null : MAX_IP == to ? null : DocValueFormat.IP.format(to);
         }
 
         @Override
@@ -122,7 +122,7 @@ public class InternalIPv4Range extends InternalRange<InternalIPv4Range.Bucket, I
         }
 
         @Override
-        public InternalIPv4Range create(String name, List<Bucket> ranges, ValueFormatter formatter, boolean keyed,
+        public InternalIPv4Range create(String name, List<Bucket> ranges, DocValueFormat formatter, boolean keyed,
                 List<PipelineAggregator> pipelineAggregators, Map<String, Object> metaData) {
             return new InternalIPv4Range(name, ranges, keyed, pipelineAggregators, metaData);
         }
@@ -134,7 +134,7 @@ public class InternalIPv4Range extends InternalRange<InternalIPv4Range.Bucket, I
 
         @Override
         public Bucket createBucket(String key, double from, double to, long docCount, InternalAggregations aggregations, boolean keyed,
-                ValueFormatter formatter) {
+                DocValueFormat formatter) {
             return new Bucket(key, from, to, docCount, aggregations, keyed);
         }
 
@@ -149,7 +149,7 @@ public class InternalIPv4Range extends InternalRange<InternalIPv4Range.Bucket, I
 
     public InternalIPv4Range(String name, List<InternalIPv4Range.Bucket> ranges, boolean keyed, List<PipelineAggregator> pipelineAggregators,
             Map<String, Object> metaData) {
-        super(name, ranges, ValueFormatter.IPv4, keyed, pipelineAggregators, metaData);
+        super(name, ranges, DocValueFormat.IP, keyed, pipelineAggregators, metaData);
     }
 
     @Override