From 234a761645cd7e3e6d1e4f4e6abbca5e070ccc37 Mon Sep 17 00:00:00 2001 From: leventov Date: Wed, 29 Mar 2017 00:01:41 -0600 Subject: [PATCH 1/7] Add queryMetrics property to Query interface; Fix bugs and removed unused code in Druids --- .../java/io/druid/query/scan/ScanQuery.java | 94 +++-- .../main/java/io/druid/query/BaseQuery.java | 25 +- .../druid/query/CPUTimeMetricQueryRunner.java | 6 +- .../src/main/java/io/druid/query/Druids.java | 180 +++++--- .../query/MetricsEmittingQueryRunner.java | 15 +- .../src/main/java/io/druid/query/Query.java | 9 + .../DataSourceMetadataQuery.java | 49 ++- .../io/druid/query/groupby/GroupByQuery.java | 399 ++++++++---------- .../query/groupby/orderby/NoopLimitSpec.java | 15 +- .../groupby/strategy/GroupByStrategyV1.java | 40 +- .../groupby/strategy/GroupByStrategyV2.java | 32 +- .../metadata/SegmentMetadataQuery.java | 86 ++-- .../query/search/search/SearchQuery.java | 92 ++-- .../io/druid/query/select/SelectQuery.java | 110 +++-- .../query/timeboundary/TimeBoundaryQuery.java | 54 ++- .../query/timeseries/TimeseriesQuery.java | 100 ++--- .../java/io/druid/query/topn/TopNQuery.java | 155 +++---- .../io/druid/query/topn/TopNQueryBuilder.java | 13 +- .../server/log/LoggingRequestLoggerTest.java | 9 +- 19 files changed, 723 insertions(+), 760 deletions(-) diff --git a/extensions-contrib/scan-query/src/main/java/io/druid/query/scan/ScanQuery.java b/extensions-contrib/scan-query/src/main/java/io/druid/query/scan/ScanQuery.java index df6a4079d5cb..fb3033f2545a 100644 --- a/extensions-contrib/scan-query/src/main/java/io/druid/query/scan/ScanQuery.java +++ b/extensions-contrib/scan-query/src/main/java/io/druid/query/scan/ScanQuery.java @@ -26,6 +26,7 @@ import io.druid.query.BaseQuery; import io.druid.query.DataSource; import io.druid.query.Query; +import io.druid.query.QueryMetrics; import io.druid.query.TableDataSource; import io.druid.query.filter.DimFilter; import io.druid.query.filter.InDimFilter; @@ -64,7 +65,22 @@ public ScanQuery( @JsonProperty("context") Map context ) { - super(dataSource, querySegmentSpec, false, context); + this(dataSource, querySegmentSpec, resultFormat, batchSize, limit, dimFilter, columns, context, null); + } + + private ScanQuery( + final DataSource dataSource, + final QuerySegmentSpec querySegmentSpec, + final String resultFormat, + final int batchSize, + final long limit, + final DimFilter dimFilter, + final List columns, + final Map context, + final QueryMetrics queryMetrics + ) + { + super(dataSource, querySegmentSpec, false, context, queryMetrics); this.resultFormat = resultFormat == null ? RESULT_FORMAT_LIST : resultFormat; this.batchSize = (batchSize == 0) ? 4096 * 5 : batchSize; this.limit = (limit == 0) ? Long.MAX_VALUE : limit; @@ -125,60 +141,31 @@ public List getColumns() @Override public Query withQuerySegmentSpec(QuerySegmentSpec querySegmentSpec) { - return new ScanQuery( - getDataSource(), - querySegmentSpec, - resultFormat, - batchSize, - limit, - dimFilter, - columns, - getContext() - ); + return ScanQueryBuilder.copy(this).intervals(querySegmentSpec).build(); } @Override public Query withDataSource(DataSource dataSource) { - return new ScanQuery( - dataSource, - getQuerySegmentSpec(), - resultFormat, - batchSize, - limit, - dimFilter, - columns, - getContext() - ); + return ScanQueryBuilder.copy(this).dataSource(dataSource).build(); } @Override public Query withOverriddenContext(Map contextOverrides) { - return new ScanQuery( - getDataSource(), - getQuerySegmentSpec(), - resultFormat, - batchSize, - limit, - dimFilter, - columns, - computeOverridenContext(contextOverrides) - ); + return ScanQueryBuilder.copy(this).context(computeOverriddenContext(getContext(), contextOverrides)).build(); } public ScanQuery withDimFilter(DimFilter dimFilter) { - return new ScanQuery( - getDataSource(), - getQuerySegmentSpec(), - resultFormat, - batchSize, - limit, - dimFilter, - columns, - getContext() - ); + return ScanQueryBuilder.copy(this).filters(dimFilter).build(); + } + + @Override + public Query withQueryMetrics(QueryMetrics queryMetrics) + { + Preconditions.checkNotNull(queryMetrics); + return ScanQueryBuilder.copy(this).queryMetrics(queryMetrics).build(); } @Override @@ -263,6 +250,7 @@ public static class ScanQueryBuilder private long limit; private DimFilter dimFilter; private List columns; + private QueryMetrics queryMetrics; public ScanQueryBuilder() { @@ -274,6 +262,7 @@ public ScanQueryBuilder() limit = 0; dimFilter = null; columns = Lists.newArrayList(); + queryMetrics = null; } public ScanQuery build() @@ -286,16 +275,23 @@ public ScanQuery build() limit, dimFilter, columns, - context + context, + queryMetrics ); } - public ScanQueryBuilder copy(ScanQueryBuilder builder) + public static ScanQueryBuilder copy(ScanQuery query) { return new ScanQueryBuilder() - .dataSource(builder.dataSource) - .intervals(builder.querySegmentSpec) - .context(builder.context); + .dataSource(query.getDataSource()) + .intervals(query.getQuerySegmentSpec()) + .resultFormat(query.getResultFormat()) + .batchSize(query.getBatchSize()) + .limit(query.getLimit()) + .filters(query.getFilter()) + .columns(query.getColumns()) + .context(query.getContext()) + .queryMetrics(query.getQueryMetrics()); } public ScanQueryBuilder dataSource(String ds) @@ -381,6 +377,12 @@ public ScanQueryBuilder columns(String... c) columns = Arrays.asList(c); return this; } + + public ScanQueryBuilder queryMetrics(QueryMetrics m) + { + queryMetrics = m; + return this; + } } public static ScanQueryBuilder newScanQueryBuilder() diff --git a/processing/src/main/java/io/druid/query/BaseQuery.java b/processing/src/main/java/io/druid/query/BaseQuery.java index 22d0fb1a4ba5..09c66f200175 100644 --- a/processing/src/main/java/io/druid/query/BaseQuery.java +++ b/processing/src/main/java/io/druid/query/BaseQuery.java @@ -30,6 +30,7 @@ import org.joda.time.Duration; import org.joda.time.Interval; +import javax.annotation.Nullable; import java.util.List; import java.util.Map; @@ -105,26 +106,31 @@ public static void checkInterrupted() } public static final String QUERYID = "queryId"; + private final DataSource dataSource; private final boolean descending; private final Map context; private final QuerySegmentSpec querySegmentSpec; + @Nullable + private final QueryMetrics queryMetrics; private volatile Duration duration; public BaseQuery( DataSource dataSource, QuerySegmentSpec querySegmentSpec, boolean descending, - Map context + Map context, + QueryMetrics queryMetrics ) { Preconditions.checkNotNull(dataSource, "dataSource can't be null"); Preconditions.checkNotNull(querySegmentSpec, "querySegmentSpec can't be null"); this.dataSource = dataSource; + this.descending = descending; this.context = context; this.querySegmentSpec = querySegmentSpec; - this.descending = descending; + this.queryMetrics = queryMetrics; } @JsonProperty @@ -206,10 +212,12 @@ public boolean getContextBoolean(String key, boolean defaultValue) return parseBoolean(this, key, defaultValue); } - protected Map computeOverridenContext(Map overrides) + protected static Map computeOverriddenContext( + final Map context, + final Map overrides + ) { Map overridden = Maps.newTreeMap(); - final Map context = getContext(); if (context != null) { overridden.putAll(context); } @@ -225,6 +233,13 @@ public Ordering getResultOrdering() return descending ? retVal.reverse() : retVal; } + @Override + @Nullable + public QueryMetrics getQueryMetrics() + { + return queryMetrics; + } + @Override public String getId() { @@ -234,7 +249,7 @@ public String getId() @Override public Query withId(String id) { - return withOverriddenContext(ImmutableMap.of(QUERYID, id)); + return withOverriddenContext(ImmutableMap.of(QUERYID, id)); } @Override diff --git a/processing/src/main/java/io/druid/query/CPUTimeMetricQueryRunner.java b/processing/src/main/java/io/druid/query/CPUTimeMetricQueryRunner.java index 7aefbb7ad472..ccac4d94f0c3 100644 --- a/processing/src/main/java/io/druid/query/CPUTimeMetricQueryRunner.java +++ b/processing/src/main/java/io/druid/query/CPUTimeMetricQueryRunner.java @@ -62,7 +62,9 @@ public Sequence run( final Query query, final Map responseContext ) { - final Sequence baseSequence = delegate.run(query, responseContext); + QueryMetrics> queryMetrics = queryToolChest.makeMetrics(query); + Query queryWithMetrics = query.withQueryMetrics(queryMetrics); + final Sequence baseSequence = delegate.run(queryWithMetrics, responseContext); return Sequences.wrap( baseSequence, new SequenceWrapper() @@ -84,7 +86,7 @@ public void after(boolean isDone, Throwable thrown) throws Exception if (report) { final long cpuTimeNs = cpuTimeAccumulator.get(); if (cpuTimeNs > 0) { - queryToolChest.makeMetrics(query).reportCpuTime(cpuTimeNs).emit(emitter); + queryMetrics.reportCpuTime(cpuTimeNs).emit(emitter); } } } diff --git a/processing/src/main/java/io/druid/query/Druids.java b/processing/src/main/java/io/druid/query/Druids.java index a835394052f3..0ee78389de56 100644 --- a/processing/src/main/java/io/druid/query/Druids.java +++ b/processing/src/main/java/io/druid/query/Druids.java @@ -338,6 +338,7 @@ public static class TimeseriesQueryBuilder private List aggregatorSpecs; private List postAggregatorSpecs; private Map context; + private QueryMetrics queryMetrics; private TimeseriesQueryBuilder() { @@ -350,6 +351,7 @@ private TimeseriesQueryBuilder() aggregatorSpecs = Lists.newArrayList(); postAggregatorSpecs = Lists.newArrayList(); context = null; + queryMetrics = null; } public TimeseriesQuery build() @@ -363,34 +365,24 @@ public TimeseriesQuery build() granularity, aggregatorSpecs, postAggregatorSpecs, - context + context, + queryMetrics ); } - public TimeseriesQueryBuilder copy(TimeseriesQuery query) + public static TimeseriesQueryBuilder copy(TimeseriesQuery query) { return new TimeseriesQueryBuilder() .dataSource(query.getDataSource()) - .intervals(query.getIntervals()) - .filters(query.getDimensionsFilter()) + .intervals(query.getQuerySegmentSpec()) .descending(query.isDescending()) + .virtualColumns(query.getVirtualColumns()) + .filters(query.getDimensionsFilter()) .granularity(query.getGranularity()) .aggregators(query.getAggregatorSpecs()) .postAggregators(query.getPostAggregatorSpecs()) - .context(query.getContext()); - } - - public TimeseriesQueryBuilder copy(TimeseriesQueryBuilder builder) - { - return new TimeseriesQueryBuilder() - .dataSource(builder.dataSource) - .intervals(builder.querySegmentSpec) - .filters(builder.dimFilter) - .descending(builder.descending) - .granularity(builder.granularity) - .aggregators(builder.aggregatorSpecs) - .postAggregators(builder.postAggregatorSpecs) - .context(builder.context); + .context(query.getContext()) + .queryMetrics(query.getQueryMetrics()); } public DataSource getDataSource() @@ -532,6 +524,12 @@ public TimeseriesQueryBuilder context(Map c) context = c; return this; } + + public TimeseriesQueryBuilder queryMetrics(QueryMetrics m) + { + queryMetrics = m; + return this; + } } public static TimeseriesQueryBuilder newTimeseriesQueryBuilder() @@ -569,6 +567,7 @@ public static class SearchQueryBuilder private SearchQuerySpec querySpec; private SearchSortSpec sortSpec; private Map context; + private QueryMetrics queryMetrics; public SearchQueryBuilder() { @@ -579,7 +578,9 @@ public SearchQueryBuilder() querySegmentSpec = null; dimensions = null; querySpec = null; + sortSpec = null; context = null; + queryMetrics = null; } public SearchQuery build() @@ -593,34 +594,24 @@ public SearchQuery build() dimensions, querySpec, sortSpec, - context + context, + queryMetrics ); } - public SearchQueryBuilder copy(SearchQuery query) + public static SearchQueryBuilder copy(SearchQuery query) { return new SearchQueryBuilder() .dataSource(query.getDataSource()) - .intervals(query.getQuerySegmentSpec()) .filters(query.getDimensionsFilter()) .granularity(query.getGranularity()) .limit(query.getLimit()) + .intervals(query.getQuerySegmentSpec()) .dimensions(query.getDimensions()) .query(query.getQuery()) - .context(query.getContext()); - } - - public SearchQueryBuilder copy(SearchQueryBuilder builder) - { - return new SearchQueryBuilder() - .dataSource(builder.dataSource) - .intervals(builder.querySegmentSpec) - .filters(builder.dimFilter) - .granularity(builder.granularity) - .limit(builder.limit) - .dimensions(builder.dimensions) - .query(builder.querySpec) - .context(builder.context); + .sortSpec(query.getSort()) + .context(query.getContext()) + .queryMetrics(query.getQueryMetrics()); } public SearchQueryBuilder dataSource(String d) @@ -770,6 +761,12 @@ public SearchQueryBuilder context(Map c) context = c; return this; } + + public SearchQueryBuilder queryMetrics(QueryMetrics m) + { + queryMetrics = m; + return this; + } } public static SearchQueryBuilder newSearchQueryBuilder() @@ -798,6 +795,7 @@ public static class TimeBoundaryQueryBuilder private String bound; private DimFilter dimFilter; private Map context; + private QueryMetrics queryMetrics; public TimeBoundaryQueryBuilder() { @@ -806,6 +804,7 @@ public TimeBoundaryQueryBuilder() bound = null; dimFilter = null; context = null; + queryMetrics = null; } public TimeBoundaryQuery build() @@ -815,18 +814,20 @@ public TimeBoundaryQuery build() querySegmentSpec, bound, dimFilter, - context + context, + queryMetrics ); } - public TimeBoundaryQueryBuilder copy(TimeBoundaryQueryBuilder builder) + public static TimeBoundaryQueryBuilder copy(TimeBoundaryQuery query) { return new TimeBoundaryQueryBuilder() - .dataSource(builder.dataSource) - .intervals(builder.querySegmentSpec) - .bound(builder.bound) - .filters(builder.dimFilter) - .context(builder.context); + .dataSource(query.getDataSource()) + .intervals(query.getQuerySegmentSpec()) + .bound(query.getBound()) + .filters(query.getFilter()) + .context(query.getContext()) + .queryMetrics(query.getQueryMetrics()); } public TimeBoundaryQueryBuilder dataSource(String ds) @@ -888,6 +889,12 @@ public TimeBoundaryQueryBuilder context(Map c) context = c; return this; } + + public TimeBoundaryQueryBuilder queryMetrics(QueryMetrics m) + { + queryMetrics = m; + return this; + } } public static TimeBoundaryQueryBuilder newTimeBoundaryQueryBuilder() @@ -985,6 +992,7 @@ public static class SegmentMetadataQueryBuilder private Boolean merge; private Boolean lenientAggregatorMerge; private Map context; + private QueryMetrics queryMetrics; public SegmentMetadataQueryBuilder() { @@ -993,8 +1001,9 @@ public SegmentMetadataQueryBuilder() toInclude = null; analysisTypes = null; merge = null; - context = null; lenientAggregatorMerge = null; + context = null; + queryMetrics = null; } public SegmentMetadataQuery build() @@ -1007,24 +1016,22 @@ public SegmentMetadataQuery build() context, analysisTypes, false, - lenientAggregatorMerge + lenientAggregatorMerge, + queryMetrics ); } - public SegmentMetadataQueryBuilder copy(SegmentMetadataQueryBuilder builder) + public static SegmentMetadataQueryBuilder copy(SegmentMetadataQuery query) { - final SegmentMetadataQuery.AnalysisType[] analysisTypesArray = - analysisTypes != null - ? analysisTypes.toArray(new SegmentMetadataQuery.AnalysisType[analysisTypes.size()]) - : null; return new SegmentMetadataQueryBuilder() - .dataSource(builder.dataSource) - .intervals(builder.querySegmentSpec) - .toInclude(toInclude) - .analysisTypes(analysisTypesArray) - .merge(merge) - .lenientAggregatorMerge(lenientAggregatorMerge) - .context(builder.context); + .dataSource(query.getDataSource()) + .intervals(query.getQuerySegmentSpec()) + .toInclude(query.getToInclude()) + .analysisTypes(query.getAnalysisTypes()) + .merge(query.isMerge()) + .lenientAggregatorMerge(query.isLenientAggregatorMerge()) + .context(query.getContext()) + .queryMetrics(query.getQueryMetrics()); } public SegmentMetadataQueryBuilder dataSource(String ds) @@ -1075,6 +1082,12 @@ public SegmentMetadataQueryBuilder analysisTypes(SegmentMetadataQuery.AnalysisTy return this; } + public SegmentMetadataQueryBuilder analysisTypes(EnumSet analysisTypes) + { + this.analysisTypes = analysisTypes; + return this; + } + public SegmentMetadataQueryBuilder merge(boolean merge) { this.merge = merge; @@ -1092,6 +1105,12 @@ public SegmentMetadataQueryBuilder context(Map c) context = c; return this; } + + public SegmentMetadataQueryBuilder queryMetrics(QueryMetrics m) + { + queryMetrics = m; + return this; + } } public static SegmentMetadataQueryBuilder newSegmentMetadataQueryBuilder() @@ -1126,17 +1145,21 @@ public static class SelectQueryBuilder private List metrics; private VirtualColumns virtualColumns; private PagingSpec pagingSpec; + private QueryMetrics queryMetrics; public SelectQueryBuilder() { dataSource = null; querySegmentSpec = null; + descending = false; context = null; dimFilter = null; granularity = Granularities.ALL; dimensions = Lists.newArrayList(); metrics = Lists.newArrayList(); + virtualColumns = null; pagingSpec = null; + queryMetrics = null; } public SelectQuery build() @@ -1151,16 +1174,25 @@ public SelectQuery build() metrics, virtualColumns, pagingSpec, - context + context, + queryMetrics ); } - public SelectQueryBuilder copy(SelectQueryBuilder builder) + public static SelectQueryBuilder copy(SelectQuery query) { return new SelectQueryBuilder() - .dataSource(builder.dataSource) - .intervals(builder.querySegmentSpec) - .context(builder.context); + .dataSource(query.getDataSource()) + .intervals(query.getQuerySegmentSpec()) + .descending(query.isDescending()) + .filters(query.getFilter()) + .granularity(query.getGranularity()) + .dimensionSpecs(query.getDimensions()) + .metrics(query.getMetrics()) + .virtualColumns(query.getVirtualColumns()) + .pagingSpec(query.getPagingSpec()) + .context(query.getContext()) + .queryMetrics(query.getQueryMetrics()); } public SelectQueryBuilder dataSource(String ds) @@ -1274,6 +1306,12 @@ public SelectQueryBuilder pagingSpec(PagingSpec p) pagingSpec = p; return this; } + + public SelectQueryBuilder queryMetrics(QueryMetrics m) + { + queryMetrics = m; + return this; + } } public static SelectQueryBuilder newSelectQueryBuilder() @@ -1300,12 +1338,14 @@ public static class DataSourceMetadataQueryBuilder private DataSource dataSource; private QuerySegmentSpec querySegmentSpec; private Map context; + private QueryMetrics queryMetrics; public DataSourceMetadataQueryBuilder() { dataSource = null; querySegmentSpec = null; context = null; + queryMetrics = null; } public DataSourceMetadataQuery build() @@ -1313,16 +1353,18 @@ public DataSourceMetadataQuery build() return new DataSourceMetadataQuery( dataSource, querySegmentSpec, - context + context, + queryMetrics ); } - public DataSourceMetadataQueryBuilder copy(DataSourceMetadataQueryBuilder builder) + public static DataSourceMetadataQueryBuilder copy(DataSourceMetadataQuery query) { return new DataSourceMetadataQueryBuilder() - .dataSource(builder.dataSource) - .intervals(builder.querySegmentSpec) - .context(builder.context); + .dataSource(query.getDataSource()) + .intervals(query.getQuerySegmentSpec()) + .context(query.getContext()) + .queryMetrics(query.getQueryMetrics()); } public DataSourceMetadataQueryBuilder dataSource(String ds) @@ -1360,6 +1402,12 @@ public DataSourceMetadataQueryBuilder context(Map c) context = c; return this; } + + public DataSourceMetadataQueryBuilder queryMetrics(QueryMetrics m) + { + queryMetrics = m; + return this; + } } public static DataSourceMetadataQueryBuilder newDataSourceMetadataQueryBuilder() diff --git a/processing/src/main/java/io/druid/query/MetricsEmittingQueryRunner.java b/processing/src/main/java/io/druid/query/MetricsEmittingQueryRunner.java index 096fb39d45fc..2c7421871546 100644 --- a/processing/src/main/java/io/druid/query/MetricsEmittingQueryRunner.java +++ b/processing/src/main/java/io/druid/query/MetricsEmittingQueryRunner.java @@ -19,7 +19,6 @@ package io.druid.query; -import com.google.common.base.Supplier; import com.metamx.emitter.service.ServiceEmitter; import io.druid.java.util.common.guava.LazySequence; import io.druid.java.util.common.guava.Sequence; @@ -85,21 +84,13 @@ public MetricsEmittingQueryRunner withWaitMeasuredFromNow() public Sequence run(final Query query, final Map responseContext) { final QueryMetrics> queryMetrics = queryToolChest.makeMetrics(query); - applyCustomDimensions.accept(queryMetrics); - + final Query queryWithMetrics = query.withQueryMetrics(queryMetrics); return Sequences.wrap( // Use LazySequence because want to account execution time of queryRunner.run() (it prepares the underlying // Sequence) as part of the reported query time, i. e. we want to execute queryRunner.run() after - // `startTime = System.currentTimeMillis();` (see below). - new LazySequence<>(new Supplier>() - { - @Override - public Sequence get() - { - return queryRunner.run(query, responseContext); - } - }), + // `startTime = System.nanoTime();` (see below). + new LazySequence<>(() -> queryRunner.run(queryWithMetrics, responseContext)), new SequenceWrapper() { private long startTimeNs; diff --git a/processing/src/main/java/io/druid/query/Query.java b/processing/src/main/java/io/druid/query/Query.java index 9ad178161ead..3fc9d8691267 100644 --- a/processing/src/main/java/io/druid/query/Query.java +++ b/processing/src/main/java/io/druid/query/Query.java @@ -36,6 +36,7 @@ import org.joda.time.Duration; import org.joda.time.Interval; +import javax.annotation.Nullable; import java.util.List; import java.util.Map; @@ -99,4 +100,12 @@ public interface Query String getId(); Query withDataSource(DataSource dataSource); + + /** + * @throws NullPointerException if the given queryMetrics is null + */ + Query withQueryMetrics(QueryMetrics queryMetrics); + + @Nullable + QueryMetrics getQueryMetrics(); } diff --git a/processing/src/main/java/io/druid/query/datasourcemetadata/DataSourceMetadataQuery.java b/processing/src/main/java/io/druid/query/datasourcemetadata/DataSourceMetadataQuery.java index 186fcaf2f6ef..54ca3f0b009f 100644 --- a/processing/src/main/java/io/druid/query/datasourcemetadata/DataSourceMetadataQuery.java +++ b/processing/src/main/java/io/druid/query/datasourcemetadata/DataSourceMetadataQuery.java @@ -21,11 +21,14 @@ import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonProperty; +import com.google.common.base.Preconditions; import com.google.common.collect.Lists; import io.druid.common.utils.JodaUtils; import io.druid.query.BaseQuery; import io.druid.query.DataSource; +import io.druid.query.Druids; import io.druid.query.Query; +import io.druid.query.QueryMetrics; import io.druid.query.Result; import io.druid.query.filter.DimFilter; import io.druid.query.spec.MultipleIntervalSegmentSpec; @@ -34,6 +37,7 @@ import org.joda.time.Interval; import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.Map; @@ -51,13 +55,28 @@ public DataSourceMetadataQuery( @JsonProperty("intervals") QuerySegmentSpec querySegmentSpec, @JsonProperty("context") Map context ) + { + this(dataSource, querySegmentSpec, context, null); + } + + /** + * This constructor is public only because {@link Druids.DataSourceMetadataQueryBuilder} needs to access this + * constructor, and it is defined in Druids rather than in as an inner class of DataSourceMetadataQuery. + */ + public DataSourceMetadataQuery( + final DataSource dataSource, + final QuerySegmentSpec querySegmentSpec, + final Map context, + final QueryMetrics queryMetrics + ) { super( dataSource, - (querySegmentSpec == null) ? new MultipleIntervalSegmentSpec(Arrays.asList(MY_Y2K_INTERVAL)) + (querySegmentSpec == null) ? new MultipleIntervalSegmentSpec(Collections.singletonList(MY_Y2K_INTERVAL)) : querySegmentSpec, false, - context + context, + queryMetrics ); } @@ -82,31 +101,27 @@ public String getType() @Override public DataSourceMetadataQuery withOverriddenContext(Map contextOverrides) { - return new DataSourceMetadataQuery( - getDataSource(), - getQuerySegmentSpec(), - computeOverridenContext(contextOverrides) - ); + Map newContext = computeOverriddenContext(getContext(), contextOverrides); + return Druids.DataSourceMetadataQueryBuilder.copy(this).context(newContext).build(); } @Override public DataSourceMetadataQuery withQuerySegmentSpec(QuerySegmentSpec spec) { - return new DataSourceMetadataQuery( - getDataSource(), - spec, - getContext() - ); + return Druids.DataSourceMetadataQueryBuilder.copy(this).intervals(spec).build(); } @Override public Query> withDataSource(DataSource dataSource) { - return new DataSourceMetadataQuery( - dataSource, - getQuerySegmentSpec(), - getContext() - ); + return Druids.DataSourceMetadataQueryBuilder.copy(this).dataSource(dataSource).build(); + } + + @Override + public Query> withQueryMetrics(QueryMetrics queryMetrics) + { + Preconditions.checkNotNull(queryMetrics); + return Druids.DataSourceMetadataQueryBuilder.copy(this).queryMetrics(queryMetrics).build(); } public Iterable> buildResult(DateTime timestamp, DateTime maxIngestedEventTime) diff --git a/processing/src/main/java/io/druid/query/groupby/GroupByQuery.java b/processing/src/main/java/io/druid/query/groupby/GroupByQuery.java index 9184ad386893..80623cd5d9a3 100644 --- a/processing/src/main/java/io/druid/query/groupby/GroupByQuery.java +++ b/processing/src/main/java/io/druid/query/groupby/GroupByQuery.java @@ -25,7 +25,6 @@ import com.google.common.base.Function; import com.google.common.base.Functions; import com.google.common.base.Preconditions; -import com.google.common.base.Predicate; import com.google.common.collect.ImmutableList; import com.google.common.collect.Lists; import com.google.common.collect.Ordering; @@ -43,6 +42,7 @@ import io.druid.query.Queries; import io.druid.query.Query; import io.druid.query.QueryDataSource; +import io.druid.query.QueryMetrics; import io.druid.query.TableDataSource; import io.druid.query.aggregation.AggregatorFactory; import io.druid.query.aggregation.PostAggregator; @@ -61,6 +61,7 @@ import io.druid.segment.column.Column; import org.joda.time.Interval; +import java.util.ArrayList; import java.util.Arrays; import java.util.Comparator; import java.util.List; @@ -76,17 +77,10 @@ public class GroupByQuery extends BaseQuery private final static Comparator NATURAL_NULLS_FIRST = Ordering.natural().nullsFirst(); - private final static Comparator NON_GRANULAR_TIME_COMP = new Comparator() - { - @Override - public int compare(Row lhs, Row rhs) - { - return Longs.compare( - lhs.getTimestampFromEpoch(), - rhs.getTimestampFromEpoch() - ); - } - }; + private final static Comparator NON_GRANULAR_TIME_COMP = (Row lhs, Row rhs) -> Longs.compare( + lhs.getTimestampFromEpoch(), + rhs.getTimestampFromEpoch() + ); public static Builder builder() { @@ -119,11 +113,44 @@ public GroupByQuery( @JsonProperty("context") Map context ) { - super(dataSource, querySegmentSpec, false, context); + this( + dataSource, + querySegmentSpec, + virtualColumns, + dimFilter, + granularity, + dimensions, + aggregatorSpecs, + postAggregatorSpecs, + havingSpec, + limitSpec, + context, + null + ); + } + + private GroupByQuery( + final DataSource dataSource, + final QuerySegmentSpec querySegmentSpec, + final VirtualColumns virtualColumns, + final DimFilter dimFilter, + final Granularity granularity, + final List dimensions, + final List aggregatorSpecs, + final List postAggregatorSpecs, + final HavingSpec havingSpec, + final LimitSpec limitSpec, + final Map context, + final QueryMetrics queryMetrics + ) + { + super(dataSource, querySegmentSpec, false, context, queryMetrics); + GroupByQueryMetrics.class.cast(queryMetrics); // ClassCastException if not + this.virtualColumns = VirtualColumns.nullToEmpty(virtualColumns); this.dimFilter = dimFilter; this.granularity = granularity; - this.dimensions = dimensions == null ? ImmutableList.of() : dimensions; + this.dimensions = dimensions == null ? ImmutableList.of() : dimensions; for (DimensionSpec spec : this.dimensions) { Preconditions.checkArgument(spec != null, "dimensions has null DimensionSpec"); } @@ -133,7 +160,7 @@ public GroupByQuery( postAggregatorSpecs == null ? ImmutableList.of() : postAggregatorSpecs ); this.havingSpec = havingSpec; - this.limitSpec = (limitSpec == null) ? new NoopLimitSpec() : limitSpec; + this.limitSpec = nullToNoopLimitSpec(limitSpec); Preconditions.checkNotNull(this.granularity, "Must specify a granularity"); @@ -143,35 +170,29 @@ public GroupByQuery( // We're not counting __time, even though that name is problematic. See: https://github.com/druid-io/druid/pull/3684 verifyOutputNames(this.dimensions, this.aggregatorSpecs, this.postAggregatorSpecs); + limitFn = makeLimitFn(this.limitSpec); + } + + private Function, Sequence> makeLimitFn(LimitSpec limitSpec) + { Function, Sequence> postProcFn = - this.limitSpec.build(this.dimensions, this.aggregatorSpecs, this.postAggregatorSpecs); + limitSpec.build(dimensions, aggregatorSpecs, postAggregatorSpecs); if (havingSpec != null) { postProcFn = Functions.compose( postProcFn, - new Function, Sequence>() - { - @Override - public Sequence apply(Sequence input) - { - GroupByQuery.this.havingSpec.setRowSignature(GroupByQueryHelper.rowSignatureFor(GroupByQuery.this)); - return Sequences.filter( - input, - new Predicate() - { - @Override - public boolean apply(Row input) - { - return GroupByQuery.this.havingSpec.eval(input); - } - } - ); - } + (Sequence input) -> { + havingSpec.setRowSignature(GroupByQueryHelper.rowSignatureFor(GroupByQuery.this)); + return Sequences.filter(input, havingSpec::eval); } ); } + return postProcFn; + } - limitFn = postProcFn; + private static LimitSpec nullToNoopLimitSpec(LimitSpec limitSpec) + { + return (limitSpec == null) ? NoopLimitSpec.instance() : limitSpec; } /** @@ -179,21 +200,23 @@ public boolean apply(Row input) * have already passed in order for the object to exist. */ private GroupByQuery( - DataSource dataSource, - QuerySegmentSpec querySegmentSpec, - VirtualColumns virtualColumns, - DimFilter dimFilter, - Granularity granularity, - List dimensions, - List aggregatorSpecs, - List postAggregatorSpecs, - HavingSpec havingSpec, - LimitSpec orderBySpec, - Function, Sequence> limitFn, - Map context + final DataSource dataSource, + final QuerySegmentSpec querySegmentSpec, + final VirtualColumns virtualColumns, + final DimFilter dimFilter, + final Granularity granularity, + final List dimensions, + final List aggregatorSpecs, + final List postAggregatorSpecs, + final HavingSpec havingSpec, + final LimitSpec orderBySpec, + final Function, Sequence> limitFn, + final Map context, + final QueryMetrics queryMetrics ) { - super(dataSource, querySegmentSpec, false, context); + super(dataSource, querySegmentSpec, false, context, queryMetrics); + GroupByQueryMetrics.class.cast(queryMetrics); // ClassCastException if not this.virtualColumns = virtualColumns; this.dimFilter = dimFilter; @@ -284,17 +307,12 @@ public Ordering getResultOrdering() final Ordering rowOrdering = getRowOrdering(false); return Ordering.from( - new Comparator() - { - @Override - public int compare(Object lhs, Object rhs) - { - if (lhs instanceof Row) { - return rowOrdering.compare((Row) lhs, (Row) rhs); - } else { - // Probably bySegment queries - return NATURAL_NULLS_FIRST.compare(lhs, rhs); - } + (lhs, rhs) -> { + if (lhs instanceof Row) { + return rowOrdering.compare((Row) lhs, (Row) rhs); + } else { + // Probably bySegment queries + return NATURAL_NULLS_FIRST.compare(lhs, rhs); } } ); @@ -307,47 +325,28 @@ public Ordering getRowOrdering(final boolean granular) final Comparator timeComparator = getTimeComparator(granular); if (timeComparator == null) { - return Ordering.from( - new Comparator() - { - @Override - public int compare(Row lhs, Row rhs) - { - return compareDims(dimensions, lhs, rhs); - } - } - ); + return Ordering.from((lhs, rhs) -> compareDims(dimensions, lhs, rhs)); } else if (sortByDimsFirst) { return Ordering.from( - new Comparator() - { - @Override - public int compare(Row lhs, Row rhs) - { - final int cmp = compareDims(dimensions, lhs, rhs); - if (cmp != 0) { - return cmp; - } - - return timeComparator.compare(lhs, rhs); + (lhs, rhs) -> { + final int cmp = compareDims(dimensions, lhs, rhs); + if (cmp != 0) { + return cmp; } + + return timeComparator.compare(lhs, rhs); } ); } else { return Ordering.from( - new Comparator() - { - @Override - public int compare(Row lhs, Row rhs) - { - final int timeCompare = timeComparator.compare(lhs, rhs); - - if (timeCompare != 0) { - return timeCompare; - } - - return compareDims(dimensions, lhs, rhs); + (lhs, rhs) -> { + final int timeCompare = timeComparator.compare(lhs, rhs); + + if (timeCompare != 0) { + return timeCompare; } + + return compareDims(dimensions, lhs, rhs); } ); } @@ -358,17 +357,10 @@ private Comparator getTimeComparator(boolean granular) if (Granularities.ALL.equals(granularity)) { return null; } else if (granular) { - return new Comparator() - { - @Override - public int compare(Row lhs, Row rhs) - { - return Longs.compare( - granularity.bucketStart(lhs.getTimestamp()).getMillis(), - granularity.bucketStart(rhs.getTimestamp()).getMillis() - ); - } - }; + return (lhs, rhs) -> Longs.compare( + granularity.bucketStart(lhs.getTimestamp()).getMillis(), + granularity.bucketStart(rhs.getTimestamp()).getMillis() + ); } else { return NON_GRANULAR_TIME_COMP; } @@ -406,147 +398,51 @@ public Sequence applyLimit(Sequence results) @Override public GroupByQuery withOverriddenContext(Map contextOverride) { - return new GroupByQuery( - getDataSource(), - getQuerySegmentSpec(), - virtualColumns, - dimFilter, - granularity, - dimensions, - aggregatorSpecs, - postAggregatorSpecs, - havingSpec, - limitSpec, - limitFn, - computeOverridenContext(contextOverride) - ); + return new Builder(this).overrideContext(contextOverride).build(); } @Override public GroupByQuery withQuerySegmentSpec(QuerySegmentSpec spec) { - return new GroupByQuery( - getDataSource(), - spec, - virtualColumns, - dimFilter, - granularity, - dimensions, - aggregatorSpecs, - postAggregatorSpecs, - havingSpec, - limitSpec, - limitFn, - getContext() - ); + return new Builder(this).setQuerySegmentSpec(spec).build(); } public GroupByQuery withDimFilter(final DimFilter dimFilter) { - return new GroupByQuery( - getDataSource(), - getQuerySegmentSpec(), - virtualColumns, - dimFilter, - getGranularity(), - getDimensions(), - getAggregatorSpecs(), - getPostAggregatorSpecs(), - getHavingSpec(), - getLimitSpec(), - limitFn, - getContext() - ); + return new Builder(this).setDimFilter(dimFilter).build(); } @Override public Query withDataSource(DataSource dataSource) { - return new GroupByQuery( - dataSource, - getQuerySegmentSpec(), - virtualColumns, - dimFilter, - granularity, - dimensions, - aggregatorSpecs, - postAggregatorSpecs, - havingSpec, - limitSpec, - limitFn, - getContext() - ); + return new Builder(this).setDataSource(dataSource).build(); } public GroupByQuery withDimensionSpecs(final List dimensionSpecs) { - return new GroupByQuery( - getDataSource(), - getQuerySegmentSpec(), - virtualColumns, - getDimFilter(), - getGranularity(), - dimensionSpecs, - getAggregatorSpecs(), - getPostAggregatorSpecs(), - getHavingSpec(), - getLimitSpec(), - limitFn, - getContext() - ); + return new Builder(this).setDimensions(dimensionSpecs).build(); } - public GroupByQuery withLimitSpec(final LimitSpec limitSpec) + public GroupByQuery withLimitSpec(LimitSpec limitSpec) { - return new GroupByQuery( - getDataSource(), - getQuerySegmentSpec(), - virtualColumns, - getDimFilter(), - getGranularity(), - getDimensions(), - getAggregatorSpecs(), - getPostAggregatorSpecs(), - getHavingSpec(), - limitSpec, - getContext() - ); + return new Builder(this).setLimitSpec(limitSpec).build(); } public GroupByQuery withAggregatorSpecs(final List aggregatorSpecs) { - return new GroupByQuery( - getDataSource(), - getQuerySegmentSpec(), - virtualColumns, - getDimFilter(), - getGranularity(), - getDimensions(), - aggregatorSpecs, - getPostAggregatorSpecs(), - getHavingSpec(), - getLimitSpec(), - limitFn, - getContext() - ); + return new Builder(this).setAggregatorSpecs(aggregatorSpecs).build(); } public GroupByQuery withPostAggregatorSpecs(final List postAggregatorSpecs) { - return new GroupByQuery( - getDataSource(), - getQuerySegmentSpec(), - virtualColumns, - getDimFilter(), - getGranularity(), - getDimensions(), - getAggregatorSpecs(), - postAggregatorSpecs, - getHavingSpec(), - getLimitSpec(), - limitFn, - getContext() - ); + return new Builder(this).setPostAggregatorSpecs(postAggregatorSpecs).build(); + } + + @Override + public Query withQueryMetrics(QueryMetrics queryMetrics) + { + Preconditions.checkNotNull(queryMetrics); + return new Builder(this).setQueryMetrics(queryMetrics).build(); } private static void verifyOutputNames( @@ -597,8 +493,10 @@ public static class Builder private Map context; private LimitSpec limitSpec = null; + private Function, Sequence> limitFn; private List orderByColumnSpecs = Lists.newArrayList(); private int limit = Integer.MAX_VALUE; + private QueryMetrics queryMetrics; public Builder() { @@ -609,14 +507,16 @@ public Builder(GroupByQuery query) dataSource = query.getDataSource(); querySegmentSpec = query.getQuerySegmentSpec(); virtualColumns = query.getVirtualColumns(); - limitSpec = query.getLimitSpec(); dimFilter = query.getDimFilter(); granularity = query.getGranularity(); dimensions = query.getDimensions(); aggregatorSpecs = query.getAggregatorSpecs(); postAggregatorSpecs = query.getPostAggregatorSpecs(); havingSpec = query.getHavingSpec(); + limitSpec = query.getLimitSpec(); + limitFn = query.limitFn; context = query.getContext(); + queryMetrics = query.getQueryMetrics(); } public Builder(Builder builder) @@ -624,16 +524,18 @@ public Builder(Builder builder) dataSource = builder.dataSource; querySegmentSpec = builder.querySegmentSpec; virtualColumns = builder.virtualColumns; - limitSpec = builder.limitSpec; dimFilter = builder.dimFilter; granularity = builder.granularity; dimensions = builder.dimensions; aggregatorSpecs = builder.aggregatorSpecs; postAggregatorSpecs = builder.postAggregatorSpecs; havingSpec = builder.havingSpec; + limitSpec = builder.limitSpec; + limitFn = builder.limitFn; limit = builder.limit; - + orderByColumnSpecs = new ArrayList<>(builder.orderByColumnSpecs); context = builder.context; + queryMetrics = builder.queryMetrics; } public Builder setDataSource(DataSource dataSource) @@ -718,8 +620,10 @@ public Builder addOrderByColumn(OrderByColumnSpec columnSpec) public Builder setLimitSpec(LimitSpec limitSpec) { + Preconditions.checkNotNull(limitSpec); ensureFluentLimitsNotSet(); this.limitSpec = limitSpec; + this.limitFn = null; return this; } @@ -819,17 +723,27 @@ public Builder setContext(Map context) return this; } + public Builder overrideContext(Map contextOverride) + { + this.context = computeOverriddenContext(context, contextOverride); + return this; + } + public Builder setHavingSpec(HavingSpec havingSpec) { this.havingSpec = havingSpec; - return this; } public Builder setLimit(Integer limit) { this.limit = limit; + return this; + } + public Builder setQueryMetrics(QueryMetrics queryMetrics) + { + this.queryMetrics = queryMetrics; return this; } @@ -843,7 +757,7 @@ public GroupByQuery build() final LimitSpec theLimitSpec; if (limitSpec == null) { if (orderByColumnSpecs.isEmpty() && limit == Integer.MAX_VALUE) { - theLimitSpec = new NoopLimitSpec(); + theLimitSpec = NoopLimitSpec.instance(); } else { theLimitSpec = new DefaultLimitSpec(orderByColumnSpecs, limit); } @@ -851,19 +765,38 @@ public GroupByQuery build() theLimitSpec = limitSpec; } - return new GroupByQuery( - dataSource, - querySegmentSpec, - virtualColumns, - dimFilter, - granularity, - dimensions, - aggregatorSpecs, - postAggregatorSpecs, - havingSpec, - theLimitSpec, - context - ); + if (limitFn != null) { + return new GroupByQuery( + dataSource, + querySegmentSpec, + virtualColumns, + dimFilter, + granularity, + dimensions, + aggregatorSpecs, + postAggregatorSpecs, + havingSpec, + theLimitSpec, + limitFn, + context, + queryMetrics + ); + } else { + return new GroupByQuery( + dataSource, + querySegmentSpec, + virtualColumns, + dimFilter, + granularity, + dimensions, + aggregatorSpecs, + postAggregatorSpecs, + havingSpec, + theLimitSpec, + context, + queryMetrics + ); + } } } diff --git a/processing/src/main/java/io/druid/query/groupby/orderby/NoopLimitSpec.java b/processing/src/main/java/io/druid/query/groupby/orderby/NoopLimitSpec.java index c8d770e87d55..b14224d4cbb3 100644 --- a/processing/src/main/java/io/druid/query/groupby/orderby/NoopLimitSpec.java +++ b/processing/src/main/java/io/druid/query/groupby/orderby/NoopLimitSpec.java @@ -19,6 +19,7 @@ package io.druid.query.groupby.orderby; +import com.fasterxml.jackson.annotation.JsonCreator; import com.google.common.base.Function; import com.google.common.base.Functions; import io.druid.data.input.Row; @@ -31,10 +32,22 @@ /** */ -public class NoopLimitSpec implements LimitSpec +public final class NoopLimitSpec implements LimitSpec { private static final byte CACHE_KEY = 0x0; + public static final NoopLimitSpec INSTANCE = new NoopLimitSpec(); + + @JsonCreator + public static NoopLimitSpec instance() + { + return INSTANCE; + } + + private NoopLimitSpec() + { + } + @Override public Function, Sequence> build( List dimensions, List aggs, List postAggs diff --git a/processing/src/main/java/io/druid/query/groupby/strategy/GroupByStrategyV1.java b/processing/src/main/java/io/druid/query/groupby/strategy/GroupByStrategyV1.java index 9b2c9163818b..c51caaf1c5e8 100644 --- a/processing/src/main/java/io/druid/query/groupby/strategy/GroupByStrategyV1.java +++ b/processing/src/main/java/io/druid/query/groupby/strategy/GroupByStrategyV1.java @@ -47,6 +47,7 @@ import io.druid.query.groupby.GroupByQueryEngine; import io.druid.query.groupby.GroupByQueryHelper; import io.druid.query.groupby.GroupByQueryQueryToolChest; +import io.druid.query.groupby.orderby.NoopLimitSpec; import io.druid.query.groupby.resource.GroupByQueryResource; import io.druid.query.spec.MultipleIntervalSegmentSpec; import io.druid.segment.StorageAdapter; @@ -119,32 +120,25 @@ public Sequence mergeResults( configSupplier.get(), bufferPool, baseRunner.run( - new GroupByQuery( - query.getDataSource(), - query.getQuerySegmentSpec(), - query.getVirtualColumns(), - query.getDimFilter(), - query.getGranularity(), - query.getDimensions(), - query.getAggregatorSpecs(), + new GroupByQuery.Builder(query) // Don't do post aggs until the end of this method. - ImmutableList.of(), + .setPostAggregatorSpecs(ImmutableList.of()) // Don't do "having" clause until the end of this method. - null, - null, - query.getContext() - ).withOverriddenContext( - ImmutableMap.of( - "finalize", false, - //setting sort to false avoids unnecessary sorting while merging results. we only need to sort - //in the end when returning results to user. (note this is only respected by groupBy v1) - GroupByQueryHelper.CTX_KEY_SORT_RESULTS, false, - //no merging needed at historicals because GroupByQueryRunnerFactory.mergeRunners(..) would return - //merged results. (note this is only respected by groupBy v1) - GroupByQueryQueryToolChest.GROUP_BY_MERGE_KEY, false, - GroupByQueryConfig.CTX_KEY_STRATEGY, GroupByStrategySelector.STRATEGY_V1 + .setHavingSpec(null) + .setLimitSpec(NoopLimitSpec.instance()) + .overrideContext( + ImmutableMap.of( + "finalize", false, + //setting sort to false avoids unnecessary sorting while merging results. we only need to sort + //in the end when returning results to user. (note this is only respected by groupBy v1) + GroupByQueryHelper.CTX_KEY_SORT_RESULTS, false, + //no merging needed at historicals because GroupByQueryRunnerFactory.mergeRunners(..) would + //return merged results. (note this is only respected by groupBy v1) + GroupByQueryQueryToolChest.GROUP_BY_MERGE_KEY, false, + GroupByQueryConfig.CTX_KEY_STRATEGY, GroupByStrategySelector.STRATEGY_V1 + ) ) - ), + .build(), responseContext ), true diff --git a/processing/src/main/java/io/druid/query/groupby/strategy/GroupByStrategyV2.java b/processing/src/main/java/io/druid/query/groupby/strategy/GroupByStrategyV2.java index 45dbd954d1c7..6c7f0192df2b 100644 --- a/processing/src/main/java/io/druid/query/groupby/strategy/GroupByStrategyV2.java +++ b/processing/src/main/java/io/druid/query/groupby/strategy/GroupByStrategyV2.java @@ -62,6 +62,7 @@ import io.druid.query.groupby.epinephelinae.GroupByMergingQueryRunnerV2; import io.druid.query.groupby.epinephelinae.GroupByQueryEngineV2; import io.druid.query.groupby.epinephelinae.GroupByRowProcessor; +import io.druid.query.groupby.orderby.NoopLimitSpec; import io.druid.query.groupby.resource.GroupByQueryResource; import io.druid.segment.StorageAdapter; import org.joda.time.DateTime; @@ -231,28 +232,21 @@ protected BinaryFn createMergeFn(Query queryParam) return query.applyLimit( Sequences.map( mergingQueryRunner.run( - new GroupByQuery( - query.getDataSource(), - query.getQuerySegmentSpec(), - query.getVirtualColumns(), - query.getDimFilter(), - query.getGranularity(), - query.getDimensions(), - query.getAggregatorSpecs(), + new GroupByQuery.Builder(query) // Don't do post aggs until the end of this method. - ImmutableList.of(), + .setPostAggregatorSpecs(ImmutableList.of()) // Don't do "having" clause until the end of this method. - null, - null, - query.getContext() - ).withOverriddenContext( - ImmutableMap.of( - "finalize", false, - GroupByQueryConfig.CTX_KEY_STRATEGY, GroupByStrategySelector.STRATEGY_V2, - CTX_KEY_FUDGE_TIMESTAMP, fudgeTimestamp == null ? "" : String.valueOf(fudgeTimestamp.getMillis()), - CTX_KEY_OUTERMOST, false + .setHavingSpec(null) + .setLimitSpec(NoopLimitSpec.instance()) + .overrideContext( + ImmutableMap.of( + "finalize", false, + GroupByQueryConfig.CTX_KEY_STRATEGY, GroupByStrategySelector.STRATEGY_V2, + CTX_KEY_FUDGE_TIMESTAMP, fudgeTimestamp == null ? "" : String.valueOf(fudgeTimestamp.getMillis()), + CTX_KEY_OUTERMOST, false + ) ) - ), + .build(), responseContext ), new Function() diff --git a/processing/src/main/java/io/druid/query/metadata/metadata/SegmentMetadataQuery.java b/processing/src/main/java/io/druid/query/metadata/metadata/SegmentMetadataQuery.java index 3d9ab5b117ed..12a05b4656c8 100644 --- a/processing/src/main/java/io/druid/query/metadata/metadata/SegmentMetadataQuery.java +++ b/processing/src/main/java/io/druid/query/metadata/metadata/SegmentMetadataQuery.java @@ -27,7 +27,9 @@ import io.druid.common.utils.JodaUtils; import io.druid.query.BaseQuery; import io.druid.query.DataSource; +import io.druid.query.Druids; import io.druid.query.Query; +import io.druid.query.QueryMetrics; import io.druid.query.TableDataSource; import io.druid.query.UnionDataSource; import io.druid.query.filter.DimFilter; @@ -107,13 +109,43 @@ public SegmentMetadataQuery( @JsonProperty("usingDefaultInterval") Boolean useDefaultInterval, @JsonProperty("lenientAggregatorMerge") Boolean lenientAggregatorMerge ) + { + this( + dataSource, + querySegmentSpec, + toInclude, + merge, + context, + analysisTypes, + useDefaultInterval, + lenientAggregatorMerge, + null + ); + } + + /** + * This constructor is public only because {@link Druids.SegmentMetadataQueryBuilder} needs to access this + * constructor, and it is defined in Druids rather than in as an inner class of SegmentMetadataQuery. + */ + public SegmentMetadataQuery( + final DataSource dataSource, + final QuerySegmentSpec querySegmentSpec, + final ColumnIncluderator toInclude, + final Boolean merge, + final Map context, + final EnumSet analysisTypes, + final Boolean useDefaultInterval, + final Boolean lenientAggregatorMerge, + final QueryMetrics queryMetrics + ) { super( dataSource, (querySegmentSpec == null) ? new MultipleIntervalSegmentSpec(Arrays.asList(DEFAULT_INTERVAL)) : querySegmentSpec, false, - context + context, + queryMetrics ); if (querySegmentSpec == null) { @@ -232,60 +264,32 @@ public byte[] getAnalysisTypesCacheKey() @Override public Query withOverriddenContext(Map contextOverride) { - return new SegmentMetadataQuery( - getDataSource(), - getQuerySegmentSpec(), - toInclude, - merge, - computeOverridenContext(contextOverride), - analysisTypes, - usingDefaultInterval, - lenientAggregatorMerge - ); + Map newContext = computeOverriddenContext(getContext(), contextOverride); + return Druids.SegmentMetadataQueryBuilder.copy(this).context(newContext).build(); } @Override public Query withQuerySegmentSpec(QuerySegmentSpec spec) { - return new SegmentMetadataQuery( - getDataSource(), - spec, - toInclude, - merge, - getContext(), - analysisTypes, - usingDefaultInterval, - lenientAggregatorMerge - ); + return Druids.SegmentMetadataQueryBuilder.copy(this).intervals(spec).build(); } @Override public Query withDataSource(DataSource dataSource) { - return new SegmentMetadataQuery( - dataSource, - getQuerySegmentSpec(), - toInclude, - merge, - getContext(), - analysisTypes, - usingDefaultInterval, - lenientAggregatorMerge - ); + return Druids.SegmentMetadataQueryBuilder.copy(this).dataSource(dataSource).build(); } public Query withColumns(ColumnIncluderator includerator) { - return new SegmentMetadataQuery( - getDataSource(), - getQuerySegmentSpec(), - includerator, - merge, - getContext(), - analysisTypes, - usingDefaultInterval, - lenientAggregatorMerge - ); + return Druids.SegmentMetadataQueryBuilder.copy(this).toInclude(includerator).build(); + } + + @Override + public Query withQueryMetrics(QueryMetrics queryMetrics) + { + Preconditions.checkNotNull(queryMetrics); + return Druids.SegmentMetadataQueryBuilder.copy(this).queryMetrics(queryMetrics).build(); } @Override diff --git a/processing/src/main/java/io/druid/query/search/search/SearchQuery.java b/processing/src/main/java/io/druid/query/search/search/SearchQuery.java index c45a21cddb0d..7abde034f3cf 100644 --- a/processing/src/main/java/io/druid/query/search/search/SearchQuery.java +++ b/processing/src/main/java/io/druid/query/search/search/SearchQuery.java @@ -26,7 +26,9 @@ import io.druid.java.util.common.granularity.Granularity; import io.druid.query.BaseQuery; import io.druid.query.DataSource; +import io.druid.query.Druids; import io.druid.query.Query; +import io.druid.query.QueryMetrics; import io.druid.query.Result; import io.druid.query.dimension.DimensionSpec; import io.druid.query.filter.DimFilter; @@ -63,7 +65,27 @@ public SearchQuery( @JsonProperty("context") Map context ) { - super(dataSource, querySegmentSpec, false, context); + this(dataSource, dimFilter, granularity, limit, querySegmentSpec, dimensions, querySpec, sortSpec, context, null); + } + + /** + * This constructor is public only because {@link Druids.SearchQueryBuilder} needs to access this constructor, and it + * is defined in Druids rather than in as an inner class of SearchQuery. + */ + public SearchQuery( + final DataSource dataSource, + final DimFilter dimFilter, + final Granularity granularity, + final int limit, + final QuerySegmentSpec querySegmentSpec, + final List dimensions, + final SearchQuerySpec querySpec, + final SearchSortSpec sortSpec, + final Map context, + final QueryMetrics queryMetrics + ) + { + super(dataSource, querySegmentSpec, false, context, queryMetrics); Preconditions.checkNotNull(querySegmentSpec, "Must specify an interval"); this.dimFilter = dimFilter; @@ -95,64 +117,32 @@ public String getType() @Override public SearchQuery withQuerySegmentSpec(QuerySegmentSpec spec) { - return new SearchQuery( - getDataSource(), - dimFilter, - granularity, - limit, - spec, - dimensions, - querySpec, - sortSpec, - getContext() - ); + return Druids.SearchQueryBuilder.copy(this).intervals(spec).build(); } @Override public Query> withDataSource(DataSource dataSource) { - return new SearchQuery( - dataSource, - dimFilter, - granularity, - limit, - getQuerySegmentSpec(), - dimensions, - querySpec, - sortSpec, - getContext() - ); + return Druids.SearchQueryBuilder.copy(this).dataSource(dataSource).build(); } @Override public SearchQuery withOverriddenContext(Map contextOverrides) { - return new SearchQuery( - getDataSource(), - dimFilter, - granularity, - limit, - getQuerySegmentSpec(), - dimensions, - querySpec, - sortSpec, - computeOverridenContext(contextOverrides) - ); + Map newContext = computeOverriddenContext(getContext(), contextOverrides); + return Druids.SearchQueryBuilder.copy(this).context(newContext).build(); } public SearchQuery withDimFilter(DimFilter dimFilter) { - return new SearchQuery( - getDataSource(), - dimFilter, - granularity, - limit, - getQuerySegmentSpec(), - dimensions, - querySpec, - sortSpec, - getContext() - ); + return Druids.SearchQueryBuilder.copy(this).filters(dimFilter).build(); + } + + @Override + public Query> withQueryMetrics(QueryMetrics queryMetrics) + { + Preconditions.checkNotNull(queryMetrics); + return Druids.SearchQueryBuilder.copy(this).queryMetrics(queryMetrics).build(); } @JsonProperty("filter") @@ -193,17 +183,7 @@ public SearchSortSpec getSort() public SearchQuery withLimit(int newLimit) { - return new SearchQuery( - getDataSource(), - dimFilter, - granularity, - newLimit, - getQuerySegmentSpec(), - dimensions, - querySpec, - sortSpec, - getContext() - ); + return Druids.SearchQueryBuilder.copy(this).limit(newLimit).build(); } @Override diff --git a/processing/src/main/java/io/druid/query/select/SelectQuery.java b/processing/src/main/java/io/druid/query/select/SelectQuery.java index 4c0154857c6a..728bde8416f7 100644 --- a/processing/src/main/java/io/druid/query/select/SelectQuery.java +++ b/processing/src/main/java/io/druid/query/select/SelectQuery.java @@ -26,7 +26,9 @@ import io.druid.java.util.common.granularity.Granularity; import io.druid.query.BaseQuery; import io.druid.query.DataSource; +import io.druid.query.Druids; import io.druid.query.Query; +import io.druid.query.QueryMetrics; import io.druid.query.Result; import io.druid.query.dimension.DimensionSpec; import io.druid.query.filter.DimFilter; @@ -63,7 +65,40 @@ public SelectQuery( @JsonProperty("context") Map context ) { - super(dataSource, querySegmentSpec, descending, context); + this( + dataSource, + querySegmentSpec, + descending, + dimFilter, + granularity, + dimensions, + metrics, + virtualColumns, + pagingSpec, + context, + null + ); + } + + /** + * This constructor is public only because {@link Druids.SelectQueryBuilder} needs to access this constructor, and it + * is defined in Druids rather than in as an inner class of SelectQuery. + */ + public SelectQuery( + final DataSource dataSource, + final QuerySegmentSpec querySegmentSpec, + final boolean descending, + final DimFilter dimFilter, + final Granularity granularity, + final List dimensions, + final List metrics, + final VirtualColumns virtualColumns, + final PagingSpec pagingSpec, + final Map context, + final QueryMetrics queryMetrics + ) + { + super(dataSource, querySegmentSpec, descending, context, queryMetrics); this.dimFilter = dimFilter; this.granularity = granularity; this.dimensions = dimensions; @@ -146,83 +181,36 @@ public PagingOffset getPagingOffset(String identifier) public SelectQuery withQuerySegmentSpec(QuerySegmentSpec querySegmentSpec) { - return new SelectQuery( - getDataSource(), - querySegmentSpec, - isDescending(), - dimFilter, - granularity, - dimensions, - metrics, - virtualColumns, - pagingSpec, - getContext() - ); + return Druids.SelectQueryBuilder.copy(this).intervals(querySegmentSpec).build(); } @Override public Query> withDataSource(DataSource dataSource) { - return new SelectQuery( - dataSource, - getQuerySegmentSpec(), - isDescending(), - dimFilter, - granularity, - dimensions, - metrics, - virtualColumns, - pagingSpec, - getContext() - ); + return Druids.SelectQueryBuilder.copy(this).dataSource(dataSource).build(); } public SelectQuery withOverriddenContext(Map contextOverrides) { - return new SelectQuery( - getDataSource(), - getQuerySegmentSpec(), - isDescending(), - dimFilter, - granularity, - dimensions, - metrics, - virtualColumns, - pagingSpec, - computeOverridenContext(contextOverrides) - ); + Map newContext = computeOverriddenContext(getContext(), contextOverrides); + return Druids.SelectQueryBuilder.copy(this).context(newContext).build(); } public SelectQuery withPagingSpec(PagingSpec pagingSpec) { - return new SelectQuery( - getDataSource(), - getQuerySegmentSpec(), - isDescending(), - dimFilter, - granularity, - dimensions, - metrics, - virtualColumns, - pagingSpec, - getContext() - ); + return Druids.SelectQueryBuilder.copy(this).pagingSpec(pagingSpec).build(); } public SelectQuery withDimFilter(DimFilter dimFilter) { - return new SelectQuery( - getDataSource(), - getQuerySegmentSpec(), - isDescending(), - dimFilter, - granularity, - dimensions, - metrics, - virtualColumns, - pagingSpec, - getContext() - ); + return Druids.SelectQueryBuilder.copy(this).filters(dimFilter).build(); + } + + @Override + public Query> withQueryMetrics(QueryMetrics queryMetrics) + { + Preconditions.checkNotNull(queryMetrics); + return Druids.SelectQueryBuilder.copy(this).queryMetrics(queryMetrics).build(); } @Override diff --git a/processing/src/main/java/io/druid/query/timeboundary/TimeBoundaryQuery.java b/processing/src/main/java/io/druid/query/timeboundary/TimeBoundaryQuery.java index 4501e40c5602..698b7ffb0f7e 100644 --- a/processing/src/main/java/io/druid/query/timeboundary/TimeBoundaryQuery.java +++ b/processing/src/main/java/io/druid/query/timeboundary/TimeBoundaryQuery.java @@ -21,13 +21,16 @@ import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonProperty; +import com.google.common.base.Preconditions; import com.google.common.collect.Lists; import com.google.common.collect.Maps; import io.druid.common.utils.JodaUtils; import io.druid.java.util.common.StringUtils; import io.druid.query.BaseQuery; import io.druid.query.DataSource; +import io.druid.query.Druids; import io.druid.query.Query; +import io.druid.query.QueryMetrics; import io.druid.query.Result; import io.druid.query.filter.DimFilter; import io.druid.query.spec.MultipleIntervalSegmentSpec; @@ -64,13 +67,30 @@ public TimeBoundaryQuery( @JsonProperty("filter") DimFilter dimFilter, @JsonProperty("context") Map context ) + { + this(dataSource, querySegmentSpec, bound, dimFilter, context, null); + } + + /** + * This constructor is public only because {@link Druids.TimeBoundaryQueryBuilder} needs to access this constructor, + * and it is defined in Druids rather than in as an inner class of TimeBoundaryQuery. + */ + public TimeBoundaryQuery( + final DataSource dataSource, + final QuerySegmentSpec querySegmentSpec, + final String bound, + final DimFilter dimFilter, + final Map context, + final QueryMetrics queryMetrics + ) { super( dataSource, (querySegmentSpec == null) ? new MultipleIntervalSegmentSpec(Arrays.asList(MY_Y2K_INTERVAL)) : querySegmentSpec, false, - context + context, + queryMetrics ); this.dimFilter = dimFilter; @@ -109,37 +129,27 @@ public String getBound() @Override public TimeBoundaryQuery withOverriddenContext(Map contextOverrides) { - return new TimeBoundaryQuery( - getDataSource(), - getQuerySegmentSpec(), - bound, - dimFilter, - computeOverridenContext(contextOverrides) - ); + Map newContext = computeOverriddenContext(getContext(), contextOverrides); + return Druids.TimeBoundaryQueryBuilder.copy(this).context(newContext).build(); } @Override public TimeBoundaryQuery withQuerySegmentSpec(QuerySegmentSpec spec) { - return new TimeBoundaryQuery( - getDataSource(), - spec, - bound, - dimFilter, - getContext() - ); + return Druids.TimeBoundaryQueryBuilder.copy(this).intervals(spec).build(); } @Override public Query> withDataSource(DataSource dataSource) { - return new TimeBoundaryQuery( - dataSource, - getQuerySegmentSpec(), - bound, - dimFilter, - getContext() - ); + return Druids.TimeBoundaryQueryBuilder.copy(this).dataSource(dataSource).build(); + } + + @Override + public Query> withQueryMetrics(QueryMetrics queryMetrics) + { + Preconditions.checkNotNull(queryMetrics); + return Druids.TimeBoundaryQueryBuilder.copy(this).queryMetrics(queryMetrics).build(); } public byte[] getCacheKey() diff --git a/processing/src/main/java/io/druid/query/timeseries/TimeseriesQuery.java b/processing/src/main/java/io/druid/query/timeseries/TimeseriesQuery.java index ee3b375001a1..d244bbf8f54d 100644 --- a/processing/src/main/java/io/druid/query/timeseries/TimeseriesQuery.java +++ b/processing/src/main/java/io/druid/query/timeseries/TimeseriesQuery.java @@ -22,12 +22,15 @@ import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonProperty; import com.fasterxml.jackson.annotation.JsonTypeName; +import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableList; import io.druid.java.util.common.granularity.Granularity; import io.druid.query.BaseQuery; import io.druid.query.DataSource; +import io.druid.query.Druids; import io.druid.query.Queries; import io.druid.query.Query; +import io.druid.query.QueryMetrics; import io.druid.query.Result; import io.druid.query.aggregation.AggregatorFactory; import io.druid.query.aggregation.PostAggregator; @@ -63,16 +66,47 @@ public TimeseriesQuery( @JsonProperty("context") Map context ) { - super(dataSource, querySegmentSpec, descending, context); + this( + dataSource, + querySegmentSpec, + descending, + virtualColumns, + dimFilter, + granularity, + aggregatorSpecs, + postAggregatorSpecs, + context, + null + ); + } + + /** + * This constructor is public only because {@link Druids.TimeseriesQueryBuilder} needs to access this constructor, and + * it is defined in Druids rather than in as an inner class of TimeseriesQuery. + */ + public TimeseriesQuery( + final DataSource dataSource, + final QuerySegmentSpec querySegmentSpec, + final boolean descending, + final VirtualColumns virtualColumns, + final DimFilter dimFilter, + final Granularity granularity, + final List aggregatorSpecs, + final List postAggregatorSpecs, + final Map context, + final QueryMetrics queryMetrics + ) + { + super(dataSource, querySegmentSpec, descending, context, queryMetrics); + TimeseriesQueryMetrics.class.cast(queryMetrics); // ClassCastException if not + this.virtualColumns = VirtualColumns.nullToEmpty(virtualColumns); this.dimFilter = dimFilter; this.granularity = granularity; - this.aggregatorSpecs = aggregatorSpecs == null ? ImmutableList.of() : aggregatorSpecs; + this.aggregatorSpecs = aggregatorSpecs == null ? ImmutableList.of() : aggregatorSpecs; this.postAggregatorSpecs = Queries.prepareAggregations( this.aggregatorSpecs, - postAggregatorSpecs == null - ? ImmutableList.of() - : postAggregatorSpecs + postAggregatorSpecs == null ? ImmutableList.of() : postAggregatorSpecs ); } @@ -131,63 +165,31 @@ public boolean isSkipEmptyBuckets() public TimeseriesQuery withQuerySegmentSpec(QuerySegmentSpec querySegmentSpec) { - return new TimeseriesQuery( - getDataSource(), - querySegmentSpec, - isDescending(), - virtualColumns, - dimFilter, - granularity, - aggregatorSpecs, - postAggregatorSpecs, - getContext() - ); + return Druids.TimeseriesQueryBuilder.copy(this).intervals(querySegmentSpec).build(); } @Override public Query> withDataSource(DataSource dataSource) { - return new TimeseriesQuery( - dataSource, - getQuerySegmentSpec(), - isDescending(), - virtualColumns, - dimFilter, - granularity, - aggregatorSpecs, - postAggregatorSpecs, - getContext() - ); + return Druids.TimeseriesQueryBuilder.copy(this).dataSource(dataSource).build(); } public TimeseriesQuery withOverriddenContext(Map contextOverrides) { - return new TimeseriesQuery( - getDataSource(), - getQuerySegmentSpec(), - isDescending(), - virtualColumns, - dimFilter, - granularity, - aggregatorSpecs, - postAggregatorSpecs, - computeOverridenContext(contextOverrides) - ); + Map newContext = computeOverriddenContext(getContext(), contextOverrides); + return Druids.TimeseriesQueryBuilder.copy(this).context(newContext).build(); } public TimeseriesQuery withDimFilter(DimFilter dimFilter) { - return new TimeseriesQuery( - getDataSource(), - getQuerySegmentSpec(), - isDescending(), - virtualColumns, - dimFilter, - granularity, - aggregatorSpecs, - postAggregatorSpecs, - getContext() - ); + return Druids.TimeseriesQueryBuilder.copy(this).filters(dimFilter).build(); + } + + @Override + public Query> withQueryMetrics(QueryMetrics queryMetrics) + { + Preconditions.checkNotNull(queryMetrics); + return Druids.TimeseriesQueryBuilder.copy(this).queryMetrics(queryMetrics).build(); } @Override diff --git a/processing/src/main/java/io/druid/query/topn/TopNQuery.java b/processing/src/main/java/io/druid/query/topn/TopNQuery.java index 29ca0cddcba5..a8177a016ba7 100644 --- a/processing/src/main/java/io/druid/query/topn/TopNQuery.java +++ b/processing/src/main/java/io/druid/query/topn/TopNQuery.java @@ -28,6 +28,7 @@ import io.druid.query.DataSource; import io.druid.query.Queries; import io.druid.query.Query; +import io.druid.query.QueryMetrics; import io.druid.query.Result; import io.druid.query.aggregation.AggregatorFactory; import io.druid.query.aggregation.PostAggregator; @@ -70,7 +71,40 @@ public TopNQuery( @JsonProperty("context") Map context ) { - super(dataSource, querySegmentSpec, false, context); + this( + dataSource, + virtualColumns, + dimensionSpec, + topNMetricSpec, + threshold, + querySegmentSpec, + dimFilter, + granularity, + aggregatorSpecs, + postAggregatorSpecs, + context, + null + ); + } + + TopNQuery( + final DataSource dataSource, + final VirtualColumns virtualColumns, + final DimensionSpec dimensionSpec, + final TopNMetricSpec topNMetricSpec, + final int threshold, + final QuerySegmentSpec querySegmentSpec, + final DimFilter dimFilter, + final Granularity granularity, + final List aggregatorSpecs, + final List postAggregatorSpecs, + final Map context, + final QueryMetrics queryMetrics + ) + { + super(dataSource, querySegmentSpec, false, context, queryMetrics); + TopNQueryMetrics.class.cast(queryMetrics); // ClassCastException if not + this.virtualColumns = VirtualColumns.nullToEmpty(virtualColumns); this.dimensionSpec = dimensionSpec; this.topNMetricSpec = topNMetricSpec; @@ -169,139 +203,50 @@ public void initTopNAlgorithmSelector(TopNAlgorithmSelector selector) public TopNQuery withQuerySegmentSpec(QuerySegmentSpec querySegmentSpec) { - return new TopNQuery( - getDataSource(), - virtualColumns, - dimensionSpec, - topNMetricSpec, - threshold, - querySegmentSpec, - dimFilter, - granularity, - aggregatorSpecs, - postAggregatorSpecs, - getContext() - ); + return new TopNQueryBuilder(this).intervals(querySegmentSpec).build(); } public TopNQuery withDimensionSpec(DimensionSpec spec) { - return new TopNQuery( - getDataSource(), - virtualColumns, - spec, - topNMetricSpec, - threshold, - getQuerySegmentSpec(), - dimFilter, - granularity, - aggregatorSpecs, - postAggregatorSpecs, - getContext() - ); + return new TopNQueryBuilder(this).dimension(spec).build(); } public TopNQuery withAggregatorSpecs(List aggregatorSpecs) { - return new TopNQuery( - getDataSource(), - virtualColumns, - getDimensionSpec(), - topNMetricSpec, - threshold, - getQuerySegmentSpec(), - dimFilter, - granularity, - aggregatorSpecs, - postAggregatorSpecs, - getContext() - ); + return new TopNQueryBuilder(this).aggregators(aggregatorSpecs).build(); } public TopNQuery withPostAggregatorSpecs(List postAggregatorSpecs) { - return new TopNQuery( - getDataSource(), - virtualColumns, - getDimensionSpec(), - topNMetricSpec, - threshold, - getQuerySegmentSpec(), - dimFilter, - granularity, - aggregatorSpecs, - postAggregatorSpecs, - getContext() - ); + return new TopNQueryBuilder(this).postAggregators(postAggregatorSpecs).build(); } @Override public Query> withDataSource(DataSource dataSource) { - return new TopNQuery( - dataSource, - virtualColumns, - dimensionSpec, - topNMetricSpec, - threshold, - getQuerySegmentSpec(), - dimFilter, - granularity, - aggregatorSpecs, - postAggregatorSpecs, - getContext() - ); + return new TopNQueryBuilder(this).dataSource(dataSource).build(); } public TopNQuery withThreshold(int threshold) { - return new TopNQuery( - getDataSource(), - virtualColumns, - dimensionSpec, - topNMetricSpec, - threshold, - getQuerySegmentSpec(), - dimFilter, - granularity, - aggregatorSpecs, - postAggregatorSpecs, - getContext() - ); + return new TopNQueryBuilder(this).threshold(threshold).build(); } public TopNQuery withOverriddenContext(Map contextOverrides) { - return new TopNQuery( - getDataSource(), - virtualColumns, - dimensionSpec, - topNMetricSpec, - threshold, - getQuerySegmentSpec(), - dimFilter, - granularity, - aggregatorSpecs, - postAggregatorSpecs, - computeOverridenContext(contextOverrides) - ); + return new TopNQueryBuilder(this).context(computeOverriddenContext(getContext(), contextOverrides)).build(); } public TopNQuery withDimFilter(DimFilter dimFilter) { - return new TopNQuery( - getDataSource(), - virtualColumns, - getDimensionSpec(), - topNMetricSpec, - threshold, - getQuerySegmentSpec(), - dimFilter, - granularity, - aggregatorSpecs, - postAggregatorSpecs, - getContext() - ); + return new TopNQueryBuilder(this).filters(dimFilter).build(); + } + + @Override + public Query> withQueryMetrics(QueryMetrics queryMetrics) + { + Preconditions.checkNotNull(queryMetrics); + return new TopNQueryBuilder(this).queryMetrics(queryMetrics).build(); } @Override diff --git a/processing/src/main/java/io/druid/query/topn/TopNQueryBuilder.java b/processing/src/main/java/io/druid/query/topn/TopNQueryBuilder.java index bdf8dc6153a9..6bc33a8bec59 100644 --- a/processing/src/main/java/io/druid/query/topn/TopNQueryBuilder.java +++ b/processing/src/main/java/io/druid/query/topn/TopNQueryBuilder.java @@ -23,6 +23,7 @@ import io.druid.java.util.common.granularity.Granularities; import io.druid.java.util.common.granularity.Granularity; import io.druid.query.DataSource; +import io.druid.query.QueryMetrics; import io.druid.query.TableDataSource; import io.druid.query.aggregation.AggregatorFactory; import io.druid.query.aggregation.PostAggregator; @@ -75,6 +76,7 @@ public class TopNQueryBuilder private List aggregatorSpecs; private List postAggregatorSpecs; private Map context; + private QueryMetrics queryMetrics; public TopNQueryBuilder() { @@ -89,6 +91,7 @@ public TopNQueryBuilder() aggregatorSpecs = Lists.newArrayList(); postAggregatorSpecs = Lists.newArrayList(); context = null; + queryMetrics = null; } public TopNQueryBuilder(final TopNQuery query) @@ -104,6 +107,7 @@ public TopNQueryBuilder(final TopNQuery query) this.aggregatorSpecs = query.getAggregatorSpecs(); this.postAggregatorSpecs = query.getPostAggregatorSpecs(); this.context = query.getContext(); + this.queryMetrics = query.getQueryMetrics(); } public DataSource getDataSource() @@ -174,7 +178,8 @@ public TopNQuery build() granularity, aggregatorSpecs, postAggregatorSpecs, - context + context, + queryMetrics ); } @@ -328,4 +333,10 @@ public TopNQueryBuilder context(Map c) context = c; return this; } + + public TopNQueryBuilder queryMetrics(QueryMetrics m) + { + queryMetrics = m; + return this; + } } diff --git a/server/src/test/java/io/druid/server/log/LoggingRequestLoggerTest.java b/server/src/test/java/io/druid/server/log/LoggingRequestLoggerTest.java index 18e330fcb02c..cc0f1cd38ceb 100644 --- a/server/src/test/java/io/druid/server/log/LoggingRequestLoggerTest.java +++ b/server/src/test/java/io/druid/server/log/LoggingRequestLoggerTest.java @@ -29,6 +29,7 @@ import io.druid.query.DataSource; import io.druid.query.LegacyDataSource; import io.druid.query.Query; +import io.druid.query.QueryMetrics; import io.druid.query.QueryRunner; import io.druid.query.QuerySegmentWalker; import io.druid.query.filter.DimFilter; @@ -183,7 +184,7 @@ class FakeQuery extends BaseQuery { public FakeQuery(DataSource dataSource, QuerySegmentSpec querySegmentSpec, boolean descending, Map context) { - super(dataSource, querySegmentSpec, descending, context); + super(dataSource, querySegmentSpec, descending, context, null); } @Override @@ -221,4 +222,10 @@ public Query withOverriddenContext(Map contextOverride) { throw new UnsupportedOperationException("shouldn't be here"); } + + @Override + public Query withQueryMetrics(QueryMetrics queryMetrics) + { + throw new UnsupportedOperationException("shouldn't be here"); + } } From 07703ff35fb9f325c314848e1db64f7f4015110e Mon Sep 17 00:00:00 2001 From: leventov Date: Wed, 29 Mar 2017 03:14:46 -0600 Subject: [PATCH 2/7] Fix a bug in TimeBoundaryQuery.getFilter() and remove TimeBoundaryQuery.getDimensionsFilter() --- .../io/druid/query/timeboundary/TimeBoundaryQuery.java | 9 ++------- .../timeboundary/TimeBoundaryQueryRunnerFactory.java | 5 +++-- 2 files changed, 5 insertions(+), 9 deletions(-) diff --git a/processing/src/main/java/io/druid/query/timeboundary/TimeBoundaryQuery.java b/processing/src/main/java/io/druid/query/timeboundary/TimeBoundaryQuery.java index 698b7ffb0f7e..9b4e14f4bcb6 100644 --- a/processing/src/main/java/io/druid/query/timeboundary/TimeBoundaryQuery.java +++ b/processing/src/main/java/io/druid/query/timeboundary/TimeBoundaryQuery.java @@ -102,10 +102,11 @@ public boolean hasFilters() { return dimFilter != null; } + @JsonProperty("filter") @Override public DimFilter getFilter() { - return null; + return dimFilter; } @Override @@ -114,12 +115,6 @@ public String getType() return Query.TIME_BOUNDARY; } - @JsonProperty("filter") - public DimFilter getDimensionsFilter() - { - return dimFilter; - } - @JsonProperty public String getBound() { diff --git a/processing/src/main/java/io/druid/query/timeboundary/TimeBoundaryQueryRunnerFactory.java b/processing/src/main/java/io/druid/query/timeboundary/TimeBoundaryQueryRunnerFactory.java index e3b383cc78e3..913933ffd333 100644 --- a/processing/src/main/java/io/druid/query/timeboundary/TimeBoundaryQueryRunnerFactory.java +++ b/processing/src/main/java/io/druid/query/timeboundary/TimeBoundaryQueryRunnerFactory.java @@ -112,7 +112,8 @@ private DateTime getTimeBoundary(StorageAdapter adapter, TimeBoundaryQuery legac final Sequence> resultSequence = QueryRunnerHelper.makeCursorBasedQuery( adapter, legacyQuery.getQuerySegmentSpec().getIntervals(), - Filters.toFilter(legacyQuery.getDimensionsFilter()), VirtualColumns.EMPTY, + Filters.toFilter(legacyQuery.getFilter()), + VirtualColumns.EMPTY, descending, Granularities.ALL, this.skipToFirstMatching @@ -154,7 +155,7 @@ public Iterator> make() final DateTime minTime; final DateTime maxTime; - if (legacyQuery.getDimensionsFilter() != null) { + if (legacyQuery.getFilter() != null) { minTime = getTimeBoundary(adapter, legacyQuery, false); if (minTime == null) { maxTime = null; From bac6957c102a4f4553b496c6b6d8f4475b9242ca Mon Sep 17 00:00:00 2001 From: leventov Date: Wed, 29 Mar 2017 15:11:51 -0600 Subject: [PATCH 3/7] Don't reassign query's queryMetrics if already present in CPUTimeMetricQueryRunner and MetricsEmittingQueryRunner --- .../io/druid/query/CPUTimeMetricQueryRunner.java | 15 ++++++++++----- .../druid/query/MetricsEmittingQueryRunner.java | 12 ++++++++++-- 2 files changed, 20 insertions(+), 7 deletions(-) diff --git a/processing/src/main/java/io/druid/query/CPUTimeMetricQueryRunner.java b/processing/src/main/java/io/druid/query/CPUTimeMetricQueryRunner.java index ccac4d94f0c3..80f9f48873fd 100644 --- a/processing/src/main/java/io/druid/query/CPUTimeMetricQueryRunner.java +++ b/processing/src/main/java/io/druid/query/CPUTimeMetricQueryRunner.java @@ -58,12 +58,17 @@ private CPUTimeMetricQueryRunner( @Override - public Sequence run( - final Query query, final Map responseContext - ) + public Sequence run(final Query query, final Map responseContext) { - QueryMetrics> queryMetrics = queryToolChest.makeMetrics(query); - Query queryWithMetrics = query.withQueryMetrics(queryMetrics); + final QueryMetrics> queryMetrics; + final Query queryWithMetrics; + if (query.getQueryMetrics() == null) { + queryMetrics = queryToolChest.makeMetrics(query); + queryWithMetrics = query.withQueryMetrics(queryMetrics); + } else { + queryMetrics = (QueryMetrics>) query.getQueryMetrics(); + queryWithMetrics = query; + } final Sequence baseSequence = delegate.run(queryWithMetrics, responseContext); return Sequences.wrap( baseSequence, diff --git a/processing/src/main/java/io/druid/query/MetricsEmittingQueryRunner.java b/processing/src/main/java/io/druid/query/MetricsEmittingQueryRunner.java index 2c7421871546..bc5340602292 100644 --- a/processing/src/main/java/io/druid/query/MetricsEmittingQueryRunner.java +++ b/processing/src/main/java/io/druid/query/MetricsEmittingQueryRunner.java @@ -83,9 +83,17 @@ public MetricsEmittingQueryRunner withWaitMeasuredFromNow() @Override public Sequence run(final Query query, final Map responseContext) { - final QueryMetrics> queryMetrics = queryToolChest.makeMetrics(query); + final QueryMetrics> queryMetrics; + final Query queryWithMetrics; + if (query.getQueryMetrics() == null) { + queryMetrics = queryToolChest.makeMetrics(query); + queryWithMetrics = query.withQueryMetrics(queryMetrics); + } else { + queryMetrics = (QueryMetrics>) query.getQueryMetrics(); + queryWithMetrics = query; + } applyCustomDimensions.accept(queryMetrics); - final Query queryWithMetrics = query.withQueryMetrics(queryMetrics); + return Sequences.wrap( // Use LazySequence because want to account execution time of queryRunner.run() (it prepares the underlying // Sequence) as part of the reported query time, i. e. we want to execute queryRunner.run() after From 72103371faed69b61b6c8e1ed42c9df7578b55b6 Mon Sep 17 00:00:00 2001 From: leventov Date: Wed, 29 Mar 2017 15:14:29 -0600 Subject: [PATCH 4/7] Add compatibility constructor to BaseQuery --- .../src/main/java/io/druid/query/BaseQuery.java | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/processing/src/main/java/io/druid/query/BaseQuery.java b/processing/src/main/java/io/druid/query/BaseQuery.java index 09c66f200175..c924e6f37de2 100644 --- a/processing/src/main/java/io/druid/query/BaseQuery.java +++ b/processing/src/main/java/io/druid/query/BaseQuery.java @@ -115,6 +115,21 @@ public static void checkInterrupted() private final QueryMetrics queryMetrics; private volatile Duration duration; + /** + * @deprecated compatibility constructor for extensions, {@link + * BaseQuery#BaseQuery(DataSource, QuerySegmentSpec, boolean, Map, QueryMetrics)} should be used instead. + */ + @Deprecated + public BaseQuery( + DataSource dataSource, + QuerySegmentSpec querySegmentSpec, + boolean descending, + Map context + ) + { + this(dataSource, querySegmentSpec, descending, context, null); + } + public BaseQuery( DataSource dataSource, QuerySegmentSpec querySegmentSpec, From 3c15d1d4989eaf31fd0d6754fe8647dcc52df355 Mon Sep 17 00:00:00 2001 From: leventov Date: Wed, 19 Apr 2017 18:09:01 +0300 Subject: [PATCH 5/7] Remove Query.queryMetrics property --- .../java/io/druid/query/scan/ScanQuery.java | 39 +-------- .../main/java/io/druid/query/BaseQuery.java | 30 +------ .../druid/query/CPUTimeMetricQueryRunner.java | 14 +--- .../src/main/java/io/druid/query/Druids.java | 84 +++---------------- .../query/MetricsEmittingQueryRunner.java | 13 +-- .../src/main/java/io/druid/query/Query.java | 9 -- .../DataSourceMetadataQuery.java | 26 +----- .../io/druid/query/groupby/GroupByQuery.java | 63 ++------------ .../metadata/SegmentMetadataQuery.java | 40 +-------- .../query/search/search/SearchQuery.java | 30 +------ .../io/druid/query/select/SelectQuery.java | 43 +--------- .../query/timeboundary/TimeBoundaryQuery.java | 28 +------ .../query/timeseries/TimeseriesQuery.java | 55 +----------- .../java/io/druid/query/topn/TopNQuery.java | 42 +--------- .../io/druid/query/topn/TopNQueryBuilder.java | 5 +- .../io/druid/query/QueryContextsTest.java | 2 +- .../server/log/LoggingRequestLoggerTest.java | 9 +- 17 files changed, 37 insertions(+), 495 deletions(-) diff --git a/extensions-contrib/scan-query/src/main/java/io/druid/query/scan/ScanQuery.java b/extensions-contrib/scan-query/src/main/java/io/druid/query/scan/ScanQuery.java index fb3033f2545a..a526868c3639 100644 --- a/extensions-contrib/scan-query/src/main/java/io/druid/query/scan/ScanQuery.java +++ b/extensions-contrib/scan-query/src/main/java/io/druid/query/scan/ScanQuery.java @@ -26,7 +26,6 @@ import io.druid.query.BaseQuery; import io.druid.query.DataSource; import io.druid.query.Query; -import io.druid.query.QueryMetrics; import io.druid.query.TableDataSource; import io.druid.query.filter.DimFilter; import io.druid.query.filter.InDimFilter; @@ -65,22 +64,7 @@ public ScanQuery( @JsonProperty("context") Map context ) { - this(dataSource, querySegmentSpec, resultFormat, batchSize, limit, dimFilter, columns, context, null); - } - - private ScanQuery( - final DataSource dataSource, - final QuerySegmentSpec querySegmentSpec, - final String resultFormat, - final int batchSize, - final long limit, - final DimFilter dimFilter, - final List columns, - final Map context, - final QueryMetrics queryMetrics - ) - { - super(dataSource, querySegmentSpec, false, context, queryMetrics); + super(dataSource, querySegmentSpec, false, context); this.resultFormat = resultFormat == null ? RESULT_FORMAT_LIST : resultFormat; this.batchSize = (batchSize == 0) ? 4096 * 5 : batchSize; this.limit = (limit == 0) ? Long.MAX_VALUE : limit; @@ -161,13 +145,6 @@ public ScanQuery withDimFilter(DimFilter dimFilter) return ScanQueryBuilder.copy(this).filters(dimFilter).build(); } - @Override - public Query withQueryMetrics(QueryMetrics queryMetrics) - { - Preconditions.checkNotNull(queryMetrics); - return ScanQueryBuilder.copy(this).queryMetrics(queryMetrics).build(); - } - @Override public boolean equals(Object o) { @@ -250,7 +227,6 @@ public static class ScanQueryBuilder private long limit; private DimFilter dimFilter; private List columns; - private QueryMetrics queryMetrics; public ScanQueryBuilder() { @@ -262,7 +238,6 @@ public ScanQueryBuilder() limit = 0; dimFilter = null; columns = Lists.newArrayList(); - queryMetrics = null; } public ScanQuery build() @@ -275,8 +250,7 @@ public ScanQuery build() limit, dimFilter, columns, - context, - queryMetrics + context ); } @@ -290,8 +264,7 @@ public static ScanQueryBuilder copy(ScanQuery query) .limit(query.getLimit()) .filters(query.getFilter()) .columns(query.getColumns()) - .context(query.getContext()) - .queryMetrics(query.getQueryMetrics()); + .context(query.getContext()); } public ScanQueryBuilder dataSource(String ds) @@ -377,12 +350,6 @@ public ScanQueryBuilder columns(String... c) columns = Arrays.asList(c); return this; } - - public ScanQueryBuilder queryMetrics(QueryMetrics m) - { - queryMetrics = m; - return this; - } } public static ScanQueryBuilder newScanQueryBuilder() diff --git a/processing/src/main/java/io/druid/query/BaseQuery.java b/processing/src/main/java/io/druid/query/BaseQuery.java index 2c4a07da196b..eea117f59051 100644 --- a/processing/src/main/java/io/druid/query/BaseQuery.java +++ b/processing/src/main/java/io/druid/query/BaseQuery.java @@ -29,7 +29,6 @@ import org.joda.time.Duration; import org.joda.time.Interval; -import javax.annotation.Nullable; import java.util.List; import java.util.Map; @@ -45,46 +44,26 @@ public static void checkInterrupted() } public static final String QUERYID = "queryId"; - private final DataSource dataSource; private final boolean descending; private final Map context; private final QuerySegmentSpec querySegmentSpec; - @Nullable - private final QueryMetrics queryMetrics; private volatile Duration duration; - /** - * @deprecated compatibility constructor for extensions, {@link - * BaseQuery#BaseQuery(DataSource, QuerySegmentSpec, boolean, Map, QueryMetrics)} should be used instead. - */ - @Deprecated public BaseQuery( DataSource dataSource, QuerySegmentSpec querySegmentSpec, boolean descending, Map context ) - { - this(dataSource, querySegmentSpec, descending, context, null); - } - - public BaseQuery( - DataSource dataSource, - QuerySegmentSpec querySegmentSpec, - boolean descending, - Map context, - QueryMetrics queryMetrics - ) { Preconditions.checkNotNull(dataSource, "dataSource can't be null"); Preconditions.checkNotNull(querySegmentSpec, "querySegmentSpec can't be null"); this.dataSource = dataSource; - this.descending = descending; this.context = context; this.querySegmentSpec = querySegmentSpec; - this.queryMetrics = queryMetrics; + this.descending = descending; } @JsonProperty @@ -187,13 +166,6 @@ public Ordering getResultOrdering() return descending ? retVal.reverse() : retVal; } - @Override - @Nullable - public QueryMetrics getQueryMetrics() - { - return queryMetrics; - } - @Override public String getId() { diff --git a/processing/src/main/java/io/druid/query/CPUTimeMetricQueryRunner.java b/processing/src/main/java/io/druid/query/CPUTimeMetricQueryRunner.java index 80f9f48873fd..0d42ab983d4a 100644 --- a/processing/src/main/java/io/druid/query/CPUTimeMetricQueryRunner.java +++ b/processing/src/main/java/io/druid/query/CPUTimeMetricQueryRunner.java @@ -60,16 +60,7 @@ private CPUTimeMetricQueryRunner( @Override public Sequence run(final Query query, final Map responseContext) { - final QueryMetrics> queryMetrics; - final Query queryWithMetrics; - if (query.getQueryMetrics() == null) { - queryMetrics = queryToolChest.makeMetrics(query); - queryWithMetrics = query.withQueryMetrics(queryMetrics); - } else { - queryMetrics = (QueryMetrics>) query.getQueryMetrics(); - queryWithMetrics = query; - } - final Sequence baseSequence = delegate.run(queryWithMetrics, responseContext); + final Sequence baseSequence = delegate.run(query, responseContext); return Sequences.wrap( baseSequence, new SequenceWrapper() @@ -91,14 +82,13 @@ public void after(boolean isDone, Throwable thrown) throws Exception if (report) { final long cpuTimeNs = cpuTimeAccumulator.get(); if (cpuTimeNs > 0) { - queryMetrics.reportCpuTime(cpuTimeNs).emit(emitter); + queryToolChest.makeMetrics(query).reportCpuTime(cpuTimeNs).emit(emitter); } } } } ); } - public static QueryRunner safeBuild( QueryRunner delegate, QueryToolChest> queryToolChest, diff --git a/processing/src/main/java/io/druid/query/Druids.java b/processing/src/main/java/io/druid/query/Druids.java index 0ee78389de56..da81fa9a80ac 100644 --- a/processing/src/main/java/io/druid/query/Druids.java +++ b/processing/src/main/java/io/druid/query/Druids.java @@ -338,7 +338,6 @@ public static class TimeseriesQueryBuilder private List aggregatorSpecs; private List postAggregatorSpecs; private Map context; - private QueryMetrics queryMetrics; private TimeseriesQueryBuilder() { @@ -351,7 +350,6 @@ private TimeseriesQueryBuilder() aggregatorSpecs = Lists.newArrayList(); postAggregatorSpecs = Lists.newArrayList(); context = null; - queryMetrics = null; } public TimeseriesQuery build() @@ -365,8 +363,7 @@ public TimeseriesQuery build() granularity, aggregatorSpecs, postAggregatorSpecs, - context, - queryMetrics + context ); } @@ -381,8 +378,7 @@ public static TimeseriesQueryBuilder copy(TimeseriesQuery query) .granularity(query.getGranularity()) .aggregators(query.getAggregatorSpecs()) .postAggregators(query.getPostAggregatorSpecs()) - .context(query.getContext()) - .queryMetrics(query.getQueryMetrics()); + .context(query.getContext()); } public DataSource getDataSource() @@ -524,12 +520,6 @@ public TimeseriesQueryBuilder context(Map c) context = c; return this; } - - public TimeseriesQueryBuilder queryMetrics(QueryMetrics m) - { - queryMetrics = m; - return this; - } } public static TimeseriesQueryBuilder newTimeseriesQueryBuilder() @@ -567,7 +557,6 @@ public static class SearchQueryBuilder private SearchQuerySpec querySpec; private SearchSortSpec sortSpec; private Map context; - private QueryMetrics queryMetrics; public SearchQueryBuilder() { @@ -580,7 +569,6 @@ public SearchQueryBuilder() querySpec = null; sortSpec = null; context = null; - queryMetrics = null; } public SearchQuery build() @@ -594,8 +582,7 @@ public SearchQuery build() dimensions, querySpec, sortSpec, - context, - queryMetrics + context ); } @@ -610,8 +597,7 @@ public static SearchQueryBuilder copy(SearchQuery query) .dimensions(query.getDimensions()) .query(query.getQuery()) .sortSpec(query.getSort()) - .context(query.getContext()) - .queryMetrics(query.getQueryMetrics()); + .context(query.getContext()); } public SearchQueryBuilder dataSource(String d) @@ -761,12 +747,6 @@ public SearchQueryBuilder context(Map c) context = c; return this; } - - public SearchQueryBuilder queryMetrics(QueryMetrics m) - { - queryMetrics = m; - return this; - } } public static SearchQueryBuilder newSearchQueryBuilder() @@ -795,7 +775,6 @@ public static class TimeBoundaryQueryBuilder private String bound; private DimFilter dimFilter; private Map context; - private QueryMetrics queryMetrics; public TimeBoundaryQueryBuilder() { @@ -804,7 +783,6 @@ public TimeBoundaryQueryBuilder() bound = null; dimFilter = null; context = null; - queryMetrics = null; } public TimeBoundaryQuery build() @@ -814,8 +792,7 @@ public TimeBoundaryQuery build() querySegmentSpec, bound, dimFilter, - context, - queryMetrics + context ); } @@ -826,8 +803,7 @@ public static TimeBoundaryQueryBuilder copy(TimeBoundaryQuery query) .intervals(query.getQuerySegmentSpec()) .bound(query.getBound()) .filters(query.getFilter()) - .context(query.getContext()) - .queryMetrics(query.getQueryMetrics()); + .context(query.getContext()); } public TimeBoundaryQueryBuilder dataSource(String ds) @@ -889,12 +865,6 @@ public TimeBoundaryQueryBuilder context(Map c) context = c; return this; } - - public TimeBoundaryQueryBuilder queryMetrics(QueryMetrics m) - { - queryMetrics = m; - return this; - } } public static TimeBoundaryQueryBuilder newTimeBoundaryQueryBuilder() @@ -992,7 +962,6 @@ public static class SegmentMetadataQueryBuilder private Boolean merge; private Boolean lenientAggregatorMerge; private Map context; - private QueryMetrics queryMetrics; public SegmentMetadataQueryBuilder() { @@ -1003,7 +972,6 @@ public SegmentMetadataQueryBuilder() merge = null; lenientAggregatorMerge = null; context = null; - queryMetrics = null; } public SegmentMetadataQuery build() @@ -1016,8 +984,7 @@ public SegmentMetadataQuery build() context, analysisTypes, false, - lenientAggregatorMerge, - queryMetrics + lenientAggregatorMerge ); } @@ -1030,8 +997,7 @@ public static SegmentMetadataQueryBuilder copy(SegmentMetadataQuery query) .analysisTypes(query.getAnalysisTypes()) .merge(query.isMerge()) .lenientAggregatorMerge(query.isLenientAggregatorMerge()) - .context(query.getContext()) - .queryMetrics(query.getQueryMetrics()); + .context(query.getContext()); } public SegmentMetadataQueryBuilder dataSource(String ds) @@ -1105,12 +1071,6 @@ public SegmentMetadataQueryBuilder context(Map c) context = c; return this; } - - public SegmentMetadataQueryBuilder queryMetrics(QueryMetrics m) - { - queryMetrics = m; - return this; - } } public static SegmentMetadataQueryBuilder newSegmentMetadataQueryBuilder() @@ -1145,7 +1105,6 @@ public static class SelectQueryBuilder private List metrics; private VirtualColumns virtualColumns; private PagingSpec pagingSpec; - private QueryMetrics queryMetrics; public SelectQueryBuilder() { @@ -1159,7 +1118,6 @@ public SelectQueryBuilder() metrics = Lists.newArrayList(); virtualColumns = null; pagingSpec = null; - queryMetrics = null; } public SelectQuery build() @@ -1174,8 +1132,7 @@ public SelectQuery build() metrics, virtualColumns, pagingSpec, - context, - queryMetrics + context ); } @@ -1191,8 +1148,7 @@ public static SelectQueryBuilder copy(SelectQuery query) .metrics(query.getMetrics()) .virtualColumns(query.getVirtualColumns()) .pagingSpec(query.getPagingSpec()) - .context(query.getContext()) - .queryMetrics(query.getQueryMetrics()); + .context(query.getContext()); } public SelectQueryBuilder dataSource(String ds) @@ -1306,12 +1262,6 @@ public SelectQueryBuilder pagingSpec(PagingSpec p) pagingSpec = p; return this; } - - public SelectQueryBuilder queryMetrics(QueryMetrics m) - { - queryMetrics = m; - return this; - } } public static SelectQueryBuilder newSelectQueryBuilder() @@ -1338,14 +1288,12 @@ public static class DataSourceMetadataQueryBuilder private DataSource dataSource; private QuerySegmentSpec querySegmentSpec; private Map context; - private QueryMetrics queryMetrics; public DataSourceMetadataQueryBuilder() { dataSource = null; querySegmentSpec = null; context = null; - queryMetrics = null; } public DataSourceMetadataQuery build() @@ -1353,8 +1301,7 @@ public DataSourceMetadataQuery build() return new DataSourceMetadataQuery( dataSource, querySegmentSpec, - context, - queryMetrics + context ); } @@ -1363,8 +1310,7 @@ public static DataSourceMetadataQueryBuilder copy(DataSourceMetadataQuery query) return new DataSourceMetadataQueryBuilder() .dataSource(query.getDataSource()) .intervals(query.getQuerySegmentSpec()) - .context(query.getContext()) - .queryMetrics(query.getQueryMetrics()); + .context(query.getContext()); } public DataSourceMetadataQueryBuilder dataSource(String ds) @@ -1402,12 +1348,6 @@ public DataSourceMetadataQueryBuilder context(Map c) context = c; return this; } - - public DataSourceMetadataQueryBuilder queryMetrics(QueryMetrics m) - { - queryMetrics = m; - return this; - } } public static DataSourceMetadataQueryBuilder newDataSourceMetadataQueryBuilder() diff --git a/processing/src/main/java/io/druid/query/MetricsEmittingQueryRunner.java b/processing/src/main/java/io/druid/query/MetricsEmittingQueryRunner.java index bc5340602292..1f4041973b4e 100644 --- a/processing/src/main/java/io/druid/query/MetricsEmittingQueryRunner.java +++ b/processing/src/main/java/io/druid/query/MetricsEmittingQueryRunner.java @@ -83,22 +83,15 @@ public MetricsEmittingQueryRunner withWaitMeasuredFromNow() @Override public Sequence run(final Query query, final Map responseContext) { - final QueryMetrics> queryMetrics; - final Query queryWithMetrics; - if (query.getQueryMetrics() == null) { - queryMetrics = queryToolChest.makeMetrics(query); - queryWithMetrics = query.withQueryMetrics(queryMetrics); - } else { - queryMetrics = (QueryMetrics>) query.getQueryMetrics(); - queryWithMetrics = query; - } + final QueryMetrics> queryMetrics = queryToolChest.makeMetrics(query); + applyCustomDimensions.accept(queryMetrics); return Sequences.wrap( // Use LazySequence because want to account execution time of queryRunner.run() (it prepares the underlying // Sequence) as part of the reported query time, i. e. we want to execute queryRunner.run() after // `startTime = System.nanoTime();` (see below). - new LazySequence<>(() -> queryRunner.run(queryWithMetrics, responseContext)), + new LazySequence<>(() -> queryRunner.run(query, responseContext)), new SequenceWrapper() { private long startTimeNs; diff --git a/processing/src/main/java/io/druid/query/Query.java b/processing/src/main/java/io/druid/query/Query.java index 522adff25be7..cfbf6f2d3403 100644 --- a/processing/src/main/java/io/druid/query/Query.java +++ b/processing/src/main/java/io/druid/query/Query.java @@ -36,7 +36,6 @@ import org.joda.time.Duration; import org.joda.time.Interval; -import javax.annotation.Nullable; import java.util.List; import java.util.Map; @@ -101,13 +100,5 @@ public interface Query Query withDataSource(DataSource dataSource); - /** - * @throws NullPointerException if the given queryMetrics is null - */ - Query withQueryMetrics(QueryMetrics queryMetrics); - - @Nullable - QueryMetrics getQueryMetrics(); - Query withDefaultTimeout(long defaultTimeout); } diff --git a/processing/src/main/java/io/druid/query/datasourcemetadata/DataSourceMetadataQuery.java b/processing/src/main/java/io/druid/query/datasourcemetadata/DataSourceMetadataQuery.java index 54ca3f0b009f..e987fe2c4fb6 100644 --- a/processing/src/main/java/io/druid/query/datasourcemetadata/DataSourceMetadataQuery.java +++ b/processing/src/main/java/io/druid/query/datasourcemetadata/DataSourceMetadataQuery.java @@ -21,14 +21,12 @@ import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonProperty; -import com.google.common.base.Preconditions; import com.google.common.collect.Lists; import io.druid.common.utils.JodaUtils; import io.druid.query.BaseQuery; import io.druid.query.DataSource; import io.druid.query.Druids; import io.druid.query.Query; -import io.druid.query.QueryMetrics; import io.druid.query.Result; import io.druid.query.filter.DimFilter; import io.druid.query.spec.MultipleIntervalSegmentSpec; @@ -55,28 +53,13 @@ public DataSourceMetadataQuery( @JsonProperty("intervals") QuerySegmentSpec querySegmentSpec, @JsonProperty("context") Map context ) - { - this(dataSource, querySegmentSpec, context, null); - } - - /** - * This constructor is public only because {@link Druids.DataSourceMetadataQueryBuilder} needs to access this - * constructor, and it is defined in Druids rather than in as an inner class of DataSourceMetadataQuery. - */ - public DataSourceMetadataQuery( - final DataSource dataSource, - final QuerySegmentSpec querySegmentSpec, - final Map context, - final QueryMetrics queryMetrics - ) { super( dataSource, (querySegmentSpec == null) ? new MultipleIntervalSegmentSpec(Collections.singletonList(MY_Y2K_INTERVAL)) : querySegmentSpec, false, - context, - queryMetrics + context ); } @@ -117,13 +100,6 @@ public Query> withDataSource(DataSource da return Druids.DataSourceMetadataQueryBuilder.copy(this).dataSource(dataSource).build(); } - @Override - public Query> withQueryMetrics(QueryMetrics queryMetrics) - { - Preconditions.checkNotNull(queryMetrics); - return Druids.DataSourceMetadataQueryBuilder.copy(this).queryMetrics(queryMetrics).build(); - } - public Iterable> buildResult(DateTime timestamp, DateTime maxIngestedEventTime) { return Arrays.asList(new Result<>(timestamp, new DataSourceMetadataResultValue(maxIngestedEventTime))); diff --git a/processing/src/main/java/io/druid/query/groupby/GroupByQuery.java b/processing/src/main/java/io/druid/query/groupby/GroupByQuery.java index 80623cd5d9a3..0fa37f690568 100644 --- a/processing/src/main/java/io/druid/query/groupby/GroupByQuery.java +++ b/processing/src/main/java/io/druid/query/groupby/GroupByQuery.java @@ -42,7 +42,6 @@ import io.druid.query.Queries; import io.druid.query.Query; import io.druid.query.QueryDataSource; -import io.druid.query.QueryMetrics; import io.druid.query.TableDataSource; import io.druid.query.aggregation.AggregatorFactory; import io.druid.query.aggregation.PostAggregator; @@ -113,39 +112,7 @@ public GroupByQuery( @JsonProperty("context") Map context ) { - this( - dataSource, - querySegmentSpec, - virtualColumns, - dimFilter, - granularity, - dimensions, - aggregatorSpecs, - postAggregatorSpecs, - havingSpec, - limitSpec, - context, - null - ); - } - - private GroupByQuery( - final DataSource dataSource, - final QuerySegmentSpec querySegmentSpec, - final VirtualColumns virtualColumns, - final DimFilter dimFilter, - final Granularity granularity, - final List dimensions, - final List aggregatorSpecs, - final List postAggregatorSpecs, - final HavingSpec havingSpec, - final LimitSpec limitSpec, - final Map context, - final QueryMetrics queryMetrics - ) - { - super(dataSource, querySegmentSpec, false, context, queryMetrics); - GroupByQueryMetrics.class.cast(queryMetrics); // ClassCastException if not + super(dataSource, querySegmentSpec, false, context); this.virtualColumns = VirtualColumns.nullToEmpty(virtualColumns); this.dimFilter = dimFilter; @@ -211,12 +178,10 @@ private GroupByQuery( final HavingSpec havingSpec, final LimitSpec orderBySpec, final Function, Sequence> limitFn, - final Map context, - final QueryMetrics queryMetrics + final Map context ) { - super(dataSource, querySegmentSpec, false, context, queryMetrics); - GroupByQueryMetrics.class.cast(queryMetrics); // ClassCastException if not + super(dataSource, querySegmentSpec, false, context); this.virtualColumns = virtualColumns; this.dimFilter = dimFilter; @@ -438,13 +403,6 @@ public GroupByQuery withPostAggregatorSpecs(final List postAggre return new Builder(this).setPostAggregatorSpecs(postAggregatorSpecs).build(); } - @Override - public Query withQueryMetrics(QueryMetrics queryMetrics) - { - Preconditions.checkNotNull(queryMetrics); - return new Builder(this).setQueryMetrics(queryMetrics).build(); - } - private static void verifyOutputNames( List dimensions, List aggregators, @@ -496,7 +454,6 @@ public static class Builder private Function, Sequence> limitFn; private List orderByColumnSpecs = Lists.newArrayList(); private int limit = Integer.MAX_VALUE; - private QueryMetrics queryMetrics; public Builder() { @@ -516,7 +473,6 @@ public Builder(GroupByQuery query) limitSpec = query.getLimitSpec(); limitFn = query.limitFn; context = query.getContext(); - queryMetrics = query.getQueryMetrics(); } public Builder(Builder builder) @@ -535,7 +491,6 @@ public Builder(Builder builder) limit = builder.limit; orderByColumnSpecs = new ArrayList<>(builder.orderByColumnSpecs); context = builder.context; - queryMetrics = builder.queryMetrics; } public Builder setDataSource(DataSource dataSource) @@ -741,12 +696,6 @@ public Builder setLimit(Integer limit) return this; } - public Builder setQueryMetrics(QueryMetrics queryMetrics) - { - this.queryMetrics = queryMetrics; - return this; - } - public Builder copy() { return new Builder(this); @@ -778,8 +727,7 @@ public GroupByQuery build() havingSpec, theLimitSpec, limitFn, - context, - queryMetrics + context ); } else { return new GroupByQuery( @@ -793,8 +741,7 @@ public GroupByQuery build() postAggregatorSpecs, havingSpec, theLimitSpec, - context, - queryMetrics + context ); } } diff --git a/processing/src/main/java/io/druid/query/metadata/metadata/SegmentMetadataQuery.java b/processing/src/main/java/io/druid/query/metadata/metadata/SegmentMetadataQuery.java index 12a05b4656c8..130025c60770 100644 --- a/processing/src/main/java/io/druid/query/metadata/metadata/SegmentMetadataQuery.java +++ b/processing/src/main/java/io/druid/query/metadata/metadata/SegmentMetadataQuery.java @@ -29,7 +29,6 @@ import io.druid.query.DataSource; import io.druid.query.Druids; import io.druid.query.Query; -import io.druid.query.QueryMetrics; import io.druid.query.TableDataSource; import io.druid.query.UnionDataSource; import io.druid.query.filter.DimFilter; @@ -109,43 +108,13 @@ public SegmentMetadataQuery( @JsonProperty("usingDefaultInterval") Boolean useDefaultInterval, @JsonProperty("lenientAggregatorMerge") Boolean lenientAggregatorMerge ) - { - this( - dataSource, - querySegmentSpec, - toInclude, - merge, - context, - analysisTypes, - useDefaultInterval, - lenientAggregatorMerge, - null - ); - } - - /** - * This constructor is public only because {@link Druids.SegmentMetadataQueryBuilder} needs to access this - * constructor, and it is defined in Druids rather than in as an inner class of SegmentMetadataQuery. - */ - public SegmentMetadataQuery( - final DataSource dataSource, - final QuerySegmentSpec querySegmentSpec, - final ColumnIncluderator toInclude, - final Boolean merge, - final Map context, - final EnumSet analysisTypes, - final Boolean useDefaultInterval, - final Boolean lenientAggregatorMerge, - final QueryMetrics queryMetrics - ) { super( dataSource, (querySegmentSpec == null) ? new MultipleIntervalSegmentSpec(Arrays.asList(DEFAULT_INTERVAL)) : querySegmentSpec, false, - context, - queryMetrics + context ); if (querySegmentSpec == null) { @@ -285,13 +254,6 @@ public Query withColumns(ColumnIncluderator includerator) return Druids.SegmentMetadataQueryBuilder.copy(this).toInclude(includerator).build(); } - @Override - public Query withQueryMetrics(QueryMetrics queryMetrics) - { - Preconditions.checkNotNull(queryMetrics); - return Druids.SegmentMetadataQueryBuilder.copy(this).queryMetrics(queryMetrics).build(); - } - @Override public String toString() { diff --git a/processing/src/main/java/io/druid/query/search/search/SearchQuery.java b/processing/src/main/java/io/druid/query/search/search/SearchQuery.java index 7abde034f3cf..a9971907dd5d 100644 --- a/processing/src/main/java/io/druid/query/search/search/SearchQuery.java +++ b/processing/src/main/java/io/druid/query/search/search/SearchQuery.java @@ -28,7 +28,6 @@ import io.druid.query.DataSource; import io.druid.query.Druids; import io.druid.query.Query; -import io.druid.query.QueryMetrics; import io.druid.query.Result; import io.druid.query.dimension.DimensionSpec; import io.druid.query.filter.DimFilter; @@ -65,27 +64,7 @@ public SearchQuery( @JsonProperty("context") Map context ) { - this(dataSource, dimFilter, granularity, limit, querySegmentSpec, dimensions, querySpec, sortSpec, context, null); - } - - /** - * This constructor is public only because {@link Druids.SearchQueryBuilder} needs to access this constructor, and it - * is defined in Druids rather than in as an inner class of SearchQuery. - */ - public SearchQuery( - final DataSource dataSource, - final DimFilter dimFilter, - final Granularity granularity, - final int limit, - final QuerySegmentSpec querySegmentSpec, - final List dimensions, - final SearchQuerySpec querySpec, - final SearchSortSpec sortSpec, - final Map context, - final QueryMetrics queryMetrics - ) - { - super(dataSource, querySegmentSpec, false, context, queryMetrics); + super(dataSource, querySegmentSpec, false, context); Preconditions.checkNotNull(querySegmentSpec, "Must specify an interval"); this.dimFilter = dimFilter; @@ -138,13 +117,6 @@ public SearchQuery withDimFilter(DimFilter dimFilter) return Druids.SearchQueryBuilder.copy(this).filters(dimFilter).build(); } - @Override - public Query> withQueryMetrics(QueryMetrics queryMetrics) - { - Preconditions.checkNotNull(queryMetrics); - return Druids.SearchQueryBuilder.copy(this).queryMetrics(queryMetrics).build(); - } - @JsonProperty("filter") public DimFilter getDimensionsFilter() { diff --git a/processing/src/main/java/io/druid/query/select/SelectQuery.java b/processing/src/main/java/io/druid/query/select/SelectQuery.java index 728bde8416f7..72059580c4f2 100644 --- a/processing/src/main/java/io/druid/query/select/SelectQuery.java +++ b/processing/src/main/java/io/druid/query/select/SelectQuery.java @@ -28,7 +28,6 @@ import io.druid.query.DataSource; import io.druid.query.Druids; import io.druid.query.Query; -import io.druid.query.QueryMetrics; import io.druid.query.Result; import io.druid.query.dimension.DimensionSpec; import io.druid.query.filter.DimFilter; @@ -65,40 +64,7 @@ public SelectQuery( @JsonProperty("context") Map context ) { - this( - dataSource, - querySegmentSpec, - descending, - dimFilter, - granularity, - dimensions, - metrics, - virtualColumns, - pagingSpec, - context, - null - ); - } - - /** - * This constructor is public only because {@link Druids.SelectQueryBuilder} needs to access this constructor, and it - * is defined in Druids rather than in as an inner class of SelectQuery. - */ - public SelectQuery( - final DataSource dataSource, - final QuerySegmentSpec querySegmentSpec, - final boolean descending, - final DimFilter dimFilter, - final Granularity granularity, - final List dimensions, - final List metrics, - final VirtualColumns virtualColumns, - final PagingSpec pagingSpec, - final Map context, - final QueryMetrics queryMetrics - ) - { - super(dataSource, querySegmentSpec, descending, context, queryMetrics); + super(dataSource, querySegmentSpec, descending, context); this.dimFilter = dimFilter; this.granularity = granularity; this.dimensions = dimensions; @@ -206,13 +172,6 @@ public SelectQuery withDimFilter(DimFilter dimFilter) return Druids.SelectQueryBuilder.copy(this).filters(dimFilter).build(); } - @Override - public Query> withQueryMetrics(QueryMetrics queryMetrics) - { - Preconditions.checkNotNull(queryMetrics); - return Druids.SelectQueryBuilder.copy(this).queryMetrics(queryMetrics).build(); - } - @Override public String toString() { diff --git a/processing/src/main/java/io/druid/query/timeboundary/TimeBoundaryQuery.java b/processing/src/main/java/io/druid/query/timeboundary/TimeBoundaryQuery.java index 9b4e14f4bcb6..d466cd211406 100644 --- a/processing/src/main/java/io/druid/query/timeboundary/TimeBoundaryQuery.java +++ b/processing/src/main/java/io/druid/query/timeboundary/TimeBoundaryQuery.java @@ -21,7 +21,6 @@ import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonProperty; -import com.google.common.base.Preconditions; import com.google.common.collect.Lists; import com.google.common.collect.Maps; import io.druid.common.utils.JodaUtils; @@ -30,7 +29,6 @@ import io.druid.query.DataSource; import io.druid.query.Druids; import io.druid.query.Query; -import io.druid.query.QueryMetrics; import io.druid.query.Result; import io.druid.query.filter.DimFilter; import io.druid.query.spec.MultipleIntervalSegmentSpec; @@ -67,30 +65,13 @@ public TimeBoundaryQuery( @JsonProperty("filter") DimFilter dimFilter, @JsonProperty("context") Map context ) - { - this(dataSource, querySegmentSpec, bound, dimFilter, context, null); - } - - /** - * This constructor is public only because {@link Druids.TimeBoundaryQueryBuilder} needs to access this constructor, - * and it is defined in Druids rather than in as an inner class of TimeBoundaryQuery. - */ - public TimeBoundaryQuery( - final DataSource dataSource, - final QuerySegmentSpec querySegmentSpec, - final String bound, - final DimFilter dimFilter, - final Map context, - final QueryMetrics queryMetrics - ) { super( dataSource, (querySegmentSpec == null) ? new MultipleIntervalSegmentSpec(Arrays.asList(MY_Y2K_INTERVAL)) : querySegmentSpec, false, - context, - queryMetrics + context ); this.dimFilter = dimFilter; @@ -140,13 +121,6 @@ public Query> withDataSource(DataSource dataSour return Druids.TimeBoundaryQueryBuilder.copy(this).dataSource(dataSource).build(); } - @Override - public Query> withQueryMetrics(QueryMetrics queryMetrics) - { - Preconditions.checkNotNull(queryMetrics); - return Druids.TimeBoundaryQueryBuilder.copy(this).queryMetrics(queryMetrics).build(); - } - public byte[] getCacheKey() { final byte[] filterBytes = dimFilter == null ? new byte[]{} : dimFilter.getCacheKey(); diff --git a/processing/src/main/java/io/druid/query/timeseries/TimeseriesQuery.java b/processing/src/main/java/io/druid/query/timeseries/TimeseriesQuery.java index 63b29511dab6..328400291348 100644 --- a/processing/src/main/java/io/druid/query/timeseries/TimeseriesQuery.java +++ b/processing/src/main/java/io/druid/query/timeseries/TimeseriesQuery.java @@ -22,7 +22,6 @@ import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonProperty; import com.fasterxml.jackson.annotation.JsonTypeName; -import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableList; import io.druid.java.util.common.granularity.Granularity; import io.druid.query.BaseQuery; @@ -30,7 +29,6 @@ import io.druid.query.Druids; import io.druid.query.Queries; import io.druid.query.Query; -import io.druid.query.QueryMetrics; import io.druid.query.Result; import io.druid.query.aggregation.AggregatorFactory; import io.druid.query.aggregation.PostAggregator; @@ -66,39 +64,7 @@ public TimeseriesQuery( @JsonProperty("context") Map context ) { - this( - dataSource, - querySegmentSpec, - descending, - virtualColumns, - dimFilter, - granularity, - aggregatorSpecs, - postAggregatorSpecs, - context, - null - ); - } - - /** - * This constructor is public only because {@link Druids.TimeseriesQueryBuilder} needs to access this constructor, and - * it is defined in Druids rather than in as an inner class of TimeseriesQuery. - */ - public TimeseriesQuery( - final DataSource dataSource, - final QuerySegmentSpec querySegmentSpec, - final boolean descending, - final VirtualColumns virtualColumns, - final DimFilter dimFilter, - final Granularity granularity, - final List aggregatorSpecs, - final List postAggregatorSpecs, - final Map context, - final QueryMetrics queryMetrics - ) - { - super(dataSource, querySegmentSpec, descending, context, queryMetrics); - TimeseriesQueryMetrics.class.cast(queryMetrics); // ClassCastException if not + super(dataSource, querySegmentSpec, descending, context); this.virtualColumns = VirtualColumns.nullToEmpty(virtualColumns); this.dimFilter = dimFilter; @@ -185,26 +151,9 @@ public TimeseriesQuery withDimFilter(DimFilter dimFilter) return Druids.TimeseriesQueryBuilder.copy(this).filters(dimFilter).build(); } - @Override - public Query> withQueryMetrics(QueryMetrics queryMetrics) - { - Preconditions.checkNotNull(queryMetrics); - return Druids.TimeseriesQueryBuilder.copy(this).queryMetrics(queryMetrics).build(); - } - public TimeseriesQuery withPostAggregatorSpecs(final List postAggregatorSpecs) { - return new TimeseriesQuery( - getDataSource(), - getQuerySegmentSpec(), - isDescending(), - virtualColumns, - dimFilter, - granularity, - aggregatorSpecs, - postAggregatorSpecs, - getContext() - ); + return Druids.TimeseriesQueryBuilder.copy(this).postAggregators(postAggregatorSpecs).build(); } @Override diff --git a/processing/src/main/java/io/druid/query/topn/TopNQuery.java b/processing/src/main/java/io/druid/query/topn/TopNQuery.java index a8177a016ba7..df7583ebd760 100644 --- a/processing/src/main/java/io/druid/query/topn/TopNQuery.java +++ b/processing/src/main/java/io/druid/query/topn/TopNQuery.java @@ -28,7 +28,6 @@ import io.druid.query.DataSource; import io.druid.query.Queries; import io.druid.query.Query; -import io.druid.query.QueryMetrics; import io.druid.query.Result; import io.druid.query.aggregation.AggregatorFactory; import io.druid.query.aggregation.PostAggregator; @@ -71,39 +70,7 @@ public TopNQuery( @JsonProperty("context") Map context ) { - this( - dataSource, - virtualColumns, - dimensionSpec, - topNMetricSpec, - threshold, - querySegmentSpec, - dimFilter, - granularity, - aggregatorSpecs, - postAggregatorSpecs, - context, - null - ); - } - - TopNQuery( - final DataSource dataSource, - final VirtualColumns virtualColumns, - final DimensionSpec dimensionSpec, - final TopNMetricSpec topNMetricSpec, - final int threshold, - final QuerySegmentSpec querySegmentSpec, - final DimFilter dimFilter, - final Granularity granularity, - final List aggregatorSpecs, - final List postAggregatorSpecs, - final Map context, - final QueryMetrics queryMetrics - ) - { - super(dataSource, querySegmentSpec, false, context, queryMetrics); - TopNQueryMetrics.class.cast(queryMetrics); // ClassCastException if not + super(dataSource, querySegmentSpec, false, context); this.virtualColumns = VirtualColumns.nullToEmpty(virtualColumns); this.dimensionSpec = dimensionSpec; @@ -242,13 +209,6 @@ public TopNQuery withDimFilter(DimFilter dimFilter) return new TopNQueryBuilder(this).filters(dimFilter).build(); } - @Override - public Query> withQueryMetrics(QueryMetrics queryMetrics) - { - Preconditions.checkNotNull(queryMetrics); - return new TopNQueryBuilder(this).queryMetrics(queryMetrics).build(); - } - @Override public String toString() { diff --git a/processing/src/main/java/io/druid/query/topn/TopNQueryBuilder.java b/processing/src/main/java/io/druid/query/topn/TopNQueryBuilder.java index 6bc33a8bec59..4a06a486f390 100644 --- a/processing/src/main/java/io/druid/query/topn/TopNQueryBuilder.java +++ b/processing/src/main/java/io/druid/query/topn/TopNQueryBuilder.java @@ -91,7 +91,6 @@ public TopNQueryBuilder() aggregatorSpecs = Lists.newArrayList(); postAggregatorSpecs = Lists.newArrayList(); context = null; - queryMetrics = null; } public TopNQueryBuilder(final TopNQuery query) @@ -107,7 +106,6 @@ public TopNQueryBuilder(final TopNQuery query) this.aggregatorSpecs = query.getAggregatorSpecs(); this.postAggregatorSpecs = query.getPostAggregatorSpecs(); this.context = query.getContext(); - this.queryMetrics = query.getQueryMetrics(); } public DataSource getDataSource() @@ -178,8 +176,7 @@ public TopNQuery build() granularity, aggregatorSpecs, postAggregatorSpecs, - context, - queryMetrics + context ); } diff --git a/processing/src/test/java/io/druid/query/QueryContextsTest.java b/processing/src/test/java/io/druid/query/QueryContextsTest.java index c656f0771101..07528ba276a5 100644 --- a/processing/src/test/java/io/druid/query/QueryContextsTest.java +++ b/processing/src/test/java/io/druid/query/QueryContextsTest.java @@ -78,7 +78,7 @@ public Query withOverriddenContext(Map contextOverride) getDataSource(), getQuerySegmentSpec(), isDescending(), - computeOverridenContext(contextOverride) + BaseQuery.computeOverriddenContext(getContext(), contextOverride) ); } } diff --git a/server/src/test/java/io/druid/server/log/LoggingRequestLoggerTest.java b/server/src/test/java/io/druid/server/log/LoggingRequestLoggerTest.java index cc0f1cd38ceb..18e330fcb02c 100644 --- a/server/src/test/java/io/druid/server/log/LoggingRequestLoggerTest.java +++ b/server/src/test/java/io/druid/server/log/LoggingRequestLoggerTest.java @@ -29,7 +29,6 @@ import io.druid.query.DataSource; import io.druid.query.LegacyDataSource; import io.druid.query.Query; -import io.druid.query.QueryMetrics; import io.druid.query.QueryRunner; import io.druid.query.QuerySegmentWalker; import io.druid.query.filter.DimFilter; @@ -184,7 +183,7 @@ class FakeQuery extends BaseQuery { public FakeQuery(DataSource dataSource, QuerySegmentSpec querySegmentSpec, boolean descending, Map context) { - super(dataSource, querySegmentSpec, descending, context, null); + super(dataSource, querySegmentSpec, descending, context); } @Override @@ -222,10 +221,4 @@ public Query withOverriddenContext(Map contextOverride) { throw new UnsupportedOperationException("shouldn't be here"); } - - @Override - public Query withQueryMetrics(QueryMetrics queryMetrics) - { - throw new UnsupportedOperationException("shouldn't be here"); - } } From 472d08bc597576fe982458eb3b057b4c373413e4 Mon Sep 17 00:00:00 2001 From: leventov Date: Tue, 25 Apr 2017 00:13:27 +0300 Subject: [PATCH 6/7] Move nullToNoopLimitSpec() method to LimitSpec interface --- .../src/main/java/io/druid/query/groupby/GroupByQuery.java | 7 +------ .../java/io/druid/query/groupby/orderby/LimitSpec.java | 6 ++++++ 2 files changed, 7 insertions(+), 6 deletions(-) diff --git a/processing/src/main/java/io/druid/query/groupby/GroupByQuery.java b/processing/src/main/java/io/druid/query/groupby/GroupByQuery.java index 0fa37f690568..bf50da9d7443 100644 --- a/processing/src/main/java/io/druid/query/groupby/GroupByQuery.java +++ b/processing/src/main/java/io/druid/query/groupby/GroupByQuery.java @@ -127,7 +127,7 @@ public GroupByQuery( postAggregatorSpecs == null ? ImmutableList.of() : postAggregatorSpecs ); this.havingSpec = havingSpec; - this.limitSpec = nullToNoopLimitSpec(limitSpec); + this.limitSpec = LimitSpec.nullToNoopLimitSpec(limitSpec); Preconditions.checkNotNull(this.granularity, "Must specify a granularity"); @@ -157,11 +157,6 @@ private Function, Sequence> makeLimitFn(LimitSpec limitSpec) return postProcFn; } - private static LimitSpec nullToNoopLimitSpec(LimitSpec limitSpec) - { - return (limitSpec == null) ? NoopLimitSpec.instance() : limitSpec; - } - /** * A private constructor that avoids all of the various state checks. Used by the with*() methods where the checks * have already passed in order for the object to exist. diff --git a/processing/src/main/java/io/druid/query/groupby/orderby/LimitSpec.java b/processing/src/main/java/io/druid/query/groupby/orderby/LimitSpec.java index 9581dc35e7a5..4638e6a49dc4 100644 --- a/processing/src/main/java/io/druid/query/groupby/orderby/LimitSpec.java +++ b/processing/src/main/java/io/druid/query/groupby/orderby/LimitSpec.java @@ -29,6 +29,7 @@ import io.druid.query.aggregation.PostAggregator; import io.druid.query.dimension.DimensionSpec; +import javax.annotation.Nullable; import java.util.List; /** @@ -39,6 +40,11 @@ }) public interface LimitSpec extends Cacheable { + static LimitSpec nullToNoopLimitSpec(@Nullable LimitSpec limitSpec) + { + return (limitSpec == null) ? NoopLimitSpec.instance() : limitSpec; + } + /** * Returns a function that applies a limit to an input sequence that is assumed to be sorted on dimensions. * From eb77aabcc290ef71f15d7867c7c29902de9bcd48 Mon Sep 17 00:00:00 2001 From: leventov Date: Tue, 25 Apr 2017 00:54:15 +0300 Subject: [PATCH 7/7] Rename GroupByQuery.applyLimit() to postProcess(); Fix inconsistencies in GroupByQuery.Builder --- .../io/druid/query/groupby/GroupByQuery.java | 165 ++++++++---------- .../groupby/strategy/GroupByStrategyV1.java | 4 +- .../groupby/strategy/GroupByStrategyV2.java | 2 +- .../query/groupby/GroupByQueryRunnerTest.java | 16 +- 4 files changed, 88 insertions(+), 99 deletions(-) diff --git a/processing/src/main/java/io/druid/query/groupby/GroupByQuery.java b/processing/src/main/java/io/druid/query/groupby/GroupByQuery.java index bf50da9d7443..553e8a2a95c0 100644 --- a/processing/src/main/java/io/druid/query/groupby/GroupByQuery.java +++ b/processing/src/main/java/io/druid/query/groupby/GroupByQuery.java @@ -60,6 +60,7 @@ import io.druid.segment.column.Column; import org.joda.time.Interval; +import javax.annotation.Nullable; import java.util.ArrayList; import java.util.Arrays; import java.util.Comparator; @@ -95,7 +96,7 @@ public static Builder builder() private final List aggregatorSpecs; private final List postAggregatorSpecs; - private final Function, Sequence> limitFn; + private final Function, Sequence> postProcessingFn; @JsonCreator public GroupByQuery( @@ -112,54 +113,41 @@ public GroupByQuery( @JsonProperty("context") Map context ) { - super(dataSource, querySegmentSpec, false, context); - - this.virtualColumns = VirtualColumns.nullToEmpty(virtualColumns); - this.dimFilter = dimFilter; - this.granularity = granularity; - this.dimensions = dimensions == null ? ImmutableList.of() : dimensions; - for (DimensionSpec spec : this.dimensions) { - Preconditions.checkArgument(spec != null, "dimensions has null DimensionSpec"); - } - this.aggregatorSpecs = aggregatorSpecs == null ? ImmutableList.of() : aggregatorSpecs; - this.postAggregatorSpecs = Queries.prepareAggregations( - this.aggregatorSpecs, - postAggregatorSpecs == null ? ImmutableList.of() : postAggregatorSpecs + this( + dataSource, + querySegmentSpec, + virtualColumns, + dimFilter, + granularity, + dimensions, + aggregatorSpecs, + postAggregatorSpecs, + havingSpec, + limitSpec, + null, + context ); - this.havingSpec = havingSpec; - this.limitSpec = LimitSpec.nullToNoopLimitSpec(limitSpec); - - Preconditions.checkNotNull(this.granularity, "Must specify a granularity"); - - - // Verify no duplicate names between dimensions, aggregators, and postAggregators. - // They will all end up in the same namespace in the returned Rows and we can't have them clobbering each other. - // We're not counting __time, even though that name is problematic. See: https://github.com/druid-io/druid/pull/3684 - verifyOutputNames(this.dimensions, this.aggregatorSpecs, this.postAggregatorSpecs); - - limitFn = makeLimitFn(this.limitSpec); } - private Function, Sequence> makeLimitFn(LimitSpec limitSpec) + private Function, Sequence> makePostProcessingFn() { - Function, Sequence> postProcFn = + Function, Sequence> postProcessingFn = limitSpec.build(dimensions, aggregatorSpecs, postAggregatorSpecs); if (havingSpec != null) { - postProcFn = Functions.compose( - postProcFn, + postProcessingFn = Functions.compose( + postProcessingFn, (Sequence input) -> { havingSpec.setRowSignature(GroupByQueryHelper.rowSignatureFor(GroupByQuery.this)); return Sequences.filter(input, havingSpec::eval); } ); } - return postProcFn; + return postProcessingFn; } /** - * A private constructor that avoids all of the various state checks. Used by the with*() methods where the checks - * have already passed in order for the object to exist. + * A private constructor that avoids recomputing postProcessingFn. */ private GroupByQuery( final DataSource dataSource, @@ -171,22 +159,36 @@ private GroupByQuery( final List aggregatorSpecs, final List postAggregatorSpecs, final HavingSpec havingSpec, - final LimitSpec orderBySpec, - final Function, Sequence> limitFn, + final LimitSpec limitSpec, + final @Nullable Function, Sequence> postProcessingFn, final Map context ) { super(dataSource, querySegmentSpec, false, context); - this.virtualColumns = virtualColumns; + this.virtualColumns = VirtualColumns.nullToEmpty(virtualColumns); this.dimFilter = dimFilter; this.granularity = granularity; - this.dimensions = dimensions; - this.aggregatorSpecs = aggregatorSpecs; - this.postAggregatorSpecs = postAggregatorSpecs; + this.dimensions = dimensions == null ? ImmutableList.of() : dimensions; + for (DimensionSpec spec : this.dimensions) { + Preconditions.checkArgument(spec != null, "dimensions has null DimensionSpec"); + } + this.aggregatorSpecs = aggregatorSpecs == null ? ImmutableList.of() : aggregatorSpecs; + this.postAggregatorSpecs = Queries.prepareAggregations( + this.aggregatorSpecs, + postAggregatorSpecs == null ? ImmutableList.of() : postAggregatorSpecs + ); this.havingSpec = havingSpec; - this.limitSpec = orderBySpec; - this.limitFn = limitFn; + this.limitSpec = LimitSpec.nullToNoopLimitSpec(limitSpec); + + Preconditions.checkNotNull(this.granularity, "Must specify a granularity"); + + // Verify no duplicate names between dimensions, aggregators, and postAggregators. + // They will all end up in the same namespace in the returned Rows and we can't have them clobbering each other. + // We're not counting __time, even though that name is problematic. See: https://github.com/druid-io/druid/pull/3684 + verifyOutputNames(this.dimensions, this.aggregatorSpecs, this.postAggregatorSpecs); + + this.postProcessingFn = postProcessingFn != null ? postProcessingFn : makePostProcessingFn(); } @JsonProperty @@ -350,9 +352,9 @@ private static int compareDims(List dimensions, Row lhs, Row rhs) * * @return sequence of rows after applying havingSpec and limitSpec */ - public Sequence applyLimit(Sequence results) + public Sequence postProcess(Sequence results) { - return limitFn.apply(results); + return postProcessingFn.apply(results); } @Override @@ -446,7 +448,7 @@ public static class Builder private Map context; private LimitSpec limitSpec = null; - private Function, Sequence> limitFn; + private Function, Sequence> postProcessingFn; private List orderByColumnSpecs = Lists.newArrayList(); private int limit = Integer.MAX_VALUE; @@ -466,7 +468,7 @@ public Builder(GroupByQuery query) postAggregatorSpecs = query.getPostAggregatorSpecs(); havingSpec = query.getHavingSpec(); limitSpec = query.getLimitSpec(); - limitFn = query.limitFn; + postProcessingFn = query.postProcessingFn; context = query.getContext(); } @@ -482,7 +484,7 @@ public Builder(Builder builder) postAggregatorSpecs = builder.postAggregatorSpecs; havingSpec = builder.havingSpec; limitSpec = builder.limitSpec; - limitFn = builder.limitFn; + postProcessingFn = builder.postProcessingFn; limit = builder.limit; orderByColumnSpecs = new ArrayList<>(builder.orderByColumnSpecs); context = builder.context; @@ -544,16 +546,17 @@ public Builder setVirtualColumns(VirtualColumn... virtualColumns) return this; } - public Builder limit(int limit) + public Builder setLimit(int limit) { - ensureExplicitLimitNotSet(); + ensureExplicitLimitSpecNotSet(); this.limit = limit; + this.postProcessingFn = null; return this; } public Builder addOrderByColumn(String dimension) { - return addOrderByColumn(dimension, (OrderByColumnSpec.Direction) null); + return addOrderByColumn(dimension, null); } public Builder addOrderByColumn(String dimension, OrderByColumnSpec.Direction direction) @@ -563,8 +566,9 @@ public Builder addOrderByColumn(String dimension, OrderByColumnSpec.Direction di public Builder addOrderByColumn(OrderByColumnSpec columnSpec) { - ensureExplicitLimitNotSet(); + ensureExplicitLimitSpecNotSet(); this.orderByColumnSpecs.add(columnSpec); + this.postProcessingFn = null; return this; } @@ -573,11 +577,11 @@ public Builder setLimitSpec(LimitSpec limitSpec) Preconditions.checkNotNull(limitSpec); ensureFluentLimitsNotSet(); this.limitSpec = limitSpec; - this.limitFn = null; + this.postProcessingFn = null; return this; } - private void ensureExplicitLimitNotSet() + private void ensureExplicitLimitSpecNotSet() { if (limitSpec != null) { throw new ISE("Ambiguous build, limitSpec[%s] already set", limitSpec); @@ -626,12 +630,14 @@ public Builder addDimension(DimensionSpec dimension) } dimensions.add(dimension); + this.postProcessingFn = null; return this; } public Builder setDimensions(List dimensions) { this.dimensions = Lists.newArrayList(dimensions); + this.postProcessingFn = null; return this; } @@ -642,12 +648,14 @@ public Builder addAggregator(AggregatorFactory aggregator) } aggregatorSpecs.add(aggregator); + this.postProcessingFn = null; return this; } public Builder setAggregatorSpecs(List aggregatorSpecs) { this.aggregatorSpecs = Lists.newArrayList(aggregatorSpecs); + this.postProcessingFn = null; return this; } @@ -658,12 +666,14 @@ public Builder addPostAggregator(PostAggregator postAgg) } postAggregatorSpecs.add(postAgg); + this.postProcessingFn = null; return this; } public Builder setPostAggregatorSpecs(List postAggregatorSpecs) { this.postAggregatorSpecs = Lists.newArrayList(postAggregatorSpecs); + this.postProcessingFn = null; return this; } @@ -682,12 +692,7 @@ public Builder overrideContext(Map contextOverride) public Builder setHavingSpec(HavingSpec havingSpec) { this.havingSpec = havingSpec; - return this; - } - - public Builder setLimit(Integer limit) - { - this.limit = limit; + this.postProcessingFn = null; return this; } @@ -709,36 +714,20 @@ public GroupByQuery build() theLimitSpec = limitSpec; } - if (limitFn != null) { - return new GroupByQuery( - dataSource, - querySegmentSpec, - virtualColumns, - dimFilter, - granularity, - dimensions, - aggregatorSpecs, - postAggregatorSpecs, - havingSpec, - theLimitSpec, - limitFn, - context - ); - } else { - return new GroupByQuery( - dataSource, - querySegmentSpec, - virtualColumns, - dimFilter, - granularity, - dimensions, - aggregatorSpecs, - postAggregatorSpecs, - havingSpec, - theLimitSpec, - context - ); - } + return new GroupByQuery( + dataSource, + querySegmentSpec, + virtualColumns, + dimFilter, + granularity, + dimensions, + aggregatorSpecs, + postAggregatorSpecs, + havingSpec, + theLimitSpec, + postProcessingFn, + context + ); } } diff --git a/processing/src/main/java/io/druid/query/groupby/strategy/GroupByStrategyV1.java b/processing/src/main/java/io/druid/query/groupby/strategy/GroupByStrategyV1.java index c51caaf1c5e8..4ba495356ba3 100644 --- a/processing/src/main/java/io/druid/query/groupby/strategy/GroupByStrategyV1.java +++ b/processing/src/main/java/io/druid/query/groupby/strategy/GroupByStrategyV1.java @@ -144,7 +144,7 @@ public Sequence mergeResults( true ); - return Sequences.withBaggage(query.applyLimit(GroupByQueryHelper.postAggregate(query, index)), index); + return Sequences.withBaggage(query.postProcess(GroupByQueryHelper.postAggregate(query, index)), index); } @Override @@ -247,7 +247,7 @@ public Sequence apply(Interval interval) innerQueryResultIndex.close(); return Sequences.withBaggage( - outerQuery.applyLimit(GroupByQueryHelper.postAggregate(query, outerQueryResultIndex)), + outerQuery.postProcess(GroupByQueryHelper.postAggregate(query, outerQueryResultIndex)), outerQueryResultIndex ); } diff --git a/processing/src/main/java/io/druid/query/groupby/strategy/GroupByStrategyV2.java b/processing/src/main/java/io/druid/query/groupby/strategy/GroupByStrategyV2.java index 39e6e850bc89..e13b4e7c6fb5 100644 --- a/processing/src/main/java/io/druid/query/groupby/strategy/GroupByStrategyV2.java +++ b/processing/src/main/java/io/druid/query/groupby/strategy/GroupByStrategyV2.java @@ -230,7 +230,7 @@ protected BinaryFn createMergeFn(Query queryParam) // Fudge timestamp, maybe. final DateTime fudgeTimestamp = getUniversalTimestamp(query); - return query.applyLimit( + return query.postProcess( Sequences.map( mergingQueryRunner.run( new GroupByQuery.Builder(query) diff --git a/processing/src/test/java/io/druid/query/groupby/GroupByQueryRunnerTest.java b/processing/src/test/java/io/druid/query/groupby/GroupByQueryRunnerTest.java index dfcc6d2b0982..adce7763197a 100644 --- a/processing/src/test/java/io/druid/query/groupby/GroupByQueryRunnerTest.java +++ b/processing/src/test/java/io/druid/query/groupby/GroupByQueryRunnerTest.java @@ -2710,11 +2710,11 @@ public void testGroupByOrderLimit() throws Exception TestHelper.assertExpectedObjects(expectedResults, mergeRunner.run(query, context), "no-limit"); TestHelper.assertExpectedObjects( - Iterables.limit(expectedResults, 5), mergeRunner.run(builder.limit(5).build(), context), "limited" + Iterables.limit(expectedResults, 5), mergeRunner.run(builder.setLimit(5).build(), context), "limited" ); // Now try it with an expression based aggregator. - builder.limit(Integer.MAX_VALUE) + builder.setLimit(Integer.MAX_VALUE) .setAggregatorSpecs( Arrays.asList( QueryRunnerTestHelper.rowsCount, @@ -2737,11 +2737,11 @@ public void testGroupByOrderLimit() throws Exception TestHelper.assertExpectedObjects(expectedResults, mergeRunner.run(builder.build(), context), "no-limit"); TestHelper.assertExpectedObjects( - Iterables.limit(expectedResults, 5), mergeRunner.run(builder.limit(5).build(), context), "limited" + Iterables.limit(expectedResults, 5), mergeRunner.run(builder.setLimit(5).build(), context), "limited" ); // Now try it with an expression virtual column. - builder.limit(Integer.MAX_VALUE) + builder.setLimit(Integer.MAX_VALUE) .setVirtualColumns( new ExpressionVirtualColumn("expr", "index / 2 + indexMin") ) @@ -2754,7 +2754,7 @@ public void testGroupByOrderLimit() throws Exception TestHelper.assertExpectedObjects(expectedResults, mergeRunner.run(builder.build(), context), "no-limit"); TestHelper.assertExpectedObjects( - Iterables.limit(expectedResults, 5), mergeRunner.run(builder.limit(5).build(), context), "limited" + Iterables.limit(expectedResults, 5), mergeRunner.run(builder.setLimit(5).build(), context), "limited" ); } @@ -2794,7 +2794,7 @@ public void testGroupByWithOrderLimit2() throws Exception QueryRunner mergeRunner = factory.getToolchest().mergeResults(runner); TestHelper.assertExpectedObjects(expectedResults, mergeRunner.run(query, context), "no-limit"); TestHelper.assertExpectedObjects( - Iterables.limit(expectedResults, 5), mergeRunner.run(builder.limit(5).build(), context), "limited" + Iterables.limit(expectedResults, 5), mergeRunner.run(builder.setLimit(5).build(), context), "limited" ); } @@ -2835,7 +2835,7 @@ public void testGroupByWithOrderLimit3() throws Exception QueryRunner mergeRunner = factory.getToolchest().mergeResults(runner); TestHelper.assertExpectedObjects(expectedResults, mergeRunner.run(query, context), "no-limit"); TestHelper.assertExpectedObjects( - Iterables.limit(expectedResults, 5), mergeRunner.run(builder.limit(5).build(), context), "limited" + Iterables.limit(expectedResults, 5), mergeRunner.run(builder.setLimit(5).build(), context), "limited" ); } @@ -2875,7 +2875,7 @@ public void testGroupByOrderLimitNumeric() throws Exception QueryRunner mergeRunner = factory.getToolchest().mergeResults(runner); TestHelper.assertExpectedObjects(expectedResults, mergeRunner.run(query, context), "no-limit"); TestHelper.assertExpectedObjects( - Iterables.limit(expectedResults, 5), mergeRunner.run(builder.limit(5).build(), context), "limited" + Iterables.limit(expectedResults, 5), mergeRunner.run(builder.setLimit(5).build(), context), "limited" ); }