diff --git a/heroic-component/src/main/java/com/spotify/heroic/Query.kt b/heroic-component/src/main/java/com/spotify/heroic/Query.kt
index ecda0b175..1274feec3 100644
--- a/heroic-component/src/main/java/com/spotify/heroic/Query.kt
+++ b/heroic-component/src/main/java/com/spotify/heroic/Query.kt
@@ -29,10 +29,9 @@ import java.util.*
 
 data class Query(
     val aggregation: Optional<Aggregation>,
-    val source: Optional<MetricType>,
+    val metricType: Optional<MetricType>, // points or distribution points for testing
     val range: Optional<QueryDateRange>,
     val filter: Optional<Filter>,
     val options: Optional<QueryOptions>,
-    // set of experimental features to enable
-    val features: Optional<FeatureSet>
+    val features: Optional<FeatureSet> // set of experimental features to enable
 )
diff --git a/heroic-component/src/main/java/com/spotify/heroic/QueryBuilder.java b/heroic-component/src/main/java/com/spotify/heroic/QueryBuilder.java
index 6693edca4..c3a5752c5 100644
--- a/heroic-component/src/main/java/com/spotify/heroic/QueryBuilder.java
+++ b/heroic-component/src/main/java/com/spotify/heroic/QueryBuilder.java
@@ -40,11 +40,11 @@
 import org.slf4j.LoggerFactory;
 
 public class QueryBuilder {
-    private Optional<MetricType> source = Optional.empty();
     private Optional<Map<String, String>> tags = Optional.empty();
     private Optional<String> key = Optional.empty();
     private Optional<Filter> filter = Optional.empty();
     private Optional<QueryDateRange> range = Optional.empty();
+    private Optional<MetricType> metricType = Optional.empty();
     private Optional<Aggregation> aggregation = Optional.empty();
     private Optional<QueryOptions> options = Optional.empty();
     private Optional<JsonNode> clientContext = Optional.empty();
@@ -97,6 +97,15 @@ public QueryBuilder filter(final Optional<Filter> filter) {
         return this;
     }
 
+    /**
+     * Specify a metricType to use, if none are specified it will be determined in CoreQueryManager.
+     */
+    public QueryBuilder metricType(final Optional<MetricType> metricType) {
+        //checkNotNull(metricType, "metricType must not be null");
+        this.metricType = pickOptional(this.metricType, metricType);
+        return this;
+    }
+
     /**
      * Specify an aggregation to use.
      */
@@ -106,11 +115,6 @@ public QueryBuilder aggregation(final Optional<Aggregation> aggregation) {
         return this;
     }
 
-    public QueryBuilder source(Optional<MetricType> source) {
-        this.source = source;
-        return this;
-    }
-
     public QueryBuilder options(final Optional<QueryOptions> options) {
         checkNotNull(options, "options");
         this.options = pickOptional(this.options, options);
@@ -146,7 +150,7 @@ public QueryBuilder features(final Optional<FeatureSet> features) {
     }
 
     public Query build() {
-        return new Query(aggregation, source, range, legacyFilter(), options, features);
+        return new Query(aggregation, metricType, range, legacyFilter(), options, features);
     }
 
 
diff --git a/heroic-component/src/main/java/com/spotify/heroic/grammar/QueryExpression.kt b/heroic-component/src/main/java/com/spotify/heroic/grammar/QueryExpression.kt
index 7bd3c5f39..d8a559400 100644
--- a/heroic-component/src/main/java/com/spotify/heroic/grammar/QueryExpression.kt
+++ b/heroic-component/src/main/java/com/spotify/heroic/grammar/QueryExpression.kt
@@ -29,7 +29,7 @@ import java.util.*
 data class QueryExpression(
     @JvmField val context: Context,
     val select: Optional<Expression>,
-    val source: Optional<MetricType>,
+    val metricType: Optional<MetricType>,
     val range: Optional<RangeExpression>,
     val filter: Optional<Filter>,
     val with: Map<String, Expression>,
@@ -41,7 +41,7 @@ data class QueryExpression(
         QueryExpression(
             context,
             select.map { it.eval(scope) },
-            source,
+            metricType,
             range.map { it.eval(scope) },
             filter,
             Expression.evalMap(with, scope),
@@ -52,6 +52,6 @@ data class QueryExpression(
         return visitor.visitQuery(this)
     }
 
-    override fun toRepr() = """{select: $select source: $source, 
+    override fun toRepr() = """{select: $select metricType: $metricType, 
         range: $range, filter: $filter, with: $with, as: $asExpression}""".trimIndent()
 }
diff --git a/heroic-component/src/main/java/com/spotify/heroic/metric/FetchData.kt b/heroic-component/src/main/java/com/spotify/heroic/metric/FetchData.kt
index 76c39bd5b..46974d3f3 100644
--- a/heroic-component/src/main/java/com/spotify/heroic/metric/FetchData.kt
+++ b/heroic-component/src/main/java/com/spotify/heroic/metric/FetchData.kt
@@ -47,7 +47,7 @@ data class FetchData(
     }
 
     data class Request(
-        val type: MetricType,
+        val metricType: MetricType,
         val series: Series,
         val range: DateRange,
         val options: QueryOptions
diff --git a/heroic-component/src/main/java/com/spotify/heroic/metric/FullQuery.java b/heroic-component/src/main/java/com/spotify/heroic/metric/FullQuery.java
index 7cfcc6cc6..3517b6ee3 100644
--- a/heroic-component/src/main/java/com/spotify/heroic/metric/FullQuery.java
+++ b/heroic-component/src/main/java/com/spotify/heroic/metric/FullQuery.java
@@ -158,20 +158,19 @@ static Summary create(
     public abstract static class Request {
         @JsonCreator
         public static Request create(
-            @JsonProperty("source") MetricType source,
             @JsonProperty("filter") Filter filter,
             @JsonProperty("range") DateRange range,
             @JsonProperty("aggregation") AggregationInstance aggregation,
+            @JsonProperty("metricType") MetricType metricType,
             @JsonProperty("options") QueryOptions options,
             @JsonProperty("context") QueryContext context,
             @JsonProperty("features") Features features
         ) {
             return new AutoValue_FullQuery_Request(
-                source, filter, range, aggregation, options, context, features);
+              filter, range, aggregation, metricType, options, context, features);
         }
 
-        @JsonProperty
-        public abstract MetricType source();
+
         @JsonProperty
         public abstract Filter filter();
         @JsonProperty
@@ -179,6 +178,8 @@ public static Request create(
         @JsonProperty
         public abstract AggregationInstance aggregation();
         @JsonProperty
+        public abstract MetricType metricType();
+        @JsonProperty
         public abstract QueryOptions options();
         @JsonProperty
         public abstract QueryContext context();
@@ -186,16 +187,16 @@ public static Request create(
         public abstract Features features();
 
         public Summary summarize() {
-            return Summary.create(source(), filter(), range(), aggregation(), options());
+            return Summary.create(filter(), range(), aggregation(), metricType(), options());
         }
 
         public void hashTo(final ObjectHasher hasher) {
             hasher.putObject(getClass(), () -> {
-                hasher.putField("source", source(), hasher.enumValue());
                 hasher.putField("filter", filter(), hasher.with(Filter::hashTo));
                 hasher.putField("range", range(), hasher.with(DateRange::hashTo));
                 hasher.putField("aggregation", aggregation(),
                     hasher.with(AggregationInstance::hashTo));
+                hasher.putField("metricType", metricType(), hasher.enumValue());
                 hasher.putField("options", options(), hasher.with(QueryOptions::hashTo));
                 hasher.putField("features", features(), hasher.with(Features::hashTo));
             });
@@ -205,18 +206,16 @@ public void hashTo(final ObjectHasher hasher) {
         public abstract static class Summary {
             @JsonCreator
             public static Summary create(
-                @JsonProperty("source") MetricType source,
                 @JsonProperty("filter") Filter filter,
                 @JsonProperty("range") DateRange range,
                 @JsonProperty("aggregation") AggregationInstance aggregation,
+                @JsonProperty("metricType") MetricType metricType,
                 @JsonProperty("options") QueryOptions options
             ) {
                 return new AutoValue_FullQuery_Request_Summary(
-                    source, filter, range, aggregation, options);
+                    filter, range, aggregation, metricType, options);
             }
 
-            @JsonProperty
-            abstract MetricType source();
             @JsonProperty
             abstract Filter filter();
             @JsonProperty
@@ -224,6 +223,8 @@ public static Summary create(
             @JsonProperty
             abstract AggregationInstance aggregation();
             @JsonProperty
+            abstract MetricType metricType();
+            @JsonProperty
             abstract QueryOptions options();
         }
     }
diff --git a/heroic-component/src/main/java/com/spotify/heroic/metric/MetricBackend.java b/heroic-component/src/main/java/com/spotify/heroic/metric/MetricBackend.java
index 0be1bfbee..49ac1bde2 100644
--- a/heroic-component/src/main/java/com/spotify/heroic/metric/MetricBackend.java
+++ b/heroic-component/src/main/java/com/spotify/heroic/metric/MetricBackend.java
@@ -73,6 +73,7 @@ AsyncFuture<FetchData.Result> fetch(
         FetchQuotaWatcher watcher,
         Consumer<MetricReadResult> metricsConsumer,
         Span parentSpan
+
     );
 
     /**
diff --git a/heroic-component/src/main/java/com/spotify/heroic/metric/MetricType.java b/heroic-component/src/main/java/com/spotify/heroic/metric/MetricType.java
index eeff260a8..ce6a7f85e 100644
--- a/heroic-component/src/main/java/com/spotify/heroic/metric/MetricType.java
+++ b/heroic-component/src/main/java/com/spotify/heroic/metric/MetricType.java
@@ -65,7 +65,7 @@ public String identifier() {
         .stream(MetricType.values())
         .collect(Collectors.toMap(MetricType::identifier, Function.identity())));
 
-    public static Optional<MetricType> fromIdentifier(String identifier) {
+    public static Optional<MetricType> fromIdentifier(Optional<String> identifier) {
         return Optional.ofNullable(mapping.get(identifier));
     }
 
diff --git a/heroic-component/src/main/java/com/spotify/heroic/metric/QueryMetrics.java b/heroic-component/src/main/java/com/spotify/heroic/metric/QueryMetrics.java
index c48ed9b62..b0deb113e 100644
--- a/heroic-component/src/main/java/com/spotify/heroic/metric/QueryMetrics.java
+++ b/heroic-component/src/main/java/com/spotify/heroic/metric/QueryMetrics.java
@@ -45,13 +45,13 @@ public abstract class QueryMetrics {
     public static QueryMetrics create(
         Optional<String> query,
         Optional<Aggregation> aggregation,
-        Optional<String> source,
+        Optional<MetricType> metricType,
         Optional<QueryDateRange> range,
         Optional<Filter> filter,
         Optional<QueryOptions> options,
         Optional<JsonNode> clientContext
     ) {
-        return legacyCreate(query, aggregation, source, range, filter, options, clientContext,
+        return legacyCreate(query, aggregation, metricType, range, filter, options, clientContext,
              Optional.empty(), Optional.empty(), Optional.empty(),
             Optional.empty(), false);
     }
@@ -60,7 +60,7 @@ public static QueryMetrics create(
     public static QueryMetrics legacyCreate(
         @JsonProperty("query") Optional<String> query,
         @JsonProperty("aggregation") Optional<Aggregation> aggregation,
-        @JsonProperty("source") Optional<String> source,
+        @JsonProperty("metricType") Optional<MetricType> metricType,
         @JsonProperty("range") Optional<QueryDateRange> range,
         @JsonProperty("filter") Optional<Filter> filter,
         @JsonProperty("options") Optional<QueryOptions> options,
@@ -75,9 +75,10 @@ public static QueryMetrics legacyCreate(
 
         final Optional<Aggregation> legitAggregation = firstPresent(aggregation,
             aggregators.filter(c -> !c.isEmpty()).map(Chain::fromList));
-        final Optional<MetricType> sourceMetric = source.flatMap(MetricType::fromIdentifier);
 
-        return new AutoValue_QueryMetrics(query, legitAggregation, sourceMetric, range, filter,
+        //final Optional<MetricType> sourceMetric = metricType;
+
+        return new AutoValue_QueryMetrics(query, legitAggregation, metricType, range, filter,
             options, clientContext, key, tags, features);
     }
 
@@ -85,8 +86,8 @@ public static QueryMetrics legacyCreate(
     public abstract Optional<String> query();
     @JsonProperty("aggregation")
     public abstract Optional<Aggregation> aggregation();
-    @JsonProperty("source")
-    public abstract Optional<MetricType> source();
+    @JsonProperty("metricType")
+    public abstract Optional<MetricType> metricType();
     @JsonProperty("range")
     public abstract Optional<QueryDateRange> range();
     @JsonProperty("filter")
@@ -111,7 +112,7 @@ public QueryBuilder toQueryBuilder(final Function<String, QueryBuilder> stringTo
             .filter(filter())
             .range(range())
             .aggregation(aggregation())
-            .source(source())
+            .metricType(metricType())
             .options(options())
             .clientContext(clientContext());
 
diff --git a/heroic-component/src/test/java/com/spotify/heroic/grammar/QueryExpressionTest.java b/heroic-component/src/test/java/com/spotify/heroic/grammar/QueryExpressionTest.java
index 9c4c101be..575d7db3a 100644
--- a/heroic-component/src/test/java/com/spotify/heroic/grammar/QueryExpressionTest.java
+++ b/heroic-component/src/test/java/com/spotify/heroic/grammar/QueryExpressionTest.java
@@ -36,7 +36,7 @@ public void testAccessors() {
         final QueryExpression e = build();
 
         assertEquals(select, e.getSelect());
-        assertEquals(source, e.getSource());
+        assertEquals(source, e.getMetricType());
         assertEquals(range, e.getRange());
         assertEquals(filter, e.getFilter());
         assertEquals(with, e.getWith());
diff --git a/heroic-core/build.gradle b/heroic-core/build.gradle
index 374cb54af..3aa460e70 100644
--- a/heroic-core/build.gradle
+++ b/heroic-core/build.gradle
@@ -57,6 +57,7 @@ dependencies {
     // Default usage tracking module. Normally different module dependencies are handled in
     // heroic-dist, but defaults need to be accessible in heroic-core.
     implementation project(':heroic-usage-tracking-google-analytics')
+    implementation project(':heroic-aggregation-simple')
     
     testImplementation project(':heroic-test')
     testImplementation project(path: ':heroic-component', configuration: 'testRuntime')
diff --git a/heroic-core/src/main/java/com/spotify/heroic/CoreQueryManager.java b/heroic-core/src/main/java/com/spotify/heroic/CoreQueryManager.java
index 8fd916e26..f577d2f0b 100644
--- a/heroic-core/src/main/java/com/spotify/heroic/CoreQueryManager.java
+++ b/heroic-core/src/main/java/com/spotify/heroic/CoreQueryManager.java
@@ -35,7 +35,9 @@
 import com.spotify.heroic.aggregation.BucketStrategy;
 import com.spotify.heroic.aggregation.DistributedAggregationCombiner;
 import com.spotify.heroic.aggregation.Empty;
+import com.spotify.heroic.aggregation.GroupInstance;
 import com.spotify.heroic.aggregation.TDigestAggregationCombiner;
+import com.spotify.heroic.aggregation.simple.TdigestInstance;
 import com.spotify.heroic.cache.QueryCache;
 import com.spotify.heroic.cluster.ClusterManager;
 import com.spotify.heroic.cluster.ClusterNode;
@@ -100,6 +102,7 @@
 import org.slf4j.LoggerFactory;
 
 public class CoreQueryManager implements QueryManager {
+
     private static final Logger log = LoggerFactory.getLogger(CoreQueryManager.class);
     public static final long SHIFT_TOLERANCE = TimeUnit.MILLISECONDS.convert(10, TimeUnit.SECONDS);
     public static final QueryTrace.Identifier QUERY_SHARD =
@@ -171,7 +174,7 @@ public QueryBuilder newQueryFromString(final String queryString) {
         return expressions.get(0).eval(scope).visit(new Expression.Visitor<QueryBuilder>() {
             @Override
             public QueryBuilder visitQuery(final QueryExpression e) {
-                final Optional<MetricType> source = e.getSource();
+                final Optional<MetricType> metricType = e.getMetricType();
 
                 final Optional<QueryDateRange> range =
                     e.getRange().map(expr -> expr.visit(new Expression.Visitor<QueryDateRange>() {
@@ -202,15 +205,16 @@ public Aggregation visitString(final StringExpression e) {
                 final Optional<Filter> filter = e.getFilter();
 
                 return new QueryBuilder()
-                    .source(source)
                     .range(range)
                     .aggregation(aggregation)
+                    .metricType(metricType)
                     .filter(filter);
             }
         });
     }
 
     public class Group implements QueryManager.Group {
+
         private final List<ClusterShard> shards;
 
         public Group(List<ClusterShard> shards) {
@@ -237,10 +241,10 @@ public AsyncFuture<QueryResult> query(
 
             final List<AsyncFuture<QueryResultPart>> futures = new ArrayList<>();
 
-            final MetricType source = q.getSource().orElse(MetricType.POINT);
-
             final Aggregation aggregation = q.getAggregation().orElse(Empty.INSTANCE);
 
+            MetricType metricType = q.getMetricType().orElse(MetricType.POINT);
+
             final DateRange rawRange = buildRange(q);
 
             final Filter filter = q.getFilter().orElseGet(TrueFilter::get);
@@ -252,12 +256,15 @@ public AsyncFuture<QueryResult> query(
 
             final AggregationInstance aggregationInstance;
 
+            final AggregationCombiner combiner;
+
             final Features features = requestFeatures(q, queryContext);
 
             boolean isDistributed = features.hasFeature(Feature.DISTRIBUTED_AGGREGATIONS);
 
             if (isDistributed) {
                 aggregationInstance = root.distributed();
+
             } else {
                 aggregationInstance = root;
             }
@@ -278,20 +285,24 @@ public AsyncFuture<QueryResult> query(
                 .orElseGet(() -> features.withFeature(Feature.END_BUCKET, () -> BucketStrategy.END,
                     () -> BucketStrategy.START));
 
-            final AggregationCombiner combiner;
-
+            if (aggregationInstance instanceof GroupInstance) {
+                if (((GroupInstance) aggregationInstance).getEach() instanceof TdigestInstance) {
+                    metricType = MetricType.DISTRIBUTION_POINTS;
+                }
+            }
 
             if (isDistributed) {
-                combiner = (source.equals(MetricType.POINT)) ?
-                    DistributedAggregationCombiner.create(root, range, bucketStrategy) :
-                    TDigestAggregationCombiner.create(root, range, bucketStrategy);
+                combiner = (metricType.equals(MetricType.POINT)) ?
+                           DistributedAggregationCombiner.create(root, range, bucketStrategy) :
+                           TDigestAggregationCombiner.create(root, range, bucketStrategy);
             } else {
-                combiner = (source.equals(MetricType.DISTRIBUTION_POINTS)) ?
-                    AggregationCombiner.TDIGEST_DEFAULT : AggregationCombiner.DEFAULT;
+                combiner = (metricType.equals(MetricType.DISTRIBUTION_POINTS)) ?
+                           AggregationCombiner.TDIGEST_DEFAULT :
+                           AggregationCombiner.DEFAULT;
             }
 
             final FullQuery.Request request =
-                FullQuery.Request.create(source, filter, range, aggregationInstance, options,
+                FullQuery.Request.create(filter, range, aggregationInstance, metricType, options,
                     queryContext, features);
 
             queryLogger.logOutgoingRequestToShards(queryContext, request);
@@ -334,21 +345,20 @@ public AsyncFuture<QueryResult> query(
                     QueryResult.collectParts(QUERY, range, combiner, limit));
             });
 
+            return query
+                .directTransform(result -> {
+                    reportCompletedQuery(result, fullQueryWatch);
+                    if (result.getErrors().size() > 0) {
+                        queryManagerSpan.addAnnotation(result.getErrors().toString());
+                        queryManagerSpan.putAttribute("error", booleanAttributeValue(true));
+                    }
+                    queryManagerSpan.putAttribute("preAggregationSampleSize",
+                        longAttributeValue(result.getPreAggregationSampleSize()));
 
-          return query
-              .directTransform(result -> {
-                  reportCompletedQuery(result, fullQueryWatch);
-                  if (result.getErrors().size() > 0) {
-                      queryManagerSpan.addAnnotation(result.getErrors().toString());
-                      queryManagerSpan.putAttribute("error", booleanAttributeValue(true));
-                  }
-                  queryManagerSpan.putAttribute("preAggregationSampleSize",
-                      longAttributeValue(result.getPreAggregationSampleSize()));
-
-                  return result;
-              })
-              .onDone(reporter.reportQuery())
-              .onDone(new EndSpanFutureReporter(queryManagerSpan));
+                    return result;
+                })
+                .onDone(reporter.reportQuery())
+                .onDone(new EndSpanFutureReporter(queryManagerSpan));
         }
 
         private void reportCompletedQuery(
@@ -500,7 +510,7 @@ private Features requestFeatures(final Query q, final QueryContext context) {
                 hasWarnedSlicedDataFetch = true;
                 log.warn(
                     "The mandatory feature 'com.spotify.heroic.sliced_data_fetch' can't be " +
-                        "disabled!");
+                    "disabled!");
             }
         }
 
@@ -560,6 +570,7 @@ private Duration cadenceFromRange(final DateRange range) {
      * too close or after 'now'. This is useful to avoid querying non-complete buckets.
      *
      * @param rawRange Original range.
+     *
      * @return A possibly shifted range.
      */
     DateRange buildShiftedRange(DateRange rawRange, long cadence, long now) {
@@ -584,10 +595,11 @@ DateRange buildShiftedRange(DateRange rawRange, long cadence, long now) {
      * Calculate a tolerance shift period that corresponds to the given difference that needs to be
      * applied to the range to honor the tolerance shift period.
      *
-     * @param diff The time difference to apply.
+     * @param diff    The time difference to apply.
      * @param cadence The cadence period.
+     *
      * @return The number of milliseconds that the query should be shifted to get within 'now' and
-     * maintain the given cadence.
+     *     maintain the given cadence.
      */
     private long toleranceShiftPeriod(final long diff, final long cadence) {
         // raw query, only shift so that we are within now.
diff --git a/heroic-core/src/main/java/com/spotify/heroic/metric/LocalMetricManager.java b/heroic-core/src/main/java/com/spotify/heroic/metric/LocalMetricManager.java
index b43a162af..e8a34eca4 100644
--- a/heroic-core/src/main/java/com/spotify/heroic/metric/LocalMetricManager.java
+++ b/heroic-core/src/main/java/com/spotify/heroic/metric/LocalMetricManager.java
@@ -191,6 +191,7 @@ public String toString() {
         private class Transform implements LazyTransform<FindSeries, FullQuery> {
 
             private final AggregationInstance aggregation;
+            private final MetricType metricType;
             private final boolean failOnLimits;
             private final OptionalLimit seriesLimit;
             private final OptionalLimit groupLimit;
@@ -201,7 +202,6 @@ private class Transform implements LazyTransform<FindSeries, FullQuery> {
             private final QueryOptions options;
             private final DataInMemoryReporter dataInMemoryReporter;
             private final Span parentSpan;
-            private final MetricType source;
 
             private Transform(
                 final FullQuery.Request request,
@@ -213,9 +213,9 @@ private Transform(
                 final Span parentSpan
             ) {
                 this.aggregation = request.aggregation();
+                this.metricType = request.metricType();
                 this.range = request.range();
                 this.options = request.options();
-                this.source = request.source();
 
                 this.failOnLimits = failOnLimits;
                 this.seriesLimit = seriesLimit;
@@ -322,7 +322,7 @@ public QueryTrace buildTrace() {
 
                         fetchSeries.addAnnotation(series.toString());
                         fetches.add(() -> metricBackend.fetch(
-                            new FetchData.Request(source, series, range, options),
+                            new FetchData.Request(metricType, series, range, options),
                             quotaWatcher,
                             mcr -> collector.acceptMetricsCollection(series, mcr),
                             fetchSeries
diff --git a/heroic-core/src/main/java/com/spotify/heroic/shell/task/Fetch.java b/heroic-core/src/main/java/com/spotify/heroic/shell/task/Fetch.java
index 285a7ca7e..56f545ed0 100644
--- a/heroic-core/src/main/java/com/spotify/heroic/shell/task/Fetch.java
+++ b/heroic-core/src/main/java/com/spotify/heroic/shell/task/Fetch.java
@@ -100,7 +100,9 @@ public AsyncFuture<Void> run(final ShellIO io, final TaskParameters base) throws
         final DateFormat point = new SimpleDateFormat("HH:mm:ss.SSS");
 
         final MetricBackendGroup readGroup = metrics.useOptionalGroup(params.group);
-        final MetricType source = MetricType.fromIdentifier(params.source).orElse(MetricType.POINT);
+
+        final MetricType metricType =
+            MetricType.fromIdentifier(params.metricType).orElse(MetricType.POINT);
 
         final QueryOptions.Builder optionsBuilder =
             QueryOptions.builder().tracing(Tracing.fromBoolean(params.tracing));
@@ -135,7 +137,7 @@ public AsyncFuture<Void> run(final ShellIO io, final TaskParameters base) throws
         };
 
         return readGroup
-            .fetch(new FetchData.Request(source, series, range, options),
+            .fetch(new FetchData.Request(metricType, series, range, options),
                 FetchQuotaWatcher.NO_QUOTA, printMetricsCollection, BlankSpan.INSTANCE)
             .lazyTransform(handleResult);
     }
@@ -172,9 +174,9 @@ private static class Parameters extends AbstractShellTaskParams {
         @Option(name = "-s", aliases = {"--series"}, usage = "Series to fetch", metaVar = "<json>")
         private Optional<String> series = Optional.empty();
 
-        @Option(name = "--source", aliases = {"--source"}, usage = "Source to fetch",
-            metaVar = "<events|points>")
-        private String source = MetricType.POINT.identifier();
+        @Option(name = "--metricType", aliases = {"--metricType"}, usage = "MetricType to fetch",
+            metaVar = "<points|distributionPoints>")
+        private Optional<String> metricType = Optional.empty();
 
         @Option(name = "--start", usage = "Start date", metaVar = "<datetime>")
         private Optional<String> start = Optional.empty();
diff --git a/heroic-dist/src/test/java/com/spotify/heroic/AbstractClusterQueryIT.java b/heroic-dist/src/test/java/com/spotify/heroic/AbstractClusterQueryIT.java
index 57f7cde60..555f5b2f0 100644
--- a/heroic-dist/src/test/java/com/spotify/heroic/AbstractClusterQueryIT.java
+++ b/heroic-dist/src/test/java/com/spotify/heroic/AbstractClusterQueryIT.java
@@ -1,7 +1,5 @@
 package com.spotify.heroic;
 
-
-
 import static com.spotify.heroic.test.Matchers.containsChild;
 import static com.spotify.heroic.test.Matchers.hasIdentifier;
 import static com.spotify.heroic.test.Matchers.identifierContains;
@@ -164,7 +162,9 @@ protected AsyncFuture<Void> prepareEnvironment() {
 
     public QueryResult query(final String queryString) throws Exception {
         return query(query.newQueryFromString(queryString), builder -> {
-        }, MetricType.POINT, true);
+        },
+        MetricType.POINT,
+        true);
     }
 
     public QueryResult query(final String queryString,
@@ -184,10 +184,10 @@ public QueryResult query(final QueryBuilder builder,
         queryCount += 1;
 
         builder
-            .source(Optional.of(metricType))
+            .metricType(Optional.of(metricType))
             .rangeIfAbsent(Optional.of(new QueryDateRange.Absolute(0, 40)));
 
-        if ( isDistributed) {
+        if (isDistributed) {
             builder
                 .features(Optional.of(FeatureSet.of(Feature.DISTRIBUTED_AGGREGATIONS)));
         }
diff --git a/heroic-dist/src/test/java/com/spotify/heroic/AbstractConsumerIT.java b/heroic-dist/src/test/java/com/spotify/heroic/AbstractConsumerIT.java
index 59aab92f0..e89a57311 100644
--- a/heroic-dist/src/test/java/com/spotify/heroic/AbstractConsumerIT.java
+++ b/heroic-dist/src/test/java/com/spotify/heroic/AbstractConsumerIT.java
@@ -5,7 +5,6 @@
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
-import com.google.protobuf.ByteString;
 import com.spotify.heroic.common.DateRange;
 import com.spotify.heroic.common.Series;
 import com.spotify.heroic.consumer.schemas.spotify100.Version;
@@ -70,7 +69,8 @@ public void consumeOneV2Message() throws Exception {
 
             instance.inject(coreComponent -> {
                 FetchData.Request fetchDataRequest =
-                    new FetchData.Request(MetricType.DISTRIBUTION_POINTS, s1, new DateRange(0, 100),
+                    new FetchData.Request(MetricType.DISTRIBUTION_POINTS, s1, new DateRange(0,
+                     100),
                         QueryOptions.defaults());
                 return coreComponent
                     .metricManager()
diff --git a/metric/bigtable/src/main/java/com/spotify/heroic/metric/bigtable/BigtableBackend.java b/metric/bigtable/src/main/java/com/spotify/heroic/metric/bigtable/BigtableBackend.java
index 84d07453d..60cac2e47 100644
--- a/metric/bigtable/src/main/java/com/spotify/heroic/metric/bigtable/BigtableBackend.java
+++ b/metric/bigtable/src/main/java/com/spotify/heroic/metric/bigtable/BigtableBackend.java
@@ -271,7 +271,7 @@ public AsyncFuture<FetchData.Result> fetch(
         final Span parentSpan
     ) {
         return connection.doto(c -> {
-            final MetricType type = request.getType();
+            final MetricType type = request.getMetricType();
 
             if (!watcher.mayReadData()) {
                 throw new IllegalArgumentException("query violated data limit");
@@ -286,7 +286,7 @@ public AsyncFuture<FetchData.Result> fetch(
                         consumer, parentSpan);
                 default:
                     return async.resolved(new FetchData.Result(QueryTrace.of(FETCH),
-                        new QueryError("unsupported source: " + request.getType())));
+                        new QueryError("unsupported source: " + request.getMetricType())));
             }
         });
     }
diff --git a/metric/datastax/src/main/java/com/spotify/heroic/metric/datastax/DatastaxBackend.java b/metric/datastax/src/main/java/com/spotify/heroic/metric/datastax/DatastaxBackend.java
index 8a814694d..c6da8e33e 100644
--- a/metric/datastax/src/main/java/com/spotify/heroic/metric/datastax/DatastaxBackend.java
+++ b/metric/datastax/src/main/java/com/spotify/heroic/metric/datastax/DatastaxBackend.java
@@ -151,7 +151,7 @@ public AsyncFuture<FetchData.Result> fetch(
             final List<PreparedFetch> prepared =
                 c.schema.ranges(request.getSeries(), request.getRange());
 
-            if (request.getType() == MetricType.POINT) {
+            if (request.getMetricType() == MetricType.POINT) {
                 final List<AsyncFuture<FetchData>> fetches =
                     fetchDataPoints(w, limit, request.getOptions(), prepared, c);
 
@@ -166,7 +166,7 @@ public AsyncFuture<FetchData.Result> fetch(
             }
 
             return async.resolved(new FetchData.Result(w.end(FETCH),
-                new QueryError("unsupported source: " + request.getType())));
+                new QueryError("unsupported source: " + request.getMetricType())));
         });
     }
 
diff --git a/metric/memory/src/main/java/com/spotify/heroic/metric/memory/MemoryBackend.java b/metric/memory/src/main/java/com/spotify/heroic/metric/memory/MemoryBackend.java
index 3f57b39e4..a347fd2de 100644
--- a/metric/memory/src/main/java/com/spotify/heroic/metric/memory/MemoryBackend.java
+++ b/metric/memory/src/main/java/com/spotify/heroic/metric/memory/MemoryBackend.java
@@ -137,7 +137,7 @@ public AsyncFuture<FetchData.Result> fetch(
         Span parentSpan
     ) {
         final QueryTrace.NamedWatch w = QueryTrace.watch(FETCH);
-        final MemoryKey key = new MemoryKey(request.getType(), request.getSeries().getTags());
+        final MemoryKey key = new MemoryKey(request.getMetricType(), request.getSeries().getTags());
         doFetch(key, request.getRange(), watcher, metricsConsumer);
         return async.resolved(new FetchData.Result(w.end()));
     }
diff --git a/statistics/semantic/src/main/java/com/spotify/heroic/statistics/semantic/SemanticMetricBackendReporter.java b/statistics/semantic/src/main/java/com/spotify/heroic/statistics/semantic/SemanticMetricBackendReporter.java
index 764edce32..abea39407 100644
--- a/statistics/semantic/src/main/java/com/spotify/heroic/statistics/semantic/SemanticMetricBackendReporter.java
+++ b/statistics/semantic/src/main/java/com/spotify/heroic/statistics/semantic/SemanticMetricBackendReporter.java
@@ -269,7 +269,7 @@ public AsyncFuture<FetchData.Result> fetch(
             final FetchQuotaWatcher watcher,
             final Consumer<MetricReadResult> metricsConsumer,
             final Span parentSpan
-        ) {
+            ) {
             return delegate.fetch(
                 request, watcher, metricsConsumer, parentSpan).onDone(fetch.setup());
         }