You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@drill.apache.org by ja...@apache.org on 2014/03/18 04:44:42 UTC
[2/7] git commit: DRILL-326: Fixes for merge join allocations
DRILL-326: Fixes for merge join allocations
Project: http://git-wip-us.apache.org/repos/asf/incubator-drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-drill/commit/d3c01968
Tree: http://git-wip-us.apache.org/repos/asf/incubator-drill/tree/d3c01968
Diff: http://git-wip-us.apache.org/repos/asf/incubator-drill/diff/d3c01968
Branch: refs/heads/master
Commit: d3c01968ec0afd5d188f23becefaec456d59b168
Parents: e64a682
Author: Jinfeng Ni <jn...@maprtech.com>
Authored: Mon Mar 17 09:01:21 2014 -0700
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Mon Mar 17 09:05:28 2014 -0700
----------------------------------------------------------------------
.../exec/physical/impl/join/JoinStatus.java | 4 +
.../exec/physical/impl/join/JoinTemplate.java | 15 ++--
.../exec/physical/impl/join/MergeJoinBatch.java | 10 ++-
.../IteratorValidatorBatchIterator.java | 4 +
.../apache/drill/exec/record/RecordBatch.java | 3 +
.../exec/physical/impl/join/TestMergeJoin.java | 26 ++++++
.../src/test/resources/join/join_batchsize.json | 88 ++++++++++++++++++++
7 files changed, 142 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/d3c01968/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinStatus.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinStatus.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinStatus.java
index 762cce7..bf87c0a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinStatus.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinStatus.java
@@ -101,6 +101,10 @@ public final class JoinStatus {
outputPosition = 0;
}
+ public final void incOutputPos() {
+ outputPosition++;
+ }
+
public final void notifyLeftRepeating() {
leftRepeating = true;
outputBatch.resetBatchBuilder();
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/d3c01968/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinTemplate.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinTemplate.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinTemplate.java
index 7c8a51c..f43934e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinTemplate.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinTemplate.java
@@ -95,8 +95,10 @@ public abstract class JoinTemplate implements JoinWorker {
if (((MergeJoinPOP)status.outputBatch.getPopConfig()).getJoinType() == JoinRelType.LEFT) {
// we've hit the end of the right record batch; copy any remaining values from the left batch
while (status.isLeftPositionAllowed()) {
- if (!doCopyLeft(status.getLeftPosition(), status.fetchAndIncOutputPos()))
+ if (!doCopyLeft(status.getLeftPosition(), status.getOutPosition()))
return false;
+
+ status.incOutputPos();
status.advanceLeft();
}
}
@@ -110,10 +112,11 @@ public abstract class JoinTemplate implements JoinWorker {
case -1:
// left key < right key
- if (((MergeJoinPOP)status.outputBatch.getPopConfig()).getJoinType() == JoinRelType.LEFT)
- if (!doCopyLeft(status.getLeftPosition(), status.fetchAndIncOutputPos())) {
+ if (((MergeJoinPOP)status.outputBatch.getPopConfig()).getJoinType() == JoinRelType.LEFT) {
+ if (!doCopyLeft(status.getLeftPosition(), status.getOutPosition()))
return false;
- }
+ status.incOutputPos();
+ }
status.advanceLeft();
continue;
@@ -140,9 +143,11 @@ public abstract class JoinTemplate implements JoinWorker {
if (!doCopyLeft(status.getLeftPosition(), status.getOutPosition()))
return false;
- if (!doCopyRight(status.getRightPosition(), status.fetchAndIncOutputPos()))
+ if (!doCopyRight(status.getRightPosition(), status.getOutPosition()))
return false;
+ status.incOutputPos();
+
// If the left key has duplicates and we're about to cross a boundary in the right batch, add the
// right table's record batch to the sv4 builder before calling next. These records will need to be
// copied again for each duplicate left key.
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/d3c01968/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/MergeJoinBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/MergeJoinBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/MergeJoinBatch.java
index bd668e7..7680ff9 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/MergeJoinBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/MergeJoinBatch.java
@@ -395,16 +395,20 @@ public class MergeJoinBatch extends AbstractRecordBatch<MergeJoinPOP> {
private void allocateBatch() {
// allocate new batch space.
container.clear();
- // add fields from both batches
+
+ //estimation of joinBatchSize : max of left/right size, expanded by a factor of 16, which is then bounded by MAX_BATCH_SIZE.
+ int joinBatchSize = Math.min(Math.max(left.getRecordCount() , right.getRecordCount() ) * 16, MAX_BATCH_SIZE);
+
+ // add fields from both batches
for (VectorWrapper<?> w : left) {
ValueVector outgoingVector = TypeHelper.getNewVector(w.getField(), context.getAllocator());
- VectorAllocator.getAllocator(outgoingVector, (int) Math.ceil(w.getValueVector().getBufferSize() / left.getRecordCount())).alloc(left.getRecordCount() * 16);
+ VectorAllocator.getAllocator(outgoingVector, (int) Math.ceil(w.getValueVector().getBufferSize() / left.getRecordCount())).alloc(joinBatchSize);
container.add(outgoingVector);
}
for (VectorWrapper<?> w : right) {
ValueVector outgoingVector = TypeHelper.getNewVector(w.getField(), context.getAllocator());
- VectorAllocator.getAllocator(outgoingVector, (int) Math.ceil(w.getValueVector().getBufferSize() / right.getRecordCount())).alloc(right.getRecordCount() * 16);
+ VectorAllocator.getAllocator(outgoingVector, (int) Math.ceil(w.getValueVector().getBufferSize() / right.getRecordCount())).alloc(joinBatchSize);
container.add(outgoingVector);
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/d3c01968/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/validate/IteratorValidatorBatchIterator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/validate/IteratorValidatorBatchIterator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/validate/IteratorValidatorBatchIterator.java
index 8552465..379fad2 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/validate/IteratorValidatorBatchIterator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/validate/IteratorValidatorBatchIterator.java
@@ -105,6 +105,10 @@ public class IteratorValidatorBatchIterator implements RecordBatch{
public IterOutcome next() {
if(state == IterOutcome.NONE ) throw new IllegalStateException("The incoming iterator has previously moved to a state of NONE. You should not be attempting to call next() again.");
state = incoming.next();
+
+ if ((state == IterOutcome.OK || state == IterOutcome.OK_NEW_SCHEMA) && incoming.getRecordCount() > MAX_BATCH_SIZE)
+ throw new IllegalStateException (String.format("Incoming batch of %s has size %d, which is beyond the limit of %d", incoming.getClass().getName(), incoming.getRecordCount(), MAX_BATCH_SIZE));
+
return state;
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/d3c01968/exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordBatch.java
index b41b733..b77a6a8 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/RecordBatch.java
@@ -33,6 +33,9 @@ import org.apache.drill.exec.vector.ValueVector;
*/
public interface RecordBatch extends VectorAccessible {
+ /* max batch size, limited by 2-byte-lentgh in SV2 : 65535 = 2^16 -1 */
+ public static final int MAX_BATCH_SIZE = 65535;
+
/**
* Describes the outcome of a RecordBatch being incremented forward.
*/
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/d3c01968/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestMergeJoin.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestMergeJoin.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestMergeJoin.java
index 27fae08..6e681e1 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestMergeJoin.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestMergeJoin.java
@@ -42,6 +42,7 @@ import org.apache.drill.exec.proto.CoordinationProtos;
import org.apache.drill.exec.proto.ExecProtos;
import org.apache.drill.exec.proto.BitControl.PlanFragment;
import org.apache.drill.exec.rpc.user.UserServer;
+import org.apache.drill.exec.rpc.user.UserServer.UserClientConnection;
import org.apache.drill.exec.server.DrillbitContext;
import org.apache.drill.exec.store.StoragePluginRegistry;
import org.apache.drill.exec.vector.ValueVector;
@@ -266,6 +267,31 @@ public class TestMergeJoin {
}
+ @Test
+ public void testJoinBatchSize(@Injectable final DrillbitContext bitContext, @Injectable UserClientConnection connection) throws Throwable{
+ new NonStrictExpectations(){{
+ bitContext.getMetrics(); result = new MetricRegistry();
+ bitContext.getAllocator(); result = new TopLevelAllocator();;
+ bitContext.getConfig(); result = c;
+ bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
+ }};
+
+ PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/join/join_batchsize.json"), Charsets.UTF_8));
+ FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+ while(exec.next()){
+ assertEquals(100, exec.getRecordCount());
+ }
+
+ if(context.getFailureCause() != null){
+ throw context.getFailureCause();
+ }
+ assertTrue(!context.isFailed());
+
+ }
+
@AfterClass
public static void tearDown() throws Exception{
// pause to get logger to catch up.
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/d3c01968/exec/java-exec/src/test/resources/join/join_batchsize.json
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/resources/join/join_batchsize.json b/exec/java-exec/src/test/resources/join/join_batchsize.json
new file mode 100644
index 0000000..a3be355
--- /dev/null
+++ b/exec/java-exec/src/test/resources/join/join_batchsize.json
@@ -0,0 +1,88 @@
+{
+ head:{
+ type:"APACHE_DRILL_PHYSICAL",
+ version:"1",
+ generator:{
+ type:"manual"
+ }
+ },
+ graph:[
+ {
+ @id:1,
+ pop:"mock-sub-scan",
+ url: "http://source1.apache.org",
+ entries:[
+ {records: 100, types: [
+ {name: "blue", type: "INT", mode: "REQUIRED"},
+ {name: "red", type: "INT", mode: "REQUIRED"},
+ {name: "green", type: "INT", mode: "REQUIRED"}
+ ]}
+ ]
+ },
+ {
+ pop : "sort",
+ @id : 2,
+ child : 1,
+ orderings : [ {
+ order : "ASC",
+ expr : "blue"
+ } ],
+ reverse : false
+ },
+ {
+ pop : "selection-vector-remover",
+ @id : 3,
+ child : 2
+ },
+ {
+ @id:4,
+ pop:"mock-sub-scan",
+ url: "http://source2.apache.org",
+ entries:[
+ {records: 2, types: [
+ {name: "blue1", type: "INT", mode: "REQUIRED"},
+ {name: "red1", type: "INT", mode: "REQUIRED"},
+ {name: "green1", type: "INT", mode: "REQUIRED"}
+ ]}
+ ]
+ },
+ {
+ pop : "sort",
+ @id : 5,
+ child : 4,
+ orderings : [ {
+ order : "ASC",
+ expr : "blue1"
+ } ],
+ reverse : false
+ },
+ {
+ pop : "selection-vector-remover",
+ @id : 6,
+ child : 5
+ },
+ {
+ @id: 7,
+ right: 6,
+ left: 3,
+ pop: "merge-join",
+ join-conditions: [ {relationship: "==", left: "blue", right: "blue1"} ]
+ },
+ {
+ pop : "limit",
+ @id : 8,
+ child : 7,
+ first : 0,
+ last : 100
+ }, {
+ pop : "selection-vector-remover",
+ @id : 9,
+ child : 8
+ },
+ {
+ @id: 10,
+ child: 9,
+ pop: "screen"
+ }
+ ]
+}