From 79e8319383eddbb49ecb4c1785dcd3eed14a0634 Mon Sep 17 00:00:00 2001
From: Justin Borromeo <jborrome@edu.uwaterloo.ca>
Date: Fri, 1 Feb 2019 18:22:58 -0800
Subject: [PATCH] Move ScanResultValue timestamp comparator to a separate class
 for testing

---
 .../query/scan/ScanQueryQueryToolChest.java   | 28 +--------
 .../ScanResultValueTimestampComparator.java   | 59 +++++++++++++++++++
 2 files changed, 61 insertions(+), 26 deletions(-)
 create mode 100644 processing/src/main/java/org/apache/druid/query/scan/ScanResultValueTimestampComparator.java

diff --git a/processing/src/main/java/org/apache/druid/query/scan/ScanQueryQueryToolChest.java b/processing/src/main/java/org/apache/druid/query/scan/ScanQueryQueryToolChest.java
index 8330da59fa8..c8504d3648a 100644
--- a/processing/src/main/java/org/apache/druid/query/scan/ScanQueryQueryToolChest.java
+++ b/processing/src/main/java/org/apache/druid/query/scan/ScanQueryQueryToolChest.java
@@ -106,37 +106,13 @@ public class ScanQueryQueryToolChest extends QueryToolChest<ScanResultValue, Sca
           return new BaseSequence<>(scanQueryLimitRowIteratorMaker);
         } else if (scanQuery.getTimeOrder().equals(ScanQuery.TIME_ORDER_ASCENDING) ||
                    scanQuery.getTimeOrder().equals(ScanQuery.TIME_ORDER_DESCENDING)) {
-          Comparator priorityQComparator = (val1, val2) -> {
-            int comparison;
-            ScanResultValue val1SRV = (ScanResultValue) val1,
-                val2SRV = (ScanResultValue) val2;
-            if (scanQuery.getResultFormat().equals(ScanQuery.RESULT_FORMAT_LIST)) {
-              comparison = Longs.compare(
-                  (Long) ((Map<String, Object>) ((List) val1SRV.getEvents()).get(0)).get(ColumnHolder.TIME_COLUMN_NAME),
-                  (Long) ((Map<String, Object>) ((List) val2SRV.getEvents()).get(0)).get(ColumnHolder.TIME_COLUMN_NAME)
-              );
-            } else if (scanQuery.getResultFormat().equals(ScanQuery.RESULT_FORMAT_COMPACTED_LIST)) {
-              int val1TimeColumnIndex = val1SRV.getColumns().indexOf(ColumnHolder.TIME_COLUMN_NAME);
-              int val2TimeColumnIndex = val2SRV.getColumns().indexOf(ColumnHolder.TIME_COLUMN_NAME);
-              List<Object> event1 = (List<Object>) ((List<Object>) val1SRV.getEvents()).get(0);
-              List<Object> event2 = (List<Object>) ((List<Object>) val2SRV.getEvents()).get(0);
-              comparison = Longs.compare(
-                  (Long) event1.get(val1TimeColumnIndex),
-                  (Long) event2.get(val2TimeColumnIndex)
-              );
-            } else {
-              throw new UOE("Result format [%s] is not supported", scanQuery.getResultFormat());
-            }
-            if (scanQuery.getTimeOrder().equals(ScanQuery.TIME_ORDER_DESCENDING)) {
-              return comparison * -1;
-            }
-            return comparison;
-          };
+          Comparator priorityQComparator = new ScanResultValueTimestampComparator(scanQuery);
 
           // Converting the limit from long to int could theoretically throw an ArithmeticException but this branch
           // only runs if limit < MAX_LIMIT_FOR_IN_MEMORY_TIME_ORDERING (which should be < Integer.MAX_VALUE)
           PriorityQueue<Object> q = new PriorityQueue<>(Math.toIntExact(scanQuery.getLimit()), priorityQComparator);
           Iterator<ScanResultValue> scanResultIterator = scanQueryLimitRowIteratorMaker.make();
+          
           while (scanResultIterator.hasNext()) {
             ScanResultValue next = scanResultIterator.next();
             List<Object> events = (List<Object>) next.getEvents();
diff --git a/processing/src/main/java/org/apache/druid/query/scan/ScanResultValueTimestampComparator.java b/processing/src/main/java/org/apache/druid/query/scan/ScanResultValueTimestampComparator.java
new file mode 100644
index 00000000000..53f7949bb35
--- /dev/null
+++ b/processing/src/main/java/org/apache/druid/query/scan/ScanResultValueTimestampComparator.java
@@ -0,0 +1,59 @@
+package org.apache.druid.query.scan;
+
+import com.google.common.primitives.Longs;
+import org.apache.druid.java.util.common.UOE;
+import org.apache.druid.segment.column.ColumnHolder;
+
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+
+public class ScanResultValueTimestampComparator implements Comparator<ScanResultValue>
+{
+  private ScanQuery scanQuery;
+
+  public ScanResultValueTimestampComparator(ScanQuery scanQuery) {
+    this.scanQuery = scanQuery;
+  }
+
+  @Override
+  public int compare(ScanResultValue o1, ScanResultValue o2)
+  {
+    int comparison;
+    if (scanQuery.getResultFormat().equals(ScanQuery.RESULT_FORMAT_LIST)) {
+      comparison = Longs.compare(
+          (Long) ((Map<String, Object>) ((List) o1.getEvents()).get(0)).get(ColumnHolder.TIME_COLUMN_NAME),
+          (Long) ((Map<String, Object>) ((List) o2.getEvents()).get(0)).get(ColumnHolder.TIME_COLUMN_NAME)
+      );
+    } else if (scanQuery.getResultFormat().equals(ScanQuery.RESULT_FORMAT_COMPACTED_LIST)) {
+      int val1TimeColumnIndex = o1.getColumns().indexOf(ColumnHolder.TIME_COLUMN_NAME);
+      int val2TimeColumnIndex = o2.getColumns().indexOf(ColumnHolder.TIME_COLUMN_NAME);
+      List<Object> event1 = (List<Object>) ((List<Object>) o1.getEvents()).get(0);
+      List<Object> event2 = (List<Object>) ((List<Object>) o2.getEvents()).get(0);
+      comparison = Longs.compare(
+          (Long) event1.get(val1TimeColumnIndex),
+          (Long) event2.get(val2TimeColumnIndex)
+      );
+    } else {
+      throw new UOE("Result format [%s] is not supported", scanQuery.getResultFormat());
+    }
+    if (scanQuery.getTimeOrder().equals(ScanQuery.TIME_ORDER_DESCENDING)) {
+      return comparison * -1;
+    }
+    return comparison;
+  }
+
+  @Override
+  public boolean equals(Object obj)
+  {
+    if (this == obj) {
+      return true;
+    }
+    if (obj == null || getClass() != obj.getClass()) {
+      return false;
+    }
+    ScanResultValueTimestampComparator comp = (ScanResultValueTimestampComparator) obj
+    return this.scanQuery.equals(comp.scanQuery);
+  }
+
+}