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 8dc28e0 Allow Metadata and Dictionary Based Plans for No Op Filters (#7563) 8dc28e0 is described below commit 8dc28e010be4a9870db27da8765ec0b825c2cef9 Author: Atri Sharma <atri.j...@gmail.com> AuthorDate: Tue Oct 26 06:27:48 2021 +0530 Allow Metadata and Dictionary Based Plans for No Op Filters (#7563) When building aggregation plans, we ought to also consider the fact that some filters might be no-ops on certain segments. For those segments, we should consider using metadata or dictionary based plans. --- .../plan/AggregationGroupByOrderByPlanNode.java | 12 ++- .../core/plan/AggregationGroupByPlanNode.java | 12 ++- .../pinot/core/plan/AggregationPlanNode.java | 15 ++-- .../apache/pinot/core/plan/DocIdSetPlanNode.java | 11 ++- .../org/apache/pinot/core/plan/FilterPlanNode.java | 49 +++++++---- .../apache/pinot/core/plan/ProjectionPlanNode.java | 9 +- .../apache/pinot/core/plan/TransformPlanNode.java | 11 ++- .../apache/pinot/core/startree/StarTreeUtils.java | 21 +++-- ...adataAndDictionaryAggregationPlanMakerTest.java | 53 +++++++----- .../pinot/core/startree/v2/BaseStarTreeV2Test.java | 11 +-- .../pinot/queries/DistinctCountQueriesTest.java | 77 ++++++++++------- ...SegmentPartitionedDistinctCountQueriesTest.java | 97 ++++++++++++---------- 12 files changed, 238 insertions(+), 140 deletions(-) diff --git a/pinot-core/src/main/java/org/apache/pinot/core/plan/AggregationGroupByOrderByPlanNode.java b/pinot-core/src/main/java/org/apache/pinot/core/plan/AggregationGroupByOrderByPlanNode.java index 81d3857..58118f4 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/plan/AggregationGroupByOrderByPlanNode.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/plan/AggregationGroupByOrderByPlanNode.java @@ -22,6 +22,7 @@ import java.util.List; import java.util.Map; import java.util.Set; import org.apache.pinot.common.request.context.ExpressionContext; +import org.apache.pinot.core.operator.filter.BaseFilterOperator; import org.apache.pinot.core.operator.query.AggregationGroupByOrderByOperator; import org.apache.pinot.core.operator.transform.TransformOperator; import org.apache.pinot.core.query.aggregation.function.AggregationFunction; @@ -58,6 +59,9 @@ public class AggregationGroupByOrderByPlanNode implements PlanNode { AggregationFunction[] aggregationFunctions = _queryContext.getAggregationFunctions(); ExpressionContext[] groupByExpressions = _queryContext.getGroupByExpressions().toArray(new ExpressionContext[0]); + FilterPlanNode filterPlanNode = new FilterPlanNode(_indexSegment, _queryContext); + BaseFilterOperator filterOperator = filterPlanNode.run(); + // Use star-tree to solve the query if possible List<StarTreeV2> starTrees = _indexSegment.getStarTrees(); if (starTrees != null && !StarTreeUtils.isStarTreeDisabled(_queryContext)) { @@ -65,7 +69,8 @@ public class AggregationGroupByOrderByPlanNode implements PlanNode { StarTreeUtils.extractAggregationFunctionPairs(aggregationFunctions); if (aggregationFunctionColumnPairs != null) { Map<String, List<CompositePredicateEvaluator>> predicateEvaluatorsMap = - StarTreeUtils.extractPredicateEvaluatorsMap(_indexSegment, _queryContext.getFilter()); + StarTreeUtils.extractPredicateEvaluatorsMap(_indexSegment, _queryContext.getFilter(), + filterPlanNode.getPredicateEvaluatorMap()); if (predicateEvaluatorsMap != null) { for (StarTreeV2 starTreeV2 : starTrees) { if (StarTreeUtils.isFitForStarTree(starTreeV2.getMetadata(), aggregationFunctionColumnPairs, @@ -83,8 +88,9 @@ public class AggregationGroupByOrderByPlanNode implements PlanNode { Set<ExpressionContext> expressionsToTransform = AggregationFunctionUtils.collectExpressionsToTransform(aggregationFunctions, groupByExpressions); - TransformOperator transformPlanNode = new TransformPlanNode(_indexSegment, _queryContext, expressionsToTransform, - DocIdSetPlanNode.MAX_DOC_PER_CALL).run(); + TransformOperator transformPlanNode = + new TransformPlanNode(_indexSegment, _queryContext, expressionsToTransform, DocIdSetPlanNode.MAX_DOC_PER_CALL, + filterOperator).run(); return new AggregationGroupByOrderByOperator(aggregationFunctions, groupByExpressions, transformPlanNode, numTotalDocs, _queryContext, false); } diff --git a/pinot-core/src/main/java/org/apache/pinot/core/plan/AggregationGroupByPlanNode.java b/pinot-core/src/main/java/org/apache/pinot/core/plan/AggregationGroupByPlanNode.java index 33d83de..f9d0e09 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/plan/AggregationGroupByPlanNode.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/plan/AggregationGroupByPlanNode.java @@ -22,6 +22,7 @@ import java.util.List; import java.util.Map; import java.util.Set; import org.apache.pinot.common.request.context.ExpressionContext; +import org.apache.pinot.core.operator.filter.BaseFilterOperator; import org.apache.pinot.core.operator.query.AggregationGroupByOperator; import org.apache.pinot.core.operator.transform.TransformOperator; import org.apache.pinot.core.query.aggregation.function.AggregationFunction; @@ -58,6 +59,9 @@ public class AggregationGroupByPlanNode implements PlanNode { AggregationFunction[] aggregationFunctions = _queryContext.getAggregationFunctions(); ExpressionContext[] groupByExpressions = _queryContext.getGroupByExpressions().toArray(new ExpressionContext[0]); + FilterPlanNode filterPlanNode = new FilterPlanNode(_indexSegment, _queryContext); + BaseFilterOperator filterOperator = filterPlanNode.run(); + // Use star-tree to solve the query if possible List<StarTreeV2> starTrees = _indexSegment.getStarTrees(); if (starTrees != null && !StarTreeUtils.isStarTreeDisabled(_queryContext)) { @@ -65,7 +69,8 @@ public class AggregationGroupByPlanNode implements PlanNode { StarTreeUtils.extractAggregationFunctionPairs(aggregationFunctions); if (aggregationFunctionColumnPairs != null) { Map<String, List<CompositePredicateEvaluator>> predicateEvaluatorsMap = - StarTreeUtils.extractPredicateEvaluatorsMap(_indexSegment, _queryContext.getFilter()); + StarTreeUtils.extractPredicateEvaluatorsMap(_indexSegment, _queryContext.getFilter(), + filterPlanNode.getPredicateEvaluatorMap()); if (predicateEvaluatorsMap != null) { for (StarTreeV2 starTreeV2 : starTrees) { if (StarTreeUtils.isFitForStarTree(starTreeV2.getMetadata(), aggregationFunctionColumnPairs, @@ -83,8 +88,9 @@ public class AggregationGroupByPlanNode implements PlanNode { Set<ExpressionContext> expressionsToTransform = AggregationFunctionUtils.collectExpressionsToTransform(aggregationFunctions, groupByExpressions); - TransformOperator transformPlanNode = new TransformPlanNode(_indexSegment, _queryContext, expressionsToTransform, - DocIdSetPlanNode.MAX_DOC_PER_CALL).run(); + TransformOperator transformPlanNode = + new TransformPlanNode(_indexSegment, _queryContext, expressionsToTransform, DocIdSetPlanNode.MAX_DOC_PER_CALL, + filterOperator).run(); return new AggregationGroupByOperator(_queryContext, groupByExpressions, transformPlanNode, numTotalDocs, false); } } diff --git a/pinot-core/src/main/java/org/apache/pinot/core/plan/AggregationPlanNode.java b/pinot-core/src/main/java/org/apache/pinot/core/plan/AggregationPlanNode.java index c1cec0b..d7318d9 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/plan/AggregationPlanNode.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/plan/AggregationPlanNode.java @@ -26,6 +26,7 @@ import java.util.Set; import org.apache.pinot.common.request.context.ExpressionContext; import org.apache.pinot.core.common.Operator; import org.apache.pinot.core.operator.blocks.IntermediateResultsBlock; +import org.apache.pinot.core.operator.filter.BaseFilterOperator; import org.apache.pinot.core.operator.query.AggregationOperator; import org.apache.pinot.core.operator.query.DictionaryBasedAggregationOperator; import org.apache.pinot.core.operator.query.MetadataBasedAggregationOperator; @@ -64,10 +65,12 @@ public class AggregationPlanNode implements PlanNode { int numTotalDocs = _indexSegment.getSegmentMetadata().getTotalDocs(); AggregationFunction[] aggregationFunctions = _queryContext.getAggregationFunctions(); + FilterPlanNode filterPlanNode = new FilterPlanNode(_indexSegment, _queryContext); + BaseFilterOperator filterOperator = filterPlanNode.run(); + // Use metadata/dictionary to solve the query if possible - // NOTE: Skip the segment with valid doc index because the valid doc index is equivalent to a filter // TODO: Use the same operator for both of them so that COUNT(*), MAX(col) can be optimized - if (_queryContext.getFilter() == null && _indexSegment.getValidDocIds() == null) { + if (filterOperator.isResultMatchingAll()) { if (isFitForMetadataBasedPlan(aggregationFunctions)) { return new MetadataBasedAggregationOperator(aggregationFunctions, _indexSegment.getSegmentMetadata(), Collections.emptyMap()); @@ -88,7 +91,8 @@ public class AggregationPlanNode implements PlanNode { StarTreeUtils.extractAggregationFunctionPairs(aggregationFunctions); if (aggregationFunctionColumnPairs != null) { Map<String, List<CompositePredicateEvaluator>> predicateEvaluatorsMap = - StarTreeUtils.extractPredicateEvaluatorsMap(_indexSegment, _queryContext.getFilter()); + StarTreeUtils.extractPredicateEvaluatorsMap(_indexSegment, _queryContext.getFilter(), + filterPlanNode.getPredicateEvaluatorMap()); if (predicateEvaluatorsMap != null) { for (StarTreeV2 starTreeV2 : starTrees) { if (StarTreeUtils.isFitForStarTree(starTreeV2.getMetadata(), aggregationFunctionColumnPairs, null, @@ -105,8 +109,9 @@ public class AggregationPlanNode implements PlanNode { Set<ExpressionContext> expressionsToTransform = AggregationFunctionUtils.collectExpressionsToTransform(aggregationFunctions, null); - TransformOperator transformOperator = new TransformPlanNode(_indexSegment, _queryContext, expressionsToTransform, - DocIdSetPlanNode.MAX_DOC_PER_CALL).run(); + TransformOperator transformOperator = + new TransformPlanNode(_indexSegment, _queryContext, expressionsToTransform, DocIdSetPlanNode.MAX_DOC_PER_CALL, + filterOperator).run(); return new AggregationOperator(aggregationFunctions, transformOperator, numTotalDocs, false); } diff --git a/pinot-core/src/main/java/org/apache/pinot/core/plan/DocIdSetPlanNode.java b/pinot-core/src/main/java/org/apache/pinot/core/plan/DocIdSetPlanNode.java index 6745c5f..fd58e8f 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/plan/DocIdSetPlanNode.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/plan/DocIdSetPlanNode.java @@ -18,7 +18,9 @@ */ package org.apache.pinot.core.plan; +import javax.annotation.Nullable; import org.apache.pinot.core.operator.DocIdSetOperator; +import org.apache.pinot.core.operator.filter.BaseFilterOperator; import org.apache.pinot.core.query.request.context.QueryContext; import org.apache.pinot.segment.spi.IndexSegment; @@ -29,17 +31,22 @@ public class DocIdSetPlanNode implements PlanNode { private final IndexSegment _indexSegment; private final QueryContext _queryContext; private final int _maxDocPerCall; + private final BaseFilterOperator _filterOperator; - public DocIdSetPlanNode(IndexSegment indexSegment, QueryContext queryContext, int maxDocPerCall) { + public DocIdSetPlanNode(IndexSegment indexSegment, QueryContext queryContext, int maxDocPerCall, + @Nullable BaseFilterOperator filterOperator) { assert maxDocPerCall > 0 && maxDocPerCall <= MAX_DOC_PER_CALL; _indexSegment = indexSegment; _queryContext = queryContext; _maxDocPerCall = maxDocPerCall; + _filterOperator = filterOperator; } @Override public DocIdSetOperator run() { - return new DocIdSetOperator(new FilterPlanNode(_indexSegment, _queryContext).run(), _maxDocPerCall); + return new DocIdSetOperator( + _filterOperator != null ? _filterOperator : new FilterPlanNode(_indexSegment, _queryContext).run(), + _maxDocPerCall); } } diff --git a/pinot-core/src/main/java/org/apache/pinot/core/plan/FilterPlanNode.java b/pinot-core/src/main/java/org/apache/pinot/core/plan/FilterPlanNode.java index da818f1..276cc57 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/plan/FilterPlanNode.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/plan/FilterPlanNode.java @@ -21,9 +21,9 @@ package org.apache.pinot.core.plan; import com.google.common.base.Preconditions; import java.util.ArrayList; import java.util.Arrays; +import java.util.HashMap; import java.util.List; import java.util.Map; -import javax.annotation.Nullable; import org.apache.pinot.common.request.context.ExpressionContext; import org.apache.pinot.common.request.context.FilterContext; import org.apache.pinot.common.request.context.FunctionContext; @@ -60,6 +60,9 @@ public class FilterPlanNode implements PlanNode { private final QueryContext _queryContext; private final int _numDocs; + // Cache the predicate evaluators + private final Map<Predicate, PredicateEvaluator> _predicateEvaluatorMap = new HashMap<>(); + public FilterPlanNode(IndexSegment indexSegment, QueryContext queryContext) { _indexSegment = indexSegment; _queryContext = queryContext; @@ -74,7 +77,7 @@ public class FilterPlanNode implements PlanNode { ThreadSafeMutableRoaringBitmap validDocIds = _indexSegment.getValidDocIds(); boolean applyValidDocIds = validDocIds != null && !QueryOptionsUtils.isSkipUpsert(_queryContext.getQueryOptions()); if (filter != null) { - BaseFilterOperator filterOperator = constructPhysicalOperator(filter, _queryContext.getDebugOptions()); + BaseFilterOperator filterOperator = constructPhysicalOperator(filter); if (applyValidDocIds) { BaseFilterOperator validDocFilter = new BitmapBasedFilterOperator(validDocIds.getMutableRoaringBitmap(), false, _numDocs); @@ -91,6 +94,13 @@ public class FilterPlanNode implements PlanNode { } /** + * Returns a map from predicates to their evaluators. + */ + public Map<Predicate, PredicateEvaluator> getPredicateEvaluatorMap() { + return _predicateEvaluatorMap; + } + + /** * H3 index can be applied iff: * <ul> * <li>Predicate is of type RANGE</li> @@ -126,14 +136,13 @@ public class FilterPlanNode implements PlanNode { /** * Helper method to build the operator tree from the filter. */ - private BaseFilterOperator constructPhysicalOperator(FilterContext filter, - @Nullable Map<String, String> debugOptions) { + private BaseFilterOperator constructPhysicalOperator(FilterContext filter) { switch (filter.getType()) { case AND: List<FilterContext> childFilters = filter.getChildren(); List<BaseFilterOperator> childFilterOperators = new ArrayList<>(childFilters.size()); for (FilterContext childFilter : childFilters) { - BaseFilterOperator childFilterOperator = constructPhysicalOperator(childFilter, debugOptions); + BaseFilterOperator childFilterOperator = constructPhysicalOperator(childFilter); if (childFilterOperator.isResultEmpty()) { // Return empty filter operator if any of the child filter operator's result is empty return EmptyFilterOperator.getInstance(); @@ -142,12 +151,13 @@ public class FilterPlanNode implements PlanNode { childFilterOperators.add(childFilterOperator); } } - return FilterOperatorUtils.getAndFilterOperator(childFilterOperators, _numDocs, debugOptions); + return FilterOperatorUtils.getAndFilterOperator(childFilterOperators, _numDocs, + _queryContext.getDebugOptions()); case OR: childFilters = filter.getChildren(); childFilterOperators = new ArrayList<>(childFilters.size()); for (FilterContext childFilter : childFilters) { - BaseFilterOperator childFilterOperator = constructPhysicalOperator(childFilter, debugOptions); + BaseFilterOperator childFilterOperator = constructPhysicalOperator(childFilter); if (childFilterOperator.isResultMatchingAll()) { // Return match all filter operator if any of the child filter operator matches all records return new MatchAllFilterOperator(_numDocs); @@ -156,7 +166,7 @@ public class FilterPlanNode implements PlanNode { childFilterOperators.add(childFilterOperator); } } - return FilterOperatorUtils.getOrFilterOperator(childFilterOperators, _numDocs, debugOptions); + return FilterOperatorUtils.getOrFilterOperator(childFilterOperators, _numDocs, _queryContext.getDebugOptions()); case PREDICATE: Predicate predicate = filter.getPredicate(); ExpressionContext lhs = predicate.getLhs(); @@ -170,6 +180,10 @@ public class FilterPlanNode implements PlanNode { } else { String column = lhs.getIdentifier(); DataSource dataSource = _indexSegment.getDataSource(column); + PredicateEvaluator predicateEvaluator = _predicateEvaluatorMap.get(predicate); + if (predicateEvaluator != null) { + return FilterOperatorUtils.getLeafFilterOperator(predicateEvaluator, dataSource, _numDocs); + } switch (predicate.getType()) { case TEXT_MATCH: return new TextMatchFilterOperator(dataSource.getTextIndex(), ((TextMatchPredicate) predicate).getValue(), @@ -183,19 +197,21 @@ public class FilterPlanNode implements PlanNode { // // Consuming segments: When FST is enabled, use AutomatonBasedEvaluator so that regexp matching logic is // similar to that of FSTBasedEvaluator, else use regular flow of getting predicate evaluator. - PredicateEvaluator evaluator; if (dataSource.getFSTIndex() != null) { - evaluator = FSTBasedRegexpPredicateEvaluatorFactory.newFSTBasedEvaluator(dataSource.getFSTIndex(), - dataSource.getDictionary(), ((RegexpLikePredicate) predicate).getValue()); + predicateEvaluator = + FSTBasedRegexpPredicateEvaluatorFactory.newFSTBasedEvaluator(dataSource.getFSTIndex(), + dataSource.getDictionary(), ((RegexpLikePredicate) predicate).getValue()); } else if (dataSource instanceof MutableDataSource && ((MutableDataSource) dataSource).isFSTEnabled()) { - evaluator = + predicateEvaluator = FSTBasedRegexpPredicateEvaluatorFactory.newAutomatonBasedEvaluator(dataSource.getDictionary(), ((RegexpLikePredicate) predicate).getValue()); } else { - evaluator = PredicateEvaluatorProvider.getPredicateEvaluator(predicate, dataSource.getDictionary(), - dataSource.getDataSourceMetadata().getDataType()); + predicateEvaluator = + PredicateEvaluatorProvider.getPredicateEvaluator(predicate, dataSource.getDictionary(), + dataSource.getDataSourceMetadata().getDataType()); } - return FilterOperatorUtils.getLeafFilterOperator(evaluator, dataSource, _numDocs); + _predicateEvaluatorMap.put(predicate, predicateEvaluator); + return FilterOperatorUtils.getLeafFilterOperator(predicateEvaluator, dataSource, _numDocs); case JSON_MATCH: JsonIndexReader jsonIndex = dataSource.getJsonIndex(); Preconditions.checkState(jsonIndex != null, "Cannot apply JSON_MATCH on column: %s without json index", @@ -216,9 +232,10 @@ public class FilterPlanNode implements PlanNode { return new MatchAllFilterOperator(_numDocs); } default: - PredicateEvaluator predicateEvaluator = + predicateEvaluator = PredicateEvaluatorProvider.getPredicateEvaluator(predicate, dataSource.getDictionary(), dataSource.getDataSourceMetadata().getDataType()); + _predicateEvaluatorMap.put(predicate, predicateEvaluator); return FilterOperatorUtils.getLeafFilterOperator(predicateEvaluator, dataSource, _numDocs); } } diff --git a/pinot-core/src/main/java/org/apache/pinot/core/plan/ProjectionPlanNode.java b/pinot-core/src/main/java/org/apache/pinot/core/plan/ProjectionPlanNode.java index 27db4bf..5f286b4 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/plan/ProjectionPlanNode.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/plan/ProjectionPlanNode.java @@ -21,8 +21,10 @@ package org.apache.pinot.core.plan; import java.util.HashMap; import java.util.Map; import java.util.Set; +import javax.annotation.Nullable; import org.apache.pinot.core.operator.DocIdSetOperator; import org.apache.pinot.core.operator.ProjectionOperator; +import org.apache.pinot.core.operator.filter.BaseFilterOperator; import org.apache.pinot.core.query.request.context.QueryContext; import org.apache.pinot.segment.spi.IndexSegment; import org.apache.pinot.segment.spi.datasource.DataSource; @@ -37,13 +39,15 @@ public class ProjectionPlanNode implements PlanNode { private final QueryContext _queryContext; private final Set<String> _projectionColumns; private final int _maxDocsPerCall; + private final BaseFilterOperator _filterOperator; public ProjectionPlanNode(IndexSegment indexSegment, QueryContext queryContext, Set<String> projectionColumns, - int maxDocsPerCall) { + int maxDocsPerCall, @Nullable BaseFilterOperator filterOperator) { _indexSegment = indexSegment; _queryContext = queryContext; _projectionColumns = projectionColumns; _maxDocsPerCall = maxDocsPerCall; + _filterOperator = filterOperator; } @Override @@ -54,7 +58,8 @@ public class ProjectionPlanNode implements PlanNode { } // NOTE: Skip creating DocIdSetOperator when maxDocsPerCall is 0 (for selection query with LIMIT 0) DocIdSetOperator docIdSetOperator = - _maxDocsPerCall > 0 ? new DocIdSetPlanNode(_indexSegment, _queryContext, _maxDocsPerCall).run() : null; + _maxDocsPerCall > 0 ? new DocIdSetPlanNode(_indexSegment, _queryContext, _maxDocsPerCall, _filterOperator).run() + : null; return new ProjectionOperator(dataSourceMap, docIdSetOperator); } } diff --git a/pinot-core/src/main/java/org/apache/pinot/core/plan/TransformPlanNode.java b/pinot-core/src/main/java/org/apache/pinot/core/plan/TransformPlanNode.java index 6a60adb..f6f750c 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/plan/TransformPlanNode.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/plan/TransformPlanNode.java @@ -21,8 +21,10 @@ package org.apache.pinot.core.plan; import java.util.Collection; import java.util.HashSet; import java.util.Set; +import javax.annotation.Nullable; import org.apache.pinot.common.request.context.ExpressionContext; import org.apache.pinot.core.operator.ProjectionOperator; +import org.apache.pinot.core.operator.filter.BaseFilterOperator; import org.apache.pinot.core.operator.transform.PassThroughTransformOperator; import org.apache.pinot.core.operator.transform.TransformOperator; import org.apache.pinot.core.query.request.context.QueryContext; @@ -37,13 +39,20 @@ public class TransformPlanNode implements PlanNode { private final QueryContext _queryContext; private final Collection<ExpressionContext> _expressions; private final int _maxDocsPerCall; + private final BaseFilterOperator _filterOperator; public TransformPlanNode(IndexSegment indexSegment, QueryContext queryContext, Collection<ExpressionContext> expressions, int maxDocsPerCall) { + this(indexSegment, queryContext, expressions, maxDocsPerCall, null); + } + + public TransformPlanNode(IndexSegment indexSegment, QueryContext queryContext, + Collection<ExpressionContext> expressions, int maxDocsPerCall, @Nullable BaseFilterOperator filterOperator) { _indexSegment = indexSegment; _queryContext = queryContext; _expressions = expressions; _maxDocsPerCall = maxDocsPerCall; + _filterOperator = filterOperator; } @Override @@ -57,7 +66,7 @@ public class TransformPlanNode implements PlanNode { } } ProjectionOperator projectionOperator = - new ProjectionPlanNode(_indexSegment, _queryContext, projectionColumns, _maxDocsPerCall).run(); + new ProjectionPlanNode(_indexSegment, _queryContext, projectionColumns, _maxDocsPerCall, _filterOperator).run(); if (hasNonIdentifierExpression) { return new TransformOperator(projectionOperator, _expressions); } else { diff --git a/pinot-core/src/main/java/org/apache/pinot/core/startree/StarTreeUtils.java b/pinot-core/src/main/java/org/apache/pinot/core/startree/StarTreeUtils.java index 0dee1e4..049da07 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/startree/StarTreeUtils.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/startree/StarTreeUtils.java @@ -33,7 +33,6 @@ import org.apache.pinot.common.request.context.ExpressionContext; import org.apache.pinot.common.request.context.FilterContext; import org.apache.pinot.common.request.context.predicate.Predicate; import org.apache.pinot.core.operator.filter.predicate.PredicateEvaluator; -import org.apache.pinot.core.operator.filter.predicate.PredicateEvaluatorProvider; import org.apache.pinot.core.query.aggregation.function.AggregationFunction; import org.apache.pinot.core.query.aggregation.function.AggregationFunctionUtils; import org.apache.pinot.core.query.request.context.QueryContext; @@ -90,10 +89,12 @@ public class StarTreeUtils { * (d1 > 50 AND (d2 > 10 OR d2 < 35)). * This method represents a list of CompositePredicates per dimension. For each dimension, all CompositePredicates in * the list are implicitly ANDed together. Any OR predicates are nested within a CompositePredicate. + * + * A map from predicates to their evaluators is passed in to accelerate the computation. */ @Nullable public static Map<String, List<CompositePredicateEvaluator>> extractPredicateEvaluatorsMap(IndexSegment indexSegment, - @Nullable FilterContext filter) { + @Nullable FilterContext filter, Map<Predicate, PredicateEvaluator> predicateEvaluatorMap) { if (filter == null) { return Collections.emptyMap(); } @@ -108,7 +109,8 @@ public class StarTreeUtils { queue.addAll(filterNode.getChildren()); break; case OR: - Pair<String, List<PredicateEvaluator>> pair = isOrClauseValidForStarTree(indexSegment, filterNode); + Pair<String, List<PredicateEvaluator>> pair = + isOrClauseValidForStarTree(indexSegment, filterNode, predicateEvaluatorMap); if (pair == null) { return null; } @@ -121,8 +123,9 @@ public class StarTreeUtils { break; case PREDICATE: Predicate predicate = filterNode.getPredicate(); - PredicateEvaluator predicateEvaluator = getPredicateEvaluatorForPredicate(indexSegment, predicate); + PredicateEvaluator predicateEvaluator = getPredicateEvaluator(indexSegment, predicate, predicateEvaluatorMap); if (predicateEvaluator == null) { + // The predicate cannot be solved with star-tree return null; } if (!predicateEvaluator.isAlwaysTrue()) { @@ -181,7 +184,7 @@ public class StarTreeUtils { */ @Nullable private static Pair<String, List<PredicateEvaluator>> isOrClauseValidForStarTree(IndexSegment indexSegment, - FilterContext filter) { + FilterContext filter, Map<Predicate, PredicateEvaluator> predicateEvaluatorMap) { assert filter.getType() == FilterContext.Type.OR; List<Predicate> predicates = new ArrayList<>(); @@ -190,7 +193,7 @@ public class StarTreeUtils { String identifier = null; List<PredicateEvaluator> predicateEvaluators = new ArrayList<>(); for (Predicate predicate : predicates) { - PredicateEvaluator predicateEvaluator = getPredicateEvaluatorForPredicate(indexSegment, predicate); + PredicateEvaluator predicateEvaluator = getPredicateEvaluator(indexSegment, predicate, predicateEvaluatorMap); if (predicateEvaluator == null) { // The predicate cannot be solved with star-tree return null; @@ -246,7 +249,8 @@ public class StarTreeUtils { * star-tree. */ @Nullable - private static PredicateEvaluator getPredicateEvaluatorForPredicate(IndexSegment indexSegment, Predicate predicate) { + private static PredicateEvaluator getPredicateEvaluator(IndexSegment indexSegment, Predicate predicate, + Map<Predicate, PredicateEvaluator> predicateEvaluatorMap) { ExpressionContext lhs = predicate.getLhs(); if (lhs.getType() != ExpressionContext.Type.IDENTIFIER) { // Star-tree does not support non-identifier expression @@ -271,7 +275,6 @@ public class StarTreeUtils { default: break; } - return PredicateEvaluatorProvider - .getPredicateEvaluator(predicate, dictionary, dataSource.getDataSourceMetadata().getDataType()); + return predicateEvaluatorMap.get(predicate); } } diff --git a/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/MetadataAndDictionaryAggregationPlanMakerTest.java b/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/MetadataAndDictionaryAggregationPlanMakerTest.java index d9ad71a..2b5b401 100644 --- a/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/MetadataAndDictionaryAggregationPlanMakerTest.java +++ b/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/MetadataAndDictionaryAggregationPlanMakerTest.java @@ -151,54 +151,63 @@ public class MetadataAndDictionaryAggregationPlanMakerTest { @DataProvider(name = "testPlanMakerDataProvider") public Object[][] testPlanMakerDataProvider() { List<Object[]> entries = new ArrayList<>(); + // Selection entries.add(new Object[]{ - "select * from testTable", /*selection query*/ - SelectionOnlyOperator.class, SelectionOnlyOperator.class + "select * from testTable", SelectionOnlyOperator.class, SelectionOnlyOperator.class }); + // Selection entries.add(new Object[]{ - "select column1,column5 from testTable", /*selection query*/ - SelectionOnlyOperator.class, SelectionOnlyOperator.class + "select column1,column5 from testTable", SelectionOnlyOperator.class, SelectionOnlyOperator.class }); + // Selection with filter entries.add(new Object[]{ - "select * from testTable where daysSinceEpoch > 100", /*selection query with filters*/ - SelectionOnlyOperator.class, SelectionOnlyOperator.class + "select * from testTable where daysSinceEpoch > 100", SelectionOnlyOperator.class, SelectionOnlyOperator.class }); + // COUNT from metadata entries.add(new Object[]{ - "select count(*) from testTable", /*count(*) from metadata*/ - MetadataBasedAggregationOperator.class, AggregationOperator.class + "select count(*) from testTable", MetadataBasedAggregationOperator.class, AggregationOperator.class }); + // COUNT from metadata with match all filter entries.add(new Object[]{ - "select max(daysSinceEpoch),min(daysSinceEpoch) from testTable", /*min max from dictionary*/ - DictionaryBasedAggregationOperator.class, AggregationOperator.class + "select count(*) from testTable where column1 > 10", MetadataBasedAggregationOperator.class, + AggregationOperator.class }); + // MIN/MAX from dictionary entries.add(new Object[]{ - "select minmaxrange(daysSinceEpoch) from testTable", /*min max from dictionary*/ - DictionaryBasedAggregationOperator.class, AggregationOperator.class + "select max(daysSinceEpoch),min(daysSinceEpoch) from testTable", DictionaryBasedAggregationOperator.class, + AggregationOperator.class }); + // MIN/MAX from dictionary with match all filter entries.add(new Object[]{ - "select max(column17),min(column17) from testTable", /* minmax from dictionary*/ + "select max(daysSinceEpoch),min(daysSinceEpoch) from testTable where column1 > 10", DictionaryBasedAggregationOperator.class, AggregationOperator.class }); + // MINMAXRANGE from dictionary entries.add(new Object[]{ - "select minmaxrange(column17) from testTable", /*no minmax metadata, go to dictionary*/ + "select minmaxrange(daysSinceEpoch) from testTable", DictionaryBasedAggregationOperator.class, + AggregationOperator.class + }); + // MINMAXRANGE from dictionary with match all filter + entries.add(new Object[]{ + "select minmaxrange(daysSinceEpoch) from testTable where column1 > 10", DictionaryBasedAggregationOperator.class, AggregationOperator.class }); + // Aggregation entries.add(new Object[]{ - "select sum(column1) from testTable", /*aggregation query*/ - AggregationOperator.class, AggregationOperator.class + "select sum(column1) from testTable", AggregationOperator.class, AggregationOperator.class }); + // Aggregation group-by entries.add(new Object[]{ - "select sum(column1) from testTable group by daysSinceEpoch", /*aggregation with group by query*/ - AggregationGroupByOperator.class, AggregationGroupByOperator.class + "select sum(column1) from testTable group by daysSinceEpoch", AggregationGroupByOperator.class, + AggregationGroupByOperator.class }); + // COUNT from metadata, MIN from dictionary entries.add(new Object[]{ - "select count(*),min(column17) from testTable", - /*multiple aggregations query, one from metadata, one from dictionary*/ - AggregationOperator.class, AggregationOperator.class + "select count(*),min(column17) from testTable", AggregationOperator.class, AggregationOperator.class }); + // Aggregation group-by entries.add(new Object[]{ "select count(*),min(daysSinceEpoch) from testTable group by daysSinceEpoch", - /*multiple aggregations with group by*/ AggregationGroupByOperator.class, AggregationGroupByOperator.class }); diff --git a/pinot-core/src/test/java/org/apache/pinot/core/startree/v2/BaseStarTreeV2Test.java b/pinot-core/src/test/java/org/apache/pinot/core/startree/v2/BaseStarTreeV2Test.java index 648934e..8cb2e7a 100644 --- a/pinot-core/src/test/java/org/apache/pinot/core/startree/v2/BaseStarTreeV2Test.java +++ b/pinot-core/src/test/java/org/apache/pinot/core/startree/v2/BaseStarTreeV2Test.java @@ -31,7 +31,6 @@ import java.util.Random; import java.util.Set; import org.apache.commons.io.FileUtils; import org.apache.pinot.common.request.context.ExpressionContext; -import org.apache.pinot.common.request.context.FilterContext; import org.apache.pinot.core.common.BlockDocIdIterator; import org.apache.pinot.core.plan.FilterPlanNode; import org.apache.pinot.core.plan.PlanNode; @@ -224,9 +223,11 @@ abstract class BaseStarTreeV2Test<R, A> { List<String> groupByColumns = new ArrayList<>(groupByColumnSet); // Filter - FilterContext filter = queryContext.getFilter(); + FilterPlanNode filterPlanNode = new FilterPlanNode(_indexSegment, queryContext); + filterPlanNode.run(); Map<String, List<CompositePredicateEvaluator>> predicateEvaluatorsMap = - StarTreeUtils.extractPredicateEvaluatorsMap(_indexSegment, filter); + StarTreeUtils.extractPredicateEvaluatorsMap(_indexSegment, queryContext.getFilter(), + filterPlanNode.getPredicateEvaluatorMap()); assertNotNull(predicateEvaluatorsMap); // Extract values with star-tree @@ -234,8 +235,8 @@ abstract class BaseStarTreeV2Test<R, A> { new StarTreeFilterPlanNode(_starTreeV2, predicateEvaluatorsMap, groupByColumnSet, null); List<ForwardIndexReader> starTreeAggregationColumnReaders = new ArrayList<>(numAggregations); for (AggregationFunctionColumnPair aggregationFunctionColumnPair : aggregationFunctionColumnPairs) { - starTreeAggregationColumnReaders - .add(_starTreeV2.getDataSource(aggregationFunctionColumnPair.toColumnName()).getForwardIndex()); + starTreeAggregationColumnReaders.add( + _starTreeV2.getDataSource(aggregationFunctionColumnPair.toColumnName()).getForwardIndex()); } List<ForwardIndexReader> starTreeGroupByColumnReaders = new ArrayList<>(numGroupByColumns); for (String groupByColumn : groupByColumns) { diff --git a/pinot-core/src/test/java/org/apache/pinot/queries/DistinctCountQueriesTest.java b/pinot-core/src/test/java/org/apache/pinot/queries/DistinctCountQueriesTest.java index afa2181..3c045bc 100644 --- a/pinot-core/src/test/java/org/apache/pinot/queries/DistinctCountQueriesTest.java +++ b/pinot-core/src/test/java/org/apache/pinot/queries/DistinctCountQueriesTest.java @@ -96,7 +96,7 @@ public class DistinctCountQueriesTest extends BaseQueriesTest { @Override protected String getFilter() { - // NOTE: Use a match all filter to switch between DictionaryBasedAggregationOperator and AggregationOperator + // NOTE: This is a match all filter return " WHERE intColumn >= 0"; } @@ -185,31 +185,23 @@ public class DistinctCountQueriesTest extends BaseQueriesTest { @Test public void testAggregationOnly() { + // Dictionary based String query = "SELECT DISTINCTCOUNT(intColumn), DISTINCTCOUNT(longColumn), DISTINCTCOUNT(floatColumn), DISTINCTCOUNT" + "(doubleColumn), DISTINCTCOUNT(stringColumn), DISTINCTCOUNT(bytesColumn) FROM testTable"; // Inner segment - Operator operator = getOperatorForPqlQuery(query); - assertTrue(operator instanceof DictionaryBasedAggregationOperator); - IntermediateResultsBlock resultsBlock = ((DictionaryBasedAggregationOperator) operator).nextBlock(); - QueriesTestUtils - .testInnerSegmentExecutionStatistics(operator.getExecutionStatistics(), NUM_RECORDS, 0, 0, NUM_RECORDS); - List<Object> aggregationResult = resultsBlock.getAggregationResult(); - - operator = getOperatorForPqlQueryWithFilter(query); - assertTrue(operator instanceof AggregationOperator); - IntermediateResultsBlock resultsBlockWithFilter = ((AggregationOperator) operator).nextBlock(); - QueriesTestUtils - .testInnerSegmentExecutionStatistics(operator.getExecutionStatistics(), NUM_RECORDS, 0, 6 * NUM_RECORDS, - NUM_RECORDS); - List<Object> aggregationResultWithFilter = resultsBlockWithFilter.getAggregationResult(); - - assertNotNull(aggregationResult); - assertNotNull(aggregationResultWithFilter); - assertEquals(aggregationResult, aggregationResultWithFilter); - for (int i = 0; i < 6; i++) { - assertEquals(((Set) aggregationResult.get(i)).size(), _values.size()); + for (Object operator : Arrays.asList(getOperatorForSqlQuery(query), getOperatorForSqlQueryWithFilter(query))) { + assertTrue(operator instanceof DictionaryBasedAggregationOperator); + IntermediateResultsBlock resultsBlock = ((DictionaryBasedAggregationOperator) operator).nextBlock(); + QueriesTestUtils.testInnerSegmentExecutionStatistics(((Operator) operator).getExecutionStatistics(), NUM_RECORDS, + 0, 0, NUM_RECORDS); + List<Object> aggregationResult = resultsBlock.getAggregationResult(); + assertNotNull(aggregationResult); + assertEquals(aggregationResult.size(), 6); + for (int i = 0; i < 6; i++) { + assertEquals(((Set) aggregationResult.get(i)).size(), _values.size()); + } } // Inter segments @@ -217,13 +209,39 @@ public class DistinctCountQueriesTest extends BaseQueriesTest { for (int i = 0; i < 6; i++) { expectedResults[i] = Integer.toString(_values.size()); } + for (BrokerResponseNative brokerResponse : Arrays.asList(getBrokerResponseForPqlQuery(query), + getBrokerResponseForPqlQueryWithFilter(query))) { + QueriesTestUtils.testInterSegmentAggregationResult(brokerResponse, 4 * NUM_RECORDS, 0, 0, 4 * NUM_RECORDS, + expectedResults); + } + + // Regular aggregation + query = query + " WHERE intColumn >= 500"; + + // Inner segment + int expectedResult = 0; + for (Integer value : _values) { + if (value >= 500) { + expectedResult++; + } + } + Operator operator = getOperatorForSqlQuery(query); + assertTrue(operator instanceof AggregationOperator); + List<Object> aggregationResult = ((AggregationOperator) operator).nextBlock().getAggregationResult(); + assertNotNull(aggregationResult); + assertEquals(aggregationResult.size(), 6); + for (int i = 0; i < 6; i++) { + assertEquals(((Set) aggregationResult.get(i)).size(), expectedResult); + } + + // Inter segment BrokerResponseNative brokerResponse = getBrokerResponseForPqlQuery(query); - QueriesTestUtils - .testInterSegmentAggregationResult(brokerResponse, 4 * NUM_RECORDS, 0, 0, 4 * NUM_RECORDS, expectedResults); - brokerResponse = getBrokerResponseForPqlQueryWithFilter(query); - QueriesTestUtils - .testInterSegmentAggregationResult(brokerResponse, 4 * NUM_RECORDS, 0, 4 * 6 * NUM_RECORDS, 4 * NUM_RECORDS, - expectedResults); + List<AggregationResult> aggregationResults = brokerResponse.getAggregationResults(); + assertNotNull(aggregationResults); + assertEquals(aggregationResults.size(), 6); + for (int i = 0; i < 6; i++) { + assertEquals(aggregationResults.get(i).getValue(), Integer.toString(expectedResult)); + } } @Test @@ -237,9 +255,8 @@ public class DistinctCountQueriesTest extends BaseQueriesTest { Operator operator = getOperatorForPqlQuery(query); assertTrue(operator instanceof AggregationGroupByOperator); IntermediateResultsBlock resultsBlock = ((AggregationGroupByOperator) operator).nextBlock(); - QueriesTestUtils - .testInnerSegmentExecutionStatistics(operator.getExecutionStatistics(), NUM_RECORDS, 0, 6 * NUM_RECORDS, - NUM_RECORDS); + QueriesTestUtils.testInnerSegmentExecutionStatistics(operator.getExecutionStatistics(), NUM_RECORDS, 0, + 6 * NUM_RECORDS, NUM_RECORDS); AggregationGroupByResult aggregationGroupByResult = resultsBlock.getAggregationGroupByResult(); assertNotNull(aggregationGroupByResult); int numGroups = 0; diff --git a/pinot-core/src/test/java/org/apache/pinot/queries/SegmentPartitionedDistinctCountQueriesTest.java b/pinot-core/src/test/java/org/apache/pinot/queries/SegmentPartitionedDistinctCountQueriesTest.java index 3bc4148..0520086 100644 --- a/pinot-core/src/test/java/org/apache/pinot/queries/SegmentPartitionedDistinctCountQueriesTest.java +++ b/pinot-core/src/test/java/org/apache/pinot/queries/SegmentPartitionedDistinctCountQueriesTest.java @@ -92,13 +92,12 @@ public class SegmentPartitionedDistinctCountQueriesTest extends BaseQueriesTest new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).build(); private Set<Integer> _values; - private long _expectedResult; private IndexSegment _indexSegment; private List<IndexSegment> _indexSegments; @Override protected String getFilter() { - // NOTE: Use a match all filter to switch between DictionaryBasedAggregationOperator and AggregationOperator + // NOTE: This is a match all filter return " WHERE intColumn >= 0"; } @@ -135,7 +134,6 @@ public class SegmentPartitionedDistinctCountQueriesTest extends BaseQueriesTest record.putValue(BYTES_COLUMN, bytesValue); records.add(record); } - _expectedResult = _values.size(); SegmentGeneratorConfig segmentGeneratorConfig = new SegmentGeneratorConfig(TABLE_CONFIG, SCHEMA); segmentGeneratorConfig.setTableName(RAW_TABLE_NAME); @@ -153,64 +151,79 @@ public class SegmentPartitionedDistinctCountQueriesTest extends BaseQueriesTest @Test public void testAggregationOnly() { - String query = - "SELECT SEGMENTPARTITIONEDDISTINCTCOUNT(intColumn), SEGMENTPARTITIONEDDISTINCTCOUNT(longColumn), " - + "SEGMENTPARTITIONEDDISTINCTCOUNT(floatColumn), SEGMENTPARTITIONEDDISTINCTCOUNT(doubleColumn), " - + "SEGMENTPARTITIONEDDISTINCTCOUNT(stringColumn), SEGMENTPARTITIONEDDISTINCTCOUNT(bytesColumn) FROM " - + "testTable"; + // Dictionary based + String query = "SELECT SEGMENTPARTITIONEDDISTINCTCOUNT(intColumn), SEGMENTPARTITIONEDDISTINCTCOUNT(longColumn), " + + "SEGMENTPARTITIONEDDISTINCTCOUNT(floatColumn), SEGMENTPARTITIONEDDISTINCTCOUNT(doubleColumn), " + + "SEGMENTPARTITIONEDDISTINCTCOUNT(stringColumn), SEGMENTPARTITIONEDDISTINCTCOUNT(bytesColumn) FROM " + + "testTable"; // Inner segment - Operator operator = getOperatorForPqlQuery(query); - assertTrue(operator instanceof DictionaryBasedAggregationOperator); - IntermediateResultsBlock resultsBlock = ((DictionaryBasedAggregationOperator) operator).nextBlock(); - QueriesTestUtils - .testInnerSegmentExecutionStatistics(operator.getExecutionStatistics(), NUM_RECORDS, 0, 0, NUM_RECORDS); - List<Object> aggregationResult = resultsBlock.getAggregationResult(); + for (Object operator : Arrays.asList(getOperatorForSqlQuery(query), getOperatorForSqlQueryWithFilter(query))) { + assertTrue(operator instanceof DictionaryBasedAggregationOperator); + IntermediateResultsBlock resultsBlock = ((DictionaryBasedAggregationOperator) operator).nextBlock(); + QueriesTestUtils.testInnerSegmentExecutionStatistics(((Operator) operator).getExecutionStatistics(), NUM_RECORDS, + 0, 0, NUM_RECORDS); + List<Object> aggregationResult = resultsBlock.getAggregationResult(); + assertNotNull(aggregationResult); + assertEquals(aggregationResult.size(), 6); + for (int i = 0; i < 6; i++) { + assertEquals(((Long) aggregationResult.get(i)).intValue(), _values.size()); + } + } - operator = getOperatorForPqlQueryWithFilter(query); - assertTrue(operator instanceof AggregationOperator); - IntermediateResultsBlock resultsBlockWithFilter = ((AggregationOperator) operator).nextBlock(); - QueriesTestUtils - .testInnerSegmentExecutionStatistics(operator.getExecutionStatistics(), NUM_RECORDS, 0, 6 * NUM_RECORDS, - NUM_RECORDS); - List<Object> aggregationResultWithFilter = resultsBlockWithFilter.getAggregationResult(); + // Inter segments (expect 4 * inner segment result) + String[] expectedResults = new String[6]; + for (int i = 0; i < 6; i++) { + expectedResults[i] = Integer.toString(4 * _values.size()); + } + for (BrokerResponseNative brokerResponse : Arrays.asList(getBrokerResponseForPqlQuery(query), + getBrokerResponseForPqlQueryWithFilter(query))) { + QueriesTestUtils.testInterSegmentAggregationResult(brokerResponse, 4 * NUM_RECORDS, 0, 0, 4 * NUM_RECORDS, + expectedResults); + } + // Regular aggregation + query = query + " WHERE intColumn >= 500"; + + // Inner segment + int expectedResult = 0; + for (Integer value : _values) { + if (value >= 500) { + expectedResult++; + } + } + Operator operator = getOperatorForSqlQuery(query); + assertTrue(operator instanceof AggregationOperator); + List<Object> aggregationResult = ((AggregationOperator) operator).nextBlock().getAggregationResult(); assertNotNull(aggregationResult); - assertNotNull(aggregationResultWithFilter); - assertEquals(aggregationResult, aggregationResultWithFilter); + assertEquals(aggregationResult.size(), 6); for (int i = 0; i < 6; i++) { - assertEquals((long) aggregationResult.get(i), _expectedResult); + assertEquals(((Long) aggregationResult.get(i)).intValue(), expectedResult); } - // Inter segments (expect 4 * inner segment result) - String[] expectedResults = new String[6]; + // Inter segment (expect 4 * inner segment result) + BrokerResponseNative brokerResponse = getBrokerResponseForPqlQuery(query); + List<AggregationResult> aggregationResults = brokerResponse.getAggregationResults(); + assertNotNull(aggregationResults); + assertEquals(aggregationResults.size(), 6); for (int i = 0; i < 6; i++) { - expectedResults[i] = Long.toString(4 * _expectedResult); + assertEquals(aggregationResults.get(i).getValue(), Integer.toString(4 * expectedResult)); } - BrokerResponseNative brokerResponse = getBrokerResponseForPqlQuery(query); - QueriesTestUtils - .testInterSegmentAggregationResult(brokerResponse, 4 * NUM_RECORDS, 0, 0, 4 * NUM_RECORDS, expectedResults); - brokerResponse = getBrokerResponseForPqlQueryWithFilter(query); - QueriesTestUtils - .testInterSegmentAggregationResult(brokerResponse, 4 * NUM_RECORDS, 0, 4 * 6 * NUM_RECORDS, 4 * NUM_RECORDS, - expectedResults); } @Test public void testAggregationGroupBy() { - String query = - "SELECT SEGMENTPARTITIONEDDISTINCTCOUNT(intColumn), SEGMENTPARTITIONEDDISTINCTCOUNT(longColumn), " - + "SEGMENTPARTITIONEDDISTINCTCOUNT(floatColumn), SEGMENTPARTITIONEDDISTINCTCOUNT(doubleColumn), " - + "SEGMENTPARTITIONEDDISTINCTCOUNT(stringColumn), SEGMENTPARTITIONEDDISTINCTCOUNT(bytesColumn) FROM " - + "testTable GROUP BY intColumn"; + String query = "SELECT SEGMENTPARTITIONEDDISTINCTCOUNT(intColumn), SEGMENTPARTITIONEDDISTINCTCOUNT(longColumn), " + + "SEGMENTPARTITIONEDDISTINCTCOUNT(floatColumn), SEGMENTPARTITIONEDDISTINCTCOUNT(doubleColumn), " + + "SEGMENTPARTITIONEDDISTINCTCOUNT(stringColumn), SEGMENTPARTITIONEDDISTINCTCOUNT(bytesColumn) FROM " + + "testTable GROUP BY intColumn"; // Inner segment Operator operator = getOperatorForPqlQuery(query); assertTrue(operator instanceof AggregationGroupByOperator); IntermediateResultsBlock resultsBlock = ((AggregationGroupByOperator) operator).nextBlock(); - QueriesTestUtils - .testInnerSegmentExecutionStatistics(operator.getExecutionStatistics(), NUM_RECORDS, 0, 6 * NUM_RECORDS, - NUM_RECORDS); + QueriesTestUtils.testInnerSegmentExecutionStatistics(operator.getExecutionStatistics(), NUM_RECORDS, 0, + 6 * NUM_RECORDS, NUM_RECORDS); AggregationGroupByResult aggregationGroupByResult = resultsBlock.getAggregationGroupByResult(); assertNotNull(aggregationGroupByResult); int numGroups = 0; --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org For additional commands, e-mail: commits-h...@pinot.apache.org