Merge pull request apache#1732 from jon-wei/segmentmeta

Add support for a configurable default segment history period for segmentMetadata queries and GET /datasources/<datasourceName> lookups

docs/content/configuration/broker.md

@@ -75,6 +75,12 @@ The broker uses processing configs for nested groupBy queries. And, optionally,
 |--------|-----------|-------|
 |`druid.query.search.maxSearchLimit`|Maximum number of search results to return.|1000|
 
+##### Segment Metadata Query Config
+
+|Property|Description|Default|
+|--------|-----------|-------|
+|`druid.query.segmentMetadata.defaultHistory`|When no interval is specified in the query, use a default interval of defaultHistory before the end time of the most recent segment, specified in ISO8601 format. This property also controls the duration of the default interval used by GET /druid/v2/datasources/{dataSourceName} interactions for retrieving datasource dimensions/metrics.|P1W|
+
 ### Caching
 
 You can optionally only configure caching to be enabled on the broker by setting caching configs here.

docs/content/design/broker.md

@@ -47,6 +47,10 @@ Returns a list of queryable datasources.
 
 Returns the dimensions and metrics of the datasource. Optionally, you can provide request parameter "full" to get list of served intervals with dimensions and metrics being served for those intervals. You can also provide request param "interval" explicitly to refer to a particular interval.
 
+If no interval is specified, a default interval spanning a configurable period before the current time will be used. The duration of this interval is specified in ISO8601 format via:
+
+druid.query.segmentMetadata.defaultHistory
+
 * `/druid/v2/datasources/{dataSourceName}/dimensions`
 
 Returns the dimensions of the datasource.

docs/content/querying/segmentmetadataquery.md

@@ -25,7 +25,7 @@ There are several main parts to a segment metadata query:
 |--------|-----------|---------|
 |queryType|This String should always be "segmentMetadata"; this is the first thing Druid looks at to figure out how to interpret the query|yes|
 |dataSource|A String or Object defining the data source to query, very similar to a table in a relational database. See [DataSource](../querying/datasource.html) for more information.|yes|
-|intervals|A JSON Object representing ISO-8601 Intervals. This defines the time ranges to run the query over.|yes|
+|intervals|A JSON Object representing ISO-8601 Intervals. This defines the time ranges to run the query over.|no|
 |toInclude|A JSON Object representing what columns should be included in the result. Defaults to "all".|no|
 |merge|Merge all individual segment metadata results into a single result|no|
 |context|See [Context](../querying/query-context.html)|no|
@@ -52,6 +52,13 @@ Timestamp column will have type `LONG`.
 
 Only columns which are dimensions (ie, have type `STRING`) will have any cardinality. Rest of the columns (timestamp and metric columns) will show cardinality as `null`.
 
+### intervals
+
+If an interval is not specified, the query will use a default interval that spans a configurable period before the end time of the most recent segment.
+
+The length of this default time period is set in the broker configuration via:
+  druid.query.segmentMetadata.defaultHistory
+
 ### toInclude
 
 There are 3 types of toInclude objects.

processing/src/main/java/io/druid/query/Druids.java

@@ -904,7 +904,8 @@ public SegmentMetadataQuery build()
           querySegmentSpec,
           toInclude,
           merge,
-          context
+          context,
+          false
       );
     }
 

processing/src/main/java/io/druid/query/metadata/SegmentMetadataQueryConfig.java

@@ -0,0 +1,49 @@
+/*
+* Licensed to Metamarkets Group Inc. (Metamarkets) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. Metamarkets licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*/
+
+package io.druid.query.metadata;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.joda.time.Period;
+import org.joda.time.format.ISOPeriodFormat;
+import org.joda.time.format.PeriodFormatter;
+
+
+public class SegmentMetadataQueryConfig
+{
+  private static final String DEFAULT_PERIOD_STRING = "P1W";
+  private static final PeriodFormatter ISO_FORMATTER = ISOPeriodFormat.standard();
+
+  @JsonProperty
+  private Period defaultHistory = ISO_FORMATTER.parsePeriod(DEFAULT_PERIOD_STRING);
+
+  public SegmentMetadataQueryConfig(String period)
+  {
+    defaultHistory = ISO_FORMATTER.parsePeriod(period);
+  }
+
+  public SegmentMetadataQueryConfig()
+  {
+  }
+
+  public Period getDefaultHistory()
+  {
+    return defaultHistory;
+  }
+}

processing/src/main/java/io/druid/query/metadata/SegmentMetadataQueryQueryToolChest.java

@@ -20,10 +20,13 @@
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.google.common.base.Function;
 import com.google.common.base.Functions;
+import com.google.common.base.Predicate;
 import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Ordering;
 import com.google.common.collect.Sets;
+import com.google.inject.Inject;
 import com.metamx.common.ISE;
 import com.metamx.common.guava.MergeSequence;
 import com.metamx.common.guava.Sequence;
@@ -41,6 +44,8 @@
 import io.druid.query.metadata.metadata.ColumnAnalysis;
 import io.druid.query.metadata.metadata.SegmentAnalysis;
 import io.druid.query.metadata.metadata.SegmentMetadataQuery;
+import io.druid.timeline.LogicalSegment;
+import org.joda.time.DateTime;
 import org.joda.time.Interval;
 
 import javax.annotation.Nullable;
@@ -56,6 +61,16 @@ public class SegmentMetadataQueryQueryToolChest extends QueryToolChest<SegmentAn
   };
   private static final byte[] SEGMENT_METADATA_CACHE_PREFIX = new byte[]{0x4};
 
+  private final SegmentMetadataQueryConfig config;
+
+  @Inject
+  public SegmentMetadataQueryQueryToolChest(
+      SegmentMetadataQueryConfig config
+  )
+  {
+    this.config = config;
+  }
+
   @Override
   public QueryRunner<SegmentAnalysis> mergeResults(final QueryRunner<SegmentAnalysis> runner)
   {
@@ -216,6 +231,37 @@ public Sequence<SegmentAnalysis> mergeSequences(Sequence<Sequence<SegmentAnalysi
     };
   }
 
+  @Override
+  public <T extends LogicalSegment> List<T> filterSegments(SegmentMetadataQuery query, List<T> segments)
+  {
+    if (!query.isUsingDefaultInterval()) {
+      return segments;
+    }
+
+    if (segments.size() <= 1) {
+      return segments;
+    }
+
+    final T max = segments.get(segments.size() - 1);
+
+    DateTime targetEnd = max.getInterval().getEnd();
+    final Interval targetInterval = new Interval(config.getDefaultHistory(), targetEnd);
+
+    return Lists.newArrayList(
+        Iterables.filter(
+            segments,
+            new Predicate<T>()
+            {
+              @Override
+              public boolean apply(T input)
+              {
+                return (input.getInterval().overlaps(targetInterval));
+              }
+            }
+        )
+    );
+  }
+
   private Ordering<SegmentAnalysis> getOrdering()
   {
     return new Ordering<SegmentAnalysis>()

processing/src/main/java/io/druid/query/metadata/metadata/SegmentMetadataQuery.java

@@ -20,33 +20,57 @@
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.base.Preconditions;
+import io.druid.common.utils.JodaUtils;
 import io.druid.query.BaseQuery;
 import io.druid.query.DataSource;
 import io.druid.query.Query;
 import io.druid.query.TableDataSource;
+import io.druid.query.spec.MultipleIntervalSegmentSpec;
 import io.druid.query.spec.QuerySegmentSpec;
+import org.joda.time.Interval;
 
+import java.util.Arrays;
 import java.util.Map;
 
 public class SegmentMetadataQuery extends BaseQuery<SegmentAnalysis>
 {
+  public static final Interval DEFAULT_INTERVAL = new Interval(
+      JodaUtils.MIN_INSTANT, JodaUtils.MAX_INSTANT
+  );
+
   private final ColumnIncluderator toInclude;
   private final boolean merge;
+  private final boolean usingDefaultInterval;
 
   @JsonCreator
   public SegmentMetadataQuery(
       @JsonProperty("dataSource") DataSource dataSource,
       @JsonProperty("intervals") QuerySegmentSpec querySegmentSpec,
       @JsonProperty("toInclude") ColumnIncluderator toInclude,
       @JsonProperty("merge") Boolean merge,
-      @JsonProperty("context") Map<String, Object> context
+      @JsonProperty("context") Map<String, Object> context,
+      @JsonProperty("usingDefaultInterval") Boolean useDefaultInterval
   )
   {
-    super(dataSource, querySegmentSpec, context);
+    super(
+        dataSource,
+        (querySegmentSpec == null) ? new MultipleIntervalSegmentSpec(Arrays.asList(DEFAULT_INTERVAL))
+                                   : querySegmentSpec,
+        context
+    );
+
+    if (querySegmentSpec == null) {
+      this.usingDefaultInterval = true;
+    } else {
+      this.usingDefaultInterval = useDefaultInterval == null ? false : useDefaultInterval;
+    }
 
     this.toInclude = toInclude == null ? new AllColumnIncluderator() : toInclude;
     this.merge = merge == null ? false : merge;
-    Preconditions.checkArgument(dataSource instanceof TableDataSource, "SegmentMetadataQuery only supports table datasource");
+    Preconditions.checkArgument(
+        dataSource instanceof TableDataSource,
+        "SegmentMetadataQuery only supports table datasource"
+    );
   }
 
   @JsonProperty
@@ -61,6 +85,12 @@ public boolean isMerge()
     return merge;
   }
 
+  @JsonProperty
+  public boolean isUsingDefaultInterval()
+  {
+    return usingDefaultInterval;
+  }
+
   @Override
   public boolean hasFilters()
   {
@@ -78,7 +108,11 @@ public Query<SegmentAnalysis> withOverriddenContext(Map<String, Object> contextO
   {
     return new SegmentMetadataQuery(
         getDataSource(),
-        getQuerySegmentSpec(), toInclude, merge, computeOverridenContext(contextOverride)
+        getQuerySegmentSpec(),
+        toInclude,
+        merge,
+        computeOverridenContext(contextOverride),
+        usingDefaultInterval
     );
   }
 
@@ -87,7 +121,12 @@ public Query<SegmentAnalysis> withQuerySegmentSpec(QuerySegmentSpec spec)
   {
     return new SegmentMetadataQuery(
         getDataSource(),
-        spec, toInclude, merge, getContext());
+        spec,
+        toInclude,
+        merge,
+        getContext(),
+        usingDefaultInterval
+    );
   }
 
   @Override
@@ -98,22 +137,34 @@ public Query<SegmentAnalysis> withDataSource(DataSource dataSource)
         getQuerySegmentSpec(),
         toInclude,
         merge,
-        getContext());
+        getContext(),
+        usingDefaultInterval
+    );
   }
 
   @Override
   public boolean equals(Object o)
   {
-    if (this == o) return true;
-    if (o == null || getClass() != o.getClass()) return false;
-    if (!super.equals(o)) return false;
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    if (!super.equals(o)) {
+      return false;
+    }
 
     SegmentMetadataQuery that = (SegmentMetadataQuery) o;
 
-    if (merge != that.merge) return false;
-    if (toInclude != null ? !toInclude.equals(that.toInclude) : that.toInclude != null) return false;
+    if (merge != that.merge) {
+      return false;
+    }
+    if (usingDefaultInterval != that.usingDefaultInterval) {
+      return false;
+    }
+    return !(toInclude != null ? !toInclude.equals(that.toInclude) : that.toInclude != null);
 
-    return true;
   }
 
   @Override
@@ -122,6 +173,7 @@ public int hashCode()
     int result = super.hashCode();
     result = 31 * result + (toInclude != null ? toInclude.hashCode() : 0);
     result = 31 * result + (merge ? 1 : 0);
+    result = 31 * result + (usingDefaultInterval ? 1 : 0);
     return result;
   }
 }

processing/src/test/java/io/druid/query/metadata/SegmentAnalyzerTest.java

@@ -99,13 +99,13 @@ private List<SegmentAnalysis> getSegmentAnalysises(Segment index)
   {
     final QueryRunner runner = QueryRunnerTestHelper.makeQueryRunner(
         (QueryRunnerFactory) new SegmentMetadataQueryRunnerFactory(
-            new SegmentMetadataQueryQueryToolChest(),
+            new SegmentMetadataQueryQueryToolChest(new SegmentMetadataQueryConfig()),
             QueryRunnerTestHelper.NOOP_QUERYWATCHER
         ), index
     );
 
     final SegmentMetadataQuery query = new SegmentMetadataQuery(
-        new LegacyDataSource("test"), QuerySegmentSpecs.create("2011/2012"), null, null, null
+        new LegacyDataSource("test"), QuerySegmentSpecs.create("2011/2012"), null, null, null, false
     );
     HashMap<String,Object> context = new HashMap<String, Object>();
     return Sequences.toList(query.run(runner, context), Lists.<SegmentAnalysis>newArrayList());