time-descending result of select queries

docs/content/development/select-query.md

@@ -8,6 +8,7 @@ Select queries return raw Druid rows and support pagination.
  {
    "queryType": "select",
    "dataSource": "wikipedia",
+   "descending": "false",
    "dimensions":[],
    "metrics":[],
    "granularity": "all",
@@ -25,6 +26,7 @@ There are several main parts to a select query:
 |queryType|This String should always be "select"; 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|
+|descending|Whether to make descending ordered result. Default is `false`(ascending). When this is `true`, page identifier and offsets will be negative value.|no|
 |filter|See [Filters](../querying/filters.html)|no|
 |dimensions|A String array of dimensions to select. If left empty, all dimensions are returned.|no|
 |metrics|A String array of metrics to select. If left empty, all metrics are returned.|no|
@@ -140,7 +142,7 @@ The format of the result is:
 } ]
 ```
 
-The `threshold` determines how many hits are returned, with each hit indexed by an offset.
+The `threshold` determines how many hits are returned, with each hit indexed by an offset. When `descending` is true, the offset will be negative value.
 
 The results above include:
 
@@ -166,4 +168,4 @@ This can be used with the next query's pagingSpec:
 
  }
 
-Note that in the second query, an offset is specified and that it is 1 greater than the largest offset found in the initial results. To return the next "page", this offset must be incremented by 1 with each new query. When an empty results set is received, the very last page has been returned.
+Note that in the second query, an offset is specified and that it is 1 greater than the largest offset found in the initial results. To return the next "page", this offset must be incremented by 1 (should be decremented by 1 for descending query), with each new query. When an empty results set is received, the very last page has been returned.

docs/content/querying/timeseriesquery.md

@@ -50,7 +50,7 @@ There are 7 main parts to a timeseries query:
 |--------|-----------|---------|
 |queryType|This String should always be "timeseries"; 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|
-|descending|Whether to make descending ordered result.|no|
+|descending|Whether to make descending ordered result. Default is `false`(ascending).|no|
 |intervals|A JSON Object representing ISO-8601 Intervals. This defines the time ranges to run the query over.|yes|
 |granularity|Defines the granularity to bucket query results. See [Granularities](../querying/granularities.html)|yes|
 |filter|See [Filters](../querying/filters.html)|no|

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

@@ -1073,6 +1073,7 @@ public static class SelectQueryBuilder
   {
     private DataSource dataSource;
     private QuerySegmentSpec querySegmentSpec;
+    private boolean descending;
     private Map<String, Object> context;
     private DimFilter dimFilter;
     private QueryGranularity granularity;
@@ -1097,6 +1098,7 @@ public SelectQuery build()
       return new SelectQuery(
           dataSource,
           querySegmentSpec,
+          descending,
           dimFilter,
           granularity,
           dimensions,
@@ -1144,6 +1146,12 @@ public SelectQueryBuilder intervals(List<Interval> l)
       return this;
     }
 
+    public SelectQueryBuilder descending(boolean descending)
+    {
+      this.descending = descending;
+      return this;
+    }
+
     public SelectQueryBuilder context(Map<String, Object> c)
     {
       context = c;

processing/src/main/java/io/druid/query/select/PagingOffset.java

@@ -0,0 +1,125 @@
+/*
+ * 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.select;
+
+import com.google.common.annotations.VisibleForTesting;
+
+/**
+ * offset iterator for select query
+ */
+public abstract class PagingOffset
+{
+  protected final int startOffset;
+  protected final int threshold;
+
+  protected int counter;
+
+  public PagingOffset(int startOffset, int threshold)
+  {
+    this.startOffset = startOffset;
+    this.threshold = threshold;
+  }
+
+  public abstract boolean isDescending();
+
+  public final int startOffset()
+  {
+    return startOffset;
+  }
+
+  public abstract int startDelta();
+
+  public final int threshold()
+  {
+    return threshold;
+  }
+
+  public final boolean hasNext()
+  {
+    return counter < threshold;
+  }
+
+  public final void next()
+  {
+    counter++;
+  }
+
+  public abstract int current();
+
+  private static class Ascending extends PagingOffset
+  {
+    public Ascending(int offset, int threshold)
+    {
+      super(offset, threshold);
+    }
+
+    public final boolean isDescending()
+    {
+      return false;
+    }
+
+    public final int startDelta()
+    {
+      return startOffset;
+    }
+
+    public final int current()
+    {
+      return startOffset + counter;
+    }
+  }
+
+  private static class Descending extends PagingOffset
+  {
+    public Descending(int offset, int threshold)
+    {
+      super(offset, threshold);
+    }
+
+    public final boolean isDescending()
+    {
+      return true;
+    }
+
+    public final int startDelta()
+    {
+      return -startOffset - 1;
+    }
+
+    public final int current()
+    {
+      return startOffset - counter;
+    }
+  }
+
+  public static PagingOffset of(int startOffset, int threshold)
+  {
+    return startOffset < 0 ? new Descending(startOffset, threshold) : new Ascending(startOffset, threshold);
+  }
+
+  @VisibleForTesting
+  static int toOffset(int delta, boolean descending)
+  {
+    if (delta < 0) {
+      throw new IllegalArgumentException("Delta should not be negative");
+    }
+    return descending ? -delta - 1 : delta;
+  }
+}

processing/src/main/java/io/druid/query/select/PagingSpec.java

@@ -98,4 +98,14 @@ public String toString()
            ", threshold=" + threshold +
            '}';
   }
+
+  public PagingOffset getOffset(String identifier, boolean descending)
+  {
+    Integer offset = pagingIdentifiers.get(identifier);
+    if (offset == null) {
+      offset = PagingOffset.toOffset(0, descending);
+    }
+    return PagingOffset.of(offset, threshold);
+  }
+
 }

processing/src/main/java/io/druid/query/select/SelectBinaryFn.java

@@ -62,7 +62,7 @@ public Result<SelectResultValue> apply(
                                ? arg1.getTimestamp()
                                : gran.toDateTime(gran.truncate(arg1.getTimestamp().getMillis()));
 
-    SelectResultValueBuilder builder = new SelectResultValueBuilder(timestamp, pagingSpec.getThreshold(), descending);
+    SelectResultValueBuilder builder = new SelectResultValueBuilder(timestamp, pagingSpec, descending);
 
     SelectResultValue arg1Val = arg1.getValue();
     SelectResultValue arg2Val = arg2.getValue();

processing/src/main/java/io/druid/query/select/SelectQuery.java

@@ -49,6 +49,7 @@ public class SelectQuery extends BaseQuery<Result<SelectResultValue>>
   public SelectQuery(
       @JsonProperty("dataSource") DataSource dataSource,
       @JsonProperty("intervals") QuerySegmentSpec querySegmentSpec,
+      @JsonProperty("descending") boolean descending,
       @JsonProperty("filter") DimFilter dimFilter,
       @JsonProperty("granularity") QueryGranularity granularity,
       @JsonProperty("dimensions") List<String> dimensions,
@@ -57,14 +58,25 @@ public SelectQuery(
       @JsonProperty("context") Map<String, Object> context
   )
   {
-    super(dataSource, querySegmentSpec, false, context);
+    super(dataSource, querySegmentSpec, descending, context);
     this.dimFilter = dimFilter;
     this.granularity = granularity;
     this.dimensions = dimensions;
     this.metrics = metrics;
     this.pagingSpec = pagingSpec;
 
     Preconditions.checkNotNull(pagingSpec, "must specify a pagingSpec");
+    Preconditions.checkArgument(checkPagingSpec(pagingSpec, descending), "invalid pagingSpec");
+  }
+
+  private boolean checkPagingSpec(PagingSpec pagingSpec, boolean descending)
+  {
+    for (Integer value : pagingSpec.getPagingIdentifiers().values()) {
+      if (descending ^ (value < 0)) {
+        return false;
+      }
+    }
+    return pagingSpec.getThreshold() >= 0;
   }
 
   @Override
@@ -109,11 +121,17 @@ public List<String> getMetrics()
     return metrics;
   }
 
+  public PagingOffset getPagingOffset(String identifier)
+  {
+    return pagingSpec.getOffset(identifier, isDescending());
+  }
+
   public SelectQuery withQuerySegmentSpec(QuerySegmentSpec querySegmentSpec)
   {
     return new SelectQuery(
         getDataSource(),
         querySegmentSpec,
+        isDescending(),
         dimFilter,
         granularity,
         dimensions,
@@ -129,6 +147,7 @@ public Query<Result<SelectResultValue>> withDataSource(DataSource dataSource)
     return new SelectQuery(
         dataSource,
         getQuerySegmentSpec(),
+        isDescending(),
         dimFilter,
         granularity,
         dimensions,
@@ -143,6 +162,7 @@ public SelectQuery withOverriddenContext(Map<String, Object> contextOverrides)
     return new SelectQuery(
         getDataSource(),
         getQuerySegmentSpec(),
+        isDescending(),
         dimFilter,
         granularity,
         dimensions,
@@ -158,6 +178,7 @@ public String toString()
     return "SelectQuery{" +
            "dataSource='" + getDataSource() + '\'' +
            ", querySegmentSpec=" + getQuerySegmentSpec() +
+           ", descending=" + isDescending() +
            ", dimFilter=" + dimFilter +
            ", granularity=" + granularity +
            ", dimensions=" + dimensions +

processing/src/main/java/io/druid/query/select/SelectQueryEngine.java

@@ -82,8 +82,7 @@ public Result<SelectResultValue> apply(Cursor cursor)
           {
             final SelectResultValueBuilder builder = new SelectResultValueBuilder(
                 cursor.getTime(),
-                query.getPagingSpec()
-                     .getThreshold(),
+                query.getPagingSpec(),
                 query.isDescending()
             );
 
@@ -102,18 +101,11 @@ public Result<SelectResultValue> apply(Cursor cursor)
               metSelectors.put(metric, metricSelector);
             }
 
-            int startOffset;
-            if (query.getPagingSpec().getPagingIdentifiers() == null) {
-              startOffset = 0;
-            } else {
-              Integer offset = query.getPagingSpec().getPagingIdentifiers().get(segment.getIdentifier());
-              startOffset = (offset == null) ? 0 : offset;
-            }
+            final PagingOffset offset = query.getPagingOffset(segment.getIdentifier());
 
-            cursor.advanceTo(startOffset);
+            cursor.advanceTo(offset.startDelta());
 
-            int offset = 0;
-            while (!cursor.isDone() && offset < query.getPagingSpec().getThreshold()) {
+            for (; !cursor.isDone() && offset.hasNext(); cursor.advance(), offset.next()) {
               final Map<String, Object> theEvent = Maps.newLinkedHashMap();
               theEvent.put(EventHolder.timestampKey, new DateTime(timestampColumnSelector.get()));
 
@@ -153,12 +145,10 @@ public Result<SelectResultValue> apply(Cursor cursor)
               builder.addEntry(
                   new EventHolder(
                       segment.getIdentifier(),
-                      startOffset + offset,
+                      offset.current(),
                       theEvent
                   )
               );
-              cursor.advance();
-              offset++;
             }
 
             return builder.build();