Apache
/
druid
mirror of https://github.com/apache/druid.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Move ScanResultValue timestamp comparator to a separate class for testing

This commit is contained in:
Justin Borromeo 2019-02-01 18:22:58 -08:00
parent 7a6080f636
commit 79e8319383
2 changed files with 61 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
processing/src/main/java/org/apache/druid/query/scan/ScanQueryQueryToolChest.java
View File

@ -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();

59
processing/src/main/java/org/apache/druid/query/scan/ScanResultValueTimestampComparator.java Normal file
View File

@ -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);
}
}