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/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new a2fe26b6d9 Revert "[multistage] Fix leaf stage return (#9892)" (#9916)
a2fe26b6d9 is described below
commit a2fe26b6d9d738be3482e2fcecc80b8608f37dbb
Author: Rong Rong <[email protected]>
AuthorDate: Mon Dec 5 18:58:34 2022 -0800
Revert "[multistage] Fix leaf stage return (#9892)" (#9916)
This reverts commit e6d3dcdbdaf9e08f0ffd939e55361bb946b03f93.
---
.../apache/pinot/query/runtime/QueryRunner.java | 46 +++-
.../LeafStageTransferableBlockOperator.java | 234 ++-------------------
.../LeafStageTransferableBlockOperatorTest.java | 78 ++-----
.../src/test/resources/queries/Aggregates.json | 93 +++-----
.../src/test/resources/queries/Distincts.json | 47 -----
.../test/resources/queries/LexicalStructure.json | 21 +-
.../test/resources/queries/SelectExpressions.json | 15 +-
7 files changed, 120 insertions(+), 414 deletions(-)
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/QueryRunner.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/QueryRunner.java
index 5cf1cfa999..7b69aa40ce 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/QueryRunner.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/QueryRunner.java
@@ -20,8 +20,10 @@ package org.apache.pinot.query.runtime;
import com.google.common.base.Preconditions;
import java.util.ArrayList;
+import java.util.LinkedList;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.concurrent.ExecutorService;
import org.apache.helix.HelixManager;
import org.apache.helix.store.zk.ZkHelixPropertyStore;
@@ -29,10 +31,13 @@ import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.pinot.common.exception.QueryException;
import org.apache.pinot.common.metadata.ZKMetadataProvider;
import org.apache.pinot.common.metrics.ServerMetrics;
+import org.apache.pinot.common.utils.DataSchema;
import org.apache.pinot.core.data.manager.InstanceDataManager;
import org.apache.pinot.core.operator.blocks.InstanceResponseBlock;
+import org.apache.pinot.core.operator.blocks.results.SelectionResultsBlock;
import org.apache.pinot.core.query.executor.ServerQueryExecutorV1Impl;
import org.apache.pinot.core.query.request.ServerQueryRequest;
+import org.apache.pinot.core.query.selection.SelectionOperatorUtils;
import org.apache.pinot.core.transport.ServerInstance;
import org.apache.pinot.query.mailbox.MailboxService;
import org.apache.pinot.query.mailbox.MultiplexingMailboxService;
@@ -184,10 +189,47 @@ public class QueryRunner {
private InstanceResponseBlock processServerQuery(ServerQueryRequest
serverQueryRequest,
ExecutorService executorService) {
try {
- return _serverExecutor.execute(serverQueryRequest, executorService);
+ InstanceResponseBlock result =
_serverExecutor.execute(serverQueryRequest, executorService);
+
+ if (result.getRows() != null &&
serverQueryRequest.getQueryContext().getOrderByExpressions() != null) {
+ // we only re-arrange columns to match the projection in the case of
order by - this is to ensure
+ // that V1 results match what the expected projection schema in the
calcite logical operator; if
+ // we realize that there are other situations where we need to
post-process v1 results to adhere to
+ // the expected results we should factor this out and also apply the
canonicalization of the data
+ // types during this post-process step (also see
LeafStageTransferableBlockOperator#canonicalizeRow)
+ DataSchema dataSchema = result.getDataSchema();
+ List<String> selectionColumns =
+
SelectionOperatorUtils.getSelectionColumns(serverQueryRequest.getQueryContext(),
dataSchema);
+
+ int[] columnIndices =
SelectionOperatorUtils.getColumnIndices(selectionColumns, dataSchema);
+ int numColumns = columnIndices.length;
+
+ DataSchema resultDataSchema =
SelectionOperatorUtils.getSchemaForProjection(dataSchema, columnIndices);
+
+ // Extract the result rows
+ LinkedList<Object[]> rowsInSelectionResults = new LinkedList<>();
+ for (Object[] row : result.getRows()) {
+ assert row != null;
+ Object[] extractedRow = new Object[numColumns];
+ for (int i = 0; i < numColumns; i++) {
+ Object value = row[columnIndices[i]];
+ if (value != null) {
+ extractedRow[i] = value;
+ }
+ }
+
+ rowsInSelectionResults.addFirst(extractedRow);
+ }
+
+ return new InstanceResponseBlock(
+ new SelectionResultsBlock(resultDataSchema,
rowsInSelectionResults),
+ serverQueryRequest.getQueryContext());
+ } else {
+ return result;
+ }
} catch (Exception e) {
InstanceResponseBlock errorResponse = new InstanceResponseBlock();
-
errorResponse.getExceptions().put(QueryException.QUERY_EXECUTION_ERROR_CODE,
e.getMessage());
+
errorResponse.getExceptions().put(QueryException.QUERY_EXECUTION_ERROR_CODE,
Objects.toString(e.getMessage()));
return errorResponse;
}
}
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/LeafStageTransferableBlockOperator.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/LeafStageTransferableBlockOperator.java
index 8def59dd32..551955c388 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/LeafStageTransferableBlockOperator.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/LeafStageTransferableBlockOperator.java
@@ -18,14 +18,11 @@
*/
package org.apache.pinot.query.runtime.operator;
-import com.clearspring.analytics.util.Preconditions;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.PriorityQueue;
-import java.util.stream.Collectors;
import javax.annotation.Nullable;
import org.apache.pinot.common.datablock.DataBlock;
import org.apache.pinot.common.datablock.DataBlockUtils;
@@ -35,13 +32,7 @@ import org.apache.pinot.common.utils.DataSchema;
import org.apache.pinot.core.common.Operator;
import org.apache.pinot.core.operator.BaseOperator;
import org.apache.pinot.core.operator.blocks.InstanceResponseBlock;
-import org.apache.pinot.core.operator.blocks.results.AggregationResultsBlock;
import org.apache.pinot.core.operator.blocks.results.BaseResultsBlock;
-import org.apache.pinot.core.operator.blocks.results.DistinctResultsBlock;
-import org.apache.pinot.core.operator.blocks.results.GroupByResultsBlock;
-import org.apache.pinot.core.operator.blocks.results.SelectionResultsBlock;
-import
org.apache.pinot.core.query.aggregation.function.DistinctAggregationFunction;
-import org.apache.pinot.core.query.selection.SelectionOperatorUtils;
import org.apache.pinot.query.runtime.blocks.TransferableBlock;
@@ -63,12 +54,10 @@ public class LeafStageTransferableBlockOperator extends
BaseOperator<Transferabl
private final InstanceResponseBlock _errorBlock;
private final List<InstanceResponseBlock> _baseResultBlock;
- private final DataSchema _desiredDataSchema;
private int _currentIndex;
public LeafStageTransferableBlockOperator(List<InstanceResponseBlock>
baseResultBlock, DataSchema dataSchema) {
_baseResultBlock = baseResultBlock;
- _desiredDataSchema = dataSchema;
_errorBlock = baseResultBlock.stream().filter(e ->
!e.getExceptions().isEmpty()).findFirst().orElse(null);
_currentIndex = 0;
}
@@ -95,10 +84,13 @@ public class LeafStageTransferableBlockOperator extends
BaseOperator<Transferabl
} else {
if (_currentIndex < _baseResultBlock.size()) {
InstanceResponseBlock responseBlock =
_baseResultBlock.get(_currentIndex++);
- if (responseBlock.getResultsBlock() != null) {
- return composeTransferableBlock(responseBlock, _desiredDataSchema);
+ BaseResultsBlock resultsBlock = responseBlock.getResultsBlock();
+ if (resultsBlock != null) {
+ List<Object[]> rows =
+ toList(resultsBlock.getRows(responseBlock.getQueryContext()),
responseBlock.getDataSchema());
+ return new TransferableBlock(rows, responseBlock.getDataSchema(),
DataBlock.Type.ROW);
} else {
- return new TransferableBlock(Collections.emptyList(),
_desiredDataSchema, DataBlock.Type.ROW);
+ return new TransferableBlock(Collections.emptyList(),
responseBlock.getDataSchema(), DataBlock.Type.ROW);
}
} else {
_currentIndex = -1;
@@ -107,181 +99,6 @@ public class LeafStageTransferableBlockOperator extends
BaseOperator<Transferabl
}
}
- /**
- * this is data transfer block compose method is here to ensure that V1
results match what the expected projection
- * schema in the calcite logical operator.
- * <p> it applies different clean up mechanism based on different type of
{@link BaseResultsBlock} and the
- * {@link org.apache.pinot.core.query.request.context.QueryContext}.</p>
- * <p> this also applies to the canonicalization of the data types during
post post-process step.</p>
- *
- * @param responseBlock result block from leaf stage
- * @param desiredDataSchema the desired schema for send operator
- * @return the converted {@link TransferableBlock} that conform with the
desiredDataSchema
- */
- private static TransferableBlock
composeTransferableBlock(InstanceResponseBlock responseBlock,
- DataSchema desiredDataSchema) {
- BaseResultsBlock resultsBlock = responseBlock.getResultsBlock();
- if (resultsBlock instanceof SelectionResultsBlock) {
- return composeSelectTransferableBlock(responseBlock, desiredDataSchema);
- } else if (resultsBlock instanceof AggregationResultsBlock) {
- return composeAggregationTransferableBlock(responseBlock,
desiredDataSchema);
- } else if (resultsBlock instanceof GroupByResultsBlock) {
- return composeGroupByTransferableBlock(responseBlock, desiredDataSchema);
- } else if (resultsBlock instanceof DistinctResultsBlock) {
- return composeDistinctTransferableBlock(responseBlock,
desiredDataSchema);
- } else {
- throw new IllegalArgumentException("Unsupported result block type: " +
resultsBlock);
- }
- }
-
- /**
- * we only need to rearrange columns when distinct is not conforming with
selection columns, specifically:
- * <ul>
- * <li> when distinct is not returning final result:
- * it should never happen as non-final result contains Object opaque
columns v2 engine can't process.</li>
- * <li> when distinct columns are not all being selected:
- * it should never happen as leaf stage MUST return the entire
list.</li>
- * </ul>
- *
- * @see
LeafStageTransferableBlockOperator#composeTransferableBlock(InstanceResponseBlock,
DataSchema).
- */
- @SuppressWarnings("ConstantConditions")
- private static TransferableBlock
composeDistinctTransferableBlock(InstanceResponseBlock responseBlock,
- DataSchema desiredDataSchema) {
- List<String> selectionColumns = Arrays.asList(
- ((DistinctAggregationFunction)
responseBlock.getQueryContext().getAggregationFunctions()[0]).getColumns());
- int[] columnIndices =
SelectionOperatorUtils.getColumnIndices(selectionColumns, desiredDataSchema);
- Preconditions.checkState(inOrder(columnIndices), "Incompatible distinct
table schema for leaf stage."
- + "Expected: " + desiredDataSchema + ". Actual: " + desiredDataSchema);
- return composeDirectTransferableBlock(responseBlock, desiredDataSchema);
- }
-
- /**
- * Calcite generated {@link DataSchema} should conform with Pinot's group by
result schema thus we only need to check
- * for correctness similar to distinct case.
- *
- * @see
LeafStageTransferableBlockOperator#composeDirectTransferableBlock(InstanceResponseBlock,
DataSchema).
- * @see
LeafStageTransferableBlockOperator#composeTransferableBlock(InstanceResponseBlock,
DataSchema).
- */
- @SuppressWarnings("ConstantConditions")
- private static TransferableBlock
composeGroupByTransferableBlock(InstanceResponseBlock responseBlock,
- DataSchema desiredDataSchema) {
- DataSchema resultSchema = responseBlock.getDataSchema();
- // GROUP-BY column names conforms with selection expression
- List<String> selectionColumns =
responseBlock.getQueryContext().getSelectExpressions().stream()
- .map(e -> e.toString()).collect(Collectors.toList());
- int[] columnIndices =
SelectionOperatorUtils.getColumnIndices(selectionColumns, resultSchema);
- Preconditions.checkState(inOrder(columnIndices), "Incompatible group by
result schema for leaf stage."
- + "Expected: " + desiredDataSchema + ". Actual: " + resultSchema);
- return composeDirectTransferableBlock(responseBlock, desiredDataSchema);
- }
-
-
- /**
- * Calcite generated {@link DataSchema} should conform with Pinot's agg
result schema thus we only need to check
- * for correctness similar to distinct case.
- *
- * @see
LeafStageTransferableBlockOperator#composeDirectTransferableBlock(InstanceResponseBlock,
DataSchema).
- * @see
LeafStageTransferableBlockOperator#composeTransferableBlock(InstanceResponseBlock,
DataSchema).
- */
- @SuppressWarnings("ConstantConditions")
- private static TransferableBlock
composeAggregationTransferableBlock(InstanceResponseBlock responseBlock,
- DataSchema desiredDataSchema) {
- DataSchema resultSchema = responseBlock.getDataSchema();
- // AGG-ONLY column names are derived from AggFunction.getColumnName()
- List<String> selectionColumns =
Arrays.stream(responseBlock.getQueryContext().getAggregationFunctions()).map(
- a -> a.getColumnName()).collect(Collectors.toList());
- int[] columnIndices =
SelectionOperatorUtils.getColumnIndices(selectionColumns, resultSchema);
- Preconditions.checkState(inOrder(columnIndices), "Incompatible aggregate
result schema for leaf stage."
- + "Expected: " + desiredDataSchema + ". Actual: " + resultSchema);
- return composeDirectTransferableBlock(responseBlock, desiredDataSchema);
- }
-
- /**
- * Only re-arrange columns to match the projection in the case of select /
order-by, when the desiredDataSchema
- * doesn't conform with the result block schema exactly.
- *
- * @see
LeafStageTransferableBlockOperator#composeTransferableBlock(InstanceResponseBlock,
DataSchema).
- */
- @SuppressWarnings("ConstantConditions")
- private static TransferableBlock
composeSelectTransferableBlock(InstanceResponseBlock responseBlock,
- DataSchema desiredDataSchema) {
- DataSchema resultSchema = responseBlock.getDataSchema();
- List<String> selectionColumns =
SelectionOperatorUtils.getSelectionColumns(responseBlock.getQueryContext(),
- resultSchema);
- int[] columnIndices =
SelectionOperatorUtils.getColumnIndices(selectionColumns, resultSchema);
- if (!inOrder(columnIndices)) {
- DataSchema adjustedResultSchema =
SelectionOperatorUtils.getSchemaForProjection(resultSchema, columnIndices);
-
Preconditions.checkState(isDataSchemaColumnTypesCompatible(desiredDataSchema.getColumnDataTypes(),
- adjustedResultSchema.getColumnDataTypes()), "Incompatible result
data schema: "
- + "Expecting: " + desiredDataSchema + " Actual: " +
adjustedResultSchema);
- // Extract the result rows
- Collection<Object[]> resultRows = responseBlock.getRows();
- List<Object[]> extractedRows = new ArrayList<>(resultRows.size());
- return composeColumnIndexedTransferableBlock(responseBlock,
adjustedResultSchema, columnIndices);
- } else {
- return composeDirectTransferableBlock(responseBlock, desiredDataSchema);
- }
- }
-
- /**
- * Created {@link TransferableBlock} using column indices.
- *
- * @see
LeafStageTransferableBlockOperator#composeTransferableBlock(InstanceResponseBlock,
DataSchema).
- */
- private static TransferableBlock
composeColumnIndexedTransferableBlock(InstanceResponseBlock responseBlock,
- DataSchema desiredDataSchema, int[] columnIndices) {
- Collection<Object[]> resultRows = responseBlock.getRows();
- List<Object[]> extractedRows = new ArrayList<>(resultRows.size());
- if (responseBlock.getQueryContext().getOrderByExpressions() != null) {
- // extract result row in ordered fashion
- PriorityQueue<Object[]> priorityQueue = (PriorityQueue<Object[]>)
resultRows;
- while (!priorityQueue.isEmpty()) {
- extractedRows.add(canonicalizeRow(priorityQueue.poll(),
desiredDataSchema, columnIndices));
- }
- } else {
- // extract result row in non-ordered fashion
- for (Object[] row : resultRows) {
- extractedRows.add(canonicalizeRow(row, desiredDataSchema,
columnIndices));
- }
- }
- return new TransferableBlock(extractedRows, desiredDataSchema,
DataBlock.Type.ROW);
- }
-
- /**
- * Fallback mechanism for {@link TransferableBlock}, used when no special
handling is necessary. This method only
- * performs {@link DataSchema.ColumnDataType} canonicalization.
- *
- * @see
LeafStageTransferableBlockOperator#composeTransferableBlock(InstanceResponseBlock,
DataSchema).
- */
- private static TransferableBlock
composeDirectTransferableBlock(InstanceResponseBlock responseBlock,
- DataSchema desiredDataSchema) {
- Collection<Object[]> resultRows = responseBlock.getRows();
- List<Object[]> extractedRows = new ArrayList<>(resultRows.size());
- if (resultRows instanceof List) {
- for (Object[] orgRow : resultRows) {
- extractedRows.add(canonicalizeRow(orgRow, desiredDataSchema));
- }
- } else if (resultRows instanceof PriorityQueue) {
- PriorityQueue<Object[]> priorityQueue = (PriorityQueue<Object[]>)
resultRows;
- while (!priorityQueue.isEmpty()) {
- extractedRows.add(canonicalizeRow(priorityQueue.poll(),
desiredDataSchema));
- }
- } else {
- throw new UnsupportedOperationException("Unsupported collection type: "
+ resultRows.getClass());
- }
- return new TransferableBlock(extractedRows, desiredDataSchema,
DataBlock.Type.ROW);
- }
-
- private static boolean inOrder(int[] columnIndices) {
- for (int i = 0; i < columnIndices.length; i++) {
- if (columnIndices[i] != i) {
- return false;
- }
- }
- return true;
- }
-
/**
* This util is used to canonicalize row generated from V1 engine, which is
stored using
* {@link DataSchema#getStoredColumnDataTypes()} format. However, the
transferable block ser/de stores data in the
@@ -294,35 +111,28 @@ public class LeafStageTransferableBlockOperator extends
BaseOperator<Transferabl
private static Object[] canonicalizeRow(Object[] row, DataSchema dataSchema)
{
Object[] resultRow = new Object[row.length];
for (int colId = 0; colId < row.length; colId++) {
- Object value = row[colId];
- if (value != null) {
- resultRow[colId] = dataSchema.getColumnDataType(colId).convert(value);
- }
+ resultRow[colId] =
dataSchema.getColumnDataType(colId).convert(row[colId]);
}
return resultRow;
}
- private static Object[] canonicalizeRow(Object[] row, DataSchema dataSchema,
int[] columnIndices) {
- Object[] resultRow = new Object[columnIndices.length];
- for (int colId = 0; colId < columnIndices.length; colId++) {
- Object value = row[columnIndices[colId]];
- if (value != null) {
- resultRow[colId] = dataSchema.getColumnDataType(colId).convert(value);
- }
+ private static List<Object[]> toList(Collection<Object[]> collection,
DataSchema dataSchema) {
+ if (collection == null || collection.isEmpty()) {
+ return new ArrayList<>();
}
- return resultRow;
- }
-
- private static boolean
isDataSchemaColumnTypesCompatible(DataSchema.ColumnDataType[] desiredTypes,
- DataSchema.ColumnDataType[] givenTypes) {
- if (desiredTypes.length != givenTypes.length) {
- return false;
- }
- for (int i = 0; i < desiredTypes.length; i++) {
- if (desiredTypes[i] != givenTypes[i] &&
!givenTypes[i].isSuperTypeOf(desiredTypes[i])) {
- return false;
+ List<Object[]> resultRows = new ArrayList<>(collection.size());
+ if (collection instanceof List) {
+ for (Object[] orgRow : collection) {
+ resultRows.add(canonicalizeRow(orgRow, dataSchema));
}
+ } else if (collection instanceof PriorityQueue) {
+ PriorityQueue<Object[]> priorityQueue = (PriorityQueue<Object[]>)
collection;
+ while (!priorityQueue.isEmpty()) {
+ resultRows.add(canonicalizeRow(priorityQueue.poll(), dataSchema));
+ }
+ } else {
+ throw new UnsupportedOperationException("Unsupported collection type: "
+ collection.getClass());
}
- return true;
+ return resultRows;
}
}
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/LeafStageTransferableBlockOperatorTest.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/LeafStageTransferableBlockOperatorTest.java
index 89df15c62b..496a09e4d5 100644
---
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/LeafStageTransferableBlockOperatorTest.java
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/LeafStageTransferableBlockOperatorTest.java
@@ -26,57 +26,28 @@ import org.apache.pinot.common.exception.QueryException;
import org.apache.pinot.common.utils.DataSchema;
import org.apache.pinot.core.operator.blocks.InstanceResponseBlock;
import org.apache.pinot.core.operator.blocks.results.SelectionResultsBlock;
-import org.apache.pinot.core.query.request.context.QueryContext;
-import
org.apache.pinot.core.query.request.context.utils.QueryContextConverterUtils;
import org.apache.pinot.query.runtime.blocks.TransferableBlock;
import org.testng.Assert;
import org.testng.annotations.Test;
-// TODO: add tests for Agg / GroupBy / Distinct result blocks
public class LeafStageTransferableBlockOperatorTest {
@Test
public void shouldReturnDataBlockThenMetadataBlock() {
// Given:
- QueryContext queryContext =
QueryContextConverterUtils.getQueryContext("SELECT strCol, intCol FROM tbl");
DataSchema schema = new DataSchema(new String[]{"strCol", "intCol"},
new DataSchema.ColumnDataType[]{DataSchema.ColumnDataType.STRING,
DataSchema.ColumnDataType.INT});
List<InstanceResponseBlock> resultsBlockList =
Collections.singletonList(new InstanceResponseBlock(
- new SelectionResultsBlock(schema, Arrays.asList(new Object[]{"foo",
1}, new Object[]{"", 2})), queryContext));
+ new SelectionResultsBlock(schema, Arrays.asList(new Object[]{"foo",
1}, new Object[]{"", 2})), null));
LeafStageTransferableBlockOperator operator = new
LeafStageTransferableBlockOperator(resultsBlockList, schema);
// When:
- TransferableBlock resultBlock = operator.nextBlock();
+ TransferableBlock transferableBlock = operator.nextBlock();
// Then:
- Assert.assertEquals(resultBlock.getContainer().get(0), new Object[]{"foo",
1});
- Assert.assertEquals(resultBlock.getContainer().get(1), new Object[]{"",
2});
- Assert.assertTrue(operator.nextBlock().isEndOfStreamBlock(), "Expected EOS
after reading two rows");
- }
-
- @Test
- public void shouldHandleDesiredDataSchemaConversionCorrectly() {
- // Given:
- QueryContext queryContext = QueryContextConverterUtils.getQueryContext(
- "SELECT boolCol, tsCol, boolCol AS newNamedBoolCol FROM tbl");
- DataSchema resultSchema = new DataSchema(new String[]{"boolCol", "tsCol"},
- new DataSchema.ColumnDataType[]{DataSchema.ColumnDataType.BOOLEAN,
DataSchema.ColumnDataType.TIMESTAMP});
- DataSchema desiredSchema = new DataSchema(new String[]{"boolCol", "tsCol",
"newNamedBoolCol"},
- new DataSchema.ColumnDataType[]{DataSchema.ColumnDataType.BOOLEAN,
DataSchema.ColumnDataType.TIMESTAMP,
- DataSchema.ColumnDataType.BOOLEAN});
- List<InstanceResponseBlock> resultsBlockList =
Collections.singletonList(new InstanceResponseBlock(
- new SelectionResultsBlock(resultSchema, Arrays.asList(new Object[]{1,
1660000000000L},
- new Object[]{0, 1600000000000L})), queryContext));
- LeafStageTransferableBlockOperator operator = new
LeafStageTransferableBlockOperator(resultsBlockList,
- desiredSchema);
-
- // When:
- TransferableBlock resultBlock = operator.nextBlock();
-
- // Then:
- Assert.assertEquals(resultBlock.getContainer().get(0), new Object[]{true,
new Timestamp(1660000000000L), true});
- Assert.assertEquals(resultBlock.getContainer().get(1), new Object[]{false,
new Timestamp(1600000000000L), false});
+ Assert.assertEquals(transferableBlock.getContainer().get(0), new
Object[]{"foo", 1});
+ Assert.assertEquals(transferableBlock.getContainer().get(1), new
Object[]{"", 2});
Assert.assertTrue(operator.nextBlock().isEndOfStreamBlock(), "Expected EOS
after reading two rows");
}
@@ -84,70 +55,67 @@ public class LeafStageTransferableBlockOperatorTest {
public void shouldHandleCanonicalizationCorrectly() {
// TODO: not all stored types are supported, add additional datatype when
they are supported.
// Given:
- QueryContext queryContext =
QueryContextConverterUtils.getQueryContext("SELECT boolCol, tsCol FROM tbl");
DataSchema schema = new DataSchema(new String[]{"boolCol", "tsCol"},
new DataSchema.ColumnDataType[]{DataSchema.ColumnDataType.BOOLEAN,
DataSchema.ColumnDataType.TIMESTAMP});
List<InstanceResponseBlock> resultsBlockList =
Collections.singletonList(new InstanceResponseBlock(
new SelectionResultsBlock(schema, Arrays.asList(new Object[]{1,
1660000000000L},
- new Object[]{0, 1600000000000L})), queryContext));
+ new Object[]{0, 1600000000000L})), null));
LeafStageTransferableBlockOperator operator = new
LeafStageTransferableBlockOperator(resultsBlockList, schema);
// When:
- TransferableBlock resultBlock = operator.nextBlock();
+ TransferableBlock transferableBlock = operator.nextBlock();
// Then:
- Assert.assertEquals(resultBlock.getContainer().get(0), new Object[]{true,
new Timestamp(1660000000000L)});
- Assert.assertEquals(resultBlock.getContainer().get(1), new Object[]{false,
new Timestamp(1600000000000L)});
+ Assert.assertEquals(transferableBlock.getContainer().get(0), new
Object[]{true, new Timestamp(1660000000000L)});
+ Assert.assertEquals(transferableBlock.getContainer().get(1), new
Object[]{false, new Timestamp(1600000000000L)});
Assert.assertTrue(operator.nextBlock().isEndOfStreamBlock(), "Expected EOS
after reading two rows");
}
@Test
public void shouldReturnMultipleDataBlockThenMetadataBlock() {
// Given:
- QueryContext queryContext =
QueryContextConverterUtils.getQueryContext("SELECT strCol, intCol FROM tbl");
DataSchema schema = new DataSchema(new String[]{"strCol", "intCol"},
new DataSchema.ColumnDataType[]{DataSchema.ColumnDataType.STRING,
DataSchema.ColumnDataType.INT});
List<InstanceResponseBlock> resultsBlockList = Arrays.asList(
new InstanceResponseBlock(new SelectionResultsBlock(schema,
- Arrays.asList(new Object[]{"foo", 1}, new Object[]{"", 2})),
queryContext),
+ Arrays.asList(new Object[]{"foo", 1}, new Object[]{"", 2})), null),
new InstanceResponseBlock(new SelectionResultsBlock(schema,
- Arrays.asList(new Object[]{"bar", 3}, new Object[]{"foo", 4})),
queryContext),
- new InstanceResponseBlock(new SelectionResultsBlock(schema,
Collections.emptyList()), queryContext));
+ Arrays.asList(new Object[]{"bar", 3}, new Object[]{"foo", 4})),
null),
+ new InstanceResponseBlock(new SelectionResultsBlock(schema,
Collections.emptyList()), null));
LeafStageTransferableBlockOperator operator = new
LeafStageTransferableBlockOperator(resultsBlockList, schema);
// When:
- TransferableBlock resultBlock1 = operator.nextBlock();
- TransferableBlock resultBlock2 = operator.nextBlock();
- TransferableBlock resultBlock3 = operator.nextBlock();
+ TransferableBlock transferableBlock1 = operator.nextBlock();
+ TransferableBlock transferableBlock2 = operator.nextBlock();
+ TransferableBlock transferableBlock3 = operator.nextBlock();
// Then:
- Assert.assertEquals(resultBlock1.getContainer().get(0), new
Object[]{"foo", 1});
- Assert.assertEquals(resultBlock1.getContainer().get(1), new Object[]{"",
2});
- Assert.assertEquals(resultBlock2.getContainer().get(0), new
Object[]{"bar", 3});
- Assert.assertEquals(resultBlock2.getContainer().get(1), new
Object[]{"foo", 4});
- Assert.assertEquals(resultBlock3.getContainer().size(), 0);
+ Assert.assertEquals(transferableBlock1.getContainer().get(0), new
Object[]{"foo", 1});
+ Assert.assertEquals(transferableBlock1.getContainer().get(1), new
Object[]{"", 2});
+ Assert.assertEquals(transferableBlock2.getContainer().get(0), new
Object[]{"bar", 3});
+ Assert.assertEquals(transferableBlock2.getContainer().get(1), new
Object[]{"foo", 4});
+ Assert.assertEquals(transferableBlock3.getContainer().size(), 0);
Assert.assertTrue(operator.nextBlock().isEndOfStreamBlock(), "Expected EOS
after reading two rows");
}
@Test
public void shouldGetErrorBlockWhenInstanceResponseContainsError() {
// Given:
- QueryContext queryContext =
QueryContextConverterUtils.getQueryContext("SELECT strCol, intCol FROM tbl");
DataSchema schema = new DataSchema(new String[]{"strCol", "intCol"},
new DataSchema.ColumnDataType[]{DataSchema.ColumnDataType.STRING,
DataSchema.ColumnDataType.INT});
InstanceResponseBlock errorBlock = new InstanceResponseBlock();
errorBlock.addException(QueryException.QUERY_EXECUTION_ERROR.getErrorCode(),
"foobar");
List<InstanceResponseBlock> resultsBlockList = Arrays.asList(
new InstanceResponseBlock(new SelectionResultsBlock(schema,
- Arrays.asList(new Object[]{"foo", 1}, new Object[]{"", 2})),
queryContext),
+ Arrays.asList(new Object[]{"foo", 1}, new Object[]{"", 2})), null),
errorBlock,
- new InstanceResponseBlock(new SelectionResultsBlock(schema,
Collections.emptyList()), queryContext));
+ new InstanceResponseBlock(new SelectionResultsBlock(schema,
Collections.emptyList()), null));
LeafStageTransferableBlockOperator operator = new
LeafStageTransferableBlockOperator(resultsBlockList, schema);
// When:
- TransferableBlock resultBlock = operator.nextBlock();
+ TransferableBlock transferableBlock = operator.nextBlock();
// Then:
- Assert.assertTrue(resultBlock.isErrorBlock());
+ Assert.assertTrue(transferableBlock.isErrorBlock());
}
}
diff --git a/pinot-query-runtime/src/test/resources/queries/Aggregates.json
b/pinot-query-runtime/src/test/resources/queries/Aggregates.json
index cba6bab972..b82be2cb5e 100644
--- a/pinot-query-runtime/src/test/resources/queries/Aggregates.json
+++ b/pinot-query-runtime/src/test/resources/queries/Aggregates.json
@@ -3,19 +3,19 @@
"tables": {
"tbl": {
"schema": [
- {"name": "int_col", "type": "INT"},
- {"name": "double_col", "type": "DOUBLE"},
- {"name": "string_col", "type": "STRING"},
- {"name": "bool_col", "type": "BOOLEAN"}
+ { "name": "int_col", "type": "INT" },
+ { "name": "double_col", "type": "DOUBLE" },
+ { "name": "string_col", "type": "STRING" },
+ { "name": "bool_col", "type": "BOOLEAN" }
],
"inputs": [
- [2, 300, "a", true],
- [2, 400, "a", true],
- [3, 100, "b", false],
- [100, 1, "b", false],
- [101, 1.01, "c", false],
- [150, 1.5, "c", false],
- [175, 1.75, "c", true]
+ [ 2, 300, "a", true ],
+ [ 2, 400, "a", true ],
+ [ 3, 100, "b", false ],
+ [ 100, 1, "b", false ],
+ [ 101, 1.01, "c", false ],
+ [ 150, 1.5, "c", false ],
+ [ 175, 1.75, "c", true ]
]
}
},
@@ -147,21 +147,21 @@
"tables": {
"tbl": {
"schema": [
- {"name": "int_col", "type": "INT"},
- {"name": "double_col", "type": "DOUBLE"},
- {"name": "string_col", "type": "STRING"},
- {"name": "bool_col", "type": "BOOLEAN"}
+ { "name": "int_col", "type": "INT" },
+ { "name": "double_col", "type": "DOUBLE" },
+ { "name": "string_col", "type": "STRING" },
+ { "name": "bool_col", "type": "BOOLEAN" }
],
"inputs": [
- [2, 300, "a", false],
- [2, 400, "a", true],
- [3, 100, "b", true],
- [0.001, 1, "b", false],
- [101, 1.01, "c", false],
- [150, 1.5, "c", true],
- [175, 1.75, "c", true],
- [-10000, 1.75, "c", false],
- [-2, 0.5, "c", false]
+ [ 2, 300, "a", false ],
+ [ 2, 400, "a", true ],
+ [ 3, 100, "b", true ],
+ [ 0.001, 1, "b", false ],
+ [ 101, 1.01, "c", false ],
+ [ 150, 1.5, "c", true ],
+ [ 175, 1.75, "c", true ],
+ [ -10000, 1.75, "c", false ],
+ [ -2, 0.5, "c", false ]
]
}
},
@@ -211,48 +211,5 @@
"sql": "select string_col, count(int_col) from {tbl} a group by
string_col order by string_col having (select count(*) from {tbl} b where
count(int_col) = b.int_col) > 0;"
}
]
- },
- "additional_aggregate": {
- "tables": {
- "tbl": {
- "schema": [
- {"name": "int_col", "type": "INT"},
- {"name": "double_col", "type": "DOUBLE"},
- {"name": "string_col", "type": "STRING"},
- {"name": "bool_col", "type": "BOOLEAN"}
- ],
- "inputs": [
- [2, 300, "a", false],
- [2, 400, "a", true],
- [3, 100, "b", true],
- [0.001, 1, "b", false],
- [101, 1.01, "c", false],
- [150, 1.5, "c", true],
- [175, 1.75, "c", true],
- [-10000, 1.75, "c", false],
- [-2, 0.5, "c", false]
- ]
- }
- },
- "queries": [
- { "sql": "SELECT min(int_col), min(int_col) FROM {tbl} WHERE int_col <
100" },
- { "sql": "SELECT min(int_col), count(int_col), count(*) FROM {tbl}" },
- { "sql": "SELECT string_col, min(int_col), count(int_col), count(*) FROM
{tbl} GROUP BY string_col" },
- { "sql": "SELECT string_col, bool_col, min(int_col), count(int_col),
count(*) FROM {tbl} GROUP BY string_col, bool_col" },
- { "sql": "SELECT string_col, bool_col, min(int_col), count(int_col),
count(*) FROM {tbl} GROUP BY bool_col, string_col" },
- { "sql": "SELECT string_col, string_col AS alias, count(int_col),
count(*) FROM {tbl} GROUP BY bool_col, string_col" },
- { "sql": "SELECT string_col, bool_col, bool_col AS alias, sum(int_col),
count(int_col), count(*) FROM {tbl} GROUP BY bool_col, string_col" },
- { "sql": "SELECT sum(int_col), count(int_col) FROM {tbl} GROUP BY
string_col" },
- { "sql": "SELECT string_col, min(int_col), count(int_col), count(*) FROM
{tbl} GROUP BY string_col ORDER BY string_col" },
- { "sql": "SELECT string_col, bool_col, min(int_col), count(int_col),
count(*) FROM {tbl} GROUP BY string_col, bool_col ORDER BY string_col" },
- { "sql": "SELECT string_col, bool_col, min(int_col), count(int_col),
count(*) FROM {tbl} GROUP BY string_col, bool_col ORDER BY bool_col,
string_col" },
- { "sql": "SELECT string_col, bool_col, min(int_col), count(int_col),
count(*) FROM {tbl} GROUP BY bool_col, string_col ORDER BY bool_col,
string_col" },
- { "sql": "SELECT string_col, string_col AS alias, count(int_col),
count(*) FROM {tbl} GROUP BY bool_col, string_col ORDER BY string_col" },
- { "sql": "SELECT string_col, string_col AS alias, count(int_col),
count(*) FROM {tbl} GROUP BY bool_col, string_col ORDER BY bool_col, alias"},
- { "sql": "SELECT string_col, bool_col, bool_col AS alias, sum(int_col),
count(int_col), count(*) FROM {tbl} GROUP BY bool_col, string_col ORDER BY
alias" },
- { "sql": "SELECT sum(int_col), count(int_col) FROM {tbl} GROUP BY
string_col ORDER BY string_col" },
- { "sql": "SELECT * FROM (SELECT string_col, bool_col, min(int_col) AS m,
count(int_col), count(*) AS c FROM {tbl} GROUP BY string_col, bool_col ORDER BY
string_col) WHERE c < m" },
- { "sql": "SELECT * FROM (SELECT string_col, bool_col, min(int_col) AS m,
count(int_col), count(*) AS c FROM {tbl} GROUP BY string_col, bool_col ORDER BY
bool_col, string_col) WHERE c < m" }
- ]
}
-}
\ No newline at end of file
+}
diff --git a/pinot-query-runtime/src/test/resources/queries/Distincts.json
b/pinot-query-runtime/src/test/resources/queries/Distincts.json
deleted file mode 100644
index 22ee05725c..0000000000
--- a/pinot-query-runtime/src/test/resources/queries/Distincts.json
+++ /dev/null
@@ -1,47 +0,0 @@
-{
- "general_distinct": {
- "tables": {
- "tbl": {
- "schema": [
- { "name": "intCol", "type": "INT" },
- { "name": "doubleCol", "type": "DOUBLE" },
- { "name": "strCol", "type": "STRING" },
- { "name": "boolCol", "type": "BOOLEAN" }
- ],
- "inputs": [
- [ 2, 300, "a", true ],
- [ 2, 400, "a", true ],
- [ 3, 100, "b", false ],
- [ 100, 1, "b", false ],
- [ 101, 1.01, "c", false ],
- [ 150, 1.5, "c", false ],
- [ 175, 1.75, "c", true ]
- ]
- }
- },
- "queries": [
- { "sql": "SELECT DISTINCT intCol FROM {tbl}" },
- { "sql": "SELECT intCol FROM {tbl} GROUP BY intCol" },
- { "sql": "SELECT DISTINCT strCol, boolCol FROM {tbl}" },
- { "sql": "SELECT strCol, boolCol FROM {tbl} GROUP BY strCol, boolCol" },
- { "sql": "SELECT strCol, boolCol FROM {tbl} GROUP BY boolCol, strCol" },
- { "sql": "SELECT strCol FROM {tbl} GROUP BY strCol, boolCol" },
- { "sql": "SELECT boolCol FROM {tbl} GROUP BY strCol, boolCol" },
- { "sql": "SELECT DISTINCT intCol, doubleCol FROM {tbl} ORDER BY
doubleCol" },
- { "sql": "SELECT intCol FROM {tbl} GROUP BY intCol ORDER BY intCol" },
- { "sql": "SELECT DISTINCT strCol, boolCol, intCol FROM {tbl} ORDER BY
intCol" },
- { "sql": "SELECT strCol, boolCol FROM {tbl} GROUP BY strCol, boolCol
ORDER BY boolCol" },
- { "sql": "SELECT strCol, boolCol FROM {tbl} GROUP BY boolCol, strCol
ORDER BY strCol" },
- { "sql": "SELECT strCol FROM {tbl} GROUP BY strCol, boolCol ORDER BY
boolCol" },
- { "sql": "SELECT strCol FROM (SELECT strCol FROM {tbl} GROUP BY
boolCol, strCol ORDER BY strCol) WHERE strCol <> 'foo'" },
- { "sql": "SELECT intCol FROM (SELECT DISTINCT intCol FROM {tbl}) WHERE
intCol < 100" },
- { "sql": "SELECT intCol FROM (SELECT intCol FROM {tbl} GROUP BY intCol)
WHERE intCol < 100" },
- { "sql": "SELECT strCol FROM (SELECT DISTINCT strCol, boolCol FROM
{tbl}) WHERE boolCol = false" },
- { "sql": "SELECT strCol FROM (SELECT strCol, boolCol FROM {tbl} GROUP
BY strCol, boolCol) WHERE strCol <> 'foo'" },
- { "sql": "SELECT strCol, boolCol FROM (SELECT strCol, boolCol FROM
{tbl} GROUP BY boolCol, strCol) WHERE strCol <> 'foo'" },
- { "sql": "SELECT strCol FROM (SELECT strCol FROM {tbl} GROUP BY strCol,
boolCol) WHERE strCol != 'lol'" },
- { "sql": "SELECT intCol FROM (SELECT DISTINCT intCol, doubleCol FROM
{tbl} ORDER BY doubleCol) WHERE doubleCol < 100" },
- { "sql": "SELECT strCol FROM (SELECT DISTINCT strCol, boolCol, intCol
FROM {tbl} ORDER BY intCol) WHERE intCol > 10" }
- ]
- }
-}
diff --git
a/pinot-query-runtime/src/test/resources/queries/LexicalStructure.json
b/pinot-query-runtime/src/test/resources/queries/LexicalStructure.json
index ea6e139ddc..400ed57fac 100644
--- a/pinot-query-runtime/src/test/resources/queries/LexicalStructure.json
+++ b/pinot-query-runtime/src/test/resources/queries/LexicalStructure.json
@@ -135,22 +135,11 @@
"inputs": [["1"]]
}
},
- "queries": [
- {
- "ignored": true,
- "comments": "numeric constant without specific type will cast all
floating points into BIG DECIMAL which is not supported by type-hoisting on
leaf stage",
- "psql": "4.1.2.6",
- "description": "numeric constant supports",
- "sql": "SELECT data, 42, 3.5, 4., .001, 5e2, 1.925e-3 FROM {tbl}"
- },
- {
- "ignored": true,
- "comments": "floating point comparison missing precision checker",
- "psql": "4.1.2.6",
- "description": "numeric constant supports",
- "sql": "SELECT data, 42, CAST(3.5 AS FLOAT), CAST(4. AS FLOAT),
CAST(.001 AS FLOAT), CAST(5e2 AS FLOAT), CAST(1.925e-3 AS FLOAT) FROM {tbl}"
- }
- ]
+ "queries": [{
+ "psql": "4.1.2.6",
+ "description": "numeric constant supports",
+ "sql": "SELECT data, 42, 3.5, 4., .001, 5e2, 1.925e-3 FROM {tbl}"
+ }]
},
"constant_casting": {
"tables": {
diff --git
a/pinot-query-runtime/src/test/resources/queries/SelectExpressions.json
b/pinot-query-runtime/src/test/resources/queries/SelectExpressions.json
index af925382ef..f214307b2a 100644
--- a/pinot-query-runtime/src/test/resources/queries/SelectExpressions.json
+++ b/pinot-query-runtime/src/test/resources/queries/SelectExpressions.json
@@ -37,20 +37,7 @@
{ "sql": "SELECT intCol as \"from\" FROM {tbl1}"},
{ "sql": "SELECT intCol as key, SUM(doubleCol + floatCol) AS aggSum FROM
{tbl1} GROUP BY intCol"},
{ "sql": "SELECT intCol, SUM(avgVal) FROM (SELECT strCol, intCol,
AVG(doubleCol) AS avgVal FROM {tbl1} GROUP BY intCol, strCol) GROUP BY intCol"},
- { "sql": "SELECT strCol, MAX(sumVal), MAX(sumVal + avgVal) AS transVal
FROM (SELECT strCol, intCol, SUM(floatCol + 2 * intCol) AS sumVal,
AVG(doubleCol) AS avgVal FROM {tbl1} GROUP BY strCol, intCol) GROUP BY strCol
ORDER BY MAX(sumVal)" },
- { "sql": "SELECT intCol, intCol FROM {tbl1} WHERE intCol < 100" },
- { "sql": "SELECT intCol, intCol, doubleCol, strCol, strCol FROM {tbl1}"
},
- { "sql": "SELECT intCol, intCol FROM {tbl1} WHERE intCol < 100 ORDER BY
doubleCol" },
- { "sql": "SELECT intCol, intCol FROM {tbl1} WHERE intCol < 100 ORDER BY
intCol" },
- { "sql": "SELECT intCol, intCol, doubleCol, strCol, strCol FROM {tbl1}
ORDER BY intCol" },
- { "sql": "SELECT intCol, intCol, doubleCol, strCol, strCol FROM {tbl1}
ORDER BY strCol, intCol" },
- { "sql": "SELECT intCol, intCol, doubleCol, strCol, strCol FROM {tbl1}
ORDER BY floatCol, intCol" },
- { "sql": "SELECT intCol, intCol, doubleCol, strCol, strCol FROM {tbl1}
ORDER BY floatCol" },
-
- { "sql": "SELECT intCol + alias FROM (SELECT intCol, intCol as alias
FROM {tbl1}) WHERE intCol < 100" },
- { "sql": "SELECT intCol, intCol, doubleCol, strCol, strCol FROM {tbl1}"
},
- { "sql": "SELECT {tbl1}.intCol, {tbl1}.intCol, {tbl1}.doubleCol,
{tbl2}.strCol, {tbl2}.strCol FROM {tbl1}, {tbl2} WHERE {tbl1}.intCol =
{tbl2}.intCol" },
- { "sql": "SELECT {tbl2}.intCol, {tbl2}.intCol FROM {tbl1}, {tbl2} WHERE
{tbl1}.intCol = {tbl2}.intCol AND {tbl1}.intCol < 100 ORDER BY
{tbl1}.doubleCol" }
+ { "sql": "SELECT strCol, MAX(sumVal), MAX(sumVal + avgVal) AS transVal
FROM (SELECT strCol, intCol, SUM(floatCol + 2 * intCol) AS sumVal,
AVG(doubleCol) AS avgVal FROM {tbl1} GROUP BY strCol, intCol) GROUP BY strCol
ORDER BY MAX(sumVal)" }
]
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
