You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by ro...@apache.org on 2022/12/06 02:56:39 UTC
[pinot] branch revert-9892-poc_fix_leaf_stage_return created (now baf2e2a0fb)
This is an automated email from the ASF dual-hosted git repository.
rongr pushed a change to branch revert-9892-poc_fix_leaf_stage_return
in repository https://gitbox.apache.org/repos/asf/pinot.git
at baf2e2a0fb Revert "[multistage] Fix leaf stage return (#9892)"
This branch includes the following new commits:
new baf2e2a0fb Revert "[multistage] Fix leaf stage return (#9892)"
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[pinot] 01/01: Revert "[multistage] Fix leaf stage return (#9892)"
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rongr pushed a commit to branch revert-9892-poc_fix_leaf_stage_return
in repository https://gitbox.apache.org/repos/asf/pinot.git
commit baf2e2a0fbf432b0e851c77fe894ce89edc67821
Author: Rong Rong <ro...@apache.org>
AuthorDate: Mon Dec 5 18:56:34 2022 -0800
Revert "[multistage] Fix leaf stage return (#9892)"
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: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org