[
https://issues.apache.org/jira/browse/DRILL-6804?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16661370#comment-16661370
]
ASF GitHub Bot commented on DRILL-6804:
---------------------------------------
ilooner closed pull request #1508: DRILL-6804: Simplify usage of OperatorPhase
in HashAgg.
URL: https://github.com/apache/drill/pull/1508
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggBatch.java
b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggBatch.java
index 80d25edb13a..485d3637277 100644
---
a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggBatch.java
+++
b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggBatch.java
@@ -22,6 +22,7 @@
import java.util.List;
import java.util.Map;
+import org.apache.drill.exec.planner.physical.AggPrelBase;
import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
import org.apache.drill.common.exceptions.UserException;
import org.apache.drill.common.expression.ErrorCollector;
@@ -49,7 +50,6 @@
import org.apache.drill.exec.physical.impl.common.Comparator;
import org.apache.drill.exec.physical.impl.common.HashTable;
import org.apache.drill.exec.physical.impl.common.HashTableConfig;
-import org.apache.drill.exec.planner.physical.AggPrelBase;
import org.apache.drill.exec.record.AbstractRecordBatch;
import org.apache.drill.exec.record.BatchSchema;
import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
@@ -192,9 +192,10 @@ public HashAggBatch(HashAggregate popConfig, RecordBatch
incoming, FragmentConte
long memAvail = oContext.getAllocator().getLimit();
long minBatchesPerPartition =
context.getOptions().getOption(ExecConstants.HASHAGG_MIN_BATCHES_PER_PARTITION_VALIDATOR);
long minBatchesNeeded = 2 * minBatchesPerPartition; // 2 - to cover
overheads, etc.
- boolean is2ndPhase = popConfig.getAggPhase() ==
AggPrelBase.OperatorPhase.PHASE_2of2;
boolean fallbackEnabled =
context.getOptions().getOption(ExecConstants.HASHAGG_FALLBACK_ENABLED_KEY).bool_val;
- if ( is2ndPhase && !fallbackEnabled ) {
+ final AggPrelBase.OperatorPhase phase = popConfig.getAggPhase();
+
+ if ( phase.is2nd() && !fallbackEnabled ) {
minBatchesNeeded *= 2; // 2nd phase (w/o fallback) needs at least 2
partitions
}
if ( configuredBatchSize > memAvail / minBatchesNeeded ) { // no cast -
memAvail may be bigger than max-int
diff --git
a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggTemplate.java
b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggTemplate.java
index 6709cf6ddef..32db9eaf487 100644
---
a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggTemplate.java
+++
b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggTemplate.java
@@ -106,9 +106,7 @@
private int rowsSpilledReturned = 0;
private int rowsReturnedEarly = 0;
- private boolean isTwoPhase = false; // 1 phase or 2 phase aggr?
- private boolean is2ndPhase = false;
- private boolean is1stPhase = false;
+ private AggPrelBase.OperatorPhase phase;
private boolean canSpill = true; // make it false in case can not
spill/return-early
private ChainedHashTable baseHashTable;
private boolean earlyOutput = false; // when 1st phase returns a partition
due to no memory
@@ -379,11 +377,8 @@ public void setup(HashAggregate hashAggrConfig,
HashTableConfig htConfig, Fragme
this.outgoing = outgoing;
this.outContainer = outContainer;
this.useMemoryPrediction =
context.getOptions().getOption(ExecConstants.HASHAGG_USE_MEMORY_PREDICTION_VALIDATOR);
-
- is2ndPhase = hashAggrConfig.getAggPhase() ==
AggPrelBase.OperatorPhase.PHASE_2of2;
- isTwoPhase = hashAggrConfig.getAggPhase() !=
AggPrelBase.OperatorPhase.PHASE_1of1;
- is1stPhase = isTwoPhase && !is2ndPhase;
- canSpill = isTwoPhase; // single phase can not spill
+ this.phase = hashAggrConfig.getAggPhase();
+ canSpill = phase.hasTwo(); // single phase can not spill
// Typically for testing - force a spill after a partition has more than
so many batches
minBatchesPerPartition =
context.getOptions().getOption(ExecConstants.HASHAGG_MIN_BATCHES_PER_PARTITION_VALIDATOR);
@@ -447,7 +442,7 @@ private void delayedSetup() {
// Set the number of partitions from the configuration (raise to a power
of two, if needed)
int numPartitions =
(int)context.getOptions().getOption(ExecConstants.HASHAGG_NUM_PARTITIONS_VALIDATOR);
- if ( numPartitions == 1 && is2ndPhase ) { // 1st phase can still do early
return with 1 partition
+ if ( numPartitions == 1 && phase.is2nd() ) { // 1st phase can still do
early return with 1 partition
canSpill = false;
logger.warn("Spilling is disabled due to configuration setting of
num_partitions to 1");
}
@@ -473,7 +468,7 @@ private void delayedSetup() {
while ( numPartitions * ( estMaxBatchSize * minBatchesPerPartition + 2 *
1024 * 1024) > memAvail ) {
numPartitions /= 2;
if ( numPartitions < 2) {
- if (is2ndPhase) {
+ if (phase.is2nd()) {
canSpill = false; // 2nd phase needs at least 2 to make progress
if (fallbackEnabled) {
@@ -492,7 +487,7 @@ private void delayedSetup() {
}
}
}
- logger.debug("{} phase. Number of partitions chosen: {}. {} spill",
isTwoPhase?(is2ndPhase?"2nd":"1st"):"Single",
+ logger.debug("{} phase. Number of partitions chosen: {}. {} spill",
phase.getName(),
numPartitions, canSpill ? "Can" : "Cannot");
// The following initial safety check should be revisited once we can
lower the number of rows in a batch
@@ -616,7 +611,7 @@ private void updateEstMaxBatchSize(RecordBatch incoming) {
estOutgoingAllocSize = estValuesBatchSize; // initially assume same size
logger.trace("{} phase. Estimated internal row width: {} Values row width:
{} batch size: {} memory limit: {} max column width: {}",
-
isTwoPhase?(is2ndPhase?"2nd":"1st"):"Single",estRowWidth,estValuesRowWidth,estMaxBatchSize,allocator.getLimit(),maxColumnWidth);
+
phase.getName(),estRowWidth,estValuesRowWidth,estMaxBatchSize,allocator.getLimit(),maxColumnWidth);
if ( estMaxBatchSize > allocator.getLimit() ) {
logger.warn("HashAggregate: Estimated max batch size {} is larger than
the memory limit {}",estMaxBatchSize,allocator.getLimit());
@@ -886,7 +881,7 @@ public void adjustOutputCount(int outputBatchSize, int
oldRowWidth, int newRowWi
@Override
public void cleanup() {
if ( schema == null ) { return; } // not set up; nothing to clean
- if ( is2ndPhase && spillSet.getWriteBytes() > 0 ) {
+ if ( phase.is2nd() && spillSet.getWriteBytes() > 0 ) {
stats.setLongStat(Metric.SPILL_MB, // update stats - total MB spilled
(int) Math.round(spillSet.getWriteBytes() / 1024.0D / 1024.0));
}
@@ -982,7 +977,7 @@ private boolean isSpilled(int part) {
* @return The partition (number) chosen to be spilled
*/
private int chooseAPartitionToFlush(int currPart, boolean tryAvoidCurr) {
- if ( is1stPhase && ! tryAvoidCurr) { return currPart; } // 1st phase: just
use the current partition
+ if ( phase.is1st() && ! tryAvoidCurr) { return currPart; } // 1st phase:
just use the current partition
int currPartSize = batchHolders[currPart].size();
if ( currPartSize == 1 ) { currPartSize = -1; } // don't pick current if
size is 1
// first find the largest spilled partition
@@ -1188,7 +1183,7 @@ public AggIterOutcome outputCurrentBatch() {
if (spilledState.isEmpty()) { // and no spilled partitions
allFlushed = true;
this.outcome = IterOutcome.NONE;
- if ( is2ndPhase && spillSet.getWriteBytes() > 0 ) {
+ if ( phase.is2nd() && spillSet.getWriteBytes() > 0 ) {
stats.setLongStat(Metric.SPILL_MB, // update stats - total MB
spilled
(int) Math.round(spillSet.getWriteBytes() / 1024.0D / 1024.0));
}
@@ -1243,7 +1238,7 @@ public AggIterOutcome outputCurrentBatch() {
this.outcome = IterOutcome.OK;
- if ( EXTRA_DEBUG_SPILL && is2ndPhase ) {
+ if ( EXTRA_DEBUG_SPILL && phase.is2nd() ) {
logger.debug("So far returned {} + SpilledReturned {} total {} (spilled
{})",rowsNotSpilled,rowsSpilledReturned,
rowsNotSpilled+rowsSpilledReturned,
rowsSpilled);
@@ -1322,12 +1317,12 @@ public int numGroupedRecords() {
*/
private String getOOMErrorMsg(String prefix) {
String errmsg;
- if (!isTwoPhase) {
+ if (!phase.hasTwo()) {
errmsg = "Single Phase Hash Aggregate operator can not spill.";
} else if (!canSpill) { // 2nd phase, with only 1 partition
errmsg = "Too little memory available to operator to facilitate
spilling.";
} else { // a bug ?
- errmsg = prefix + " OOM at " + (is2ndPhase ? "Second Phase" : "First
Phase") + ". Partitions: " + spilledState.getNumPartitions() +
+ errmsg = prefix + " OOM at " + phase.getName() + " Phase. Partitions: "
+ spilledState.getNumPartitions() +
". Estimated batch size: " + estMaxBatchSize + ". values size: " +
estValuesBatchSize + ". Output alloc size: " + estOutgoingAllocSize;
if ( plannedBatches > 0 ) { errmsg += ". Planned batches: " +
plannedBatches; }
if ( rowsSpilled > 0 ) { errmsg += ". Rows spilled so far: " +
rowsSpilled; }
@@ -1367,11 +1362,11 @@ private void checkGroupAndAggrValues(int
incomingRowIdx) {
hashCode >>>= spilledState.getBitsInMask();
HashTable.PutStatus putStatus = null;
long allocatedBeforeHTput = allocator.getAllocatedMemory();
+ String tryingTo = phase.is1st() ? "early return" : "spill";
// Proactive spill - in case there is no reserve memory - spill and retry
putting later
if ( reserveValueBatchMemory == 0 && canSpill ) {
- logger.trace("Reserved memory runs short, trying to {} a partition and
retry Hash Table put() again.",
- is1stPhase ? "early return" : "spill");
+ logger.trace("Reserved memory runs short, trying to {} a partition and
retry Hash Table put() again.", tryingTo);
doSpill(currentPartition); // spill to free some memory
@@ -1389,8 +1384,7 @@ private void checkGroupAndAggrValues(int incomingRowIdx) {
} catch (RetryAfterSpillException re) {
if ( ! canSpill ) { throw new OutOfMemoryException(getOOMErrorMsg("Can
not spill")); }
- logger.trace("HT put failed with an OOM, trying to {} a partition and
retry Hash Table put() again.",
- is1stPhase ? "early return" : "spill");
+ logger.trace("HT put failed with an OOM, trying to {} a partition and
retry Hash Table put() again.", tryingTo);
// for debugging - in case there's a leak
long memDiff = allocator.getAllocatedMemory() - allocatedBeforeHTput;
@@ -1493,7 +1487,7 @@ private void spillIfNeeded(int currentPartition, boolean
forceSpill) {
// log a detailed debug message explaining why a spill may be needed
logger.trace("MEMORY CHECK: Allocated mem: {}, agg phase: {}, trying to
add to partition {} with {} batches. " + "Max memory needed {}, Est batch size
{}, mem limit {}",
- allocator.getAllocatedMemory(), isTwoPhase ? (is2ndPhase ? "2ND" :
"1ST") : "Single", currentPartition, batchHolders[currentPartition].size(),
maxMemoryNeeded,
+ allocator.getAllocatedMemory(), phase.getName(), currentPartition,
batchHolders[currentPartition].size(), maxMemoryNeeded,
estMaxBatchSize, allocator.getLimit());
}
//
@@ -1516,7 +1510,7 @@ private void spillIfNeeded(int currentPartition, boolean
forceSpill) {
return;
}
- if ( is2ndPhase ) {
+ if ( phase.is2nd() ) {
long before = allocator.getAllocatedMemory();
spillAPartition(victimPartition);
@@ -1583,7 +1577,7 @@ private void updateStats(HashTable[] htables) {
this.stats.setLongStat(Metric.RESIZING_TIME_MS, htStats.resizingTime);
this.stats.setLongStat(Metric.NUM_PARTITIONS,
spilledState.getNumPartitions());
this.stats.setLongStat(Metric.SPILL_CYCLE, spilledState.getCycle()); //
Put 0 in case no spill
- if ( is2ndPhase ) {
+ if ( phase.is2nd() ) {
this.stats.setLongStat(Metric.SPILLED_PARTITIONS, numSpilled);
}
if ( rowsReturnedEarly > 0 ) {
diff --git
a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/AggPrelBase.java
b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/AggPrelBase.java
index 84f85ba2f01..f3d527e821b 100644
---
a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/AggPrelBase.java
+++
b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/AggPrelBase.java
@@ -47,7 +47,42 @@
public abstract class AggPrelBase extends DrillAggregateRelBase implements
Prel {
- public enum OperatorPhase {PHASE_1of1, PHASE_1of2, PHASE_2of2}
+ public enum OperatorPhase {
+ PHASE_1of1(false, false, false, "Single"),
+ PHASE_1of2(true, true, false, "1st"),
+ PHASE_2of2(true, false, true, "2nd");
+
+ private boolean hasTwo;
+ private boolean is1st;
+ private boolean is2nd;
+ private String name;
+
+ OperatorPhase(boolean hasTwo,
+ boolean is1st,
+ boolean is2nd,
+ String name) {
+ this.hasTwo = hasTwo;
+ this.is1st = is1st;
+ this.is2nd = is2nd;
+ this.name = name;
+ }
+
+ public boolean hasTwo() {
+ return hasTwo;
+ }
+
+ public boolean is1st() {
+ return is1st;
+ }
+
+ public boolean is2nd() {
+ return is2nd;
+ }
+
+ public String getName() {
+ return name;
+ }
+ }
protected OperatorPhase operPhase = OperatorPhase.PHASE_1of1; // default
phase
protected List<NamedExpression> keys = Lists.newArrayList();
diff --git a/exec/java-exec/src/test/resources/agg/hashagg/q6.json
b/exec/java-exec/src/test/resources/agg/hashagg/q6.json
index c15539188dc..35b200a5fa9 100644
--- a/exec/java-exec/src/test/resources/agg/hashagg/q6.json
+++ b/exec/java-exec/src/test/resources/agg/hashagg/q6.json
@@ -35,6 +35,7 @@
pop : "hash-aggregate",
@id : 3,
child : 2,
+ phase : "PHASE_1of1",
keys : [ {
ref : "$f0",
expr : "$f0"
diff --git a/exec/java-exec/src/test/resources/agg/hashagg/q7_1.json
b/exec/java-exec/src/test/resources/agg/hashagg/q7_1.json
index ef05613677f..3494c793362 100644
--- a/exec/java-exec/src/test/resources/agg/hashagg/q7_1.json
+++ b/exec/java-exec/src/test/resources/agg/hashagg/q7_1.json
@@ -38,6 +38,7 @@
pop : "hash-aggregate",
@id : 3,
child : 2,
+ phase : "PHASE_1of1",
keys : [ {
ref : "$f0",
expr : "$f0"
diff --git a/exec/java-exec/src/test/resources/agg/hashagg/q7_2.json
b/exec/java-exec/src/test/resources/agg/hashagg/q7_2.json
index 62cf5c3587c..f77c08b3d1f 100644
--- a/exec/java-exec/src/test/resources/agg/hashagg/q7_2.json
+++ b/exec/java-exec/src/test/resources/agg/hashagg/q7_2.json
@@ -38,6 +38,7 @@
pop : "hash-aggregate",
@id : 3,
child : 2,
+ phase : "PHASE_1of1",
keys : [ {
ref : "$f0",
expr : "$f0"
diff --git a/exec/java-exec/src/test/resources/agg/hashagg/q7_3.json
b/exec/java-exec/src/test/resources/agg/hashagg/q7_3.json
index 8edc1106929..37ba0ad57f6 100644
--- a/exec/java-exec/src/test/resources/agg/hashagg/q7_3.json
+++ b/exec/java-exec/src/test/resources/agg/hashagg/q7_3.json
@@ -38,6 +38,7 @@
pop : "hash-aggregate",
@id : 3,
child : 2,
+ phase : "PHASE_1of1",
keys : [ {
ref : "$f0",
expr : "$f0"
diff --git a/exec/java-exec/src/test/resources/agg/hashagg/q8.json
b/exec/java-exec/src/test/resources/agg/hashagg/q8.json
index a457aa9aec4..f67cd293121 100644
--- a/exec/java-exec/src/test/resources/agg/hashagg/q8.json
+++ b/exec/java-exec/src/test/resources/agg/hashagg/q8.json
@@ -37,6 +37,7 @@
pop : "hash-aggregate",
@id : 3,
child : 2,
+ phase : "PHASE_1of1",
keys : [ {
ref : "$f0",
expr : "$f0"
diff --git a/exec/java-exec/src/test/resources/agg/hashagg/q8_1.json
b/exec/java-exec/src/test/resources/agg/hashagg/q8_1.json
index 3461c8ca320..73ce4be7837 100644
--- a/exec/java-exec/src/test/resources/agg/hashagg/q8_1.json
+++ b/exec/java-exec/src/test/resources/agg/hashagg/q8_1.json
@@ -35,6 +35,7 @@
pop : "hash-aggregate",
@id : 3,
child : 2,
+ phase : "PHASE_1of1",
keys : [ {
ref : "$f0",
expr : "$f0"
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Simplify Usage of OperatorPhase in HashAgg Template
> ---------------------------------------------------
>
> Key: DRILL-6804
> URL: https://issues.apache.org/jira/browse/DRILL-6804
> Project: Apache Drill
> Issue Type: Sub-task
> Reporter: Timothy Farkas
> Assignee: Timothy Farkas
> Priority: Major
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
