You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by ja...@apache.org on 2020/04/08 03:39:25 UTC
[incubator-pinot] branch master updated: Support order-by on BYTES
column (#5213)
This is an automated email from the ASF dual-hosted git repository.
jackie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
The following commit(s) were added to refs/heads/master by this push:
new e84a3af Support order-by on BYTES column (#5213)
e84a3af is described below
commit e84a3afe7212196f639fb3b4023e6799a1c2de28
Author: Xiaotian (Jackie) Jiang <17...@users.noreply.github.com>
AuthorDate: Tue Apr 7 20:39:18 2020 -0700
Support order-by on BYTES column (#5213)
In order to support order-by on BYTES column everywhere, inside the system we should always use ByteArray (comparable) to store the BYTES value.
Currently BYTES value are stored as byte[], ByteArray or String in different places, which is very confusing and could cause casting errors.
Changes:
- For DisctinctCount, fix the casting issue when ordering on BYTES column
- For selection order-by, order BYTES column using ByteArray instead of String for performance improvement
- Inside Record, always store BYTES as ByteArray for clarity and also performance improvement (avoid expensive deepEquals and deepHashCode)
- On broker side, store BYTES column using ByteArray instead of String for performance improvement
- On broker side, support type compatible merges for all selection queries
No format change on the query results.
TODO: We are still returning String for BYTES column when preserving the type. Consider changing it to byte[].
---
.../org/apache/pinot/common/utils/DataTable.java | 3 +
.../core/common/RowBasedBlockValueFetcher.java | 34 ++--
.../core/common/datatable/DataTableBuilder.java | 46 ++---
.../core/common/datatable/DataTableImplV2.java | 8 +
.../java/org/apache/pinot/core/data/table/Key.java | 37 ++--
.../org/apache/pinot/core/data/table/Record.java | 37 ++--
.../operator/CombineGroupByOrderByOperator.java | 25 +--
.../pinot/core/operator/CombineOperator.java | 3 +-
.../operator/blocks/IntermediateResultsBlock.java | 41 +++--
.../core/operator/query/SelectionOnlyOperator.java | 3 +-
.../operator/query/SelectionOrderByOperator.java | 14 +-
.../core/query/aggregation/DistinctTable.java | 16 +-
.../function/DistinctAggregationFunction.java | 8 +-
.../query/reduce/AggregationDataTableReducer.java | 9 +-
.../pinot/core/query/reduce/CombineService.java | 12 +-
.../query/reduce/DistinctDataTableReducer.java | 41 +++--
.../core/query/reduce/GroupByDataTableReducer.java | 11 +-
.../query/reduce/SelectionDataTableReducer.java | 13 +-
.../query/selection/SelectionOperatorService.java | 69 +++++---
.../query/selection/SelectionOperatorUtils.java | 190 ++++++++++++++-------
.../apache/pinot/queries/DistinctQueriesTest.java | 77 ++++++++-
...InnerSegmentSelectionMultiValueQueriesTest.java | 21 ++-
...nnerSegmentSelectionSingleValueQueriesTest.java | 27 ++-
.../RangePredicateWithSortedInvertedIndexTest.java | 6 +-
.../pinot/queries/TextSearchQueriesTest.java | 10 +-
.../selection/SelectionOperatorServiceTest.java | 101 ++++++-----
26 files changed, 513 insertions(+), 349 deletions(-)
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/utils/DataTable.java b/pinot-common/src/main/java/org/apache/pinot/common/utils/DataTable.java
index 83f53d3..527669c 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/utils/DataTable.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/utils/DataTable.java
@@ -21,6 +21,7 @@ package org.apache.pinot.common.utils;
import java.io.IOException;
import java.util.Map;
import org.apache.pinot.common.response.ProcessingException;
+import org.apache.pinot.spi.utils.ByteArray;
/**
@@ -63,6 +64,8 @@ public interface DataTable {
String getString(int rowId, int colId);
+ ByteArray getBytes(int rowId, int colId);
+
<T> T getObject(int rowId, int colId);
int[] getIntArray(int rowId, int colId);
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/common/RowBasedBlockValueFetcher.java b/pinot-core/src/main/java/org/apache/pinot/core/common/RowBasedBlockValueFetcher.java
index 2eeeeee..31b5805 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/common/RowBasedBlockValueFetcher.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/common/RowBasedBlockValueFetcher.java
@@ -18,8 +18,8 @@
*/
package org.apache.pinot.core.common;
-import java.io.Serializable;
import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.utils.ByteArray;
public class RowBasedBlockValueFetcher {
@@ -33,9 +33,9 @@ public class RowBasedBlockValueFetcher {
}
}
- public Serializable[] getRow(int docId) {
+ public Object[] getRow(int docId) {
int numColumns = _valueFetchers.length;
- Serializable[] row = new Serializable[numColumns];
+ Object[] row = new Object[numColumns];
for (int i = 0; i < numColumns; i++) {
row[i] = _valueFetchers[i].getValue(docId);
}
@@ -80,10 +80,10 @@ public class RowBasedBlockValueFetcher {
}
private interface ValueFetcher {
- Serializable getValue(int docId);
+ Object getValue(int docId);
}
- private class IntSingleValueFetcher implements ValueFetcher {
+ private static class IntSingleValueFetcher implements ValueFetcher {
private final int[] _values;
IntSingleValueFetcher(int[] values) {
@@ -95,7 +95,7 @@ public class RowBasedBlockValueFetcher {
}
}
- private class LongSingleValueFetcher implements ValueFetcher {
+ private static class LongSingleValueFetcher implements ValueFetcher {
private final long[] _values;
LongSingleValueFetcher(long[] values) {
@@ -107,7 +107,7 @@ public class RowBasedBlockValueFetcher {
}
}
- private class FloatSingleValueFetcher implements ValueFetcher {
+ private static class FloatSingleValueFetcher implements ValueFetcher {
private final float[] _values;
FloatSingleValueFetcher(float[] values) {
@@ -119,7 +119,7 @@ public class RowBasedBlockValueFetcher {
}
}
- private class DoubleSingleValueFetcher implements ValueFetcher {
+ private static class DoubleSingleValueFetcher implements ValueFetcher {
private final double[] _values;
DoubleSingleValueFetcher(double[] values) {
@@ -131,7 +131,7 @@ public class RowBasedBlockValueFetcher {
}
}
- private class StringSingleValueFetcher implements ValueFetcher {
+ private static class StringSingleValueFetcher implements ValueFetcher {
private final String[] _values;
StringSingleValueFetcher(String[] values) {
@@ -143,19 +143,19 @@ public class RowBasedBlockValueFetcher {
}
}
- private class BytesValueFetcher implements ValueFetcher {
+ private static class BytesValueFetcher implements ValueFetcher {
private final byte[][] _values;
BytesValueFetcher(byte[][] values) {
_values = values;
}
- public byte[] getValue(int docId) {
- return _values[docId];
+ public ByteArray getValue(int docId) {
+ return new ByteArray(_values[docId]);
}
}
- private class IntMultiValueFetcher implements ValueFetcher {
+ private static class IntMultiValueFetcher implements ValueFetcher {
private final int[][] _values;
IntMultiValueFetcher(int[][] values) {
@@ -167,7 +167,7 @@ public class RowBasedBlockValueFetcher {
}
}
- private class LongMultiValueFetcher implements ValueFetcher {
+ private static class LongMultiValueFetcher implements ValueFetcher {
private final long[][] _values;
LongMultiValueFetcher(long[][] values) {
@@ -179,7 +179,7 @@ public class RowBasedBlockValueFetcher {
}
}
- private class FloatMultiValueFetcher implements ValueFetcher {
+ private static class FloatMultiValueFetcher implements ValueFetcher {
private final float[][] _values;
FloatMultiValueFetcher(float[][] values) {
@@ -191,7 +191,7 @@ public class RowBasedBlockValueFetcher {
}
}
- private class DoubleMultiValueFetcher implements ValueFetcher {
+ private static class DoubleMultiValueFetcher implements ValueFetcher {
private final double[][] _values;
DoubleMultiValueFetcher(double[][] values) {
@@ -203,7 +203,7 @@ public class RowBasedBlockValueFetcher {
}
}
- private class StringMultiValueFetcher implements ValueFetcher {
+ private static class StringMultiValueFetcher implements ValueFetcher {
private final String[][] _values;
StringMultiValueFetcher(String[][] values) {
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/DataTableBuilder.java b/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/DataTableBuilder.java
index 4dc32cb..cafafff 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/DataTableBuilder.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/DataTableBuilder.java
@@ -27,6 +27,7 @@ import java.util.Map;
import org.apache.pinot.common.utils.DataSchema;
import org.apache.pinot.common.utils.DataTable;
import org.apache.pinot.core.common.ObjectSerDeUtils;
+import org.apache.pinot.spi.utils.ByteArray;
/**
@@ -74,6 +75,7 @@ import org.apache.pinot.core.common.ObjectSerDeUtils;
// TODO: 1. Fix float size.
// TODO: 2. Use one dictionary for all columns (save space).
// TODO: 3. Given a data schema, write all values one by one instead of using rowId and colId to position (save time).
+// TODO: 4. Store bytes as variable size data instead of String
public class DataTableBuilder {
private final DataSchema _dataSchema;
private final int[] _columnOffsets;
@@ -162,44 +164,32 @@ public class DataTableBuilder {
_currentRowDataByteBuffer.putInt(dictId);
}
- public void setColumn(int colId, Object value)
+ public void setColumn(int colId, ByteArray value)
throws IOException {
- _currentRowDataByteBuffer.position(_columnOffsets[colId]);
- _currentRowDataByteBuffer.putInt(_variableSizeDataByteArrayOutputStream.size());
- int objectTypeValue = ObjectSerDeUtils.ObjectType.getObjectType(value).getValue();
- byte[] bytes = ObjectSerDeUtils.serialize(value, objectTypeValue);
- _currentRowDataByteBuffer.putInt(bytes.length);
- _variableSizeDataOutputStream.writeInt(objectTypeValue);
- _variableSizeDataByteArrayOutputStream.write(bytes);
- }
+ // NOTE: Use String to store bytes value in DataTable V2 for backward-compatibility
+ setColumn(colId, value.toHexString());
- public void setColumn(int colId, byte[] values) {
- _currentRowDataByteBuffer.position(_columnOffsets[colId]);
- _currentRowDataByteBuffer.putInt(_variableSizeDataByteArrayOutputStream.size());
- _currentRowDataByteBuffer.putInt(values.length);
- for (byte value : values) {
- _variableSizeDataByteArrayOutputStream.write(value);
- }
- }
+ /*
+ TODO: Store bytes as variable size data instead of String. Make the change for the next version data table for
+ backward-compatibility
- public void setColumn(int colId, char[] values)
- throws IOException {
_currentRowDataByteBuffer.position(_columnOffsets[colId]);
_currentRowDataByteBuffer.putInt(_variableSizeDataByteArrayOutputStream.size());
- _currentRowDataByteBuffer.putInt(values.length);
- for (char value : values) {
- _variableSizeDataOutputStream.writeChar(value);
- }
+ byte[] bytes = value.getBytes();
+ _currentRowDataByteBuffer.putInt(bytes.length);
+ _variableSizeDataByteArrayOutputStream.write(bytes);
+ */
}
- public void setColumn(int colId, short[] values)
+ public void setColumn(int colId, Object value)
throws IOException {
_currentRowDataByteBuffer.position(_columnOffsets[colId]);
_currentRowDataByteBuffer.putInt(_variableSizeDataByteArrayOutputStream.size());
- _currentRowDataByteBuffer.putInt(values.length);
- for (short value : values) {
- _variableSizeDataOutputStream.writeShort(value);
- }
+ int objectTypeValue = ObjectSerDeUtils.ObjectType.getObjectType(value).getValue();
+ byte[] bytes = ObjectSerDeUtils.serialize(value, objectTypeValue);
+ _currentRowDataByteBuffer.putInt(bytes.length);
+ _variableSizeDataOutputStream.writeInt(objectTypeValue);
+ _variableSizeDataByteArrayOutputStream.write(bytes);
}
public void setColumn(int colId, int[] values)
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/DataTableImplV2.java b/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/DataTableImplV2.java
index 0e95862..ac3d060 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/DataTableImplV2.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/DataTableImplV2.java
@@ -33,6 +33,8 @@ import org.apache.pinot.common.utils.DataSchema;
import org.apache.pinot.common.utils.DataTable;
import org.apache.pinot.common.utils.StringUtil;
import org.apache.pinot.core.common.ObjectSerDeUtils;
+import org.apache.pinot.spi.utils.ByteArray;
+import org.apache.pinot.spi.utils.BytesUtils;
public class DataTableImplV2 implements DataTable {
@@ -389,6 +391,12 @@ public class DataTableImplV2 implements DataTable {
}
@Override
+ public ByteArray getBytes(int rowId, int colId) {
+ // NOTE: DataTable V2 uses String to store BYTES value
+ return BytesUtils.toByteArray(getString(rowId, colId));
+ }
+
+ @Override
public <T> T getObject(int rowId, int colId) {
int size = positionCursorInVariableBuffer(rowId, colId);
int objectTypeValue = _variableSizeData.getInt();
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/table/Key.java b/pinot-core/src/main/java/org/apache/pinot/core/data/table/Key.java
index 515659a..bfe7d70 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/table/Key.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/table/Key.java
@@ -22,33 +22,36 @@ import java.util.Arrays;
/**
- * Defines the key component of the record
+ * Defines the key component of the record.
+ * <p>Key can be used as the key in a map, and may only contain single-value columns.
+ * <p>For each data type, the value should be stored as:
+ * <ul>
+ * <li>INT: Integer</li>
+ * <li>LONG: Long</li>
+ * <li>FLOAT: Float</li>
+ * <li>DOUBLE: Double</li>
+ * <li>STRING: String</li>
+ * <li>BYTES: ByteArray</li>
+ * </ul>
+ *
+ * TODO: Consider replacing Key with Record as the concept is very close and the implementation is the same
*/
public class Key {
- private Object[] _columns;
-
- public Key(Object[] columns) {
- _columns = columns;
- }
+ private final Object[] _values;
- public Object[] getColumns() {
- return _columns;
+ public Key(Object[] values) {
+ _values = values;
}
+ // NOTE: Not check class for performance concern
+ @SuppressWarnings("EqualsWhichDoesntCheckParameterClass")
@Override
public boolean equals(Object o) {
- Key that = (Key) o;
- return Arrays.deepEquals(_columns, that._columns);
+ return Arrays.equals(_values, ((Key) o)._values);
}
@Override
public int hashCode() {
- return Arrays.deepHashCode(_columns);
+ return Arrays.hashCode(_values);
}
-
- @Override
- public String toString() {
- return Arrays.deepToString(_columns);
- }
-
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/table/Record.java b/pinot-core/src/main/java/org/apache/pinot/core/data/table/Record.java
index b795db9..a2930dd 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/table/Record.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/table/Record.java
@@ -22,35 +22,46 @@ import java.util.Arrays;
/**
- * Defines a single record in Pinot
+ * Defines a single record in Pinot.
+ * <p>Record may contain both single-value and multi-value columns. In order to use the record as the key in a map, it
+ * can only contain single-value columns (to avoid using Arrays.deepEquals() and Arrays.deepHashCode() for performance
+ * concern).
+ * <p>For each data type, the value should be stored as:
+ * <ul>
+ * <li>INT: Integer</li>
+ * <li>LONG: Long</li>
+ * <li>FLOAT: Float</li>
+ * <li>DOUBLE: Double</li>
+ * <li>STRING: String</li>
+ * <li>BYTES: ByteArray</li>
+ * <li>OBJECT (intermediate aggregation result): Object</li>
+ * <li>INT_ARRAY: int[]</li>
+ * <li>LONG_ARRAY: long[]</li>
+ * <li>FLOAT_ARRAY: float[]</li>
+ * <li>DOUBLE_ARRAY: double[]</li>
+ * <li>STRING_ARRAY: String[]</li>
+ * </ul>
*/
public class Record {
- private Object[] _values;
+ private final Object[] _values;
public Record(Object[] values) {
_values = values;
}
- /**
- * Returns the column values contained in the Record
- */
public Object[] getValues() {
return _values;
}
- @Override
- public String toString() {
- return Arrays.deepToString(_values);
- }
-
+ // NOTE: Not check class for performance concern
+ @SuppressWarnings("EqualsWhichDoesntCheckParameterClass")
@Override
public boolean equals(Object o) {
- Record that = (Record) o;
- return Arrays.deepEquals(_values, that._values);
+ return Arrays.equals(_values, ((Record) o)._values);
}
@Override
public int hashCode() {
- return Arrays.deepHashCode(_values);
+ return Arrays.hashCode(_values);
}
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/operator/CombineGroupByOrderByOperator.java b/pinot-core/src/main/java/org/apache/pinot/core/operator/CombineGroupByOrderByOperator.java
index 1c7ff67..5b724f1 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/operator/CombineGroupByOrderByOperator.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/operator/CombineGroupByOrderByOperator.java
@@ -245,30 +245,23 @@ public class CombineGroupByOrderByOperator extends BaseOperator<IntermediateResu
}
private Function<String, Object> getConverterFunction(DataSchema.ColumnDataType columnDataType) {
- Function<String, Object> function;
switch (columnDataType) {
-
case INT:
- function = Integer::valueOf;
- break;
+ return Integer::valueOf;
case LONG:
- function = Long::valueOf;
- break;
+ return Long::valueOf;
case FLOAT:
- function = Float::valueOf;
- break;
+ return Float::valueOf;
case DOUBLE:
- function = Double::valueOf;
- break;
- case BYTES:
- function = BytesUtils::toByteArray;
- break;
+ return Double::valueOf;
case STRING:
+ return s -> s;
+ case BYTES:
+ return BytesUtils::toByteArray;
+ // Add other group-by column type supports here
default:
- function = s -> s;
- break;
+ throw new IllegalStateException();
}
- return function;
}
@Override
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/operator/CombineOperator.java b/pinot-core/src/main/java/org/apache/pinot/core/operator/CombineOperator.java
index b9dcdb6..7b559cc 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/operator/CombineOperator.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/operator/CombineOperator.java
@@ -18,7 +18,6 @@
*/
package org.apache.pinot.core.operator;
-import java.io.Serializable;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
@@ -214,7 +213,7 @@ public class CombineOperator extends BaseOperator<IntermediateResultsBlock> {
Selection selections = brokerRequest.getSelections();
if (selections != null && brokerRequest.getOrderBy() == null) {
// Selection-only
- Collection<Serializable[]> selectionResult = mergedBlock.getSelectionResult();
+ Collection<Object[]> selectionResult = mergedBlock.getSelectionResult();
return selectionResult != null && selectionResult.size() >= selections.getSize();
}
return false;
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/operator/blocks/IntermediateResultsBlock.java b/pinot-core/src/main/java/org/apache/pinot/core/operator/blocks/IntermediateResultsBlock.java
index b061e5e..882c8b2 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/operator/blocks/IntermediateResultsBlock.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/operator/blocks/IntermediateResultsBlock.java
@@ -19,7 +19,6 @@
package org.apache.pinot.core.operator.blocks;
import java.io.IOException;
-import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
@@ -43,6 +42,7 @@ import org.apache.pinot.core.data.table.Table;
import org.apache.pinot.core.query.aggregation.AggregationFunctionContext;
import org.apache.pinot.core.query.aggregation.groupby.AggregationGroupByResult;
import org.apache.pinot.core.query.selection.SelectionOperatorUtils;
+import org.apache.pinot.spi.utils.ByteArray;
/**
@@ -50,7 +50,7 @@ import org.apache.pinot.core.query.selection.SelectionOperatorUtils;
*/
public class IntermediateResultsBlock implements Block {
private DataSchema _dataSchema;
- private Collection<Serializable[]> _selectionResult;
+ private Collection<Object[]> _selectionResult;
private AggregationFunctionContext[] _aggregationFunctionContexts;
private List<Object> _aggregationResult;
private AggregationGroupByResult _aggregationGroupByResult;
@@ -69,7 +69,7 @@ public class IntermediateResultsBlock implements Block {
/**
* Constructor for selection result.
*/
- public IntermediateResultsBlock(DataSchema dataSchema, Collection<Serializable[]> selectionResult) {
+ public IntermediateResultsBlock(DataSchema dataSchema, Collection<Object[]> selectionResult) {
_dataSchema = dataSchema;
_selectionResult = selectionResult;
}
@@ -139,11 +139,11 @@ public class IntermediateResultsBlock implements Block {
}
@Nullable
- public Collection<Serializable[]> getSelectionResult() {
+ public Collection<Object[]> getSelectionResult() {
return _selectionResult;
}
- public void setSelectionResult(@Nullable Collection<Serializable[]> rowEventsSet) {
+ public void setSelectionResult(@Nullable Collection<Object[]> rowEventsSet) {
_selectionResult = rowEventsSet;
}
@@ -274,29 +274,44 @@ public class IntermediateResultsBlock implements Block {
Object value)
throws IOException {
switch (columnDataType) {
-
case INT:
- dataTableBuilder.setColumn(columnIndex, ((Number) value).intValue());
+ dataTableBuilder.setColumn(columnIndex, (int) value);
break;
case LONG:
- dataTableBuilder.setColumn(columnIndex, ((Number) value).longValue());
+ dataTableBuilder.setColumn(columnIndex, (long) value);
break;
case FLOAT:
- dataTableBuilder.setColumn(columnIndex, ((Number) value).floatValue());
+ dataTableBuilder.setColumn(columnIndex, (float) value);
break;
case DOUBLE:
- dataTableBuilder.setColumn(columnIndex, ((Number) value).doubleValue());
+ dataTableBuilder.setColumn(columnIndex, (double) value);
break;
case STRING:
dataTableBuilder.setColumn(columnIndex, (String) value);
break;
case BYTES:
- // FIXME: support BYTES in DataTable instead of converting to string
- dataTableBuilder.setColumn(columnIndex, value.toString()); // ByteArray::toString
+ dataTableBuilder.setColumn(columnIndex, (ByteArray) value);
break;
- default:
+ case OBJECT:
dataTableBuilder.setColumn(columnIndex, value);
break;
+ case INT_ARRAY:
+ dataTableBuilder.setColumn(columnIndex, (int[]) value);
+ break;
+ case LONG_ARRAY:
+ dataTableBuilder.setColumn(columnIndex, (long[]) value);
+ break;
+ case FLOAT_ARRAY:
+ dataTableBuilder.setColumn(columnIndex, (float[]) value);
+ break;
+ case DOUBLE_ARRAY:
+ dataTableBuilder.setColumn(columnIndex, (double[]) value);
+ break;
+ case STRING_ARRAY:
+ dataTableBuilder.setColumn(columnIndex, (String[]) value);
+ break;
+ default:
+ throw new IllegalStateException();
}
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/operator/query/SelectionOnlyOperator.java b/pinot-core/src/main/java/org/apache/pinot/core/operator/query/SelectionOnlyOperator.java
index f86d16c..6c67437 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/operator/query/SelectionOnlyOperator.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/operator/query/SelectionOnlyOperator.java
@@ -18,7 +18,6 @@
*/
package org.apache.pinot.core.operator.query;
-import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import org.apache.pinot.common.request.Selection;
@@ -45,7 +44,7 @@ public class SelectionOnlyOperator extends BaseOperator<IntermediateResultsBlock
private final BlockValSet[] _blockValSets;
private final DataSchema _dataSchema;
private final int _numRowsToKeep;
- private final List<Serializable[]> _rows;
+ private final List<Object[]> _rows;
private int _numDocsScanned = 0;
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/operator/query/SelectionOrderByOperator.java b/pinot-core/src/main/java/org/apache/pinot/core/operator/query/SelectionOrderByOperator.java
index 86b84d9..d4f98fc 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/operator/query/SelectionOrderByOperator.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/operator/query/SelectionOrderByOperator.java
@@ -18,7 +18,6 @@
*/
package org.apache.pinot.core.operator.query;
-import java.io.Serializable;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
@@ -50,7 +49,7 @@ public class SelectionOrderByOperator extends BaseOperator<IntermediateResultsBl
private final TransformResultMetadata[] _expressionMetadata;
private final DataSchema _dataSchema;
private final int _numRowsToKeep;
- private final PriorityQueue<Serializable[]> _rows;
+ private final PriorityQueue<Object[]> _rows;
private int _numDocsScanned = 0;
@@ -79,7 +78,7 @@ public class SelectionOrderByOperator extends BaseOperator<IntermediateResultsBl
getComparator(selection.getSelectionSortSequence()));
}
- private Comparator<Serializable[]> getComparator(List<SelectionSort> sortSequence) {
+ private Comparator<Object[]> getComparator(List<SelectionSort> sortSequence) {
// Compare all single-value columns
int numOrderByExpressions = sortSequence.size();
List<Integer> valueIndexList = new ArrayList<>(numOrderByExpressions);
@@ -104,8 +103,10 @@ public class SelectionOrderByOperator extends BaseOperator<IntermediateResultsBl
return (o1, o2) -> {
for (int i = 0; i < numValuesToCompare; i++) {
int index = valueIndices[i];
- Serializable v1 = o1[index];
- Serializable v2 = o2[index];
+
+ // TODO: Evaluate the performance of casting to Comparable and avoid the switch
+ Object v1 = o1[index];
+ Object v2 = o2[index];
int result;
switch (dataTypes[i]) {
case INT:
@@ -124,8 +125,9 @@ public class SelectionOrderByOperator extends BaseOperator<IntermediateResultsBl
result = ((String) v1).compareTo((String) v2);
break;
case BYTES:
- result = ByteArray.compare((byte[]) v1, (byte[]) v2);
+ result = ((ByteArray) v1).compareTo((ByteArray) v2);
break;
+ // NOTE: Multi-value columns are not comparable, so we should not reach here
default:
throw new IllegalStateException();
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/DistinctTable.java b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/DistinctTable.java
index 8dadf14..de51993 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/DistinctTable.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/DistinctTable.java
@@ -36,7 +36,7 @@ import org.apache.pinot.core.common.datatable.DataTableFactory;
import org.apache.pinot.core.data.table.BaseTable;
import org.apache.pinot.core.data.table.Key;
import org.apache.pinot.core.data.table.Record;
-import org.apache.pinot.spi.utils.BytesUtils;
+import org.apache.pinot.spi.utils.ByteArray;
/**
@@ -120,8 +120,9 @@ public class DistinctTable extends BaseTable {
return dataTable.toBytes();
}
- private void serializeColumns(final Object[] columns, final DataSchema.ColumnDataType[] columnDataTypes,
- final DataTableBuilder dataTableBuilder) {
+ private void serializeColumns(Object[] columns, DataSchema.ColumnDataType[] columnDataTypes,
+ DataTableBuilder dataTableBuilder)
+ throws IOException {
for (int colIndex = 0; colIndex < columns.length; colIndex++) {
switch (columnDataTypes[colIndex]) {
case INT:
@@ -140,7 +141,11 @@ public class DistinctTable extends BaseTable {
dataTableBuilder.setColumn(colIndex, ((String) columns[colIndex]));
break;
case BYTES:
- dataTableBuilder.setColumn(colIndex, BytesUtils.toHexString((byte[]) columns[colIndex]));
+ dataTableBuilder.setColumn(colIndex, (ByteArray) columns[colIndex]);
+ break;
+ // Add other distinct column type supports here
+ default:
+ throw new IllegalStateException();
}
}
}
@@ -188,8 +193,9 @@ public class DistinctTable extends BaseTable {
columnValues[colIndex] = dataTable.getString(rowIndex, colIndex);
break;
case BYTES:
- columnValues[colIndex] = dataTable.getString(rowIndex, colIndex);
+ columnValues[colIndex] = dataTable.getBytes(rowIndex, colIndex);
break;
+ // Add other distinct column type supports here
default:
throw new IllegalStateException(
"Unexpected column data type " + columnDataType + " while deserializing data table for DISTINCT query");
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctAggregationFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctAggregationFunction.java
index 457e431..f681e3a 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctAggregationFunction.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctAggregationFunction.java
@@ -91,17 +91,13 @@ public class DistinctAggregationFunction implements AggregationFunction<Distinct
// for DISTINCT queries without filter.
RowBasedBlockValueFetcher blockValueFetcher = new RowBasedBlockValueFetcher(blockValSets);
- int rowIndex = 0;
// TODO: Do early termination in the operator itself which should
// not call aggregate function at all if the limit has reached
// that will require the interface change since this function
// has to communicate back that required number of records have
// been collected
- while (rowIndex < length) {
- Object[] columnData = blockValueFetcher.getRow(rowIndex);
- Record record = new Record(columnData);
- distinctTable.upsert(record);
- rowIndex++;
+ for (int i = 0; i < length; i++) {
+ distinctTable.upsert(new Record(blockValueFetcher.getRow(i)));
}
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/AggregationDataTableReducer.java b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/AggregationDataTableReducer.java
index e444a62..3bde93a 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/AggregationDataTableReducer.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/AggregationDataTableReducer.java
@@ -43,12 +43,11 @@ import org.apache.pinot.core.util.QueryOptions;
* Helper class to reduce and set Aggregation results into the BrokerResponseNative
*/
public class AggregationDataTableReducer implements DataTableReducer {
-
private final AggregationFunction[] _aggregationFunctions;
private final List<AggregationInfo> _aggregationInfos;
private final int _numAggregationFunctions;
private final boolean _preserveType;
- private boolean _responseFormatSql;
+ private final boolean _responseFormatSql;
AggregationDataTableReducer(BrokerRequest brokerRequest, AggregationFunction[] aggregationFunctions,
QueryOptions queryOptions) {
@@ -112,8 +111,7 @@ public class AggregationDataTableReducer implements DataTableReducer {
if (_responseFormatSql) {
brokerResponseNative.setResultTable(reduceToResultTable(intermediateResults));
} else {
- brokerResponseNative
- .setAggregationResults(reduceToAggregationResult(intermediateResults, dataSchema));
+ brokerResponseNative.setAggregationResults(reduceToAggregationResult(intermediateResults, dataSchema));
}
}
@@ -135,8 +133,7 @@ public class AggregationDataTableReducer implements DataTableReducer {
/**
* Sets aggregation results into AggregationResults
*/
- private List<AggregationResult> reduceToAggregationResult(Object[] intermediateResults,
- DataSchema dataSchema) {
+ private List<AggregationResult> reduceToAggregationResult(Object[] intermediateResults, DataSchema dataSchema) {
// Extract final results and set them into the broker response.
List<AggregationResult> reducedAggregationResults = new ArrayList<>(_numAggregationFunctions);
for (int i = 0; i < _numAggregationFunctions; i++) {
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/CombineService.java b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/CombineService.java
index 9caa3f4..79682e9 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/CombineService.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/CombineService.java
@@ -18,11 +18,9 @@
*/
package org.apache.pinot.core.query.reduce;
-import java.io.Serializable;
import java.util.Collection;
import java.util.List;
import java.util.PriorityQueue;
-import javax.annotation.Nonnull;
import org.apache.pinot.common.exception.QueryException;
import org.apache.pinot.common.request.BrokerRequest;
import org.apache.pinot.common.request.Selection;
@@ -45,8 +43,8 @@ public class CombineService {
private static final Logger LOGGER = LoggerFactory.getLogger(CombineService.class);
- public static void mergeTwoBlocks(@Nonnull BrokerRequest brokerRequest, @Nonnull IntermediateResultsBlock mergedBlock,
- @Nonnull IntermediateResultsBlock blockToMerge) {
+ public static void mergeTwoBlocks(BrokerRequest brokerRequest, IntermediateResultsBlock mergedBlock,
+ IntermediateResultsBlock blockToMerge) {
// Combine processing exceptions.
List<ProcessingException> mergedProcessingExceptions = mergedBlock.getProcessingExceptions();
List<ProcessingException> processingExceptionsToMerge = blockToMerge.getProcessingExceptions();
@@ -95,8 +93,8 @@ public class CombineService {
// Result set size will be zero if no row matches the predicate.
DataSchema mergedBlockSchema = mergedBlock.getDataSchema();
DataSchema blockToMergeSchema = blockToMerge.getDataSchema();
- Collection<Serializable[]> mergedBlockResultSet = mergedBlock.getSelectionResult();
- Collection<Serializable[]> blockToMergeResultSet = blockToMerge.getSelectionResult();
+ Collection<Object[]> mergedBlockResultSet = mergedBlock.getSelectionResult();
+ Collection<Object[]> blockToMergeResultSet = blockToMerge.getSelectionResult();
if (mergedBlockSchema == null || mergedBlockResultSet.size() == 0) {
// No data in merged block.
@@ -130,7 +128,7 @@ public class CombineService {
if (isSelectionOrderBy) {
// Combine selection order-by.
SelectionOperatorUtils
- .mergeWithOrdering((PriorityQueue<Serializable[]>) mergedBlockResultSet, blockToMergeResultSet,
+ .mergeWithOrdering((PriorityQueue<Object[]>) mergedBlockResultSet, blockToMergeResultSet,
selection.getOffset() + selectionSize);
} else {
// Combine selection only.
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/DistinctDataTableReducer.java b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/DistinctDataTableReducer.java
index e45366a..d6e888c 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/DistinctDataTableReducer.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/DistinctDataTableReducer.java
@@ -37,7 +37,7 @@ import org.apache.pinot.common.utils.DataTable;
import org.apache.pinot.core.data.table.Record;
import org.apache.pinot.core.query.aggregation.DistinctTable;
import org.apache.pinot.core.query.aggregation.function.AggregationFunction;
-import org.apache.pinot.core.query.aggregation.function.AggregationFunctionUtils;
+import org.apache.pinot.core.query.selection.SelectionOperatorUtils;
import org.apache.pinot.core.transport.ServerRoutingInstance;
import org.apache.pinot.core.util.QueryOptions;
import org.apache.pinot.pql.parsers.pql2.ast.FunctionCallAstNode;
@@ -47,11 +47,11 @@ import org.apache.pinot.pql.parsers.pql2.ast.FunctionCallAstNode;
* Helper class to reduce data tables and set results of distinct query into the BrokerResponseNative
*/
public class DistinctDataTableReducer implements DataTableReducer {
-
private final BrokerRequest _brokerRequest;
private final AggregationFunction _aggregationFunction;
- private boolean _responseFormatSql;
+ private final boolean _responseFormatSql;
+ // TODO: queryOptions.isPreserveType() is ignored for DISTINCT queries.
DistinctDataTableReducer(BrokerRequest brokerRequest, AggregationFunction aggregationFunction,
QueryOptions queryOptions) {
_brokerRequest = brokerRequest;
@@ -129,30 +129,37 @@ public class DistinctDataTableReducer implements DataTableReducer {
}
private SelectionResults reduceToSelectionResult(DistinctTable distinctTable) {
- List<Serializable[]> resultSet = new ArrayList<>(distinctTable.size());
- String[] columnNames = distinctTable.getDataSchema().getColumnNames();
+ List<Serializable[]> rows = new ArrayList<>(distinctTable.size());
+ DataSchema dataSchema = distinctTable.getDataSchema();
+ DataSchema.ColumnDataType[] columnDataTypes = dataSchema.getColumnDataTypes();
+ int numColumns = columnDataTypes.length;
Iterator<Record> iterator = distinctTable.iterator();
while (iterator.hasNext()) {
- Record record = iterator.next();
- Object[] columns = record.getValues();
- Serializable[] distinctRow = new Serializable[columns.length];
- for (int col = 0; col < columns.length; col++) {
- final Serializable columnValue = AggregationFunctionUtils.getSerializableValue(columns[col]);
- distinctRow[col] = columnValue;
+ Object[] values = iterator.next().getValues();
+ Serializable[] row = new Serializable[numColumns];
+ for (int i = 0; i < numColumns; i++) {
+ row[i] = SelectionOperatorUtils.convertValueToType(values[i], columnDataTypes[i]);
}
- resultSet.add(distinctRow);
+ rows.add(row);
}
- return new SelectionResults(Arrays.asList(columnNames), resultSet);
+ return new SelectionResults(Arrays.asList(dataSchema.getColumnNames()), rows);
}
private ResultTable reduceToResultTable(DistinctTable distinctTable) {
- List<Object[]> resultSet = new ArrayList<>(distinctTable.size());
+ List<Object[]> rows = new ArrayList<>(distinctTable.size());
+ DataSchema dataSchema = distinctTable.getDataSchema();
+ DataSchema.ColumnDataType[] columnDataTypes = dataSchema.getColumnDataTypes();
+ int numColumns = columnDataTypes.length;
Iterator<Record> iterator = distinctTable.iterator();
while (iterator.hasNext()) {
- Record record = iterator.next();
- resultSet.add(record.getValues());
+ Object[] values = iterator.next().getValues();
+ Object[] row = new Object[numColumns];
+ for (int i = 0; i < numColumns; i++) {
+ row[i] = SelectionOperatorUtils.convertValueToType(values[i], columnDataTypes[i]);
+ }
+ rows.add(row);
}
- return new ResultTable(distinctTable.getDataSchema(), resultSet);
+ return new ResultTable(dataSchema, rows);
}
private String[] getDistinctColumns() {
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/GroupByDataTableReducer.java b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/GroupByDataTableReducer.java
index a72c7a9..97c2019 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/GroupByDataTableReducer.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/GroupByDataTableReducer.java
@@ -55,7 +55,6 @@ import org.apache.pinot.core.query.aggregation.groupby.AggregationGroupByTrimmin
import org.apache.pinot.core.transport.ServerRoutingInstance;
import org.apache.pinot.core.util.GroupByUtils;
import org.apache.pinot.core.util.QueryOptions;
-import org.apache.pinot.spi.utils.BytesUtils;
/**
@@ -315,7 +314,6 @@ public class GroupByDataTableReducer implements DataTableReducer {
DataSchema.ColumnDataType columnDataType = dataSchema.getColumnDataType(i);
BiFunction<Integer, Integer, Object> function;
switch (columnDataType) {
-
case INT:
function = dataTable::getInt;
break;
@@ -332,11 +330,14 @@ public class GroupByDataTableReducer implements DataTableReducer {
function = dataTable::getString;
break;
case BYTES:
- // FIXME: support BYTES in DataTable instead of converting to string
- function = (row, col) -> BytesUtils.toByteArray(dataTable.getString(row, col));
+ function = dataTable::getBytes;
break;
- default:
+ case OBJECT:
function = dataTable::getObject;
+ break;
+ // Add other aggregation intermediate result / group-by column type supports here
+ default:
+ throw new IllegalStateException();
}
functions[i] = function;
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/SelectionDataTableReducer.java b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/SelectionDataTableReducer.java
index 266d6db..13bd794 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/SelectionDataTableReducer.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/SelectionDataTableReducer.java
@@ -18,7 +18,6 @@
*/
package org.apache.pinot.core.query.reduce;
-import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
@@ -48,12 +47,11 @@ import org.slf4j.LoggerFactory;
* Helper class to reduce and set Selection results into the BrokerResponseNative
*/
public class SelectionDataTableReducer implements DataTableReducer {
-
private static final Logger LOGGER = LoggerFactory.getLogger(SelectionDataTableReducer.class);
private final Selection _selection;
- private boolean _preserveType;
- private boolean _responseFormatSql;
+ private final boolean _preserveType;
+ private final boolean _responseFormatSql;
SelectionDataTableReducer(BrokerRequest brokerRequest, QueryOptions queryOptions) {
_selection = brokerRequest.getSelections();
@@ -103,8 +101,6 @@ public class SelectionDataTableReducer implements DataTableReducer {
SelectionOperatorService selectionService = new SelectionOperatorService(_selection, dataSchema);
selectionService.reduceWithOrdering(dataTableMap.values());
if (_responseFormatSql) {
- // TODO: Selection uses Serializable[] in all its operations
- // Converting that to Object[] end to end would be a big change, and will be done in future PRs
brokerResponseNative.setResultTable(selectionService.renderResultTableWithOrdering());
} else {
brokerResponseNative.setSelectionResults(selectionService.renderSelectionResultsWithOrdering(_preserveType));
@@ -113,11 +109,8 @@ public class SelectionDataTableReducer implements DataTableReducer {
// Selection only
List<String> selectionColumns =
SelectionOperatorUtils.getSelectionColumns(_selection.getSelectionColumns(), dataSchema);
- List<Serializable[]> reducedRows =
- SelectionOperatorUtils.reduceWithoutOrdering(dataTableMap.values(), selectionSize);
+ List<Object[]> reducedRows = SelectionOperatorUtils.reduceWithoutOrdering(dataTableMap.values(), selectionSize);
if (_responseFormatSql) {
- // TODO: Selection uses Serializable[] in all its operations
- // Converting that to Object[] end to end would be a big change, and will be done in future PRs
brokerResponseNative
.setResultTable(SelectionOperatorUtils.renderResultTableWithoutOrdering(reducedRows, dataSchema));
} else {
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/selection/SelectionOperatorService.java b/pinot-core/src/main/java/org/apache/pinot/core/query/selection/SelectionOperatorService.java
index 1ee0b3c..edd4720 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/query/selection/SelectionOperatorService.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/selection/SelectionOperatorService.java
@@ -63,7 +63,7 @@ public class SelectionOperatorService {
private final DataSchema _dataSchema;
private final int _offset;
private final int _numRowsToKeep;
- private final PriorityQueue<Serializable[]> _rows;
+ private final PriorityQueue<Object[]> _rows;
/**
* Constructor for <code>SelectionOperatorService</code> with {@link DataSchema}. (Inter segment)
@@ -87,7 +87,7 @@ public class SelectionOperatorService {
*
* @return flexible {@link Comparator} for selection rows.
*/
- private Comparator<Serializable[]> getTypeCompatibleComparator(List<SelectionSort> sortSequence) {
+ private Comparator<Object[]> getTypeCompatibleComparator(List<SelectionSort> sortSequence) {
// Compare all single-value columns
int numOrderByExpressions = sortSequence.size();
List<Integer> valueIndexList = new ArrayList<>(numOrderByExpressions);
@@ -112,13 +112,14 @@ public class SelectionOperatorService {
return (o1, o2) -> {
for (int i = 0; i < numValuesToCompare; i++) {
int index = valueIndices[i];
- Serializable v1 = o1[index];
- Serializable v2 = o2[index];
+ Object v1 = o1[index];
+ Object v2 = o2[index];
int result;
if (isNumber[i]) {
result = Double.compare(((Number) v1).doubleValue(), ((Number) v2).doubleValue());
} else {
- result = ((String) v1).compareTo((String) v2);
+ //noinspection unchecked
+ result = ((Comparable) v1).compareTo(v2);
}
if (result != 0) {
return result * multipliers[i];
@@ -133,7 +134,7 @@ public class SelectionOperatorService {
*
* @return selection results.
*/
- public PriorityQueue<Serializable[]> getRows() {
+ public PriorityQueue<Object[]> getRows() {
return _rows;
}
@@ -145,7 +146,7 @@ public class SelectionOperatorService {
for (DataTable dataTable : dataTables) {
int numRows = dataTable.getNumberOfRows();
for (int rowId = 0; rowId < numRows; rowId++) {
- Serializable[] row = SelectionOperatorUtils.extractRowFromDataTable(dataTable, rowId);
+ Object[] row = SelectionOperatorUtils.extractRowFromDataTable(dataTable, rowId);
SelectionOperatorUtils.addToPriorityQueue(row, _rows, _numRowsToKeep);
}
}
@@ -161,22 +162,33 @@ public class SelectionOperatorService {
*/
public SelectionResults renderSelectionResultsWithOrdering(boolean preserveType) {
LinkedList<Serializable[]> rowsInSelectionResults = new LinkedList<>();
-
int[] columnIndices = SelectionOperatorUtils.getColumnIndices(_selectionColumns, _dataSchema);
- DataSchema.ColumnDataType[] columnDataTypes;
+ int numColumns = columnIndices.length;
+ DataSchema.ColumnDataType[] columnDataTypes = _dataSchema.getColumnDataTypes();
+
if (preserveType) {
- columnDataTypes = null;
+ while (_rows.size() > _offset) {
+ Object[] row = _rows.poll();
+ assert row != null;
+ Serializable[] extractedRow = new Serializable[numColumns];
+ for (int i = 0; i < numColumns; i++) {
+ int columnIndex = columnIndices[i];
+ extractedRow[i] = SelectionOperatorUtils.convertValueToType(row[columnIndex], columnDataTypes[columnIndex]);
+ }
+ rowsInSelectionResults.addFirst(extractedRow);
+ }
} else {
- int numColumns = _selectionColumns.size();
- columnDataTypes = new DataSchema.ColumnDataType[numColumns];
- for (int i = 0; i < numColumns; i++) {
- columnDataTypes[i] = _dataSchema.getColumnDataType(columnIndices[i]);
+ while (_rows.size() > _offset) {
+ Object[] row = _rows.poll();
+ assert row != null;
+ Serializable[] extractedRow = new Serializable[numColumns];
+ for (int i = 0; i < numColumns; i++) {
+ int columnIndex = columnIndices[i];
+ extractedRow[i] = SelectionOperatorUtils.getFormattedValue(row[columnIndex], columnDataTypes[columnIndex]);
+ }
+ rowsInSelectionResults.addFirst(extractedRow);
}
}
- while (_rows.size() > _offset) {
- rowsInSelectionResults
- .addFirst(SelectionOperatorUtils.extractColumns(_rows.poll(), columnIndices, columnDataTypes));
- }
return new SelectionResults(_selectionColumns, rowsInSelectionResults);
}
@@ -193,18 +205,29 @@ public class SelectionOperatorService {
LinkedList<Object[]> rowsInSelectionResults = new LinkedList<>();
int[] columnIndices = SelectionOperatorUtils.getColumnIndices(_selectionColumns, _dataSchema);
int numColumns = columnIndices.length;
+ DataSchema.ColumnDataType[] columnDataTypes = _dataSchema.getColumnDataTypes();
while (_rows.size() > _offset) {
- Serializable[] row = _rows.poll();
+ Object[] row = _rows.poll();
+ assert row != null;
Object[] extractedRow = new Object[numColumns];
for (int i = 0; i < numColumns; i++) {
- extractedRow[i] = row[columnIndices[i]];
+ int columnIndex = columnIndices[i];
+ extractedRow[i] = SelectionOperatorUtils.convertValueToType(row[columnIndex], columnDataTypes[columnIndex]);
}
rowsInSelectionResults.addFirst(extractedRow);
}
- // final data schema in ResultTable should be created based on order in _selectionColumns
- DataSchema finalDataSchema = SelectionOperatorUtils.getResultTableDataSchema(_dataSchema, _selectionColumns);
- return new ResultTable(finalDataSchema, rowsInSelectionResults);
+ // Construct the result data schema
+ String[] columnNames = _dataSchema.getColumnNames();
+ String[] resultColumnNames = new String[numColumns];
+ DataSchema.ColumnDataType[] resultColumnDataTypes = new DataSchema.ColumnDataType[numColumns];
+ for (int i = 0; i < numColumns; i++) {
+ int columnIndex = columnIndices[i];
+ resultColumnNames[i] = columnNames[columnIndex];
+ resultColumnDataTypes[i] = columnDataTypes[columnIndex];
+ }
+ DataSchema resultDataSchema = new DataSchema(resultColumnNames, resultColumnDataTypes);
+ return new ResultTable(resultDataSchema, rowsInSelectionResults);
}
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/selection/SelectionOperatorUtils.java b/pinot-core/src/main/java/org/apache/pinot/core/query/selection/SelectionOperatorUtils.java
index 8aeca7e..7bdd38c 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/query/selection/SelectionOperatorUtils.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/selection/SelectionOperatorUtils.java
@@ -32,7 +32,6 @@ import java.util.Map;
import java.util.PriorityQueue;
import java.util.Set;
import java.util.stream.Collectors;
-import javax.annotation.Nullable;
import org.apache.pinot.common.request.SelectionSort;
import org.apache.pinot.common.request.transform.TransformExpressionTree;
import org.apache.pinot.common.response.broker.ResultTable;
@@ -43,7 +42,7 @@ import org.apache.pinot.core.common.datatable.DataTableBuilder;
import org.apache.pinot.core.indexsegment.IndexSegment;
import org.apache.pinot.core.util.ArrayCopyUtils;
import org.apache.pinot.pql.parsers.pql2.ast.IdentifierAstNode;
-import org.apache.pinot.spi.utils.BytesUtils;
+import org.apache.pinot.spi.utils.ByteArray;
/**
@@ -215,9 +214,9 @@ public class SelectionOperatorUtils {
* @param rowsToMerge partial results 2.
* @param selectionSize size of the selection.
*/
- public static void mergeWithoutOrdering(Collection<Serializable[]> mergedRows, Collection<Serializable[]> rowsToMerge,
+ public static void mergeWithoutOrdering(Collection<Object[]> mergedRows, Collection<Object[]> rowsToMerge,
int selectionSize) {
- Iterator<Serializable[]> iterator = rowsToMerge.iterator();
+ Iterator<Object[]> iterator = rowsToMerge.iterator();
while (mergedRows.size() < selectionSize && iterator.hasNext()) {
mergedRows.add(iterator.next());
}
@@ -231,9 +230,9 @@ public class SelectionOperatorUtils {
* @param rowsToMerge partial results 2.
* @param maxNumRows maximum number of rows need to be stored.
*/
- public static void mergeWithOrdering(PriorityQueue<Serializable[]> mergedRows, Collection<Serializable[]> rowsToMerge,
+ public static void mergeWithOrdering(PriorityQueue<Object[]> mergedRows, Collection<Object[]> rowsToMerge,
int maxNumRows) {
- for (Serializable[] row : rowsToMerge) {
+ for (Object[] row : rowsToMerge) {
addToPriorityQueue(row, mergedRows, maxNumRows);
}
}
@@ -244,20 +243,24 @@ public class SelectionOperatorUtils {
* <p>The actual data types for each column in rows can be different but must be compatible with each other.
* <p>Before write each row into the data table, first convert it to match the data types in data schema.
*
+ * TODO: Type compatibility is not supported for selection order-by because all segments on the same server shared the
+ * same comparator. Another solution is to always use the table schema to execute the query (preferable because
+ * type compatible checks are expensive).
+ *
* @param rows {@link Collection} of selection rows.
* @param dataSchema data schema.
* @return data table.
* @throws Exception
*/
- public static DataTable getDataTableFromRows(Collection<Serializable[]> rows, DataSchema dataSchema)
+ public static DataTable getDataTableFromRows(Collection<Object[]> rows, DataSchema dataSchema)
throws Exception {
int numColumns = dataSchema.size();
DataTableBuilder dataTableBuilder = new DataTableBuilder(dataSchema);
- for (Serializable[] row : rows) {
+ for (Object[] row : rows) {
dataTableBuilder.startRow();
for (int i = 0; i < numColumns; i++) {
- Serializable columnValue = row[i];
+ Object columnValue = row[i];
DataSchema.ColumnDataType columnDataType = dataSchema.getColumnDataType(i);
switch (columnDataType) {
// Single-value column
@@ -277,7 +280,7 @@ public class SelectionOperatorUtils {
dataTableBuilder.setColumn(i, ((String) columnValue));
break;
case BYTES:
- dataTableBuilder.setColumn(i, BytesUtils.toHexString((byte[]) columnValue));
+ dataTableBuilder.setColumn(i, (ByteArray) columnValue);
break;
// Multi-value column
@@ -345,11 +348,11 @@ public class SelectionOperatorUtils {
* @param rowId row id.
* @return selection row.
*/
- public static Serializable[] extractRowFromDataTable(DataTable dataTable, int rowId) {
+ public static Object[] extractRowFromDataTable(DataTable dataTable, int rowId) {
DataSchema dataSchema = dataTable.getDataSchema();
int numColumns = dataSchema.size();
- Serializable[] row = new Serializable[numColumns];
+ Object[] row = new Object[numColumns];
for (int i = 0; i < numColumns; i++) {
DataSchema.ColumnDataType columnDataType = dataSchema.getColumnDataType(i);
switch (columnDataType) {
@@ -367,9 +370,11 @@ public class SelectionOperatorUtils {
row[i] = dataTable.getDouble(rowId, i);
break;
case STRING:
- case BYTES:
row[i] = dataTable.getString(rowId, i);
break;
+ case BYTES:
+ row[i] = dataTable.getBytes(rowId, i);
+ break;
// Multi-value column
case INT_ARRAY:
@@ -401,8 +406,8 @@ public class SelectionOperatorUtils {
* Reduces a collection of {@link DataTable}s to selection rows for selection queries without <code>ORDER BY</code>.
* (Broker side)
*/
- public static List<Serializable[]> reduceWithoutOrdering(Collection<DataTable> dataTables, int selectionSize) {
- List<Serializable[]> rows = new ArrayList<>(selectionSize);
+ public static List<Object[]> reduceWithoutOrdering(Collection<DataTable> dataTables, int selectionSize) {
+ List<Object[]> rows = new ArrayList<>(selectionSize);
for (DataTable dataTable : dataTables) {
int numRows = dataTable.getNumberOfRows();
for (int rowId = 0; rowId < numRows; rowId++) {
@@ -427,15 +432,30 @@ public class SelectionOperatorUtils {
* @param selectionColumns selection columns.
* @return {@link SelectionResults} object results.
*/
- public static SelectionResults renderSelectionResultsWithoutOrdering(List<Serializable[]> rows, DataSchema dataSchema,
+ public static SelectionResults renderSelectionResultsWithoutOrdering(List<Object[]> rows, DataSchema dataSchema,
List<String> selectionColumns, boolean preserveType) {
int numRows = rows.size();
- if (!preserveType) {
- for (int i = 0; i < numRows; i++) {
- rows.set(i, formatRowWithoutOrdering(rows.get(i), dataSchema));
+ List<Serializable[]> resultRows = new ArrayList<>(numRows);
+ DataSchema.ColumnDataType[] columnDataTypes = dataSchema.getColumnDataTypes();
+ int numColumns = columnDataTypes.length;
+ if (preserveType) {
+ for (Object[] row : rows) {
+ Serializable[] resultRow = new Serializable[numColumns];
+ for (int i = 0; i < numColumns; i++) {
+ resultRow[i] = convertValueToType(row[i], columnDataTypes[i]);
+ }
+ resultRows.add(resultRow);
+ }
+ } else {
+ for (Object[] row : rows) {
+ Serializable[] resultRow = new Serializable[numColumns];
+ for (int i = 0; i < numColumns; i++) {
+ resultRow[i] = getFormattedValue(row[i], columnDataTypes[i]);
+ }
+ resultRows.add(resultRow);
}
}
- return new SelectionResults(selectionColumns, rows);
+ return new SelectionResults(selectionColumns, resultRows);
}
/**
@@ -448,9 +468,18 @@ public class SelectionOperatorUtils {
* @param dataSchema data schema.
* @return {@link ResultTable} object results.
*/
- public static ResultTable renderResultTableWithoutOrdering(List<Serializable[]> rows, DataSchema dataSchema) {
- List<Object[]> resultRows = new ArrayList<>(rows.size());
- resultRows.addAll(rows);
+ public static ResultTable renderResultTableWithoutOrdering(List<Object[]> rows, DataSchema dataSchema) {
+ int numRows = rows.size();
+ List<Object[]> resultRows = new ArrayList<>(numRows);
+ DataSchema.ColumnDataType[] columnDataTypes = dataSchema.getColumnDataTypes();
+ int numColumns = columnDataTypes.length;
+ for (Object[] row : rows) {
+ Object[] resultRow = new Object[numColumns];
+ for (int i = 0; i < numColumns; i++) {
+ resultRow[i] = convertValueToType(row[i], columnDataTypes[i]);
+ }
+ resultRows.add(resultRow);
+ }
return new ResultTable(dataSchema, resultRows);
}
@@ -481,55 +510,81 @@ public class SelectionOperatorUtils {
}
/**
- * Extract columns from the row based on the given column indices.
- * <p>The extracted row is used to build the {@link SelectionResults}.
- *
- * @param row selection row to be extracted.
- * @param columnIndices column indices.
- * @return selection row.
+ * Converts a value into the given data type. (Broker side)
+ * <p>Actual value type can be different with data type passed in, but they must be type compatible.
*/
- public static Serializable[] extractColumns(Serializable[] row, int[] columnIndices,
- @Nullable DataSchema.ColumnDataType[] columnDataTypes) {
- int numColumns = columnIndices.length;
- Serializable[] extractedRow = new Serializable[numColumns];
- if (columnDataTypes == null) {
- for (int i = 0; i < numColumns; i++) {
- extractedRow[i] = row[columnIndices[i]];
- }
- } else {
- for (int i = 0; i < numColumns; i++) {
- extractedRow[i] = getFormattedValue(row[columnIndices[i]], columnDataTypes[i]);
- }
- }
- return extractedRow;
- }
+ public static Serializable convertValueToType(Object value, DataSchema.ColumnDataType dataType) {
+ switch (dataType) {
+ // Single-value column
+ case INT:
+ return ((Number) value).intValue();
+ case LONG:
+ return ((Number) value).longValue();
+ case FLOAT:
+ return ((Number) value).floatValue();
+ case DOUBLE:
+ return ((Number) value).doubleValue();
+ // NOTE: Return hex-encoded String for BYTES columns for backward-compatibility
+ // TODO: Revisit to see whether we should return byte[] instead
+ case BYTES:
+ return ((ByteArray) value).toHexString();
- /**
- * Helper method to format a selection row, make all values string or string array type based on data schema passed in
- * for selection queries without <code>ORDER BY</code>. (Broker side)
- * <p>Formatted row is used to build the {@link SelectionResults}.
- *
- * @param row selection row to be formatted.
- * @param dataSchema data schema.
- */
- private static Serializable[] formatRowWithoutOrdering(Serializable[] row, DataSchema dataSchema) {
- int numColumns = row.length;
- for (int i = 0; i < numColumns; i++) {
- row[i] = getFormattedValue(row[i], dataSchema.getColumnDataType(i));
+ // Multi-value column
+ case LONG_ARRAY:
+ // LONG_ARRAY type covers INT_ARRAY and LONG_ARRAY
+ if (value instanceof int[]) {
+ int[] ints = (int[]) value;
+ int length = ints.length;
+ long[] longs = new long[length];
+ for (int i = 0; i < length; i++) {
+ longs[i] = ints[i];
+ }
+ return longs;
+ } else {
+ return (long[]) value;
+ }
+ case DOUBLE_ARRAY:
+ // DOUBLE_ARRAY type covers INT_ARRAY, LONG_ARRAY, FLOAT_ARRAY and DOUBLE_ARRAY
+ if (value instanceof int[]) {
+ int[] ints = (int[]) value;
+ int length = ints.length;
+ double[] doubles = new double[length];
+ for (int i = 0; i < length; i++) {
+ doubles[i] = ints[i];
+ }
+ return doubles;
+ } else if (value instanceof long[]) {
+ long[] longs = (long[]) value;
+ int length = longs.length;
+ double[] doubles = new double[length];
+ for (int i = 0; i < length; i++) {
+ doubles[i] = longs[i];
+ }
+ return doubles;
+ } else if (value instanceof float[]) {
+ float[] floats = (float[]) value;
+ int length = floats.length;
+ double[] doubles = new double[length];
+ for (int i = 0; i < length; i++) {
+ doubles[i] = floats[i];
+ }
+ return doubles;
+ } else {
+ return (double[]) value;
+ }
+
+ default:
+ // For STRING, INT_ARRAY, FLOAT_ARRAY and STRING_ARRAY, no need to format
+ return (Serializable) value;
}
- return row;
}
/**
- * Format a {@link Serializable} value into a {@link String} or {@link String} array based on the data type.
- * (Broker side)
+ * Formats a value into a {@code String} (single-value column) or {@code String[]} (multi-value column) based on the
+ * data type. (Broker side)
* <p>Actual value type can be different with data type passed in, but they must be type compatible.
- *
- * @param value value to be formatted.
- * @param dataType data type.
- * @return formatted value.
*/
- private static Serializable getFormattedValue(Serializable value, DataSchema.ColumnDataType dataType) {
+ public static Serializable getFormattedValue(Object value, DataSchema.ColumnDataType dataType) {
switch (dataType) {
// Single-value column
case INT:
@@ -540,6 +595,9 @@ public class SelectionOperatorUtils {
return THREAD_LOCAL_FLOAT_FORMAT.get().format(((Number) value).floatValue());
case DOUBLE:
return THREAD_LOCAL_DOUBLE_FORMAT.get().format(((Number) value).doubleValue());
+ // NOTE: Return String for BYTES columns for backward-compatibility
+ case BYTES:
+ return ((ByteArray) value).toHexString();
// Multi-value column
case INT_ARRAY:
@@ -617,8 +675,8 @@ public class SelectionOperatorUtils {
}
default:
- // For STRING, BYTES and STRING_ARRAY, no need to intFormat
- return value;
+ // For STRING and STRING_ARRAY, no need to format
+ return (Serializable) value;
}
}
diff --git a/pinot-core/src/test/java/org/apache/pinot/queries/DistinctQueriesTest.java b/pinot-core/src/test/java/org/apache/pinot/queries/DistinctQueriesTest.java
index 647b68b..607288a 100644
--- a/pinot-core/src/test/java/org/apache/pinot/queries/DistinctQueriesTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/queries/DistinctQueriesTest.java
@@ -57,6 +57,7 @@ import org.apache.pinot.spi.config.TableType;
import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.utils.ByteArray;
import org.apache.pinot.spi.utils.BytesUtils;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
@@ -69,7 +70,6 @@ import static org.testng.Assert.assertTrue;
/**
* Queries test for DISTINCT queries.
- * TODO: Add a test for 'ORDER BY bytesColumn' when it is supported
*/
public class DistinctQueriesTest extends BaseQueriesTest {
private static final File INDEX_DIR = new File(FileUtils.getTempDirectory(), "DistinctQueryTest");
@@ -169,6 +169,7 @@ public class DistinctQueriesTest extends BaseQueriesTest {
* <ul>
* <li>Selecting all columns</li>
* <li>Selecting some columns with filter</li>
+ * <li>Selecting some columns order by BYTES column</li>
* <li>Selecting some columns transform, filter, order-by and limit</li>
* <li>Selecting some columns with filter that does not match any record</li>
* </ul>
@@ -207,7 +208,7 @@ public class DistinctQueriesTest extends BaseQueriesTest {
assertEquals(((Float) values[2]).intValue(), intValue);
assertEquals(((Double) values[3]).intValue(), intValue);
assertEquals(Integer.parseInt((String) values[4]), intValue);
- assertEquals(StringUtil.decodeUtf8((byte[]) values[5]), values[4]);
+ assertEquals(StringUtil.decodeUtf8(((ByteArray) values[5]).getBytes()), values[4]);
actualValues.add(intValue);
}
assertEquals(actualValues, expectedValues);
@@ -236,13 +237,42 @@ public class DistinctQueriesTest extends BaseQueriesTest {
Record record = iterator.next();
Object[] values = record.getValues();
int intValue = Integer.parseInt((String) values[0]);
- assertEquals(StringUtil.decodeUtf8((byte[]) values[1]), values[0]);
+ assertEquals(StringUtil.decodeUtf8(((ByteArray) values[1]).getBytes()), values[0]);
assertEquals(((Float) values[2]).intValue(), intValue);
actualValues.add(intValue);
}
assertEquals(actualValues, expectedValues);
}
{
+ // Test selecting some columns order by BYTES column
+ String query = "SELECT DISTINCT(intColumn, bytesColumn) FROM testTable ORDER BY bytesColumn LIMIT 5";
+
+ // Check data schema
+ DistinctTable distinctTable = getDistinctTableInnerSegment(query);
+ DataSchema dataSchema = distinctTable.getDataSchema();
+ assertEquals(dataSchema.getColumnNames(), new String[]{"intColumn", "bytesColumn"});
+ assertEquals(dataSchema.getColumnDataTypes(), new ColumnDataType[]{ColumnDataType.INT, ColumnDataType.BYTES});
+
+ // Check values, where all 100 unique values should be returned (limit won't take effect on server side)
+ // TODO: After optimizing the DistinctTable (only keep the limit number of records), only 5 values should be
+ // returned
+ assertEquals(distinctTable.size(), NUM_UNIQUE_RECORDS_PER_SEGMENT);
+ Set<Integer> expectedValues = new HashSet<>();
+ for (int i = 0; i < NUM_UNIQUE_RECORDS_PER_SEGMENT; i++) {
+ expectedValues.add(i);
+ }
+ Set<Integer> actualValues = new HashSet<>();
+ Iterator<Record> iterator = distinctTable.iterator();
+ while (iterator.hasNext()) {
+ Record record = iterator.next();
+ Object[] values = record.getValues();
+ int intValue = (int) values[0];
+ assertEquals(StringUtil.decodeUtf8(((ByteArray) values[1]).getBytes()), Integer.toString(intValue));
+ actualValues.add(intValue);
+ }
+ assertEquals(actualValues, expectedValues);
+ }
+ {
// Test selecting some columns with transform, filter, order-by and limit
String query =
"SELECT DISTINCT(ADD(intColumn, floatColumn), stringColumn) FROM testTable WHERE longColumn < 60 ORDER BY stringColumn DESC, ADD(intColumn, floatColumn) ASC LIMIT 10";
@@ -255,6 +285,8 @@ public class DistinctQueriesTest extends BaseQueriesTest {
new ColumnDataType[]{ColumnDataType.DOUBLE, ColumnDataType.STRING});
// Check values, where 60 matched values should be returned (limit won't take effect on server side)
+ // TODO: After optimizing the DistinctTable (only keep the limit number of records), only 10 values should be
+ // returned
assertEquals(distinctTable.size(), 60);
Set<Integer> expectedValues = new HashSet<>();
for (int i = 0; i < 60; i++) {
@@ -310,6 +342,7 @@ public class DistinctQueriesTest extends BaseQueriesTest {
* <ul>
* <li>Selecting all columns</li>
* <li>Selecting some columns with filter</li>
+ * <li>Selecting some columns order by BYTES column</li>
* <li>Selecting some columns transform, filter, order-by and limit</li>
* <li>Selecting some columns with filter that does not match any record</li>
* <li>
@@ -435,6 +468,44 @@ public class DistinctQueriesTest extends BaseQueriesTest {
assertEquals(sqlValues, expectedValues);
}
{
+ // Test selecting some columns order by BYTES column
+ String pqlQuery = "SELECT DISTINCT(intColumn, bytesColumn) FROM testTable ORDER BY bytesColumn LIMIT 5";
+ String sqlQuery = "SELECT DISTINCT intColumn, bytesColumn FROM testTable ORDER BY bytesColumn LIMIT 5";
+
+ // Check data schema
+ BrokerResponseNative pqlResponse = getBrokerResponseForPqlQuery(pqlQuery);
+ SelectionResults selectionResults = pqlResponse.getSelectionResults();
+ assertNotNull(selectionResults);
+ assertEquals(selectionResults.getColumns(), Arrays.asList("intColumn", "bytesColumn"));
+ BrokerResponseNative sqlResponse = getBrokerResponseForSqlQuery(sqlQuery);
+ ResultTable resultTable = sqlResponse.getResultTable();
+ assertNotNull(resultTable);
+ DataSchema dataSchema = resultTable.getDataSchema();
+ assertEquals(dataSchema.getColumnNames(), new String[]{"intColumn", "bytesColumn"});
+ assertEquals(dataSchema.getColumnDataTypes(), new ColumnDataType[]{ColumnDataType.INT, ColumnDataType.BYTES});
+
+ // Check values, where only 5 top values sorted in ByteArray format ascending order should be returned
+ List<Serializable[]> pqlRows = selectionResults.getRows();
+ assertEquals(pqlRows.size(), 5);
+ List<Object[]> sqlRows = resultTable.getRows();
+ assertEquals(sqlRows.size(), 5);
+ // ByteArray of "30", "31", "3130", "31303030", "31303031" (same as String order because all digits can be
+ // encoded with a single byte)
+ int[] expectedValues = new int[]{0, 1, 10, 1000, 1001};
+ for (int i = 0; i < 5; i++) {
+ Serializable[] row = pqlRows.get(i);
+ int intValue = (int) row[0];
+ assertEquals(intValue, expectedValues[i]);
+ assertEquals(Integer.parseInt(StringUtil.decodeUtf8(BytesUtils.toBytes((String) row[1]))), intValue);
+ }
+ for (int i = 0; i < 5; i++) {
+ Object[] row = sqlRows.get(i);
+ int intValue = (int) row[0];
+ assertEquals(intValue, expectedValues[i]);
+ assertEquals(Integer.parseInt(StringUtil.decodeUtf8(BytesUtils.toBytes((String) row[1]))), intValue);
+ }
+ }
+ {
// Test selecting some columns with transform, filter, order-by and limit
String pqlQuery =
"SELECT DISTINCT(ADD(intColumn, floatColumn), stringColumn) FROM testTable WHERE longColumn < 60 ORDER BY stringColumn DESC, ADD(intColumn, floatColumn) ASC LIMIT 10";
diff --git a/pinot-core/src/test/java/org/apache/pinot/queries/InnerSegmentSelectionMultiValueQueriesTest.java b/pinot-core/src/test/java/org/apache/pinot/queries/InnerSegmentSelectionMultiValueQueriesTest.java
index 790e954..1589cc3 100644
--- a/pinot-core/src/test/java/org/apache/pinot/queries/InnerSegmentSelectionMultiValueQueriesTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/queries/InnerSegmentSelectionMultiValueQueriesTest.java
@@ -18,11 +18,10 @@
*/
package org.apache.pinot.queries;
-import java.io.Serializable;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Queue;
+import java.util.PriorityQueue;
import org.apache.pinot.common.utils.DataSchema;
import org.apache.pinot.core.operator.BaseOperator;
import org.apache.pinot.core.operator.ExecutionStatistics;
@@ -103,9 +102,9 @@ public class InnerSegmentSelectionMultiValueQueriesTest extends BaseMultiValueQu
DataSchema.ColumnDataType.INT);
Assert.assertEquals(selectionDataSchema.getColumnDataType(columnIndexMap.get("column6")),
DataSchema.ColumnDataType.INT_ARRAY);
- List<Serializable[]> selectionResult = (List<Serializable[]>) resultsBlock.getSelectionResult();
+ List<Object[]> selectionResult = (List<Object[]>) resultsBlock.getSelectionResult();
Assert.assertEquals(selectionResult.size(), 10);
- Serializable[] firstRow = selectionResult.get(0);
+ Object[] firstRow = selectionResult.get(0);
Assert.assertEquals(firstRow.length, 10);
Assert.assertEquals(((Integer) firstRow[columnIndexMap.get("column1")]).intValue(), 890282370);
Assert.assertEquals(firstRow[columnIndexMap.get("column6")], new int[]{2147483647});
@@ -128,7 +127,7 @@ public class InnerSegmentSelectionMultiValueQueriesTest extends BaseMultiValueQu
DataSchema.ColumnDataType.INT);
Assert.assertEquals(selectionDataSchema.getColumnDataType(columnIndexMap.get("column6")),
DataSchema.ColumnDataType.INT_ARRAY);
- selectionResult = (List<Serializable[]>) resultsBlock.getSelectionResult();
+ selectionResult = (List<Object[]>) resultsBlock.getSelectionResult();
Assert.assertEquals(selectionResult.size(), 10);
firstRow = selectionResult.get(0);
Assert.assertEquals(firstRow.length, 10);
@@ -158,9 +157,9 @@ public class InnerSegmentSelectionMultiValueQueriesTest extends BaseMultiValueQu
DataSchema.ColumnDataType.INT);
Assert.assertEquals(selectionDataSchema.getColumnDataType(columnIndexMap.get("column6")),
DataSchema.ColumnDataType.INT_ARRAY);
- List<Serializable[]> selectionResult = (List<Serializable[]>) resultsBlock.getSelectionResult();
+ List<Object[]> selectionResult = (List<Object[]>) resultsBlock.getSelectionResult();
Assert.assertEquals(selectionResult.size(), 10);
- Serializable[] firstRow = selectionResult.get(0);
+ Object[] firstRow = selectionResult.get(0);
Assert.assertEquals(firstRow.length, 3);
Assert.assertEquals(((Integer) firstRow[columnIndexMap.get("column1")]).intValue(), 890282370);
Assert.assertEquals(firstRow[columnIndexMap.get("column6")], new int[]{2147483647});
@@ -183,7 +182,7 @@ public class InnerSegmentSelectionMultiValueQueriesTest extends BaseMultiValueQu
DataSchema.ColumnDataType.INT);
Assert.assertEquals(selectionDataSchema.getColumnDataType(columnIndexMap.get("column6")),
DataSchema.ColumnDataType.INT_ARRAY);
- selectionResult = (List<Serializable[]>) resultsBlock.getSelectionResult();
+ selectionResult = (List<Object[]>) resultsBlock.getSelectionResult();
Assert.assertEquals(selectionResult.size(), 10);
firstRow = selectionResult.get(0);
Assert.assertEquals(firstRow.length, 3);
@@ -213,9 +212,9 @@ public class InnerSegmentSelectionMultiValueQueriesTest extends BaseMultiValueQu
DataSchema.ColumnDataType.INT);
Assert.assertEquals(selectionDataSchema.getColumnDataType(columnIndexMap.get("column6")),
DataSchema.ColumnDataType.INT_ARRAY);
- Queue<Serializable[]> selectionResult = (Queue<Serializable[]>) resultsBlock.getSelectionResult();
+ PriorityQueue<Object[]> selectionResult = (PriorityQueue<Object[]>) resultsBlock.getSelectionResult();
Assert.assertEquals(selectionResult.size(), 10);
- Serializable[] lastRow = selectionResult.peek();
+ Object[] lastRow = selectionResult.peek();
Assert.assertEquals(lastRow.length, 4);
Assert.assertEquals((String) lastRow[columnIndexMap.get("column5")], "AKXcXcIqsqOJFsdwxZ");
Assert.assertEquals(lastRow[columnIndexMap.get("column6")], new int[]{1252});
@@ -238,7 +237,7 @@ public class InnerSegmentSelectionMultiValueQueriesTest extends BaseMultiValueQu
DataSchema.ColumnDataType.INT);
Assert.assertEquals(selectionDataSchema.getColumnDataType(columnIndexMap.get("column6")),
DataSchema.ColumnDataType.INT_ARRAY);
- selectionResult = (Queue<Serializable[]>) resultsBlock.getSelectionResult();
+ selectionResult = (PriorityQueue<Object[]>) resultsBlock.getSelectionResult();
Assert.assertEquals(selectionResult.size(), 10);
lastRow = selectionResult.peek();
Assert.assertEquals(lastRow.length, 4);
diff --git a/pinot-core/src/test/java/org/apache/pinot/queries/InnerSegmentSelectionSingleValueQueriesTest.java b/pinot-core/src/test/java/org/apache/pinot/queries/InnerSegmentSelectionSingleValueQueriesTest.java
index d06d269..7294527 100644
--- a/pinot-core/src/test/java/org/apache/pinot/queries/InnerSegmentSelectionSingleValueQueriesTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/queries/InnerSegmentSelectionSingleValueQueriesTest.java
@@ -18,11 +18,10 @@
*/
package org.apache.pinot.queries;
-import java.io.Serializable;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Queue;
+import java.util.PriorityQueue;
import org.apache.pinot.common.utils.DataSchema;
import org.apache.pinot.core.operator.BaseOperator;
import org.apache.pinot.core.operator.ExecutionStatistics;
@@ -103,9 +102,9 @@ public class InnerSegmentSelectionSingleValueQueriesTest extends BaseSingleValue
DataSchema.ColumnDataType.INT);
Assert.assertEquals(selectionDataSchema.getColumnDataType(columnIndexMap.get("column11")),
DataSchema.ColumnDataType.STRING);
- List<Serializable[]> selectionResult = (List<Serializable[]>) resultsBlock.getSelectionResult();
+ List<Object[]> selectionResult = (List<Object[]>) resultsBlock.getSelectionResult();
Assert.assertEquals(selectionResult.size(), 10);
- Serializable[] firstRow = selectionResult.get(0);
+ Object[] firstRow = selectionResult.get(0);
Assert.assertEquals(firstRow.length, 11);
Assert.assertEquals(((Integer) firstRow[columnIndexMap.get("column1")]).intValue(), 1578964907);
Assert.assertEquals((String) firstRow[columnIndexMap.get("column11")], "P");
@@ -128,7 +127,7 @@ public class InnerSegmentSelectionSingleValueQueriesTest extends BaseSingleValue
DataSchema.ColumnDataType.INT);
Assert.assertEquals(selectionDataSchema.getColumnDataType(columnIndexMap.get("column11")),
DataSchema.ColumnDataType.STRING);
- selectionResult = (List<Serializable[]>) resultsBlock.getSelectionResult();
+ selectionResult = (List<Object[]>) resultsBlock.getSelectionResult();
Assert.assertEquals(selectionResult.size(), 10);
firstRow = selectionResult.get(0);
Assert.assertEquals(firstRow.length, 11);
@@ -158,9 +157,9 @@ public class InnerSegmentSelectionSingleValueQueriesTest extends BaseSingleValue
DataSchema.ColumnDataType.INT);
Assert.assertEquals(selectionDataSchema.getColumnDataType(columnIndexMap.get("column11")),
DataSchema.ColumnDataType.STRING);
- List<Serializable[]> selectionResult = (List<Serializable[]>) resultsBlock.getSelectionResult();
+ List<Object[]> selectionResult = (List<Object[]>) resultsBlock.getSelectionResult();
Assert.assertEquals(selectionResult.size(), 10);
- Serializable[] firstRow = selectionResult.get(0);
+ Object[] firstRow = selectionResult.get(0);
Assert.assertEquals(firstRow.length, 3);
Assert.assertEquals(((Integer) firstRow[0]).intValue(), 1578964907);
Assert.assertEquals((String) firstRow[2], "P");
@@ -182,7 +181,7 @@ public class InnerSegmentSelectionSingleValueQueriesTest extends BaseSingleValue
DataSchema.ColumnDataType.INT);
Assert.assertEquals(selectionDataSchema.getColumnDataType(columnIndexMap.get("column11")),
DataSchema.ColumnDataType.STRING);
- selectionResult = (List<Serializable[]>) resultsBlock.getSelectionResult();
+ selectionResult = (List<Object[]>) resultsBlock.getSelectionResult();
Assert.assertEquals(selectionResult.size(), 10);
firstRow = selectionResult.get(0);
Assert.assertEquals(firstRow.length, 3);
@@ -212,9 +211,9 @@ public class InnerSegmentSelectionSingleValueQueriesTest extends BaseSingleValue
DataSchema.ColumnDataType.INT);
Assert.assertEquals(selectionDataSchema.getColumnDataType(columnIndexMap.get("column1")),
DataSchema.ColumnDataType.INT);
- Queue<Serializable[]> selectionResult = (Queue<Serializable[]>) resultsBlock.getSelectionResult();
+ PriorityQueue<Object[]> selectionResult = (PriorityQueue<Object[]>) resultsBlock.getSelectionResult();
Assert.assertEquals(selectionResult.size(), 10);
- Serializable[] lastRow = selectionResult.peek();
+ Object[] lastRow = selectionResult.peek();
Assert.assertEquals(lastRow.length, 4);
Assert.assertEquals(((Integer) lastRow[columnIndexMap.get("column6")]).intValue(), 6043515);
Assert.assertEquals(((Integer) lastRow[columnIndexMap.get("column1")]).intValue(), 10542595);
@@ -237,7 +236,7 @@ public class InnerSegmentSelectionSingleValueQueriesTest extends BaseSingleValue
DataSchema.ColumnDataType.INT);
Assert.assertEquals(selectionDataSchema.getColumnDataType(columnIndexMap.get("column1")),
DataSchema.ColumnDataType.INT);
- selectionResult = (Queue<Serializable[]>) resultsBlock.getSelectionResult();
+ selectionResult = (PriorityQueue<Object[]>) resultsBlock.getSelectionResult();
Assert.assertEquals(selectionResult.size(), 10);
lastRow = selectionResult.peek();
Assert.assertEquals(lastRow.length, 4);
@@ -268,9 +267,9 @@ public class InnerSegmentSelectionSingleValueQueriesTest extends BaseSingleValue
DataSchema.ColumnDataType.INT);
Assert.assertEquals(selectionDataSchema.getColumnDataType(columnIndexMap.get("column1")),
DataSchema.ColumnDataType.INT);
- Queue<Serializable[]> selectionResult = (Queue<Serializable[]>) resultsBlock.getSelectionResult();
+ PriorityQueue<Object[]> selectionResult = (PriorityQueue<Object[]>) resultsBlock.getSelectionResult();
Assert.assertEquals(selectionResult.size(), 10);
- Serializable[] lastRow = selectionResult.peek();
+ Object[] lastRow = selectionResult.peek();
Assert.assertEquals(lastRow.length, 11);
Assert.assertEquals(((Integer) lastRow[columnIndexMap.get("column6")]).intValue(), 6043515);
Assert.assertEquals(((Integer) lastRow[columnIndexMap.get("column1")]).intValue(), 10542595);
@@ -294,7 +293,7 @@ public class InnerSegmentSelectionSingleValueQueriesTest extends BaseSingleValue
DataSchema.ColumnDataType.INT);
Assert.assertEquals(selectionDataSchema.getColumnDataType(columnIndexMap.get("column1")),
DataSchema.ColumnDataType.INT);
- selectionResult = (Queue<Serializable[]>) resultsBlock.getSelectionResult();
+ selectionResult = (PriorityQueue<Object[]>) resultsBlock.getSelectionResult();
Assert.assertEquals(selectionResult.size(), 10);
lastRow = selectionResult.peek();
Assert.assertEquals(lastRow.length, 11);
diff --git a/pinot-core/src/test/java/org/apache/pinot/queries/RangePredicateWithSortedInvertedIndexTest.java b/pinot-core/src/test/java/org/apache/pinot/queries/RangePredicateWithSortedInvertedIndexTest.java
index 5c67ca9..cd534c1 100644
--- a/pinot-core/src/test/java/org/apache/pinot/queries/RangePredicateWithSortedInvertedIndexTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/queries/RangePredicateWithSortedInvertedIndexTest.java
@@ -19,7 +19,6 @@
package org.apache.pinot.queries;
import java.io.File;
-import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
@@ -37,7 +36,6 @@ import org.apache.pinot.core.indexsegment.immutable.ImmutableSegmentLoader;
import org.apache.pinot.core.operator.blocks.IntermediateResultsBlock;
import org.apache.pinot.core.operator.query.SelectionOnlyOperator;
import org.apache.pinot.core.segment.creator.impl.SegmentIndexCreationDriverImpl;
-import org.apache.pinot.pql.parsers.Pql2Compiler;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
@@ -209,14 +207,14 @@ public class RangePredicateWithSortedInvertedIndexTest extends BaseQueriesTest {
private void runQuery(String query, int count, List<Pairs.IntPair> intPairs, int numColumns) {
SelectionOnlyOperator operator = getOperatorForQuery(query);
IntermediateResultsBlock block = operator.nextBlock();
- Collection<Serializable[]> rows = block.getSelectionResult();
+ Collection<Object[]> rows = block.getSelectionResult();
Assert.assertNotNull(rows);
Assert.assertEquals(rows.size(), count);
if (count > 0) {
Pairs.IntPair pair = intPairs.get(0);
int startPos = pair.getLeft();
int pairPos = 0;
- for (Serializable[] row : rows) {
+ for (Object[] row : rows) {
Assert.assertEquals(numColumns, row.length);
Assert.assertEquals(row[0], stringValues[startPos]);
Assert.assertEquals(row[1], startPos);
diff --git a/pinot-core/src/test/java/org/apache/pinot/queries/TextSearchQueriesTest.java b/pinot-core/src/test/java/org/apache/pinot/queries/TextSearchQueriesTest.java
index 7ed16b7..72315e7 100644
--- a/pinot-core/src/test/java/org/apache/pinot/queries/TextSearchQueriesTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/queries/TextSearchQueriesTest.java
@@ -1067,7 +1067,7 @@ public class TextSearchQueriesTest extends BaseQueriesTest {
throws Exception {
SelectionOnlyOperator operator = getOperatorForQuery(query);
IntermediateResultsBlock operatorResult = operator.nextBlock();
- List<Serializable[]> resultset = (List<Serializable[]>) operatorResult.getSelectionResult();
+ List<Object[]> resultset = (List<Object[]>) operatorResult.getSelectionResult();
Assert.assertNotNull(resultset);
Assert.assertEquals(resultset.size(), expectedResultSize);
if (compareGrepOutput) {
@@ -1076,8 +1076,8 @@ public class TextSearchQueriesTest extends BaseQueriesTest {
} else if (expectedResults != null) {
// compare with expected result table
for (int i = 0; i < expectedResultSize; i++) {
- Serializable[] actualRow = resultset.get(i);
- Serializable[] expectedRow = expectedResults.get(i);
+ Object[] actualRow = resultset.get(i);
+ Object[] expectedRow = expectedResults.get(i);
Assert.assertEquals(actualRow.length, expectedRow.length);
for (int j = 0; j < actualRow.length; j++) {
Object actualColValue = actualRow[j];
@@ -1088,7 +1088,7 @@ public class TextSearchQueriesTest extends BaseQueriesTest {
}
}
- private void verifySearchOutputWithGrepResults(List<Serializable[]> actualResultSet)
+ private void verifySearchOutputWithGrepResults(List<Object[]> actualResultSet)
throws Exception {
URL resourceUrl = getClass().getClassLoader().getResource("data/text_search_data/group_by_grep_results.out");
File file = new File(resourceUrl.getFile());
@@ -1098,7 +1098,7 @@ public class TextSearchQueriesTest extends BaseQueriesTest {
int counter = 0;
while ((line = reader.readLine()) != null) {
String[] expectedRow = line.split(":");
- Serializable[] actualRow = actualResultSet.get(counter);
+ Object[] actualRow = actualResultSet.get(counter);
int expectedIntColValue = Integer.valueOf(expectedRow[0]) + INT_BASE_VALUE - 1;
Assert.assertEquals(expectedIntColValue, actualRow[0]);
Assert.assertEquals(expectedRow[1], actualRow[1]);
diff --git a/pinot-core/src/test/java/org/apache/pinot/query/selection/SelectionOperatorServiceTest.java b/pinot-core/src/test/java/org/apache/pinot/query/selection/SelectionOperatorServiceTest.java
index ccfe031..aa1fdcf 100644
--- a/pinot-core/src/test/java/org/apache/pinot/query/selection/SelectionOperatorServiceTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/query/selection/SelectionOperatorServiceTest.java
@@ -35,13 +35,13 @@ import org.apache.pinot.common.utils.DataTable;
import org.apache.pinot.core.indexsegment.IndexSegment;
import org.apache.pinot.core.query.selection.SelectionOperatorService;
import org.apache.pinot.core.query.selection.SelectionOperatorUtils;
+import org.apache.pinot.spi.utils.BytesUtils;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertNotSame;
import static org.testng.Assert.assertSame;
import static org.testng.Assert.assertTrue;
@@ -62,14 +62,18 @@ public class SelectionOperatorServiceTest {
private final DataSchema.ColumnDataType[] _upgradedColumnDataTypes =
new DataSchema.ColumnDataType[]{DataSchema.ColumnDataType.LONG, DataSchema.ColumnDataType.DOUBLE, DataSchema.ColumnDataType.DOUBLE, DataSchema.ColumnDataType.DOUBLE, DataSchema.ColumnDataType.STRING, DataSchema.ColumnDataType.LONG_ARRAY, DataSchema.ColumnDataType.DOUBLE_ARRAY, DataSchema.ColumnDataType.DOUBLE_ARRAY, DataSchema.ColumnDataType.DOUBLE_ARRAY, DataSchema.ColumnDataType.STRING_ARRAY, DataSchema.ColumnDataType.BYTES};
private final DataSchema _upgradedDataSchema = new DataSchema(_columnNames, _upgradedColumnDataTypes);
- private final Serializable[] _row1 =
- {0, 1L, 2.0F, 3.0, "4", new int[]{5}, new long[]{6L}, new float[]{7.0F}, new double[]{8.0}, new String[]{"9"}, new byte[]{0x10, 0x20}};
- private final Serializable[] _row2 =
- {10, 11L, 12.0F, 13.0, "14", new int[]{15}, new long[]{16L}, new float[]{17.0F}, new double[]{18.0}, new String[]{"19"}, new byte[]{0x30, 0x40}};
- private final Serializable[] _compatibleRow1 =
- {1L, 2.0F, 3.0, 4, "5", new long[]{6L}, new float[]{7.0F}, new double[]{8.0}, new int[]{9}, new String[]{"10"}, new byte[]{0x50, 0x60}};
- private final Serializable[] _compatibleRow2 =
- {11L, 12.0F, 13.0, 14, "15", new long[]{16L}, new float[]{17.0F}, new double[]{18.0}, new int[]{19}, new String[]{"20"}, new byte[]{0x70, 0x00}};
+ private final Object[] _row1 =
+ {0, 1L, 2.0F, 3.0, "4", new int[]{5}, new long[]{6L}, new float[]{7.0F}, new double[]{8.0}, new String[]{"9"}, BytesUtils.toByteArray(
+ "1020")};
+ private final Object[] _row2 =
+ {10, 11L, 12.0F, 13.0, "14", new int[]{15}, new long[]{16L}, new float[]{17.0F}, new double[]{18.0}, new String[]{"19"}, BytesUtils.toByteArray(
+ "3040")};
+ private final Object[] _compatibleRow1 =
+ {1L, 2.0F, 3.0, 4, "5", new long[]{6L}, new float[]{7.0F}, new double[]{8.0}, new int[]{9}, new String[]{"10"}, BytesUtils.toByteArray(
+ "5060")};
+ private final Object[] _compatibleRow2 =
+ {11L, 12.0F, 13.0, 14, "15", new long[]{16L}, new float[]{17.0F}, new double[]{18.0}, new int[]{19}, new String[]{"20"}, BytesUtils.toByteArray(
+ "7000")};
private final Selection _selectionOrderBy = new Selection();
@BeforeClass
@@ -152,10 +156,10 @@ public class SelectionOperatorServiceTest {
@Test
public void testCompatibleRowsMergeWithoutOrdering() {
- ArrayList<Serializable[]> mergedRows = new ArrayList<>(2);
+ ArrayList<Object[]> mergedRows = new ArrayList<>(2);
mergedRows.add(_row1);
mergedRows.add(_row2);
- Collection<Serializable[]> rowsToMerge = new ArrayList<>(2);
+ Collection<Object[]> rowsToMerge = new ArrayList<>(2);
rowsToMerge.add(_compatibleRow1);
rowsToMerge.add(_compatibleRow2);
SelectionOperatorUtils.mergeWithoutOrdering(mergedRows, rowsToMerge, 3);
@@ -168,13 +172,13 @@ public class SelectionOperatorServiceTest {
@Test
public void testCompatibleRowsMergeWithOrdering() {
SelectionOperatorService selectionOperatorService = new SelectionOperatorService(_selectionOrderBy, _dataSchema);
- PriorityQueue<Serializable[]> mergedRows = selectionOperatorService.getRows();
+ PriorityQueue<Object[]> mergedRows = selectionOperatorService.getRows();
int maxNumRows = _selectionOrderBy.getOffset() + _selectionOrderBy.getSize();
- Collection<Serializable[]> rowsToMerge1 = new ArrayList<>(2);
+ Collection<Object[]> rowsToMerge1 = new ArrayList<>(2);
rowsToMerge1.add(_row1);
rowsToMerge1.add(_row2);
SelectionOperatorUtils.mergeWithOrdering(mergedRows, rowsToMerge1, maxNumRows);
- Collection<Serializable[]> rowsToMerge2 = new ArrayList<>(2);
+ Collection<Object[]> rowsToMerge2 = new ArrayList<>(2);
rowsToMerge2.add(_compatibleRow1);
rowsToMerge2.add(_compatibleRow2);
SelectionOperatorUtils.mergeWithOrdering(mergedRows, rowsToMerge2, maxNumRows);
@@ -187,7 +191,7 @@ public class SelectionOperatorServiceTest {
@Test
public void testCompatibleRowsDataTableTransformation()
throws Exception {
- Collection<Serializable[]> rows = new ArrayList<>(2);
+ Collection<Object[]> rows = new ArrayList<>(2);
rows.add(_row1);
rows.add(_compatibleRow1);
DataSchema dataSchema = _dataSchema.clone();
@@ -195,35 +199,32 @@ public class SelectionOperatorServiceTest {
dataSchema.upgradeToCover(_compatibleDataSchema);
assertEquals(dataSchema, _upgradedDataSchema);
DataTable dataTable = SelectionOperatorUtils.getDataTableFromRows(rows, dataSchema);
- Serializable[] expectedRow1 =
- {0L, 1.0, 2.0, 3.0, "4", new long[]{5L}, new double[]{6.0}, new double[]{7.0}, new double[]{8.0}, new String[]{"9"}, "1020"};
- Serializable[] expectedCompatibleRow1 =
- {1L, 2.0, 3.0, 4.0, "5", new long[]{6L}, new double[]{7.0}, new double[]{8.0}, new double[]{9.0}, new String[]{"10"}, "5060"};
+ Object[] expectedRow1 =
+ {0L, 1.0, 2.0, 3.0, "4", new long[]{5L}, new double[]{6.0}, new double[]{7.0}, new double[]{8.0}, new String[]{"9"}, BytesUtils.toByteArray(
+ "1020")};
+ Object[] expectedCompatibleRow1 =
+ {1L, 2.0, 3.0, 4.0, "5", new long[]{6L}, new double[]{7.0}, new double[]{8.0}, new double[]{9.0}, new String[]{"10"}, BytesUtils.toByteArray(
+ "5060")};
assertTrue(Arrays.deepEquals(SelectionOperatorUtils.extractRowFromDataTable(dataTable, 0), expectedRow1));
assertTrue(Arrays.deepEquals(SelectionOperatorUtils.extractRowFromDataTable(dataTable, 1), expectedCompatibleRow1));
}
@Test
public void testCompatibleRowsRenderSelectionResultsWithoutOrdering() {
- // Replace byte[] with String because it is already converted to String at this stage
- Serializable[] row1 = _row1.clone();
- row1[10] = "1020";
- Serializable[] compatibleRow1 = _compatibleRow1.clone();
- compatibleRow1[10] = "5060";
-
- List<Serializable[]> rows = new ArrayList<>(2);
- rows.add(row1);
- rows.add(compatibleRow1);
+ List<Object[]> rows = new ArrayList<>(2);
+ rows.add(_row1);
+ rows.add(_compatibleRow1);
SelectionResults selectionResults = SelectionOperatorUtils
.renderSelectionResultsWithoutOrdering(rows, _upgradedDataSchema, Arrays.asList(_columnNames), true);
List<Serializable[]> resultRows = selectionResults.getRows();
- assertSame(resultRows.get(0), row1);
- assertSame(resultRows.get(1), compatibleRow1);
+ Serializable[] expectedRow1 =
+ {0L, 1.0, 2.0, 3.0, "4", new long[]{5L}, new double[]{6.0}, new double[]{7.0}, new double[]{8.0}, new String[]{"9"}, "1020"};
+ Serializable[] expectedRow2 =
+ {1L, 2.0, 3.0, 4.0, "5", new long[]{6L}, new double[]{7.0}, new double[]{8.0}, new double[]{9.0}, new String[]{"10"}, "5060"};
+ assertTrue(Arrays.deepEquals(resultRows.get(0), expectedRow1));
+ assertTrue(Arrays.deepEquals(resultRows.get(1), expectedRow2));
- rows = new ArrayList<>(2);
- rows.add(row1);
- rows.add(compatibleRow1);
- SelectionOperatorUtils
+ selectionResults = SelectionOperatorUtils
.renderSelectionResultsWithoutOrdering(rows, _upgradedDataSchema, Arrays.asList(_columnNames), false);
resultRows = selectionResults.getRows();
Serializable[] expectedFormattedRow1 =
@@ -236,32 +237,26 @@ public class SelectionOperatorServiceTest {
@Test
public void testCompatibleRowsRenderSelectionResultsWithOrdering() {
- // Replace byte[] with String because it is already converted to String at this stage
- Serializable[] row1 = _row1.clone();
- row1[10] = "1020";
- Serializable[] compatibleRow1 = _compatibleRow1.clone();
- compatibleRow1[10] = "5060";
- Serializable[] compatibleRow2 = _compatibleRow2.clone();
- compatibleRow2[10] = "7000";
-
SelectionOperatorService selectionOperatorService =
new SelectionOperatorService(_selectionOrderBy, _upgradedDataSchema);
- PriorityQueue<Serializable[]> rows = selectionOperatorService.getRows();
- rows.offer(row1);
- rows.offer(compatibleRow1);
- rows.offer(compatibleRow2);
+ PriorityQueue<Object[]> rows = selectionOperatorService.getRows();
+ rows.offer(_row1);
+ rows.offer(_compatibleRow1);
+ rows.offer(_compatibleRow2);
SelectionResults selectionResults = selectionOperatorService.renderSelectionResultsWithOrdering(true);
List<Serializable[]> resultRows = selectionResults.getRows();
- assertNotSame(resultRows.get(0), compatibleRow1);
- assertEquals(resultRows.get(0), compatibleRow1);
- assertNotSame(resultRows.get(1), row1);
- assertEquals(resultRows.get(1), row1);
+ Serializable[] expectedRow1 =
+ {1L, 2.0, 3.0, 4.0, "5", new long[]{6L}, new double[]{7.0}, new double[]{8.0}, new double[]{9.0}, new String[]{"10"}, "5060"};
+ Serializable[] expectedRow2 =
+ {0L, 1.0, 2.0, 3.0, "4", new long[]{5L}, new double[]{6.0}, new double[]{7.0}, new double[]{8.0}, new String[]{"9"}, "1020"};
+ assertTrue(Arrays.deepEquals(resultRows.get(0), expectedRow1));
+ assertTrue(Arrays.deepEquals(resultRows.get(1), expectedRow2));
selectionOperatorService = new SelectionOperatorService(_selectionOrderBy, _upgradedDataSchema);
rows = selectionOperatorService.getRows();
- rows.offer(row1);
- rows.offer(compatibleRow1);
- rows.offer(compatibleRow2);
+ rows.offer(_row1);
+ rows.offer(_compatibleRow1);
+ rows.offer(_compatibleRow2);
selectionResults = selectionOperatorService.renderSelectionResultsWithOrdering(false);
resultRows = selectionResults.getRows();
Serializable[] expectedFormattedRow1 =
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org