Fix for sort order in select/topN query cache (#4766)

When historical caching is enabled, and a select or topN query is
issued, and then a following query with "descending": true is set, the
cached query returns the ascending result (or vice versa), often
resulting in invalid paging identifiers.

The CacheKey for these queries doesn't include the "descending" flag;
this change adds it, and fixes the problem.

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

@@ -208,9 +208,11 @@ public class SelectQueryQueryToolChest extends QueryToolChest<Result<SelectResul
         }
 
         final byte[] virtualColumnsCacheKey = query.getVirtualColumns().getCacheKey();
+        final byte isDescendingByte = query.isDescending() ? (byte) 1 : 0;
+
         final ByteBuffer queryCacheKey = ByteBuffer
             .allocate(
-                1
+                2
                 + granularityBytes.length
                 + filterBytes.length
                 + query.getPagingSpec().getCacheKey().length
@@ -221,7 +223,8 @@ public class SelectQueryQueryToolChest extends QueryToolChest<Result<SelectResul
             .put(SELECT_QUERY)
             .put(granularityBytes)
             .put(filterBytes)
-            .put(query.getPagingSpec().getCacheKey());
+            .put(query.getPagingSpec().getCacheKey())
+            .put(isDescendingByte);
 
         for (byte[] dimensionsByte : dimensionsBytes) {
           queryCacheKey.put(dimensionsByte);

processing/src/test/java/io/druid/query/select/SelectQueryQueryToolChestTest.java

@@ -0,0 +1,73 @@
+/*
+ * 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.base.Supplier;
+import com.google.common.base.Suppliers;
+import io.druid.jackson.DefaultObjectMapper;
+import io.druid.query.CacheStrategy;
+import io.druid.query.Druids;
+import io.druid.query.QueryRunnerTestHelper;
+import io.druid.query.Result;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+public class SelectQueryQueryToolChestTest
+{
+  private static final Supplier<SelectQueryConfig> configSupplier = Suppliers.ofInstance(new SelectQueryConfig(true));
+
+  private static final SelectQueryQueryToolChest toolChest = new SelectQueryQueryToolChest(
+      new DefaultObjectMapper(),
+      QueryRunnerTestHelper.NoopIntervalChunkingQueryRunnerDecorator(),
+      configSupplier
+  );
+
+  @Test
+  public void testComputeCacheKeyWithDifferentSortOrer() throws Exception
+  {
+    final SelectQuery query1 = Druids.newSelectQueryBuilder()
+                                     .dataSource("dummy")
+                                     .dimensions(Collections.singletonList("testDim"))
+                                     .intervals(SelectQueryRunnerTest.I_0112_0114)
+                                     .granularity(QueryRunnerTestHelper.allGran)
+                                     .pagingSpec(PagingSpec.newSpec(3))
+                                     .descending(false)
+                                     .build();
+
+    final SelectQuery query2 = Druids.newSelectQueryBuilder()
+                                     .dataSource("dummy")
+                                     .dimensions(Collections.singletonList("testDim"))
+                                     .intervals(SelectQueryRunnerTest.I_0112_0114)
+                                     .granularity(QueryRunnerTestHelper.allGran)
+                                     .pagingSpec(PagingSpec.newSpec(3))
+                                     .descending(true)
+                                     .build();
+
+    final CacheStrategy<Result<SelectResultValue>, Object, SelectQuery> strategy1 = toolChest.getCacheStrategy(query1);
+    Assert.assertNotNull(strategy1);
+    final CacheStrategy<Result<SelectResultValue>, Object, SelectQuery> strategy2 = toolChest.getCacheStrategy(query2);
+    Assert.assertNotNull(strategy2);
+
+    Assert.assertFalse(Arrays.equals(strategy1.computeCacheKey(query1), strategy2.computeCacheKey(query2)));
+  }
+}