| // Licensed to the Apache Software Foundation (ASF) under one |
| // or more contributor license agreements. See the NOTICE file |
| // distributed with this work for additional information |
| // regarding copyright ownership. The ASF licenses this file |
| // to you under the Apache License, Version 2.0 (the |
| // "License"); you may not use this file except in compliance |
| // with the License. You may obtain a copy of the License at |
| // |
| // http://www.apache.org/licenses/LICENSE-2.0 |
| // |
| // Unless required by applicable law or agreed to in writing, |
| // software distributed under the License is distributed on an |
| // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| // KIND, either express or implied. See the License for the |
| // specific language governing permissions and limitations |
| // under the License. |
| |
| package org.apache.impala.planner; |
| |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.List; |
| |
| import org.apache.impala.analysis.AggregateInfo; |
| import org.apache.impala.analysis.Analyzer; |
| import org.apache.impala.analysis.CaseExpr; |
| import org.apache.impala.analysis.CaseWhenClause; |
| import org.apache.impala.analysis.Expr; |
| import org.apache.impala.analysis.FunctionCallExpr; |
| import org.apache.impala.analysis.MultiAggregateInfo; |
| import org.apache.impala.analysis.MultiAggregateInfo.AggPhase; |
| import org.apache.impala.analysis.NumericLiteral; |
| import org.apache.impala.analysis.TupleId; |
| import org.apache.impala.analysis.ValidTupleIdExpr; |
| import org.apache.impala.common.InternalException; |
| import org.apache.impala.thrift.TAggregationNode; |
| import org.apache.impala.thrift.TAggregator; |
| import org.apache.impala.thrift.TExplainLevel; |
| import org.apache.impala.thrift.TExpr; |
| import org.apache.impala.thrift.TPlanNode; |
| import org.apache.impala.thrift.TPlanNodeType; |
| import org.apache.impala.thrift.TQueryOptions; |
| import org.apache.impala.util.BitUtil; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import com.google.common.base.Preconditions; |
| import com.google.common.collect.Lists; |
| |
| /** |
| * Aggregation computation. |
| * |
| */ |
| public class AggregationNode extends PlanNode { |
| private static final Logger LOG = LoggerFactory.getLogger(AggregationNode.class); |
| |
| // Default per-instance memory requirement used if no valid stats are available. |
| // TODO: Come up with a more useful heuristic. |
| private final static long DEFAULT_PER_INSTANCE_MEM = 128L * 1024L * 1024L; |
| |
| // Conservative minimum size of hash table for low-cardinality aggregations. |
| private final static long MIN_HASH_TBL_MEM = 10L * 1024L * 1024L; |
| |
| // Default skew factor to account for data skew among fragment instances. |
| private final static double DEFAULT_SKEW_FACTOR = 1.5; |
| |
| private final MultiAggregateInfo multiAggInfo_; |
| private final AggPhase aggPhase_; |
| |
| // Aggregation-class infos derived from 'multiAggInfo_' and 'aggPhase_' in c'tor. |
| private final List<AggregateInfo> aggInfos_; |
| |
| // If true, this node produces intermediate aggregation tuples. |
| private boolean useIntermediateTuple_ = false; |
| |
| // If true, this node performs the finalize step. |
| private boolean needsFinalize_ = false; |
| |
| // If true, this node uses streaming preaggregation. Invalid if this is a merge agg. |
| private boolean useStreamingPreagg_ = false; |
| |
| // Resource profiles for each aggregation class. |
| private List<ResourceProfile> resourceProfiles_; |
| |
| public AggregationNode( |
| PlanNodeId id, PlanNode input, MultiAggregateInfo multiAggInfo, AggPhase aggPhase) { |
| super(id, "AGGREGATE"); |
| children_.add(input); |
| multiAggInfo_ = multiAggInfo; |
| aggInfos_ = multiAggInfo_.getMaterializedAggInfos(aggPhase); |
| aggPhase_ = aggPhase; |
| needsFinalize_ = true; |
| computeTupleIds(); |
| } |
| |
| /** |
| * Copy c'tor used in clone(). |
| */ |
| private AggregationNode(PlanNodeId id, AggregationNode src) { |
| super(id, src, "AGGREGATE"); |
| multiAggInfo_ = src.multiAggInfo_; |
| aggPhase_ = src.aggPhase_; |
| aggInfos_ = src.aggInfos_; |
| needsFinalize_ = src.needsFinalize_; |
| useIntermediateTuple_ = src.useIntermediateTuple_; |
| } |
| |
| @Override |
| public void computeTupleIds() { |
| clearTupleIds(); |
| for (AggregateInfo aggInfo : aggInfos_) { |
| TupleId aggClassTupleId = null; |
| if (useIntermediateTuple_) { |
| aggClassTupleId = aggInfo.getIntermediateTupleId(); |
| } else { |
| aggClassTupleId = aggInfo.getOutputTupleId(); |
| } |
| tupleIds_.add(aggClassTupleId); |
| tblRefIds_.add(aggClassTupleId); |
| // Nullable tuples are only required to distinguish between multiple |
| // aggregation classes. |
| if (aggInfos_.size() > 1) { |
| nullableTupleIds_.add(aggClassTupleId); |
| } |
| } |
| } |
| |
| /** |
| * Sets this node as a preaggregation. |
| */ |
| public void setIsPreagg(PlannerContext ctx) { |
| if (ctx.getQueryOptions().disable_streaming_preaggregations) { |
| useStreamingPreagg_ = false; |
| return; |
| } |
| for (AggregateInfo aggInfo : aggInfos_) { |
| if (aggInfo.getGroupingExprs().size() > 0) { |
| useStreamingPreagg_ = true; |
| return; |
| } |
| } |
| } |
| |
| /** |
| * Unsets this node as requiring finalize. Only valid to call this if it is |
| * currently marked as needing finalize. |
| */ |
| public void unsetNeedsFinalize() { |
| Preconditions.checkState(needsFinalize_); |
| needsFinalize_ = false; |
| } |
| |
| public void setIntermediateTuple() { |
| useIntermediateTuple_ = true; |
| computeTupleIds(); |
| } |
| |
| public MultiAggregateInfo getMultiAggInfo() { return multiAggInfo_; } |
| public AggPhase getAggPhase() { return aggPhase_; } |
| public boolean hasGrouping() { return multiAggInfo_.hasGrouping(); } |
| public boolean isSingleClassAgg() { return aggInfos_.size() == 1; } |
| |
| public boolean isDistinctAgg() { |
| for (AggregateInfo aggInfo : aggInfos_) { |
| if (aggInfo.isDistinctAgg()) return true; |
| } |
| return false; |
| } |
| |
| @Override |
| public boolean isBlockingNode() { return !useStreamingPreagg_; } |
| |
| @Override |
| public void init(Analyzer analyzer) throws InternalException { |
| Preconditions.checkState(tupleIds_.size() == aggInfos_.size()); |
| // Assign conjuncts to the top-most agg in the single-node plan. They are transferred |
| // to the proper place in the distributed plan via transferConjuncts(). |
| if (aggPhase_ == multiAggInfo_.getConjunctAssignmentPhase()) { |
| conjuncts_.clear(); |
| // TODO: If this is the transposition phase, then we can push conjuncts that |
| // reference a single aggregation class down into the aggregators of the |
| // previous phase. |
| conjuncts_.addAll(multiAggInfo_.collectConjuncts(analyzer, true)); |
| conjuncts_ = orderConjunctsByCost(conjuncts_); |
| } |
| |
| // Compute the mem layout for both tuples here for simplicity. |
| for (AggregateInfo aggInfo : aggInfos_) { |
| aggInfo.getOutputTupleDesc().computeMemLayout(); |
| aggInfo.getIntermediateTupleDesc().computeMemLayout(); |
| } |
| |
| // Do at the end so it can take all conjuncts into account |
| computeStats(analyzer); |
| |
| // don't call createDefaultSMap(), it would point our conjuncts (= Having clause) |
| // to our input; our conjuncts don't get substituted because they already |
| // refer to our output |
| outputSmap_ = getCombinedChildSmap(); |
| |
| // Substitute exprs and check consistency. |
| // All of the AggregationNodes corresponding to a MultiAggregationInfo will have the |
| // same outputSmap_, so just substitute it once. |
| if (aggPhase_ == AggPhase.FIRST) multiAggInfo_.substitute(outputSmap_, analyzer); |
| for (AggregateInfo aggInfo : aggInfos_) { |
| aggInfo.substitute(outputSmap_, analyzer); |
| aggInfo.checkConsistency(); |
| } |
| } |
| |
| @Override |
| public void computeStats(Analyzer analyzer) { |
| super.computeStats(analyzer); |
| // TODO: IMPALA-2945: this doesn't correctly take into account duplicate keys on |
| // multiple nodes in a pre-aggregation. |
| cardinality_ = 0; |
| for (AggregateInfo aggInfo : aggInfos_) { |
| // Compute the cardinality for this set of grouping exprs. |
| long numGroups = estimateNumGroups(aggInfo); |
| Preconditions.checkState(numGroups >= -1, numGroups); |
| if (numGroups == -1) { |
| // No estimate of the number of groups is possible, can't even make a |
| // conservative estimate. |
| cardinality_ = -1; |
| break; |
| } |
| cardinality_ = checkedAdd(cardinality_, numGroups); |
| } |
| |
| // Take conjuncts into account. |
| if (cardinality_ > 0) { |
| cardinality_ = applyConjunctsSelectivity(cardinality_); |
| } |
| cardinality_ = capCardinalityAtLimit(cardinality_); |
| } |
| |
| /** |
| * Estimate the number of groups that will be present for the aggregation class |
| * described by 'aggInfo'. |
| * Returns -1 if a reasonable cardinality estimate cannot be produced. |
| */ |
| private long estimateNumGroups(AggregateInfo aggInfo) { |
| // This is prone to overestimation, because we keep multiplying cardinalities, |
| // even if the grouping exprs are functionally dependent (example: |
| // group by the primary key of a table plus a number of other columns from that |
| // same table). We limit the estimate to the estimated number of input row to |
| // limit the potential overestimation. We could, in future, improve this further |
| // by recognizing functional dependencies. |
| List<Expr> groupingExprs = aggInfo.getGroupingExprs(); |
| if (groupingExprs.isEmpty()) { |
| // Non-grouping aggregation class - always results in one group even if there are |
| // zero input rows. |
| return 1; |
| } |
| long numGroups = Expr.getNumDistinctValues(groupingExprs); |
| // Sanity check the cardinality_ based on the input cardinality_. |
| long aggInputCardinality = getAggInputCardinality(); |
| if (LOG.isTraceEnabled()) { |
| LOG.trace("Node " + id_ + " numGroups= " + numGroups + " aggInputCardinality=" + |
| aggInputCardinality + " for agg class " + aggInfo.debugString()); |
| } |
| if (numGroups == -1) { |
| // A worst-case cardinality_ is better than an unknown cardinality_. |
| // Note that this will still be -1 if the child's cardinality is unknown. |
| return aggInputCardinality; |
| } |
| // We have a valid estimate of the number of groups. Cap it at number of input |
| // rows because an aggregation cannot increase the cardinality_. |
| if (aggInputCardinality >= 0) { |
| numGroups = Math.min(aggInputCardinality, numGroups); |
| } |
| return numGroups; |
| } |
| |
| /** |
| * Compute the input cardinality to the distributed aggregation. If this is a |
| * merge aggregation, we need to find the cardinality of the input to the |
| * preaggregation. |
| * Return -1 if unknown. |
| */ |
| private long getAggInputCardinality() { |
| PlanNode child = getChild(0); |
| if (!aggPhase_.isMerge()) return child.getCardinality(); |
| PlanNode preAgg; |
| if (child instanceof ExchangeNode) { |
| preAgg = child.getChild(0); |
| } else { |
| preAgg = child; |
| } |
| Preconditions.checkState(preAgg instanceof AggregationNode); |
| return preAgg.getChild(0).getCardinality(); |
| } |
| |
| /** |
| * Returns a list of exprs suitable for hash partitioning the output of this node |
| * before the merge aggregation step. Only valid to call if this node is not a merge |
| * or transposing aggregation. The returned exprs are bound by the intermediate tuples. |
| * Takes the SHUFFLE_DISTINCT_EXPRS query option into account. |
| * |
| * For single-class aggregations the returned exprs are typically the grouping exprs. |
| * With SHUFFLE_DISTINCT_EXPRS=true the distinct exprs are also included if this is the |
| * non-merge first-phase aggregation of a distinct aggregation. |
| * |
| * For multi-class aggregations the returned exprs are a list of CASE exprs which you |
| * can think of as a "union" of the merge partition exprs of each class. Each CASE |
| * switches on the valid tuple id of an input row to determine the aggregation class, |
| * and selects the corresponding partition expr. |
| * The main challenges with crafting these exprs are: |
| * 1. Different aggregation classes can have a different number of distinct exprs |
| * Solution: The returned list is maximally wide to accommodate the widest |
| * aggregation class. For classes that have fewer than the max distinct exprs we add |
| * constant dummy exprs in the corresponding branch of the CASE. |
| * 2. A CASE expr must return a single output type, but different aggregation classes |
| * may have incompatible distinct exprs, so selecting the distinct exprs directly |
| * in the CASE branches would not always work (unless we cast everything to STRING, |
| * which we try to avoid). Instead, we call MURMUR_HASH() on the exprs to produce |
| * a hash value. That way, all branches of a CASE return the same type. |
| * Considering that the backend will do another round of hashing, there's an |
| * unnecessary double hashing here that we deemed acceptable for now and has |
| * potential for cleanup (maybe the FE should always add a MURMUR_HASH()). |
| * The handling of SHUFFLE_DISTINCT_EXPRS is analogous to the single-class case. |
| * |
| * Example: |
| * SELECT COUNT(DISTINCT a,b), COUNT(DISTINCT c) FROM t GROUP BY d |
| * Suppose the two aggregation classes have intermediate tuple ids 0 and 1. |
| * |
| * Challenges explained on this example: |
| * 1. First class has distinct exprs a,b and second class has c. We need to accommodate |
| * the widest class (a,b) and also hash on the grouping expr (d), so there will be |
| * three cases. |
| * 2. The types of a and c might be incompatible |
| * |
| * The first-phase partition exprs are a list of the following 3 exprs: |
| * CASE valid_tid() |
| * WHEN 0 THEN murmur_hash(d) <-- d SlotRef into tuple 0 |
| * WHEN 1 THEN murmur_hash(d) <-- d SlotRef into tuple 1 |
| * END, |
| * CASE valid_tid() |
| * WHEN 0 THEN murmur_hash(a) |
| * WHEN 1 THEN murmur_hash(c) |
| * END, |
| * CASE valid_tid() |
| * WHEN 0 THEN murmur_hash(b) |
| * WHEN 1 THEN 0 <-- dummy constant integer |
| * END |
| */ |
| public List<Expr> getMergePartitionExprs(Analyzer analyzer) { |
| Preconditions.checkState(!tupleIds_.isEmpty()); |
| Preconditions.checkState(!aggPhase_.isMerge() && !aggPhase_.isTranspose()); |
| |
| boolean shuffleDistinctExprs = analyzer.getQueryOptions().shuffle_distinct_exprs; |
| if (aggInfos_.size() == 1) { |
| AggregateInfo aggInfo = aggInfos_.get(0); |
| List<Expr> groupingExprs = null; |
| if (aggPhase_.isFirstPhase() && hasGrouping() && !shuffleDistinctExprs) { |
| groupingExprs = multiAggInfo_.getSubstGroupingExprs(); |
| } else { |
| groupingExprs = aggInfo.getPartitionExprs(); |
| if (groupingExprs == null) groupingExprs = aggInfo.getGroupingExprs(); |
| } |
| return Expr.substituteList( |
| groupingExprs, aggInfo.getIntermediateSmap(), analyzer, false); |
| } |
| |
| int maxNumExprs = 0; |
| for (AggregateInfo aggInfo : aggInfos_) { |
| if (aggInfo.getGroupingExprs() == null) continue; |
| maxNumExprs = Math.max(maxNumExprs, aggInfo.getGroupingExprs().size()); |
| } |
| if (maxNumExprs == 0) return Collections.emptyList(); |
| |
| List<Expr> result = new ArrayList<>(); |
| for (int i = 0; i < maxNumExprs; ++i) { |
| List<CaseWhenClause> caseWhenClauses = new ArrayList<>(); |
| for (AggregateInfo aggInfo : aggInfos_) { |
| TupleId tid; |
| if (aggInfo.isDistinctAgg()) { |
| tid = aggInfo.getOutputTupleId(); |
| } else { |
| tid = aggInfo.getIntermediateTupleId(); |
| } |
| List<Expr> groupingExprs = aggInfo.getGroupingExprs(); |
| if (aggPhase_.isFirstPhase() && hasGrouping() && !shuffleDistinctExprs) { |
| groupingExprs = multiAggInfo_.getSubstGroupingExprs(); |
| } |
| Expr whenExpr = NumericLiteral.create(tid.asInt()); |
| Expr thenExpr; |
| if (groupingExprs == null || i >= groupingExprs.size()) { |
| thenExpr = NumericLiteral.create(0); |
| } else { |
| thenExpr = new FunctionCallExpr( |
| "murmur_hash", Lists.newArrayList(groupingExprs.get(i).clone())); |
| thenExpr.analyzeNoThrow(analyzer); |
| thenExpr = thenExpr.substitute(aggInfo.getIntermediateSmap(), analyzer, true); |
| } |
| caseWhenClauses.add(new CaseWhenClause(whenExpr, thenExpr)); |
| } |
| CaseExpr caseExpr = |
| new CaseExpr(new ValidTupleIdExpr(tupleIds_), caseWhenClauses, null); |
| caseExpr.analyzeNoThrow(analyzer); |
| result.add(caseExpr); |
| } |
| return result; |
| } |
| |
| @Override |
| protected void toThrift(TPlanNode msg) { |
| msg.agg_node = new TAggregationNode(); |
| msg.node_type = TPlanNodeType.AGGREGATION_NODE; |
| boolean replicateInput = aggPhase_ == AggPhase.FIRST && aggInfos_.size() > 1; |
| msg.agg_node.setReplicate_input(replicateInput); |
| msg.agg_node.setEstimated_input_cardinality(getChild(0).getCardinality()); |
| for (int i = 0; i < aggInfos_.size(); ++i) { |
| AggregateInfo aggInfo = aggInfos_.get(i); |
| List<TExpr> aggregateFunctions = new ArrayList<>(); |
| for (FunctionCallExpr e : aggInfo.getMaterializedAggregateExprs()) { |
| aggregateFunctions.add(e.treeToThrift()); |
| } |
| TAggregator taggregator = new TAggregator(aggregateFunctions, |
| aggInfo.getIntermediateTupleId().asInt(), aggInfo.getOutputTupleId().asInt(), |
| needsFinalize_, useStreamingPreagg_, resourceProfiles_.get(i).toThrift()); |
| List<Expr> groupingExprs = aggInfo.getGroupingExprs(); |
| if (!groupingExprs.isEmpty()) { |
| taggregator.setGrouping_exprs(Expr.treesToThrift(groupingExprs)); |
| } |
| msg.agg_node.addToAggregators(taggregator); |
| } |
| } |
| |
| @Override |
| protected String getDisplayLabelDetail() { |
| if (useStreamingPreagg_) return "STREAMING"; |
| if (needsFinalize_) return "FINALIZE"; |
| return null; |
| } |
| |
| @Override |
| protected String getNodeExplainString(String prefix, String detailPrefix, |
| TExplainLevel detailLevel) { |
| StringBuilder output = new StringBuilder(); |
| String nameDetail = getDisplayLabelDetail(); |
| output.append(String.format("%s%s", prefix, getDisplayLabel())); |
| if (nameDetail != null) output.append(" [" + nameDetail + "]"); |
| output.append("\n"); |
| |
| if (detailLevel.ordinal() >= TExplainLevel.STANDARD.ordinal()) { |
| if (aggInfos_.size() == 1) { |
| output.append( |
| getAggInfoExplainString(detailPrefix, aggInfos_.get(0), detailLevel)); |
| } else { |
| for (int i = 0; i < aggInfos_.size(); ++i) { |
| AggregateInfo aggInfo = aggInfos_.get(i); |
| output.append(String.format("%sClass %d\n", detailPrefix, i)); |
| output.append( |
| getAggInfoExplainString(detailPrefix + " ", aggInfo, detailLevel)); |
| } |
| } |
| if (!conjuncts_.isEmpty()) { |
| output.append(detailPrefix) |
| .append("having: ") |
| .append(Expr.getExplainString(conjuncts_, detailLevel)) |
| .append("\n"); |
| } |
| } |
| return output.toString(); |
| } |
| |
| private StringBuilder getAggInfoExplainString( |
| String prefix, AggregateInfo aggInfo, TExplainLevel detailLevel) { |
| StringBuilder output = new StringBuilder(); |
| List<FunctionCallExpr> aggExprs = aggInfo.getMaterializedAggregateExprs(); |
| List<Expr> groupingExprs = aggInfo.getGroupingExprs(); |
| if (!aggExprs.isEmpty()) { |
| output.append(prefix) |
| .append("output: ") |
| .append(Expr.getExplainString(aggExprs, detailLevel)) |
| .append("\n"); |
| } |
| if (!groupingExprs.isEmpty()) { |
| output.append(prefix) |
| .append("group by: ") |
| .append(Expr.getExplainString(groupingExprs, detailLevel)) |
| .append("\n"); |
| } |
| return output; |
| } |
| |
| @Override |
| public void computeNodeResourceProfile(TQueryOptions queryOptions) { |
| resourceProfiles_ = Lists.newArrayListWithCapacity(aggInfos_.size()); |
| resourceProfiles_.clear(); |
| for (AggregateInfo aggInfo : aggInfos_) { |
| resourceProfiles_.add(computeAggClassResourceProfile(queryOptions, aggInfo)); |
| } |
| if (aggInfos_.size() == 1) { |
| nodeResourceProfile_ = resourceProfiles_.get(0); |
| } else { |
| nodeResourceProfile_ = ResourceProfile.noReservation(0); |
| for (ResourceProfile aggProfile : resourceProfiles_) { |
| nodeResourceProfile_ = nodeResourceProfile_.sum(aggProfile); |
| } |
| } |
| } |
| |
| private ResourceProfile computeAggClassResourceProfile( |
| TQueryOptions queryOptions, AggregateInfo aggInfo) { |
| Preconditions.checkNotNull( |
| fragment_, "PlanNode must be placed into a fragment before calling this method."); |
| long perInstanceCardinality = |
| fragment_.getPerInstanceNdv(queryOptions.getMt_dop(), aggInfo.getGroupingExprs()); |
| long perInstanceMemEstimate; |
| long perInstanceDataBytes = -1; |
| if (perInstanceCardinality == -1) { |
| perInstanceMemEstimate = DEFAULT_PER_INSTANCE_MEM; |
| } else { |
| // Per-instance cardinality cannot be greater than the total input cardinality. |
| long inputCardinality = getChild(0).getCardinality(); |
| if (inputCardinality != -1) { |
| // Calculate the input cardinality distributed across fragment instances. |
| long numInstances = fragment_.getNumInstances(queryOptions.getMt_dop()); |
| long perInstanceInputCardinality; |
| if (numInstances > 1) { |
| perInstanceInputCardinality = |
| (long) Math.ceil((inputCardinality / numInstances) * DEFAULT_SKEW_FACTOR); |
| } else { |
| // When numInstances is 1 or unknown(-1), perInstanceInputCardinality is the |
| // same as inputCardinality. |
| perInstanceInputCardinality = inputCardinality; |
| } |
| perInstanceCardinality = |
| Math.min(perInstanceCardinality, perInstanceInputCardinality); |
| } |
| perInstanceDataBytes = (long)Math.ceil(perInstanceCardinality * avgRowSize_); |
| perInstanceMemEstimate = (long)Math.max(perInstanceDataBytes * |
| PlannerContext.HASH_TBL_SPACE_OVERHEAD, MIN_HASH_TBL_MEM); |
| } |
| |
| // Must be kept in sync with GroupingAggregator::MinReservation() in backend. |
| long perInstanceMinMemReservation; |
| long bufferSize = queryOptions.getDefault_spillable_buffer_size(); |
| long maxRowBufferSize = |
| computeMaxSpillableBufferSize(bufferSize, queryOptions.getMax_row_size()); |
| if (aggInfo.getGroupingExprs().isEmpty()) { |
| perInstanceMinMemReservation = 0; |
| } else { |
| // This is a grouping pre-aggregation or merge aggregation. |
| final int PARTITION_FANOUT = 16; |
| if (perInstanceDataBytes != -1) { |
| long bytesPerPartition = perInstanceDataBytes / PARTITION_FANOUT; |
| // Scale down the buffer size if we think there will be excess free space with the |
| // default buffer size, e.g. with small dimension tables. |
| bufferSize = Math.min(bufferSize, Math.max( |
| queryOptions.getMin_spillable_buffer_size(), |
| BitUtil.roundUpToPowerOf2(bytesPerPartition))); |
| // Recompute the max row buffer size with the smaller buffer. |
| maxRowBufferSize = |
| computeMaxSpillableBufferSize(bufferSize, queryOptions.getMax_row_size()); |
| } |
| if (useStreamingPreagg_) { |
| // We can execute a streaming preagg without any buffers by passing through rows, |
| // but that is a very low performance mode of execution if the aggregation reduces |
| // its input significantly. Instead reserve memory for one buffer per partition |
| // and at least 64kb for hash tables per partition. We must reserve at least one |
| // full buffer for hash tables for the suballocator to subdivide. We don't need to |
| // reserve memory for large rows since they can be passed through if needed. |
| perInstanceMinMemReservation = bufferSize * PARTITION_FANOUT + |
| Math.max(64 * 1024 * PARTITION_FANOUT, bufferSize); |
| } else { |
| long minBuffers = PARTITION_FANOUT + 1 + (aggInfo.needsSerialize() ? 1 : 0); |
| // Two of the buffers need to be buffers large enough to hold the maximum-sized |
| // row to serve as input and output buffers while repartitioning. |
| perInstanceMinMemReservation = bufferSize * (minBuffers - 2) + maxRowBufferSize * 2; |
| } |
| } |
| |
| return new ResourceProfileBuilder() |
| .setMemEstimateBytes(perInstanceMemEstimate) |
| .setMinMemReservationBytes(perInstanceMinMemReservation) |
| .setSpillableBufferBytes(bufferSize) |
| .setMaxRowBufferBytes(maxRowBufferSize) |
| .build(); |
| } |
| } |
