You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by ve...@apache.org on 2015/04/28 04:58:28 UTC
[1/5] drill git commit: DRILL-2857: Update StreamingAggBatch current
workspace record counter variable type to "long" to avoid overflow issues.
Repository: drill
Updated Branches:
refs/heads/master 3689522d4 -> 57a96d200
DRILL-2857: Update StreamingAggBatch current workspace record counter variable type to "long" to avoid overflow issues.
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/a8c96f66
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/a8c96f66
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/a8c96f66
Branch: refs/heads/master
Commit: a8c96f666699c26fceb0ff2081f122fcdc529fb5
Parents: fa0f1c5
Author: vkorukanti <ve...@gmail.com>
Authored: Thu Apr 23 11:48:06 2015 -0700
Committer: vkorukanti <ve...@gmail.com>
Committed: Mon Apr 27 14:11:39 2015 -0700
----------------------------------------------------------------------
.../drill/exec/physical/impl/aggregate/StreamingAggTemplate.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/a8c96f66/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggTemplate.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggTemplate.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggTemplate.java
index 1f139d4..86f3100 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggTemplate.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggTemplate.java
@@ -36,7 +36,7 @@ public abstract class StreamingAggTemplate implements StreamingAggregator {
private int previousIndex = -1;
private int underlyingIndex = 0;
private int currentIndex;
- private int addedRecordCount = 0;
+ private long addedRecordCount = 0;
private IterOutcome outcome;
private int outputCount = 0;
private RecordBatch incoming;
[4/5] drill git commit: DRILL-2858: Refactor hash expression
construction in InsertLocalExchangeVisitor and PrelUtil into one place
Posted by ve...@apache.org.
DRILL-2858: Refactor hash expression construction in InsertLocalExchangeVisitor and PrelUtil into one place
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/6878bfdb
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/6878bfdb
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/6878bfdb
Branch: refs/heads/master
Commit: 6878bfdb29634832230fe54f55581b1fee063fa5
Parents: a8c96f6
Author: vkorukanti <ve...@gmail.com>
Authored: Fri Apr 24 13:06:23 2015 -0700
Committer: vkorukanti <ve...@gmail.com>
Committed: Mon Apr 27 14:12:13 2015 -0700
----------------------------------------------------------------------
.../physical/impl/common/ChainedHashTable.java | 3 +-
.../exec/planner/physical/HashPrelUtil.java | 138 +++++++++++++++++++
.../physical/HashToMergeExchangePrel.java | 2 +-
.../physical/HashToRandomExchangePrel.java | 2 +-
.../drill/exec/planner/physical/PrelUtil.java | 42 ------
.../physical/UnorderedDeMuxExchangePrel.java | 2 +-
.../visitor/InsertLocalExchangeVisitor.java | 57 ++++----
.../exec/physical/impl/TestLocalExchange.java | 16 ++-
8 files changed, 186 insertions(+), 76 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/6878bfdb/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/ChainedHashTable.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/ChainedHashTable.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/ChainedHashTable.java
index 42bb3ec..98cf81d 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/ChainedHashTable.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/ChainedHashTable.java
@@ -44,6 +44,7 @@ import org.apache.drill.exec.expr.fn.FunctionGenerationHelper;
import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.impl.join.JoinUtils;
+import org.apache.drill.exec.planner.physical.HashPrelUtil;
import org.apache.drill.exec.planner.physical.PrelUtil;
import org.apache.drill.exec.record.MaterializedField;
import org.apache.drill.exec.record.RecordBatch;
@@ -313,7 +314,7 @@ public class ChainedHashTable {
* aggregate. For join we need to hash everything as double (both for distribution and for comparison) but
* for aggregation we can avoid the penalty of casting to double
*/
- LogicalExpression hashExpression = PrelUtil.getHashExpression(Arrays.asList(keyExprs),
+ LogicalExpression hashExpression = HashPrelUtil.getHashExpression(Arrays.asList(keyExprs),
incomingProbe != null ? true : false);
final LogicalExpression materializedExpr = ExpressionTreeMaterializer.materializeAndCheckErrors(hashExpression, batch, context.getFunctionRegistry());
HoldingContainer hash = cg.addExpr(materializedExpr);
http://git-wip-us.apache.org/repos/asf/drill/blob/6878bfdb/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/HashPrelUtil.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/HashPrelUtil.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/HashPrelUtil.java
new file mode 100644
index 0000000..5ce5ac1
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/HashPrelUtil.java
@@ -0,0 +1,138 @@
+/**
+ * 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.drill.exec.planner.physical;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexNode;
+import org.apache.drill.common.expression.ExpressionPosition;
+import org.apache.drill.common.expression.FieldReference;
+import org.apache.drill.common.expression.FunctionCall;
+import org.apache.drill.common.expression.LogicalExpression;
+import org.apache.drill.exec.planner.physical.DrillDistributionTrait.DistributionField;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Contains utility methods for creating hash expression for either distribution (in PartitionSender) or for HashTable.
+ */
+public class HashPrelUtil {
+
+ public static final String HASH_EXPR_NAME = "E_X_P_R_H_A_S_H_F_I_E_L_D";
+
+ /**
+ * Interface for creating different forms of hash expression types.
+ * @param <T>
+ */
+ public interface HashExpressionCreatorHelper<T> {
+ T createCall(String funcName, List<T> inputFiled);
+ }
+
+ /**
+ * Implementation of {@link HashExpressionCreatorHelper} for {@link LogicalExpression} type.
+ */
+ public static HashExpressionCreatorHelper<LogicalExpression> HASH_HELPER_LOGICALEXPRESSION =
+ new HashExpressionCreatorHelper<LogicalExpression>() {
+ @Override
+ public LogicalExpression createCall(String funcName, List<LogicalExpression> inputFiled) {
+ return new FunctionCall(funcName, inputFiled, ExpressionPosition.UNKNOWN);
+ }
+ };
+
+ private static final String HASH64_FUNCTION_NAME = "hash64";
+ private static final String HASH64_DOUBLE_FUNCTION_NAME = "hash64AsDouble";
+ private static final String CAST_TO_INT_FUNCTION_NAME = "castInt";
+
+ /**
+ * Create hash based partition expression based on the given distribution fields.
+ *
+ * @param distFields Field list based on which the distribution partition expression is constructed.
+ * @param helper Implementation of {@link HashExpressionCreatorHelper}
+ * which is used to create function expressions.
+ * @param <T> Input and output expression type.
+ * Currently it could be either {@link RexNode} or {@link LogicalExpression}
+ * @return
+ */
+ public static <T> T createHashBasedPartitionExpression(
+ List<T> distFields,
+ HashExpressionCreatorHelper<T> helper) {
+ return createHashExpression(distFields, helper, true /*for distribution always hash as double*/);
+ }
+
+ /**
+ * Create hash expression based on the given input fields.
+ *
+ * @param inputExprs Expression list based on which the hash expression is constructed.
+ * @param helper Implementation of {@link HashExpressionCreatorHelper}
+ * which is used to create function expressions.
+ * @param hashAsDouble Whether to use the hash as double function or regular hash64 function.
+ * @param <T> Input and output expression type.
+ * Currently it could be either {@link RexNode} or {@link LogicalExpression}
+ * @return
+ */
+ public static <T> T createHashExpression(
+ List<T> inputExprs,
+ HashExpressionCreatorHelper<T> helper,
+ boolean hashAsDouble) {
+
+ assert inputExprs.size() > 0;
+
+ final String functionName = hashAsDouble ? HASH64_DOUBLE_FUNCTION_NAME : HASH64_FUNCTION_NAME;
+
+ T func = helper.createCall(functionName, ImmutableList.of(inputExprs.get(0)));
+ for (int i = 1; i<inputExprs.size(); i++) {
+ func = helper.createCall(functionName, ImmutableList.of(inputExprs.get(i), func));
+ }
+
+ return helper.createCall(CAST_TO_INT_FUNCTION_NAME, ImmutableList.of(func));
+ }
+
+ /**
+ * Return a hash expression : (int) hash(field1, hash(field2, hash(field3, 0)));
+ */
+ public static LogicalExpression getHashExpression(List<LogicalExpression> fields, boolean hashAsDouble){
+ return createHashExpression(fields, HASH_HELPER_LOGICALEXPRESSION, hashAsDouble);
+ }
+
+
+ /**
+ * Create a distribution hash expression.
+ *
+ * @param fields Distribution fields
+ * @param rowType Row type
+ * @return
+ */
+ public static LogicalExpression getHashExpression(List<DistributionField> fields, RelDataType rowType) {
+ assert fields.size() > 0;
+
+ final List<String> childFields = rowType.getFieldNames();
+
+ // If we already included a field with hash - no need to calculate hash further down
+ if ( childFields.contains(HASH_EXPR_NAME)) {
+ return new FieldReference(HASH_EXPR_NAME);
+ }
+
+ final List<LogicalExpression> expressions = new ArrayList<LogicalExpression>(childFields.size());
+ for(int i =0; i < fields.size(); i++){
+ expressions.add(new FieldReference(childFields.get(fields.get(i).getFieldId()), ExpressionPosition.UNKNOWN));
+ }
+
+ return createHashBasedPartitionExpression(expressions, HASH_HELPER_LOGICALEXPRESSION);
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/6878bfdb/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/HashToMergeExchangePrel.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/HashToMergeExchangePrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/HashToMergeExchangePrel.java
index 7a0cbeb..4459f85 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/HashToMergeExchangePrel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/HashToMergeExchangePrel.java
@@ -84,7 +84,7 @@ public class HashToMergeExchangePrel extends ExchangePrel {
}
HashToMergeExchange g = new HashToMergeExchange(childPOP,
- PrelUtil.getHashExpression(this.distFields, getInput().getRowType()),
+ HashPrelUtil.getHashExpression(this.distFields, getInput().getRowType()),
PrelUtil.getOrdering(this.collation, getInput().getRowType()));
return creator.addMetadata(this, g);
http://git-wip-us.apache.org/repos/asf/drill/blob/6878bfdb/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/HashToRandomExchangePrel.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/HashToRandomExchangePrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/HashToRandomExchangePrel.java
index c08e6b6..bc47e25 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/HashToRandomExchangePrel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/HashToRandomExchangePrel.java
@@ -95,7 +95,7 @@ public class HashToRandomExchangePrel extends ExchangePrel {
}
// TODO - refactor to different exchange name
- HashToRandomExchange g = new HashToRandomExchange(childPOP, PrelUtil.getHashExpression(this.fields, getInput().getRowType()));
+ HashToRandomExchange g = new HashToRandomExchange(childPOP, HashPrelUtil.getHashExpression(this.fields, getInput().getRowType()));
return creator.addMetadata(this, g);
}
http://git-wip-us.apache.org/repos/asf/drill/blob/6878bfdb/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PrelUtil.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PrelUtil.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PrelUtil.java
index c8141cd..44d828b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PrelUtil.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PrelUtil.java
@@ -23,8 +23,6 @@ import java.util.Iterator;
import java.util.List;
import java.util.Set;
-
-import org.apache.drill.common.expression.CastExpression;
import org.apache.drill.common.expression.ExpressionPosition;
import org.apache.drill.common.expression.FieldReference;
import org.apache.drill.common.expression.FunctionCall;
@@ -34,8 +32,6 @@ import org.apache.drill.common.expression.PathSegment.ArraySegment;
import org.apache.drill.common.expression.PathSegment.NameSegment;
import org.apache.drill.common.expression.SchemaPath;
import org.apache.drill.common.logical.data.Order.Ordering;
-import org.apache.drill.common.types.TypeProtos.MinorType;
-import org.apache.drill.common.types.Types;
import org.apache.drill.exec.planner.physical.DrillDistributionTrait.DistributionField;
import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
@@ -64,10 +60,6 @@ import com.google.common.collect.Sets;
public class PrelUtil {
- public static final String HASH_EXPR_NAME = "E_X_P_R_H_A_S_H_F_I_E_L_D";
- private static final String HASH64_FUNCTION_NAME = "hash64";
- private static final String HASH64_DOUBLE_FUNCTION_NAME = "hash64AsDouble";
-
public static List<Ordering> getOrdering(RelCollation collation, RelDataType rowType) {
List<Ordering> orderExpr = Lists.newArrayList();
@@ -81,40 +73,6 @@ public class PrelUtil {
return orderExpr;
}
- /*
- * Return a hash expression : (int) hash(field1, hash(field2, hash(field3, 0)));
- */
- public static LogicalExpression getHashExpression(List<LogicalExpression> fields, boolean hashAsDouble){
- assert fields.size() > 0;
-
- String functionName = hashAsDouble ? HASH64_DOUBLE_FUNCTION_NAME : HASH64_FUNCTION_NAME;
- FunctionCall func = new FunctionCall(functionName, ImmutableList.of(fields.get(0)), ExpressionPosition.UNKNOWN);
- for (int i = 1; i<fields.size(); i++) {
- func = new FunctionCall(functionName, ImmutableList.of(fields.get(i), func), ExpressionPosition.UNKNOWN);
- }
-
- return new CastExpression(func, Types.required(MinorType.INT), ExpressionPosition.UNKNOWN);
-
- }
-
- public static LogicalExpression getHashExpression(List<DistributionField> fields, RelDataType rowType) {
- assert fields.size() > 0;
-
- final List<String> childFields = rowType.getFieldNames();
-
- // If we already included a field with hash - no need to calculate hash further down
- if ( childFields.contains(HASH_EXPR_NAME)) {
- return new FieldReference(HASH_EXPR_NAME);
- }
-
- final List<LogicalExpression> expressions = new ArrayList<LogicalExpression>(childFields.size());
- for(int i =0; i < fields.size(); i++){
- expressions.add(new FieldReference(childFields.get(fields.get(i).getFieldId()), ExpressionPosition.UNKNOWN));
- }
-
- // for distribution always hash as double
- return getHashExpression(expressions, true);
- }
public static Iterator<Prel> iter(RelNode... nodes) {
return (Iterator<Prel>) (Object) Arrays.asList(nodes).iterator();
http://git-wip-us.apache.org/repos/asf/drill/blob/6878bfdb/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/UnorderedDeMuxExchangePrel.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/UnorderedDeMuxExchangePrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/UnorderedDeMuxExchangePrel.java
index e04f6d0..9f03220 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/UnorderedDeMuxExchangePrel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/UnorderedDeMuxExchangePrel.java
@@ -49,7 +49,7 @@ public class UnorderedDeMuxExchangePrel extends ExchangePrel {
PhysicalOperator childPOP = child.getPhysicalOperator(creator);
- UnorderedDeMuxExchange p = new UnorderedDeMuxExchange(childPOP, PrelUtil.getHashExpression(this.fields, getInput().getRowType()));
+ UnorderedDeMuxExchange p = new UnorderedDeMuxExchange(childPOP, HashPrelUtil.getHashExpression(this.fields, getInput().getRowType()));
return creator.addMetadata(this, p);
}
http://git-wip-us.apache.org/repos/asf/drill/blob/6878bfdb/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/InsertLocalExchangeVisitor.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/InsertLocalExchangeVisitor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/InsertLocalExchangeVisitor.java
index d8311c7..ad64ed8 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/InsertLocalExchangeVisitor.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/InsertLocalExchangeVisitor.java
@@ -21,6 +21,8 @@ import com.google.common.collect.Lists;
import org.apache.drill.common.types.TypeProtos.MajorType;
import org.apache.drill.exec.planner.physical.ExchangePrel;
+import org.apache.drill.exec.planner.physical.HashPrelUtil;
+import org.apache.drill.exec.planner.physical.HashPrelUtil.HashExpressionCreatorHelper;
import org.apache.drill.exec.planner.physical.HashToRandomExchangePrel;
import org.apache.drill.exec.planner.physical.PlannerSettings;
import org.apache.drill.exec.planner.physical.Prel;
@@ -42,13 +44,25 @@ import java.util.Collections;
import java.util.List;
public class InsertLocalExchangeVisitor extends BasePrelVisitor<Prel, Void, RuntimeException> {
- private static final DrillSqlOperator SQL_OP_HASH64_WITH_NO_SEED = new DrillSqlOperator("hash64", 1, MajorType.getDefaultInstance(), true);
- private static final DrillSqlOperator SQL_OP_HASH64_WITH_SEED = new DrillSqlOperator("hash64", 2, MajorType.getDefaultInstance(), true);
- private static final DrillSqlOperator SQL_OP_CAST_INT = new DrillSqlOperator("castINT", 1, MajorType.getDefaultInstance(), true);
-
private final boolean isMuxEnabled;
private final boolean isDeMuxEnabled;
+
+ public static class RexNodeBasedHashExpressionCreatorHelper implements HashExpressionCreatorHelper<RexNode> {
+ private final RexBuilder rexBuilder;
+
+ public RexNodeBasedHashExpressionCreatorHelper(RexBuilder rexBuilder) {
+ this.rexBuilder = rexBuilder;
+ }
+
+ @Override
+ public RexNode createCall(String funcName, List<RexNode> inputFields) {
+ final DrillSqlOperator op =
+ new DrillSqlOperator(funcName, inputFields.size(), MajorType.getDefaultInstance(), true);
+ return rexBuilder.makeCall(op, inputFields);
+ }
+ }
+
public static Prel insertLocalExchanges(Prel prel, OptionManager options) {
boolean isMuxEnabled = options.getOption(PlannerSettings.MUX_EXCHANGE.getOptionName()).bool_val;
boolean isDeMuxEnabled = options.getOption(PlannerSettings.DEMUX_EXCHANGE.getOptionName()).bool_val;
@@ -77,39 +91,36 @@ public class InsertLocalExchangeVisitor extends BasePrelVisitor<Prel, Void, Runt
Prel newPrel = child;
- HashToRandomExchangePrel hashPrel = (HashToRandomExchangePrel) prel;
+ final HashToRandomExchangePrel hashPrel = (HashToRandomExchangePrel) prel;
final List<String> childFields = child.getRowType().getFieldNames();
- List <RexNode> removeUpdatedExpr = Lists.newArrayList();
- if ( isMuxEnabled ) {
+ List <RexNode> removeUpdatedExpr = null;
+
+ if (isMuxEnabled) {
// Insert Project Operator with new column that will be a hash for HashToRandomExchange fields
- List<DistributionField> fields = hashPrel.getFields();
- List<String> outputFieldNames = Lists.newArrayList(childFields);
+ final List<DistributionField> distFields = hashPrel.getFields();
+ final List<String> outputFieldNames = Lists.newArrayList(childFields);
final RexBuilder rexBuilder = prel.getCluster().getRexBuilder();
final List<RelDataTypeField> childRowTypeFields = child.getRowType().getFieldList();
- // First field has no seed argument for hash64 function.
- final int firstFieldId = fields.get(0).getFieldId();
- RexNode firstFieldInputRef = rexBuilder.makeInputRef(childRowTypeFields.get(firstFieldId).getType(), firstFieldId);
- RexNode hashExpr = rexBuilder.makeCall(SQL_OP_HASH64_WITH_NO_SEED, firstFieldInputRef);
-
- for (int i=1; i<fields.size(); i++) {
- final int fieldId = fields.get(i).getFieldId();
- RexNode inputRef = rexBuilder.makeInputRef(childRowTypeFields.get(fieldId).getType(), fieldId);
- hashExpr = rexBuilder.makeCall(SQL_OP_HASH64_WITH_SEED, inputRef, hashExpr);
+ final HashExpressionCreatorHelper<RexNode> hashHelper = new RexNodeBasedHashExpressionCreatorHelper(rexBuilder);
+ final List<RexNode> distFieldRefs = Lists.newArrayListWithExpectedSize(distFields.size());
+ for(int i=0; i<distFields.size(); i++) {
+ final int fieldId = distFields.get(i).getFieldId();
+ distFieldRefs.add(rexBuilder.makeInputRef(childRowTypeFields.get(fieldId).getType(), fieldId));
}
- hashExpr = rexBuilder.makeCall(SQL_OP_CAST_INT, hashExpr);
-
- List <RexNode> updatedExpr = Lists.newArrayList();
+ final List <RexNode> updatedExpr = Lists.newArrayListWithExpectedSize(childRowTypeFields.size());
+ removeUpdatedExpr = Lists.newArrayListWithExpectedSize(childRowTypeFields.size());
for ( RelDataTypeField field : childRowTypeFields) {
RexNode rex = rexBuilder.makeInputRef(field.getType(), field.getIndex());
updatedExpr.add(rex);
removeUpdatedExpr.add(rex);
}
- outputFieldNames.add(PrelUtil.HASH_EXPR_NAME);
- updatedExpr.add(hashExpr);
+ outputFieldNames.add(HashPrelUtil.HASH_EXPR_NAME);
+ updatedExpr.add(HashPrelUtil.createHashBasedPartitionExpression(distFieldRefs, hashHelper));
+
RelDataType rowType = RexUtil.createStructType(prel.getCluster().getTypeFactory(), updatedExpr, outputFieldNames);
ProjectPrel addColumnprojectPrel = new ProjectPrel(child.getCluster(), child.getTraitSet(), child, updatedExpr, rowType);
http://git-wip-us.apache.org/repos/asf/drill/blob/6878bfdb/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestLocalExchange.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestLocalExchange.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestLocalExchange.java
index 2080fce..9758eb0 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestLocalExchange.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestLocalExchange.java
@@ -34,6 +34,7 @@ import org.apache.drill.exec.planner.fragment.Fragment;
import org.apache.drill.exec.planner.fragment.Fragment.ExchangeFragmentPair;
import org.apache.drill.exec.planner.fragment.PlanningSet;
import org.apache.drill.exec.planner.fragment.SimpleParallelizer;
+import org.apache.drill.exec.planner.physical.HashPrelUtil;
import org.apache.drill.exec.planner.physical.PrelUtil;
import org.apache.drill.exec.pop.PopUnitTestBase;
import org.apache.drill.exec.proto.BitControl.PlanFragment;
@@ -57,6 +58,7 @@ import java.io.IOException;
import java.io.PrintWriter;
import java.util.List;
+import static org.apache.drill.exec.planner.physical.HashPrelUtil.HASH_EXPR_NAME;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
@@ -208,7 +210,7 @@ public class TestLocalExchange extends PlanTestBase {
final String plan = getPlanInString("EXPLAIN PLAN FOR " + groupByMultipleQuery, JSON_FORMAT);
System.out.println("Plan: " + plan);
- jsonExchangeOrderChecker(plan, false, 1, "castint\\(hash64\\(.*, hash64\\(.*, hash64\\(.*\\) \\) \\) \\) ");
+ jsonExchangeOrderChecker(plan, false, 1, "castint\\(hash64asdouble\\(.*, hash64asdouble\\(.*, hash64asdouble\\(.*\\) \\) \\) \\) ");
// Run the query and verify the output
final TestBuilder testBuilder = testBuilder()
@@ -285,10 +287,10 @@ public class TestLocalExchange extends PlanTestBase {
if ( isMuxOn ) {
// # of hash exchanges should be = # of mux exchanges + # of demux exchanges
- assertEquals("HashExpr on the hash column should not happen", 2*expectedNumMuxes+expectedNumDeMuxes, StringUtils.countMatches(plan, PrelUtil.HASH_EXPR_NAME));
- jsonExchangeOrderChecker(plan, isDeMuxOn, expectedNumMuxes, "castint\\(hash64\\(.*\\) \\) ");
+ assertEquals("HashExpr on the hash column should not happen", 2*expectedNumMuxes+expectedNumDeMuxes, StringUtils.countMatches(plan, HASH_EXPR_NAME));
+ jsonExchangeOrderChecker(plan, isDeMuxOn, expectedNumMuxes, "castint\\(hash64asdouble\\(.*\\) \\) ");
} else {
- assertEquals("HashExpr on the hash column should not happen", 0, StringUtils.countMatches(plan, PrelUtil.HASH_EXPR_NAME));
+ assertEquals("HashExpr on the hash column should not happen", 0, StringUtils.countMatches(plan, HASH_EXPR_NAME));
}
// Make sure the plan has mux and demux exchanges (TODO: currently testing is rudimentary,
@@ -331,7 +333,7 @@ public class TestLocalExchange extends PlanTestBase {
final JSONArray exprsArray = (JSONArray) popObj.get("exprs");
for (Object exprObj : exprsArray) {
final JSONObject expr = (JSONObject) exprObj;
- if ( expr.containsKey("ref") && expr.get("ref").equals("`"+PrelUtil.HASH_EXPR_NAME +"`")) {
+ if ( expr.containsKey("ref") && expr.get("ref").equals("`"+ HASH_EXPR_NAME +"`")) {
// found a match. Let's see if next one is the one we need
final String hashField = (String) expr.get("expr");
assertNotNull("HashExpr field can not be null", hashField);
@@ -361,7 +363,7 @@ public class TestLocalExchange extends PlanTestBase {
popObj.containsKey("pop") && popObj.get("pop").equals(HASH_EXCHANGE));
// is HashToRandom is using HashExpr
assertTrue("HashToRandomExchnage should use hashExpr",
- popObj.containsKey("expr") && popObj.get("expr").equals("`"+PrelUtil.HASH_EXPR_NAME +"`"));
+ popObj.containsKey("expr") && popObj.get("expr").equals("`"+ HASH_EXPR_NAME +"`"));
}
// if Demux is enabled it also should use HashExpr
if ( isDemuxEnabled && k == i-3) {
@@ -369,7 +371,7 @@ public class TestLocalExchange extends PlanTestBase {
popObj.containsKey("pop") && popObj.get("pop").equals(DEMUX_EXCHANGE_CONST));
// is HashToRandom is using HashExpr
assertTrue("UnorderdDemuxExchange should use hashExpr",
- popObj.containsKey("expr") && popObj.get("expr").equals("`"+PrelUtil.HASH_EXPR_NAME +"`"));
+ popObj.containsKey("expr") && popObj.get("expr").equals("`"+HASH_EXPR_NAME +"`"));
}
if ( (isDemuxEnabled && k == i-4) || (!isDemuxEnabled && k == i-3) ) {
// it should be a project without hashexpr, check if number of exprs is 1 less then in first project
[3/5] drill git commit: DRILL-2856: Fix StreamingAgg inifnite loop
problem due to state management issue.
Posted by ve...@apache.org.
DRILL-2856: Fix StreamingAgg inifnite loop problem due to state management issue.
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/3f781be0
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/3f781be0
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/3f781be0
Branch: refs/heads/master
Commit: 3f781be01451e3fe7d312de49b3c8d3c6d73e02d
Parents: 3689522
Author: Jacques Nadeau <ja...@apache.org>
Authored: Sun Apr 19 08:52:47 2015 -0700
Committer: vkorukanti <ve...@gmail.com>
Committed: Mon Apr 27 14:11:39 2015 -0700
----------------------------------------------------------------------
.../physical/impl/aggregate/StreamingAggTemplate.java | 13 +++++++------
1 file changed, 7 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/3f781be0/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggTemplate.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggTemplate.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggTemplate.java
index bf27187..05560c4 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggTemplate.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggTemplate.java
@@ -21,7 +21,6 @@ import javax.inject.Named;
import org.apache.drill.exec.exception.SchemaChangeException;
import org.apache.drill.exec.ops.FragmentContext;
-import org.apache.drill.exec.record.BatchSchema;
import org.apache.drill.exec.record.RecordBatch;
import org.apache.drill.exec.record.RecordBatch.IterOutcome;
import org.apache.drill.exec.record.VectorWrapper;
@@ -85,12 +84,12 @@ public abstract class StreamingAggTemplate implements StreamingAggregator {
if (first) {
this.currentIndex = incoming.getRecordCount() == 0 ? 0 : this.getVectorIndex(underlyingIndex);
- }
- if (incoming.getRecordCount() == 0) {
- outer: while (true) {
- IterOutcome out = outgoing.next(0, incoming);
- switch (out) {
+ // consume empty batches until we get one with data.
+ if (incoming.getRecordCount() == 0) {
+ outer: while (true) {
+ IterOutcome out = outgoing.next(0, incoming);
+ switch (out) {
case OK_NEW_SCHEMA:
case OK:
if (incoming.getRecordCount() == 0) {
@@ -106,10 +105,12 @@ public abstract class StreamingAggTemplate implements StreamingAggregator {
outcome = out;
done = true;
return AggOutcome.CLEANUP_AND_RETURN;
+ }
}
}
}
+
if (newSchema) {
return AggOutcome.UPDATE_AGGREGATOR;
}
[5/5] drill git commit: DRILL-2083: Fix bug in merging receiver
Posted by ve...@apache.org.
DRILL-2083: Fix bug in merging receiver
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/57a96d20
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/57a96d20
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/57a96d20
Branch: refs/heads/master
Commit: 57a96d200e12c0efcad3f3ca9d935c42647234b1
Parents: 6878bfd
Author: Steven Phillips <sm...@apache.org>
Authored: Mon Apr 20 15:12:57 2015 -0700
Committer: vkorukanti <ve...@gmail.com>
Committed: Mon Apr 27 14:12:13 2015 -0700
----------------------------------------------------------------------
.../impl/mergereceiver/MergingRecordBatch.java | 35 ++++++++++++++------
.../impl/mergereceiver/TestMergingReceiver.java | 28 ++++++++--------
.../test/resources/mergerecv/empty_batch.json | 2 +-
.../resources/mergerecv/merging_receiver.json | 4 +--
.../resources/mergerecv/multiple_providers.json | 10 +++---
5 files changed, 46 insertions(+), 33 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/57a96d20/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/mergereceiver/MergingRecordBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/mergereceiver/MergingRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/mergereceiver/MergingRecordBatch.java
index 3cf2857..40cbc89 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/mergereceiver/MergingRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/mergereceiver/MergingRecordBatch.java
@@ -112,7 +112,9 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
private int[] batchOffsets;
private PriorityQueue <Node> pqueue;
private RawFragmentBatch emptyBatch = null;
- private RawFragmentBatch[] tempBatchHolder; //
+ private RawFragmentBatch[] tempBatchHolder;
+ private long[] inputCounts;
+ private long[] outputCounts;
public static enum Metric implements MetricDef{
BYTES_RECEIVED,
@@ -135,15 +137,19 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
this.outgoingContainer = new VectorContainer(oContext);
this.stats.setLongStat(Metric.NUM_SENDERS, config.getNumSenders());
this.config = config;
+ this.inputCounts = new long[config.getNumSenders()];
+ this.outputCounts = new long[config.getNumSenders()];
}
- private RawFragmentBatch getNext(final RawFragmentBatchProvider provider) throws IOException{
+ private RawFragmentBatch getNext(final int providerIndex) throws IOException{
stats.startWait();
+ final RawFragmentBatchProvider provider = fragProviders[providerIndex];
try {
final RawFragmentBatch b = provider.getNext();
if (b != null) {
stats.addLongStat(Metric.BYTES_RECEIVED, b.getByteCount());
stats.batchReceived(0, b.getHeader().getDef().getRecordCount(), false);
+ inputCounts[providerIndex] += b.getHeader().getDef().getRecordCount();
}
return b;
} finally {
@@ -186,9 +192,8 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
rawBatch = tempBatchHolder[p];
tempBatchHolder[p] = null;
} else {
- rawBatch = getNext(provider);
+ rawBatch = getNext(p);
}
- p++;
if (rawBatch == null && !context.shouldContinue()) {
return IterOutcome.STOP;
}
@@ -204,7 +209,7 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
emptyBatch = rawBatch;
}
try {
- while ((rawBatch = getNext(provider)) != null && rawBatch.getHeader().getDef().getRecordCount() == 0) {
+ while ((rawBatch = getNext(p)) != null && rawBatch.getHeader().getDef().getRecordCount() == 0) {
;
}
if (rawBatch == null && !context.shouldContinue()) {
@@ -220,6 +225,7 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
rawBatches.add(emptyBatch);
}
}
+ p++;
}
// allocate the incoming record batch loaders
@@ -304,7 +310,7 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
for (int b = 0; b < senderCount; ++b) {
while (batchLoaders[b] != null && batchLoaders[b].getRecordCount() == 0) {
try {
- final RawFragmentBatch batch = getNext(fragProviders[b]);
+ final RawFragmentBatch batch = getNext(b);
incomingBatches[b] = batch;
if (batch != null) {
batchLoaders[b].load(batch.getHeader().getDef(), batch.getBody());
@@ -335,7 +341,6 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
if (!copyRecordToOutgoingBatch(node)) {
logger.debug("Outgoing vectors space is full; breaking");
prevBatchWasFull = true;
- break;
}
pqueue.poll();
@@ -349,11 +354,13 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
// reached the end of an incoming record batch
RawFragmentBatch nextBatch = null;
try {
- nextBatch = getNext(fragProviders[node.batchId]);
+ nextBatch = getNext(node.batchId);
while (nextBatch != null && nextBatch.getHeader().getDef().getRecordCount() == 0) {
- nextBatch = getNext(fragProviders[node.batchId]);
+ nextBatch = getNext(node.batchId);
}
+ assert nextBatch != null || inputCounts[node.batchId] == outputCounts[node.batchId]
+ : String.format("Stream %d input count: %d output count %d", node.batchId, inputCounts[node.batchId], outputCounts[node.batchId]);
if (nextBatch == null && !context.shouldContinue()) {
return IterOutcome.STOP;
}
@@ -383,7 +390,11 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
// this batch is empty; since the pqueue no longer references this batch, it will be
// ignored in subsequent iterations.
- continue;
+ if (prevBatchWasFull) {
+ break;
+ } else {
+ continue;
+ }
}
final UserBitShared.RecordBatchDef rbd = incomingBatches[node.batchId].getHeader().getDef();
@@ -447,7 +458,7 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
state = BatchState.DONE;
return;
}
- final RawFragmentBatch batch = getNext(fragProviders[i]);
+ final RawFragmentBatch batch = getNext(i);
if (batch.getHeader().getDef().getFieldCount() == 0) {
i++;
continue;
@@ -661,6 +672,8 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
* @param node Reference to the next record to copy from the incoming batches
*/
private boolean copyRecordToOutgoingBatch(final Node node) {
+ assert ++outputCounts[node.batchId] <= inputCounts[node.batchId]
+ : String.format("Stream %d input count: %d output count %d", node.batchId, inputCounts[node.batchId], outputCounts[node.batchId]);
final int inIndex = (node.batchId << 16) + node.valueIndex;
merger.doCopy(inIndex, outgoingPosition);
outgoingPosition++;
http://git-wip-us.apache.org/repos/asf/drill/blob/57a96d20/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/mergereceiver/TestMergingReceiver.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/mergereceiver/TestMergingReceiver.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/mergereceiver/TestMergingReceiver.java
index cf9dd84..0122c08 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/mergereceiver/TestMergingReceiver.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/mergereceiver/TestMergingReceiver.java
@@ -66,21 +66,21 @@ public class TestMergingReceiver extends PopUnitTestBase {
}
for (Object cell : row) {
if (cell == null) {
- System.out.print("<null> ");
+// System.out.print("<null> ");
continue;
}
int len = cell.toString().length();
- System.out.print(cell + " ");
+// System.out.print(cell + " ");
for (int i = 0; i < (30 - len); ++i) {
- System.out.print(" ");
+// System.out.print(" ");
}
}
- System.out.println();
+// System.out.println();
}
b.release();
batchLoader.clear();
}
- assertEquals(200, count);
+ assertEquals(200000, count);
}
}
@@ -122,17 +122,17 @@ public class TestMergingReceiver extends PopUnitTestBase {
}
for (Object cell : row) {
int len = cell.toString().length();
- System.out.print(cell + " ");
+// System.out.print(cell + " ");
for (int i = 0; i < (30 - len); ++i) {
- System.out.print(" ");
+// System.out.print(" ");
}
}
- System.out.println();
+// System.out.println();
}
b.release();
batchLoader.clear();
}
- assertEquals(400, count);
+ assertEquals(400000, count);
}
}
@@ -163,21 +163,21 @@ public class TestMergingReceiver extends PopUnitTestBase {
}
for (Object cell : row) {
if (cell == null) {
- System.out.print("<null> ");
+// System.out.print("<null> ");
continue;
}
int len = cell.toString().length();
- System.out.print(cell + " ");
+// System.out.print(cell + " ");
for (int i = 0; i < (30 - len); ++i) {
- System.out.print(" ");
+// System.out.print(" ");
}
}
- System.out.println();
+// System.out.println();
}
b.release();
batchLoader.clear();
}
- assertEquals(100, count);
+ assertEquals(100000, count);
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/57a96d20/exec/java-exec/src/test/resources/mergerecv/empty_batch.json
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/resources/mergerecv/empty_batch.json b/exec/java-exec/src/test/resources/mergerecv/empty_batch.json
index 55b3f7d..c954827 100644
--- a/exec/java-exec/src/test/resources/mergerecv/empty_batch.json
+++ b/exec/java-exec/src/test/resources/mergerecv/empty_batch.json
@@ -12,7 +12,7 @@
pop:"mock-scan",
url: "http://apache.org",
entries:[
- {records: 100, types: [
+ {records: 100000, types: [
{name: "blue", type: "BIGINT", mode: "OPTIONAL"},
{name: "red", type: "BIGINT", mode: "OPTIONAL"},
{name: "green", type: "BIGINT", mode: "OPTIONAL"}
http://git-wip-us.apache.org/repos/asf/drill/blob/57a96d20/exec/java-exec/src/test/resources/mergerecv/merging_receiver.json
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/resources/mergerecv/merging_receiver.json b/exec/java-exec/src/test/resources/mergerecv/merging_receiver.json
index 50a66f5..5a0166a 100644
--- a/exec/java-exec/src/test/resources/mergerecv/merging_receiver.json
+++ b/exec/java-exec/src/test/resources/mergerecv/merging_receiver.json
@@ -12,12 +12,12 @@
pop:"mock-scan",
url: "http://apache.org",
entries:[
- {records: 100, types: [
+ {records: 100000, types: [
{name: "blue", type: "BIGINT", mode: "OPTIONAL"},
{name: "red", type: "BIGINT", mode: "OPTIONAL"},
{name: "green", type: "BIGINT", mode: "OPTIONAL"}
]},
- {records: 100, types: [
+ {records: 100000, types: [
{name: "blue", type: "BIGINT", mode: "OPTIONAL"},
{name: "red", type: "BIGINT", mode: "OPTIONAL"},
{name: "green", type: "BIGINT", mode: "OPTIONAL"}
http://git-wip-us.apache.org/repos/asf/drill/blob/57a96d20/exec/java-exec/src/test/resources/mergerecv/multiple_providers.json
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/resources/mergerecv/multiple_providers.json b/exec/java-exec/src/test/resources/mergerecv/multiple_providers.json
index a02fe3f..b32de5c 100644
--- a/exec/java-exec/src/test/resources/mergerecv/multiple_providers.json
+++ b/exec/java-exec/src/test/resources/mergerecv/multiple_providers.json
@@ -12,27 +12,27 @@
pop:"mock-scan",
url: "http://apache.org",
entries:[
- {records: 100, types: [
+ {records: 100000, types: [
{name: "blue", type: "BIGINT", mode: "REQUIRED"},
{name: "red", type: "BIGINT", mode: "REQUIRED"},
{name: "green", type: "BIGINT", mode: "REQUIRED"}
]},
- {records: 90, types: [
+ {records: 90000, types: [
{name: "blue", type: "BIGINT", mode: "REQUIRED"},
{name: "red", type: "BIGINT", mode: "REQUIRED"},
{name: "green", type: "BIGINT", mode: "REQUIRED"}
]},
- {records: 80, types: [
+ {records: 80000, types: [
{name: "blue", type: "BIGINT", mode: "REQUIRED"},
{name: "red", type: "BIGINT", mode: "REQUIRED"},
{name: "green", type: "BIGINT", mode: "REQUIRED"}
]},
- {records: 70, types: [
+ {records: 70000, types: [
{name: "blue", type: "BIGINT", mode: "REQUIRED"},
{name: "red", type: "BIGINT", mode: "REQUIRED"},
{name: "green", type: "BIGINT", mode: "REQUIRED"}
]},
- {records: 60, types: [
+ {records: 60000, types: [
{name: "blue", type: "BIGINT", mode: "REQUIRED"},
{name: "red", type: "BIGINT", mode: "REQUIRED"},
{name: "green", type: "BIGINT", mode: "REQUIRED"}
[2/5] drill git commit: DRILL-2855: Fix invalid result issues with
StreamAggBatch
Posted by ve...@apache.org.
DRILL-2855: Fix invalid result issues with StreamAggBatch
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/fa0f1c5f
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/fa0f1c5f
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/fa0f1c5f
Branch: refs/heads/master
Commit: fa0f1c5f0feace931ddb236a7eadc57f3b4135b2
Parents: 3f781be
Author: vkorukanti <ve...@gmail.com>
Authored: Sun Apr 19 14:16:09 2015 -0700
Committer: vkorukanti <ve...@gmail.com>
Committed: Mon Apr 27 14:11:39 2015 -0700
----------------------------------------------------------------------
.../impl/aggregate/StreamingAggTemplate.java | 19 ++++++++-----------
1 file changed, 8 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/fa0f1c5f/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggTemplate.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggTemplate.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggTemplate.java
index 05560c4..1f139d4 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggTemplate.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggTemplate.java
@@ -153,21 +153,18 @@ public abstract class StreamingAggTemplate implements StreamingAggregator {
}
// Update the indices to set the state for processing next record in incoming batch in subsequent doWork calls.
- previousIndex = currentIndex;
- incIndex();
+ previousIndex = -1;
return setOkAndReturn();
}
}
previousIndex = currentIndex;
}
- InternalBatch previous = null;
+ InternalBatch previous = new InternalBatch(incoming);
+
try {
while (true) {
- if (previous != null) {
- previous.clear();
- }
- previous = new InternalBatch(incoming);
+
IterOutcome out = outgoing.next(0, incoming);
if (EXTRA_DEBUG) {
logger.debug("Received IterOutcome of {}", out);
@@ -232,16 +229,17 @@ public abstract class StreamingAggTemplate implements StreamingAggregator {
if (EXTRA_DEBUG) {
logger.debug("This is not the same as the previous, add record and continue outside.");
}
- previousIndex = currentIndex;
if (addedRecordCount > 0) {
if (outputToBatchPrev(previous, previousIndex, outputCount)) {
if (EXTRA_DEBUG) {
logger.debug("Output container is full. flushing it.");
}
+ previousIndex = -1;
return setOkAndReturn();
}
- continue outside;
}
+ previousIndex = -1;
+ continue outside;
}
}
case STOP:
@@ -250,8 +248,7 @@ public abstract class StreamingAggTemplate implements StreamingAggregator {
outcome = out;
return AggOutcome.CLEANUP_AND_RETURN;
}
-
- }
+ }
} finally {
// make sure to clear previous
if (previous != null) {