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/05/06 23:42:05 UTC
[05/15] DRILL-620: Memory consumption fixes
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionSenderRootExec.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionSenderRootExec.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionSenderRootExec.java
index 3e3157b..6048085 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionSenderRootExec.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionSenderRootExec.java
@@ -29,13 +29,17 @@ import org.apache.drill.exec.expr.ClassGenerator;
import org.apache.drill.exec.expr.CodeGenerator;
import org.apache.drill.exec.expr.ExpressionTreeMaterializer;
import org.apache.drill.exec.expr.TypeHelper;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.memory.OutOfMemoryException;
import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.OperatorContext;
import org.apache.drill.exec.physical.config.HashPartitionSender;
import org.apache.drill.exec.physical.impl.RootExec;
import org.apache.drill.exec.physical.impl.SendingAccountor;
import org.apache.drill.exec.physical.impl.filter.ReturnValueExpression;
import org.apache.drill.exec.proto.CoordinationProtos;
import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
+import org.apache.drill.exec.record.AbstractRecordBatch;
import org.apache.drill.exec.record.RecordBatch;
import org.apache.drill.exec.record.TypedFieldId;
import org.apache.drill.exec.record.VectorWrapper;
@@ -57,6 +61,7 @@ public class PartitionSenderRootExec implements RootExec {
private OutgoingRecordBatch[] outgoing;
private Partitioner partitioner;
private FragmentContext context;
+ private OperatorContext oContext;
private boolean ok = true;
private AtomicLong batchesSent = new AtomicLong(0);
private final SendingAccountor sendCount = new SendingAccountor();
@@ -64,11 +69,12 @@ public class PartitionSenderRootExec implements RootExec {
public PartitionSenderRootExec(FragmentContext context,
RecordBatch incoming,
- HashPartitionSender operator) {
+ HashPartitionSender operator) throws OutOfMemoryException {
this.incoming = incoming;
this.operator = operator;
this.context = context;
+ this.oContext = new OperatorContext(operator, context);
this.outgoing = new OutgoingRecordBatch[operator.getDestinations().size()];
int fieldId = 0;
for (CoordinationProtos.DrillbitEndpoint endpoint : operator.getDestinations()) {
@@ -77,6 +83,7 @@ public class PartitionSenderRootExec implements RootExec {
context.getDataTunnel(endpoint, opposite),
incoming,
context,
+ oContext.getAllocator(),
fieldId);
fieldId++;
}
@@ -252,16 +259,17 @@ public class PartitionSenderRootExec implements RootExec {
// ((IntVector) outgoingVectors[bucket][0]).copyFrom(inIndex,
// outgoingBatches[bucket].getRecordCount(),
// vv1);
- cg.getEvalBlock().add(
+ cg.getEvalBlock()._if(
((JExpression) JExpr.cast(vvClass,
((JExpression)
outgoingVectors
.component(bucket))
.component(JExpr.lit(fieldId))))
- .invoke("copyFrom")
+ .invoke("copyFromSafe")
.arg(inIndex)
.arg(((JExpression) outgoingBatches.component(bucket)).invoke("getRecordCount"))
- .arg(incomingVV));
+ .arg(incomingVV).not())._then().add(((JExpression) outgoingBatches.component(bucket)).invoke("flush"))
+ ._return();
++fieldId;
}
@@ -306,7 +314,8 @@ public class PartitionSenderRootExec implements RootExec {
for(OutgoingRecordBatch b : outgoing){
b.clear();
}
- incoming.cleanup();
sendCount.waitForSendComplete();
+ oContext.close();
+ incoming.cleanup();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectRecordBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectRecordBatch.java
index e8ee3cc..347092a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectRecordBatch.java
@@ -22,6 +22,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Set;
+import com.sun.codemodel.JExpr;
import org.apache.drill.common.expression.ErrorCollector;
import org.apache.drill.common.expression.ErrorCollectorImpl;
import org.apache.drill.common.expression.FieldReference;
@@ -34,13 +35,16 @@ import org.apache.drill.common.types.TypeProtos;
import org.apache.drill.exec.exception.ClassTransformationException;
import org.apache.drill.exec.exception.SchemaChangeException;
import org.apache.drill.exec.expr.ClassGenerator;
+import org.apache.drill.exec.expr.ClassGenerator.HoldingContainer;
import org.apache.drill.exec.expr.CodeGenerator;
import org.apache.drill.exec.expr.ExpressionTreeMaterializer;
import org.apache.drill.exec.expr.TypeHelper;
import org.apache.drill.exec.expr.ValueVectorReadExpression;
import org.apache.drill.exec.expr.ValueVectorWriteExpression;
+import org.apache.drill.exec.memory.OutOfMemoryException;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.config.Project;
+import org.apache.drill.exec.physical.impl.filter.ReturnValueExpression;
import org.apache.drill.exec.record.AbstractSingleRecordBatch;
import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
import org.apache.drill.exec.record.MaterializedField;
@@ -48,6 +52,7 @@ import org.apache.drill.exec.record.RecordBatch;
import org.apache.drill.exec.record.TransferPair;
import org.apache.drill.exec.record.TypedFieldId;
import org.apache.drill.exec.record.VectorWrapper;
+import org.apache.drill.exec.util.VectorUtil;
import org.apache.drill.exec.vector.AllocationHelper;
import org.apache.drill.exec.vector.ValueVector;
@@ -59,26 +64,81 @@ public class ProjectRecordBatch extends AbstractSingleRecordBatch<Project>{
private Projector projector;
private List<ValueVector> allocationVectors;
+ private boolean hasRemainder = false;
+ private int remainderIndex = 0;
+ private int recordCount;
- public ProjectRecordBatch(Project pop, RecordBatch incoming, FragmentContext context){
+ public ProjectRecordBatch(Project pop, RecordBatch incoming, FragmentContext context) throws OutOfMemoryException {
super(pop, context, incoming);
}
@Override
public int getRecordCount() {
- return incoming.getRecordCount();
+ return recordCount;
+ }
+
+ @Override
+ public IterOutcome next() {
+ if (hasRemainder) {
+ handleRemainder();
+ return IterOutcome.OK;
+ }
+ return super.next();
}
@Override
protected void doWork() {
- int recordCount = incoming.getRecordCount();
+// VectorUtil.showVectorAccessibleContent(incoming, ",");
+ int incomingRecordCount = incoming.getRecordCount();
for(ValueVector v : this.allocationVectors){
- AllocationHelper.allocate(v, recordCount, 250);
+ AllocationHelper.allocate(v, incomingRecordCount, 250);
}
- projector.projectRecords(recordCount, 0);
- for(VectorWrapper<?> v : container){
- ValueVector.Mutator m = v.getValueVector().getMutator();
- m.setValueCount(recordCount);
+ int outputRecords = projector.projectRecords(0, incomingRecordCount, 0);
+ if (outputRecords < incomingRecordCount) {
+ for(VectorWrapper<?> v : container){
+ ValueVector.Mutator m = v.getValueVector().getMutator();
+ m.setValueCount(outputRecords);
+ }
+ hasRemainder = true;
+ remainderIndex = outputRecords;
+ this.recordCount = remainderIndex;
+ } else {
+ for(VectorWrapper<?> v : container){
+ ValueVector.Mutator m = v.getValueVector().getMutator();
+ m.setValueCount(incomingRecordCount);
+ }
+ for(VectorWrapper<?> v: incoming) {
+ v.clear();
+ }
+ this.recordCount = outputRecords;
+ }
+ }
+
+ private void handleRemainder() {
+ int remainingRecordCount = incoming.getRecordCount() - remainderIndex;
+ for(ValueVector v : this.allocationVectors){
+ AllocationHelper.allocate(v, remainingRecordCount, 250);
+ }
+ int outputIndex = projector.projectRecords(remainderIndex, remainingRecordCount, 0);
+ if (outputIndex < incoming.getRecordCount()) {
+ for(VectorWrapper<?> v : container){
+ ValueVector.Mutator m = v.getValueVector().getMutator();
+ m.setValueCount(outputIndex - remainderIndex);
+ }
+ hasRemainder = true;
+ this.recordCount = outputIndex - remainderIndex;
+ remainderIndex = outputIndex;
+ } else {
+ for(VectorWrapper<?> v : container){
+ ValueVector.Mutator m = v.getValueVector().getMutator();
+ m.setValueCount(remainingRecordCount);
+ }
+ hasRemainder = false;
+ remainderIndex = 0;
+ for(VectorWrapper<?> v: incoming) {
+ v.clear();
+ }
+ this.recordCount = remainingRecordCount;
}
}
@@ -156,17 +216,20 @@ public class ProjectRecordBatch extends AbstractSingleRecordBatch<Project>{
// logger.debug("Added transfer.");
}else{
// need to do evaluation.
- ValueVector vector = TypeHelper.getNewVector(outputField, context.getAllocator());
+ ValueVector vector = TypeHelper.getNewVector(outputField, oContext.getAllocator());
allocationVectors.add(vector);
TypedFieldId fid = container.add(vector);
- ValueVectorWriteExpression write = new ValueVectorWriteExpression(fid, expr);
- cg.addExpr(write);
-// logger.debug("Added eval.");
+ ValueVectorWriteExpression write = new ValueVectorWriteExpression(fid, expr, true);
+ HoldingContainer hc = cg.addExpr(write);
+ cg.getEvalBlock()._if(hc.getValue().eq(JExpr.lit(0)))._then()._return(JExpr.FALSE);
+ logger.debug("Added eval.");
}
- }
+ }
}
+ cg.rotateBlock();
+ cg.getEvalBlock()._return(JExpr.TRUE);
container.buildSchema(incoming.getSchema().getSelectionVectorMode());
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/Projector.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/Projector.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/Projector.java
index 2857fe1..ebfce41 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/Projector.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/Projector.java
@@ -28,7 +28,7 @@ import org.apache.drill.exec.record.TransferPair;
public interface Projector {
public abstract void setup(FragmentContext context, RecordBatch incoming, RecordBatch outgoing, List<TransferPair> transfers) throws SchemaChangeException;
- public abstract int projectRecords(int recordCount, int firstOutputIndex);
+ public abstract int projectRecords(int startIndex, int recordCount, int firstOutputIndex);
public static TemplateClassDefinition<Projector> TEMPLATE_DEFINITION = new TemplateClassDefinition<Projector>(Projector.class, ProjectorTemplate.class);
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectorTemplate.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectorTemplate.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectorTemplate.java
index bd26ce4..60e5993 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectorTemplate.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectorTemplate.java
@@ -43,7 +43,7 @@ public abstract class ProjectorTemplate implements Projector {
}
@Override
- public final int projectRecords(final int recordCount, int firstOutputIndex) {
+ public final int projectRecords(int startIndex, final int recordCount, int firstOutputIndex) {
switch(svMode){
case FOUR_BYTE:
throw new UnsupportedOperationException();
@@ -60,8 +60,17 @@ public abstract class ProjectorTemplate implements Projector {
case NONE:
final int countN = recordCount;
- for (int i = 0; i < countN; i++, firstOutputIndex++) {
- doEval(i, firstOutputIndex);
+ int i;
+ for (i = startIndex; i < startIndex + countN; i++, firstOutputIndex++) {
+ if (!doEval(i, firstOutputIndex)) {
+ break;
+ }
+ }
+ if (i < recordCount || startIndex > 0) {
+ for(TransferPair t : transfers){
+ t.splitAndTransfer(startIndex, i - startIndex);
+ }
+ return i;
}
for(TransferPair t : transfers){
t.transfer();
@@ -91,7 +100,7 @@ public abstract class ProjectorTemplate implements Projector {
}
public abstract void doSetup(@Named("context") FragmentContext context, @Named("incoming") RecordBatch incoming, @Named("outgoing") RecordBatch outgoing);
- public abstract void doEval(@Named("inIndex") int inIndex, @Named("outIndex") int outIndex);
+ public abstract boolean doEval(@Named("inIndex") int inIndex, @Named("outIndex") int outIndex);
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/sort/SortBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/sort/SortBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/sort/SortBatch.java
index 29e629a..375276e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/sort/SortBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/sort/SortBatch.java
@@ -32,6 +32,8 @@ import org.apache.drill.exec.expr.ClassGenerator.HoldingContainer;
import org.apache.drill.exec.expr.CodeGenerator;
import org.apache.drill.exec.expr.ExpressionTreeMaterializer;
import org.apache.drill.exec.expr.fn.FunctionGenerationHelper;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.memory.OutOfMemoryException;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.config.Sort;
import org.apache.drill.exec.record.AbstractRecordBatch;
@@ -60,10 +62,10 @@ public class SortBatch extends AbstractRecordBatch<Sort> {
private Sorter sorter;
private BatchSchema schema;
- public SortBatch(Sort popConfig, FragmentContext context, RecordBatch incoming) {
+ public SortBatch(Sort popConfig, FragmentContext context, RecordBatch incoming) throws OutOfMemoryException {
super(popConfig, context);
this.incoming = incoming;
- this.builder = new SortRecordBatchBuilder(context.getAllocator(), MAX_SORT_BYTES);
+ this.builder = new SortRecordBatchBuilder(oContext.getAllocator(), MAX_SORT_BYTES);
}
@Override
@@ -74,7 +76,6 @@ public class SortBatch extends AbstractRecordBatch<Sort> {
@Override
public void kill() {
incoming.kill();
- cleanup();
}
@Override
@@ -91,9 +92,9 @@ public class SortBatch extends AbstractRecordBatch<Sort> {
@Override
public void cleanup() {
+ builder.clear();
super.cleanup();
incoming.cleanup();
- builder.clear();
}
@Override
@@ -116,7 +117,6 @@ public class SortBatch extends AbstractRecordBatch<Sort> {
case NOT_YET:
throw new UnsupportedOperationException();
case STOP:
- cleanup();
return upstream;
case OK_NEW_SCHEMA:
// only change in the case that the schema truly changes. Artificial schema changes are ignored.
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/Copier.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/Copier.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/Copier.java
index dad8858..0aab7b2 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/Copier.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/Copier.java
@@ -28,7 +28,7 @@ public interface Copier {
public static TemplateClassDefinition<Copier> TEMPLATE_DEFINITION4 = new TemplateClassDefinition<Copier>(Copier.class, CopierTemplate4.class);
public void setupRemover(FragmentContext context, RecordBatch incoming, RecordBatch outgoing, VectorAllocator[] allocators) throws SchemaChangeException;
- public abstract void copyRecords();
+ public abstract int copyRecords(int index, int recordCount);
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/CopierTemplate2.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/CopierTemplate2.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/CopierTemplate2.java
index 6d12731..2f589a5 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/CopierTemplate2.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/CopierTemplate2.java
@@ -48,22 +48,20 @@ public abstract class CopierTemplate2 implements Copier{
}
@Override
- public void copyRecords(){
- final int recordCount = sv2.getCount();
+ public int copyRecords(int index, int recordCount){
allocateVectors(recordCount);
int outgoingPosition = 0;
- for(int svIndex = 0; svIndex < recordCount; svIndex++, outgoingPosition++){
- doEval(sv2.getIndex(svIndex), outgoingPosition);
- }
-// logger.debug("This: {}, Incoming: {}", System.identityHashCode(this), incoming);
- for(VectorWrapper<?> v : incoming){
- v.clear();
+ for(int svIndex = index; svIndex < index + recordCount; svIndex++, outgoingPosition++){
+ if (!doEval(sv2.getIndex(svIndex), outgoingPosition)) {
+ break;
+ }
}
+ return outgoingPosition;
}
public abstract void doSetup(@Named("context") FragmentContext context, @Named("incoming") RecordBatch incoming, @Named("outgoing") RecordBatch outgoing);
- public abstract void doEval(@Named("inIndex") int inIndex, @Named("outIndex") int outIndex);
+ public abstract boolean doEval(@Named("inIndex") int inIndex, @Named("outIndex") int outIndex);
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/CopierTemplate4.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/CopierTemplate4.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/CopierTemplate4.java
index 9f4ae7e..a7aba6e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/CopierTemplate4.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/CopierTemplate4.java
@@ -50,24 +50,21 @@ public abstract class CopierTemplate4 implements Copier{
@Override
- public void copyRecords(){
+ public int copyRecords(int index, int recordCount){
// logger.debug("Copying records.");
- final int recordCount = sv4.getCount();
allocateVectors(recordCount);
int outgoingPosition = 0;
- for(int svIndex = 0; svIndex < sv4.getCount(); svIndex++, outgoingPosition++){
+ for(int svIndex = index; svIndex < index + recordCount; svIndex++, outgoingPosition++){
int deRefIndex = sv4.get(svIndex);
- doEval(deRefIndex, outgoingPosition);
+ if (!doEval(deRefIndex, outgoingPosition)) {
+ break;
+ }
}
-
-// for(VectorWrapper<?> v : incoming){
-// v.clear();
-// }
-
+ return outgoingPosition;
}
public abstract void doSetup(@Named("context") FragmentContext context, @Named("incoming") RecordBatch incoming, @Named("outgoing") RecordBatch outgoing);
- public abstract void doEval(@Named("inIndex") int inIndex, @Named("outIndex") int outIndex);
+ public abstract boolean doEval(@Named("inIndex") int inIndex, @Named("outIndex") int outIndex);
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/RemovingRecordBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/RemovingRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/RemovingRecordBatch.java
index 499f4d1..4018991 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/RemovingRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/RemovingRecordBatch.java
@@ -25,6 +25,8 @@ import org.apache.drill.exec.exception.SchemaChangeException;
import org.apache.drill.exec.expr.ClassGenerator;
import org.apache.drill.exec.expr.CodeGenerator;
import org.apache.drill.exec.expr.TypeHelper;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.memory.OutOfMemoryException;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.config.SelectionVectorRemover;
import org.apache.drill.exec.record.*;
@@ -54,8 +56,10 @@ public class RemovingRecordBatch extends AbstractSingleRecordBatch<SelectionVect
private Copier copier;
private int recordCount;
-
- public RemovingRecordBatch(SelectionVectorRemover popConfig, FragmentContext context, RecordBatch incoming) {
+ private boolean hasRemainder;
+ private int remainderIndex;
+
+ public RemovingRecordBatch(SelectionVectorRemover popConfig, FragmentContext context, RecordBatch incoming) throws OutOfMemoryException {
super(popConfig, context, incoming);
logger.debug("Created.");
}
@@ -88,12 +92,64 @@ public class RemovingRecordBatch extends AbstractSingleRecordBatch<SelectionVect
}
@Override
+ public IterOutcome next() {
+ if (hasRemainder) {
+ handleRemainder();
+ return IterOutcome.OK;
+ }
+ return super.next();
+ }
+
+ @Override
protected void doWork() {
recordCount = incoming.getRecordCount();
- copier.copyRecords();
- for(VectorWrapper<?> v : container){
- ValueVector.Mutator m = v.getValueVector().getMutator();
- m.setValueCount(recordCount);
+ int copiedRecords = copier.copyRecords(0, recordCount);
+ if (copiedRecords < recordCount) {
+ for(VectorWrapper<?> v : container){
+ ValueVector.Mutator m = v.getValueVector().getMutator();
+ m.setValueCount(copiedRecords);
+ }
+ hasRemainder = true;
+ remainderIndex = copiedRecords;
+ this.recordCount = remainderIndex;
+ } else {
+ for(VectorWrapper<?> v : container){
+ ValueVector.Mutator m = v.getValueVector().getMutator();
+ m.setValueCount(recordCount);
+ }
+ if (incoming.getSchema().getSelectionVectorMode() != SelectionVectorMode.FOUR_BYTE) {
+ for(VectorWrapper<?> v: incoming) {
+ v.clear();
+ }
+ if (incoming.getSchema().getSelectionVectorMode() == SelectionVectorMode.TWO_BYTE) {
+ incoming.getSelectionVector2().clear();
+ }
+ }
+ }
+ }
+
+ private void handleRemainder() {
+ int remainingRecordCount = incoming.getRecordCount() - remainderIndex;
+ int copiedRecords = copier.copyRecords(0, recordCount);
+ if (copiedRecords < remainingRecordCount) {
+ for(VectorWrapper<?> v : container){
+ ValueVector.Mutator m = v.getValueVector().getMutator();
+ m.setValueCount(copiedRecords);
+ }
+ remainderIndex += copiedRecords;
+ this.recordCount = copiedRecords;
+ } else {
+ for(VectorWrapper<?> v : container){
+ ValueVector.Mutator m = v.getValueVector().getMutator();
+ m.setValueCount(remainingRecordCount);
+ }
+ if (incoming.getSchema().getSelectionVectorMode() != SelectionVectorMode.FOUR_BYTE) {
+ for(VectorWrapper<?> v: incoming) {
+ v.clear();
+ }
+ }
+ remainderIndex = 0;
+ hasRemainder = false;
}
}
@@ -116,10 +172,12 @@ public class RemovingRecordBatch extends AbstractSingleRecordBatch<SelectionVect
}
@Override
- public void copyRecords() {
+ public int copyRecords(int index, int recordCount) {
+ assert index == 0 && recordCount == incoming.getRecordCount() : "Straight copier cannot split batch";
for(TransferPair tp : pairs){
tp.transfer();
}
+ return recordCount;
}
public List<ValueVector> getOut() {
@@ -140,7 +198,7 @@ public class RemovingRecordBatch extends AbstractSingleRecordBatch<SelectionVect
List<VectorAllocator> allocators = Lists.newArrayList();
for(VectorWrapper<?> i : incoming){
- ValueVector v = TypeHelper.getNewVector(i.getField(), context.getAllocator());
+ ValueVector v = TypeHelper.getNewVector(i.getField(), oContext.getAllocator());
container.add(v);
allocators.add(VectorAllocator.getAllocator(i.getValueVector(), v));
}
@@ -158,15 +216,15 @@ public class RemovingRecordBatch extends AbstractSingleRecordBatch<SelectionVect
private Copier getGenerated4Copier() throws SchemaChangeException {
Preconditions.checkArgument(incoming.getSchema().getSelectionVectorMode() == SelectionVectorMode.FOUR_BYTE);
- return getGenerated4Copier(incoming, context, container, this);
+ return getGenerated4Copier(incoming, context, oContext.getAllocator(), container, this);
}
- public static Copier getGenerated4Copier(RecordBatch batch, FragmentContext context, VectorContainer container, RecordBatch outgoing) throws SchemaChangeException{
+ public static Copier getGenerated4Copier(RecordBatch batch, FragmentContext context, BufferAllocator allocator, VectorContainer container, RecordBatch outgoing) throws SchemaChangeException{
List<VectorAllocator> allocators = Lists.newArrayList();
for(VectorWrapper<?> i : batch){
- ValueVector v = TypeHelper.getNewVector(i.getField(), context.getAllocator());
+ ValueVector v = TypeHelper.getNewVector(i.getField(), allocator);
container.add(v);
allocators.add(getAllocator4(v));
}
@@ -195,23 +253,27 @@ public class RemovingRecordBatch extends AbstractSingleRecordBatch<SelectionVect
if(hyper){
- g.getEvalBlock().add(
+ g.getEvalBlock()._if(
outVV
- .invoke("copyFrom")
+ .invoke("copyFromSafe")
.arg(
inIndex.band(JExpr.lit((int) Character.MAX_VALUE)))
.arg(outIndex)
.arg(
inVV.component(inIndex.shrz(JExpr.lit(16)))
)
- );
+ .not()
+ )
+ ._then()._return(JExpr.FALSE);
}else{
- g.getEvalBlock().add(outVV.invoke("copyFrom").arg(inIndex).arg(outIndex).arg(inVV));
+ g.getEvalBlock()._if(outVV.invoke("copyFromSafe").arg(inIndex).arg(outIndex).arg(inVV).not())._then()._return(JExpr.FALSE);
}
fieldId++;
}
+ g.rotateBlock();
+ g.getEvalBlock()._return(JExpr.TRUE);
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/trace/TraceRecordBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/trace/TraceRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/trace/TraceRecordBatch.java
index 00c3d2f..b012cec 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/trace/TraceRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/trace/TraceRecordBatch.java
@@ -111,7 +111,7 @@ public class TraceRecordBatch extends AbstractSingleRecordBatch<Trace> {
}
WritableBatch batch = WritableBatch.getBatchNoHVWrap(incoming.getRecordCount(), incoming, incomingHasSv2 ? true
: false);
- VectorAccessibleSerializable wrap = new VectorAccessibleSerializable(batch, sv, context.getAllocator());
+ VectorAccessibleSerializable wrap = new VectorAccessibleSerializable(batch, sv, oContext.getAllocator());
try {
wrap.writeToStreamAndRetain(fos);
@@ -119,6 +119,9 @@ public class TraceRecordBatch extends AbstractSingleRecordBatch<Trace> {
throw new RuntimeException(e);
}
batch.reconstructContainer(container);
+ if (incomingHasSv2) {
+ sv = wrap.getSv2();
+ }
}
@Override
@@ -161,6 +164,7 @@ public class TraceRecordBatch extends AbstractSingleRecordBatch<Trace> {
logger.error("Unable to close file descriptors for file: " + getFileName());
}
super.cleanup();
+ incoming.cleanup();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/union/UnionRecordBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/union/UnionRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/union/UnionRecordBatch.java
index 0fc5e0f..c27b3c8 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/union/UnionRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/union/UnionRecordBatch.java
@@ -18,7 +18,9 @@
package org.apache.drill.exec.physical.impl.union;
import com.google.common.collect.Lists;
+import org.apache.drill.exec.memory.OutOfMemoryException;
import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.OperatorContext;
import org.apache.drill.exec.physical.config.Union;
import org.apache.drill.exec.record.*;
import org.apache.drill.exec.record.selection.SelectionVector2;
@@ -39,7 +41,7 @@ public class UnionRecordBatch extends AbstractRecordBatch<Union> {
private ArrayList<TransferPair> transfers;
private int outRecordCount;
- public UnionRecordBatch(Union config, List<RecordBatch> children, FragmentContext context) {
+ public UnionRecordBatch(Union config, List<RecordBatch> children, FragmentContext context) throws OutOfMemoryException {
super(config, context);
this.incoming = children;
this.incomingIterator = incoming.iterator();
@@ -47,7 +49,6 @@ public class UnionRecordBatch extends AbstractRecordBatch<Union> {
sv = null;
}
-
@Override
public int getRecordCount() {
return outRecordCount;
@@ -119,10 +120,10 @@ public class UnionRecordBatch extends AbstractRecordBatch<Union> {
transfer.transfer();
}
- for (VectorWrapper<?> vw : this.container) {
- ValueVector.Mutator m = vw.getValueVector().getMutator();
- m.setValueCount(outRecordCount);
- }
+// for (VectorWrapper<?> vw : this.container) {
+// ValueVector.Mutator m = vw.getValueVector().getMutator();
+// m.setValueCount(outRecordCount);
+// }
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/BatchGroup.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/BatchGroup.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/BatchGroup.java
index 20540dd..d87a9f5 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/BatchGroup.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/BatchGroup.java
@@ -88,7 +88,7 @@ public class BatchGroup implements VectorAccessible {
watch.start();
outputBatch.writeToStream(outputStream);
newContainer.zeroVectors();
- logger.debug("Took {} us to spill {} records", watch.elapsed(TimeUnit.MICROSECONDS), recordCount);
+// logger.debug("Took {} us to spill {} records", watch.elapsed(TimeUnit.MICROSECONDS), recordCount);
spilledBatches++;
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/ExternalSortBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/ExternalSortBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/ExternalSortBatch.java
index 8bb3d43..930f851 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/ExternalSortBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/ExternalSortBatch.java
@@ -38,6 +38,7 @@ import org.apache.drill.exec.expr.ExpressionTreeMaterializer;
import org.apache.drill.exec.expr.TypeHelper;
import org.apache.drill.exec.expr.fn.FunctionGenerationHelper;
import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.memory.OutOfMemoryException;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.config.ExternalSort;
import org.apache.drill.exec.physical.impl.sort.RecordBatchData;
@@ -82,6 +83,7 @@ public class ExternalSortBatch extends AbstractRecordBatch<ExternalSort> {
private MSorter mSorter;
private PriorityQueueSelector selector;
private PriorityQueueCopier copier;
+ private BufferAllocator copierAllocator;
private LinkedList<BatchGroup> batchGroups = Lists.newLinkedList();
private SelectionVector4 sv4;
private FileSystem fs;
@@ -89,7 +91,7 @@ public class ExternalSortBatch extends AbstractRecordBatch<ExternalSort> {
private int batchesSinceLastSpill = 0;
private long uid;//used for spill files to ensure multiple sorts within same fragment don't clobber each others' files
- public ExternalSortBatch(ExternalSort popConfig, FragmentContext context, RecordBatch incoming) {
+ public ExternalSortBatch(ExternalSort popConfig, FragmentContext context, RecordBatch incoming) throws OutOfMemoryException {
super(popConfig, context);
this.incoming = incoming;
DrillConfig config = context.getConfig();
@@ -107,6 +109,7 @@ public class ExternalSortBatch extends AbstractRecordBatch<ExternalSort> {
SPILL_DIRECTORIES = config.getStringList(ExecConstants.EXTERNAL_SORT_SPILL_DIRS);
dirs = Iterators.cycle(Lists.newArrayList(SPILL_DIRECTORIES));
uid = System.nanoTime();
+ copierAllocator = oContext.getAllocator().getChildAllocator(context.getHandle(), 10000000, 20000000);
}
@Override
@@ -117,7 +120,6 @@ public class ExternalSortBatch extends AbstractRecordBatch<ExternalSort> {
@Override
public void kill() {
incoming.kill();
- cleanup();
}
@Override
@@ -134,8 +136,6 @@ public class ExternalSortBatch extends AbstractRecordBatch<ExternalSort> {
@Override
public void cleanup() {
- super.cleanup();
- incoming.cleanup();
if (batchGroups != null) {
for (BatchGroup group: batchGroups) {
try {
@@ -151,6 +151,9 @@ public class ExternalSortBatch extends AbstractRecordBatch<ExternalSort> {
if (sv4 != null) {
sv4.clear();
}
+ copierAllocator.close();
+ super.cleanup();
+ incoming.cleanup();
}
@Override
@@ -170,11 +173,11 @@ public class ExternalSortBatch extends AbstractRecordBatch<ExternalSort> {
int count = selector.next();
if(count > 0){
long t = w.elapsed(TimeUnit.MICROSECONDS);
-// logger.debug("Took {} us to merge {} records", t, count);
+ logger.debug("Took {} us to merge {} records", t, count);
container.setRecordCount(count);
return IterOutcome.OK;
}else{
- cleanup();
+ logger.debug("selector returned 0 records");
return IterOutcome.NONE;
}
}
@@ -192,7 +195,6 @@ public class ExternalSortBatch extends AbstractRecordBatch<ExternalSort> {
case NOT_YET:
throw new UnsupportedOperationException();
case STOP:
- cleanup();
return upstream;
case OK_NEW_SCHEMA:
// only change in the case that the schema truly changes. Artificial schema changes are ignored.
@@ -207,7 +209,11 @@ public class ExternalSortBatch extends AbstractRecordBatch<ExternalSort> {
if (incoming.getSchema().getSelectionVectorMode() == BatchSchema.SelectionVectorMode.TWO_BYTE) {
sv2 = incoming.getSelectionVector2();
} else {
- sv2 = newSV2();
+ try {
+ sv2 = newSV2();
+ } catch (OutOfMemoryException e) {
+ throw new RuntimeException();
+ }
}
int count = sv2.getCount();
assert sv2.getCount() > 0;
@@ -225,6 +231,10 @@ public class ExternalSortBatch extends AbstractRecordBatch<ExternalSort> {
long t = w.elapsed(TimeUnit.MICROSECONDS);
// logger.debug("Took {} us to sort {} records", t, count);
break;
+ case OUT_OF_MEMORY:
+ mergeAndSpill();
+ batchesSinceLastSpill = 0;
+ break;
default:
throw new UnsupportedOperationException();
}
@@ -243,7 +253,7 @@ public class ExternalSortBatch extends AbstractRecordBatch<ExternalSort> {
return IterOutcome.NONE;
}
- builder = new SortRecordBatchBuilder(context.getAllocator(), MAX_SORT_BYTES);
+ builder = new SortRecordBatchBuilder(oContext.getAllocator(), MAX_SORT_BYTES);
for (BatchGroup group : batchGroups) {
RecordBatchData rbd = new RecordBatchData(group.getFirstContainer());
@@ -254,7 +264,7 @@ public class ExternalSortBatch extends AbstractRecordBatch<ExternalSort> {
builder.build(context, container);
sv4 = builder.getSv4();
mSorter = createNewMSorter();
- mSorter.setup(context, getSelectionVector4(), this.container);
+ mSorter.setup(context, oContext.getAllocator(), getSelectionVector4(), this.container);
mSorter.sort(this.container);
sv4 = mSorter.getSV4();
@@ -265,7 +275,7 @@ public class ExternalSortBatch extends AbstractRecordBatch<ExternalSort> {
constructHyperBatch(batchGroups, this.container);
constructSV4();
selector = createSelector();
- selector.setup(context, this, sv4, batchGroups);
+ selector.setup(context, oContext.getAllocator(), this, sv4, batchGroups);
selector.next();
}
@@ -284,12 +294,22 @@ public class ExternalSortBatch extends AbstractRecordBatch<ExternalSort> {
}
public void mergeAndSpill() throws SchemaChangeException {
+ logger.debug("Copier allocator current allocation {}", copierAllocator.getAllocatedMemory());
VectorContainer hyperBatch = new VectorContainer();
VectorContainer outputContainer = new VectorContainer();
List<BatchGroup> batchGroupList = Lists.newArrayList();
for (int i = 0; i < SPILL_BATCH_GROUP_SIZE; i++) {
+ if (batchGroups.size() == 0) {
+ break;
+ }
+ if (batchGroups.peekLast().getSecondContainer() != null) {
+ break;
+ }
batchGroupList.add(batchGroups.pollLast());
}
+ if (batchGroupList.size() == 0) {
+ return;
+ }
constructHyperBatch(batchGroupList, hyperBatch);
createCopier(hyperBatch, batchGroupList, outputContainer);
@@ -309,7 +329,7 @@ public class ExternalSortBatch extends AbstractRecordBatch<ExternalSort> {
c2.setRecordCount(count);
String outputFile = String.format(Utilities.getFileNameForQueryFragment(context, dirs.next(), "spill" + uid + "_" + spillCount++));
- BatchGroup newGroup = new BatchGroup(c1, c2, fs, outputFile, context.getAllocator());
+ BatchGroup newGroup = new BatchGroup(c1, c2, fs, outputFile, oContext.getAllocator());
try {
while ((count = copier.next()) > 0) {
@@ -328,9 +348,19 @@ public class ExternalSortBatch extends AbstractRecordBatch<ExternalSort> {
}
}
- private SelectionVector2 newSV2() {
- SelectionVector2 sv2 = new SelectionVector2(context.getAllocator());
- sv2.allocateNew(incoming.getRecordCount());
+ private SelectionVector2 newSV2() throws OutOfMemoryException {
+ SelectionVector2 sv2 = new SelectionVector2(oContext.getAllocator());
+ if (!sv2.allocateNew(incoming.getRecordCount())) {
+ try {
+ mergeAndSpill();
+ } catch (SchemaChangeException e) {
+ throw new RuntimeException();
+ }
+ batchesSinceLastSpill = 0;
+ if (!sv2.allocateNew(incoming.getRecordCount())) {
+ throw new OutOfMemoryException();
+ }
+ }
for (int i = 0; i < incoming.getRecordCount(); i++) {
sv2.setIndex(i, (char) i);
}
@@ -360,7 +390,7 @@ public class ExternalSortBatch extends AbstractRecordBatch<ExternalSort> {
}
private void constructSV4() throws SchemaChangeException {
- BufferAllocator.PreAllocator preAlloc = context.getAllocator().getNewPreAllocator();
+ BufferAllocator.PreAllocator preAlloc = oContext.getAllocator().getNewPreAllocator();
preAlloc.preAllocate(4 * TARGET_RECORD_COUNT);
sv4 = new SelectionVector4(preAlloc.getAllocation(), TARGET_RECORD_COUNT, TARGET_RECORD_COUNT);
}
@@ -476,11 +506,11 @@ public class ExternalSortBatch extends AbstractRecordBatch<ExternalSort> {
List<VectorAllocator> allocators = Lists.newArrayList();
for(VectorWrapper<?> i : batch){
- ValueVector v = TypeHelper.getNewVector(i.getField(), context.getAllocator());
+ ValueVector v = TypeHelper.getNewVector(i.getField(), copierAllocator);
outputContainer.add(v);
allocators.add(VectorAllocator.getAllocator(v, 110));
}
- copier.setup(context, batch, batchGroupList, outputContainer, allocators);
+ copier.setup(context, copierAllocator, batch, batchGroupList, outputContainer, allocators);
} catch (ClassTransformationException e) {
throw new RuntimeException(e);
} catch (IOException e) {
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/MSortTemplate.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/MSortTemplate.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/MSortTemplate.java
index e87774b..c54b2b7 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/MSortTemplate.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/MSortTemplate.java
@@ -33,15 +33,18 @@ import java.util.Queue;
public abstract class MSortTemplate implements MSorter, IndexedSortable{
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(MSortTemplate.class);
-
+
+ private BufferAllocator allocator;
private SelectionVector4 vector4;
private SelectionVector4 aux;
private long compares;
private Queue<Integer> runStarts = Queues.newLinkedBlockingQueue();
private Queue<Integer> newRunStarts;
-
- public void setup(FragmentContext context, SelectionVector4 vector4, VectorContainer hyperBatch) throws SchemaChangeException{
+
+ @Override
+ public void setup(FragmentContext context, BufferAllocator allocator, SelectionVector4 vector4, VectorContainer hyperBatch) throws SchemaChangeException{
+ this.allocator = allocator;
// we pass in the local hyperBatch since that is where we'll be reading data.
Preconditions.checkNotNull(vector4);
this.vector4 = vector4.createNewWrapperCurrent();
@@ -60,7 +63,7 @@ public abstract class MSortTemplate implements MSorter, IndexedSortable{
throw new UnsupportedOperationException("Missing batch");
}
}
- BufferAllocator.PreAllocator preAlloc = context.getAllocator().getNewPreAllocator();
+ BufferAllocator.PreAllocator preAlloc = allocator.getNewPreAllocator();
preAlloc.preAllocate(4 * this.vector4.getTotalCount());
aux = new SelectionVector4(preAlloc.getAllocation(), this.vector4.getTotalCount(), Character.MAX_VALUE);
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/MSorter.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/MSorter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/MSorter.java
index 6ad4e3d..1300830 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/MSorter.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/MSorter.java
@@ -19,12 +19,13 @@ package org.apache.drill.exec.physical.impl.xsort;
import org.apache.drill.exec.compile.TemplateClassDefinition;
import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.record.VectorContainer;
import org.apache.drill.exec.record.selection.SelectionVector4;
public interface MSorter {
- public void setup(FragmentContext context, SelectionVector4 vector4, VectorContainer hyperBatch) throws SchemaChangeException;
+ public void setup(FragmentContext context, BufferAllocator allocator, SelectionVector4 vector4, VectorContainer hyperBatch) throws SchemaChangeException;
public void sort(VectorContainer container);
public SelectionVector4 getSV4();
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/PriorityQueueCopier.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/PriorityQueueCopier.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/PriorityQueueCopier.java
index b31e287..7122963 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/PriorityQueueCopier.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/PriorityQueueCopier.java
@@ -19,6 +19,7 @@ package org.apache.drill.exec.physical.impl.xsort;
import org.apache.drill.exec.compile.TemplateClassDefinition;
import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.impl.sort.RecordBatchData;
import org.apache.drill.exec.physical.impl.svremover.Copier;
@@ -31,7 +32,7 @@ import org.apache.drill.exec.vector.allocator.VectorAllocator;
import java.util.List;
public interface PriorityQueueCopier {
- public void setup(FragmentContext context, VectorAccessible hyperBatch, List<BatchGroup> batchGroups, VectorAccessible outgoing, List<VectorAllocator> allocators) throws SchemaChangeException;
+ public void setup(FragmentContext context, BufferAllocator allocator, VectorAccessible hyperBatch, List<BatchGroup> batchGroups, VectorAccessible outgoing, List<VectorAllocator> allocators) throws SchemaChangeException;
public int next();
public List<VectorAllocator> getAllocators();
public void cleanup();
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/PriorityQueueCopierTemplate.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/PriorityQueueCopierTemplate.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/PriorityQueueCopierTemplate.java
index bc2c19c..4221ae2 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/PriorityQueueCopierTemplate.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/PriorityQueueCopierTemplate.java
@@ -42,6 +42,7 @@ public abstract class PriorityQueueCopierTemplate implements PriorityQueueCopier
private List<BatchGroup> batchGroups;
private VectorAccessible hyperBatch;
private FragmentContext context;
+ private BufferAllocator allocator;
private VectorAccessible outgoing;
private List<VectorAllocator> allocators;
private int size;
@@ -49,19 +50,21 @@ public abstract class PriorityQueueCopierTemplate implements PriorityQueueCopier
private int targetRecordCount = ExternalSortBatch.SPILL_TARGET_RECORD_COUNT;
@Override
- public void setup(FragmentContext context, VectorAccessible hyperBatch, List<BatchGroup> batchGroups, VectorAccessible outgoing, List<VectorAllocator> allocators) throws SchemaChangeException {
+ public void setup(FragmentContext context, BufferAllocator allocator, VectorAccessible hyperBatch, List<BatchGroup> batchGroups, VectorAccessible outgoing, List<VectorAllocator> allocators) throws SchemaChangeException {
this.context = context;
+ this.allocator = allocator;
this.hyperBatch = hyperBatch;
this.batchGroups = batchGroups;
this.outgoing = outgoing;
this.allocators = allocators;
this.size = batchGroups.size();
- BufferAllocator.PreAllocator preAlloc = context.getAllocator().getNewPreAllocator();
+ BufferAllocator.PreAllocator preAlloc = allocator.getNewPreAllocator();
preAlloc.preAllocate(4 * size);
vector4 = new SelectionVector4(preAlloc.getAllocation(), size, Character.MAX_VALUE);
doSetup(context, hyperBatch, outgoing);
+ queueSize = 0;
for (int i = 0; i < size; i++) {
vector4.set(i, i * 2, batchGroups.get(i).getNextIndex());
siftUp();
@@ -79,10 +82,14 @@ public abstract class PriorityQueueCopierTemplate implements PriorityQueueCopier
}
int compoundIndex = vector4.get(0);
int batch = compoundIndex >>> 16;
- assert batch < batchGroups.size() * 2;
+ assert batch < batchGroups.size() * 2 : String.format("batch: %d batchGroups: %d", batch, batchGroups.size());
int batchGroup = batch / 2;
- doCopy(compoundIndex, outgoingIndex);
+ if (!doCopy(compoundIndex, outgoingIndex)) {
+ setValueCount(outgoingIndex);
+ return outgoingIndex;
+ }
int nextIndex = batchGroups.get(batchGroup).getNextIndex();
+ batch = batch & 0xFFFE;
batch += batchGroups.get(batchGroup).getBatchPointer();
if (nextIndex < 0) {
vector4.set(0, vector4.get(--queueSize));
@@ -172,6 +179,6 @@ public abstract class PriorityQueueCopierTemplate implements PriorityQueueCopier
public abstract void doSetup(@Named("context") FragmentContext context, @Named("incoming") VectorAccessible incoming, @Named("outgoing") VectorAccessible outgoing);
public abstract int doEval(@Named("leftIndex") int leftIndex, @Named("rightIndex") int rightIndex);
- public abstract void doCopy(@Named("inIndex") int inIndex, @Named("outIndex") int outIndex);
+ public abstract boolean doCopy(@Named("inIndex") int inIndex, @Named("outIndex") int outIndex);
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/PriorityQueueSelector.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/PriorityQueueSelector.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/PriorityQueueSelector.java
index 963b8b0..786667a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/PriorityQueueSelector.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/PriorityQueueSelector.java
@@ -19,6 +19,7 @@ package org.apache.drill.exec.physical.impl.xsort;
import org.apache.drill.exec.compile.TemplateClassDefinition;
import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.record.VectorAccessible;
import org.apache.drill.exec.record.selection.SelectionVector4;
@@ -27,7 +28,7 @@ import org.apache.drill.exec.vector.allocator.VectorAllocator;
import java.util.List;
public interface PriorityQueueSelector {
- public void setup(FragmentContext context, VectorAccessible hyperBatch, SelectionVector4 sv4, List<BatchGroup> batchGroups) throws SchemaChangeException;
+ public void setup(FragmentContext context, BufferAllocator allocator, VectorAccessible hyperBatch, SelectionVector4 sv4, List<BatchGroup> batchGroups) throws SchemaChangeException;
public int next();
public void cleanup();
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/PriorityQueueSelectorTemplate.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/PriorityQueueSelectorTemplate.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/PriorityQueueSelectorTemplate.java
index ccb26cf..65a072b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/PriorityQueueSelectorTemplate.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/PriorityQueueSelectorTemplate.java
@@ -35,20 +35,22 @@ public abstract class PriorityQueueSelectorTemplate implements PriorityQueueSele
private SelectionVector4 vector4;
private List<BatchGroup> batchGroups;
private FragmentContext context;
+ private BufferAllocator allocator;
private int size;
private int queueSize = 0;
private int targetRecordCount = ExternalSortBatch.TARGET_RECORD_COUNT;
private VectorAccessible hyperBatch;
@Override
- public void setup(FragmentContext context, VectorAccessible hyperBatch, SelectionVector4 sv4, List<BatchGroup> batchGroups) throws SchemaChangeException {
+ public void setup(FragmentContext context, BufferAllocator allocator, VectorAccessible hyperBatch, SelectionVector4 sv4, List<BatchGroup> batchGroups) throws SchemaChangeException {
this.context = context;
+ this.allocator = allocator;
this.sv4 = sv4;
this.batchGroups = batchGroups;
this.size = batchGroups.size();
this.hyperBatch = hyperBatch;
- BufferAllocator.PreAllocator preAlloc = context.getAllocator().getNewPreAllocator();
+ BufferAllocator.PreAllocator preAlloc = allocator.getNewPreAllocator();
preAlloc.preAllocate(4 * size);
vector4 = new SelectionVector4(preAlloc.getAllocation(), size, Character.MAX_VALUE);
doSetup(context, hyperBatch, null);
@@ -78,6 +80,7 @@ public abstract class PriorityQueueSelectorTemplate implements PriorityQueueSele
} else if (nextIndex == -2) {
vector4.set(0, batch - 1, 0);
sv4.setCount(outgoingIndex);
+ assert outgoingIndex != 0;
return outgoingIndex;
} else {
vector4.set(0, batch, nextIndex);
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/SingleBatchSorterTemplate.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/SingleBatchSorterTemplate.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/SingleBatchSorterTemplate.java
index d518f04..0ba84f9 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/SingleBatchSorterTemplate.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/SingleBatchSorterTemplate.java
@@ -18,6 +18,7 @@
package org.apache.drill.exec.physical.impl.xsort;
import com.google.common.base.Preconditions;
+import com.google.common.base.Stopwatch;
import org.apache.drill.exec.exception.SchemaChangeException;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.impl.sort.Sorter;
@@ -29,6 +30,7 @@ import org.apache.hadoop.util.IndexedSortable;
import org.apache.hadoop.util.QuickSort;
import javax.inject.Named;
+import java.util.concurrent.TimeUnit;
public abstract class SingleBatchSorterTemplate implements SingleBatchSorter, IndexedSortable{
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(SingleBatchSorterTemplate.class);
@@ -44,7 +46,10 @@ public abstract class SingleBatchSorterTemplate implements SingleBatchSorter, In
@Override
public void sort(SelectionVector2 vector2){
QuickSort qs = new QuickSort();
+ Stopwatch watch = new Stopwatch();
+ watch.start();
qs.sort(this, 0, vector2.getCount());
+ logger.debug("Took {} us to sort {} records", watch.elapsed(TimeUnit.MICROSECONDS), vector2.getCount());
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/Materializer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/Materializer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/Materializer.java
index 70200a9..87078a2 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/Materializer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/Materializer.java
@@ -33,6 +33,7 @@ public class Materializer extends AbstractPhysicalVisitor<PhysicalOperator, Mate
@Override
public PhysicalOperator visitExchange(Exchange exchange, IndexedFragmentNode iNode) throws ExecutionSetupException {
+ iNode.addAllocation(exchange);
if(exchange == iNode.getNode().getSendingExchange()){
// this is a sending exchange.
@@ -56,6 +57,7 @@ public class Materializer extends AbstractPhysicalVisitor<PhysicalOperator, Mate
@Override
public PhysicalOperator visitSubScan(SubScan subScan, IndexedFragmentNode value) throws ExecutionSetupException {
+ value.addAllocation(subScan);
// TODO - implement this
return super.visitOp(subScan, value);
}
@@ -63,6 +65,8 @@ public class Materializer extends AbstractPhysicalVisitor<PhysicalOperator, Mate
@Override
public PhysicalOperator visitStore(Store store, IndexedFragmentNode iNode) throws ExecutionSetupException {
PhysicalOperator child = store.getChild().accept(this, iNode);
+
+ iNode.addAllocation(store);
try {
PhysicalOperator o = store.getSpecificStore(child, iNode.getMinorFragmentId());
@@ -75,6 +79,7 @@ public class Materializer extends AbstractPhysicalVisitor<PhysicalOperator, Mate
@Override
public PhysicalOperator visitOp(PhysicalOperator op, IndexedFragmentNode iNode) throws ExecutionSetupException {
+ iNode.addAllocation(op);
// logger.debug("Visiting catch all: {}", op);
List<PhysicalOperator> children = Lists.newArrayList();
for(PhysicalOperator child : op){
@@ -104,6 +109,10 @@ public class Materializer extends AbstractPhysicalVisitor<PhysicalOperator, Mate
public Wrapper getInfo() {
return info;
}
+
+ public void addAllocation(PhysicalOperator pop) {
+ info.addAllocation(pop);
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/SimpleParallelizer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/SimpleParallelizer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/SimpleParallelizer.java
index 581499d..6e951df 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/SimpleParallelizer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/SimpleParallelizer.java
@@ -132,7 +132,9 @@ public class SimpleParallelizer {
.setAssignment(wrapper.getAssignedEndpoint(minorFragmentId)) //
.setLeafFragment(isLeafFragment) //
.setQueryStartTime(queryStartTime)
- .setTimeZone(timeZone)
+ .setTimeZone(timeZone)//
+ .setMemInitial(wrapper.getInitialAllocation())//
+ .setMemMax(wrapper.getMaxAllocation())
.build();
if (isRootNode) {
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/StatsCollector.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/StatsCollector.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/StatsCollector.java
index ca933c6..6d720a7 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/StatsCollector.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/StatsCollector.java
@@ -80,9 +80,9 @@ public class StatsCollector {
}
@Override
- public Void visitSubScan(SubScan subScan, Wrapper value) throws RuntimeException {
+ public Void visitSubScan(SubScan subScan, Wrapper wrapper) throws RuntimeException {
// TODO - implement this
- return super.visitOp(subScan, value);
+ return super.visitOp(subScan, wrapper);
}
@Override
@@ -93,9 +93,9 @@ public class StatsCollector {
}
@Override
- public Void visitLimit(Limit limit, Wrapper value) throws RuntimeException {
+ public Void visitLimit(Limit limit, Wrapper wrapper) throws RuntimeException {
// TODO: Implement this
- return visitOp(limit, value);
+ return visitOp(limit, wrapper);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/Wrapper.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/Wrapper.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/Wrapper.java
index 94fcac5..8602bf0 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/Wrapper.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/Wrapper.java
@@ -48,6 +48,8 @@ public class Wrapper {
private final Stats stats;
private boolean endpointsAssigned;
private Map<DrillbitEndpoint, EndpointAffinity> endpointAffinity = Maps.newHashMap();
+ private long initialAllocation = 0;
+ private long maxAllocation = 0;
// a list of assigned endpoints. Technically, there could repeated endpoints in this list if we'd like to assign the
// same fragment multiple times to the same endpoint.
@@ -99,6 +101,19 @@ public class Wrapper {
return node;
}
+ public long getInitialAllocation() {
+ return initialAllocation;
+ }
+
+ public long getMaxAllocation() {
+ return maxAllocation;
+ }
+
+ public void addAllocation(PhysicalOperator pop) {
+ initialAllocation += pop.getInitialAllocation();
+ maxAllocation += pop.getMaxAllocation();
+ }
+
private class AssignEndpointsToScanAndStore extends AbstractPhysicalVisitor<Void, List<DrillbitEndpoint>, PhysicalOperatorSetupException>{
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillRuleSets.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillRuleSets.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillRuleSets.java
index 7898937..5521c4e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillRuleSets.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillRuleSets.java
@@ -21,15 +21,7 @@ import java.util.Iterator;
import net.hydromatic.optiq.tools.RuleSet;
-import org.apache.drill.exec.planner.physical.FilterPrule;
-import org.apache.drill.exec.planner.physical.LimitPrule;
-import org.apache.drill.exec.planner.physical.MergeJoinPrule;
-import org.apache.drill.exec.planner.physical.ProjectPrule;
-import org.apache.drill.exec.planner.physical.ScanPrule;
-import org.apache.drill.exec.planner.physical.ScreenPrule;
-import org.apache.drill.exec.planner.physical.SortConvertPrule;
-import org.apache.drill.exec.planner.physical.SortPrule;
-import org.apache.drill.exec.planner.physical.StreamAggPrule;
+import org.apache.drill.exec.planner.physical.*;
import org.eigenbase.rel.RelFactories;
import org.eigenbase.rel.rules.MergeProjectRule;
import org.eigenbase.rel.rules.PushFilterPastJoinRule;
@@ -116,9 +108,9 @@ public class DrillRuleSets {
StreamAggPrule.INSTANCE,
MergeJoinPrule.INSTANCE,
FilterPrule.INSTANCE,
- LimitPrule.INSTANCE
+ LimitPrule.INSTANCE,
-// PushLimitToTopN.INSTANCE
+ PushLimitToTopN.INSTANCE
// ExpandConversionRule.INSTANCE,
// SwapJoinRule.INSTANCE,
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SortPrel.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SortPrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SortPrel.java
index 344be4e..998ed0a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SortPrel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SortPrel.java
@@ -51,11 +51,11 @@ public class SortPrel extends SortRel implements Prel {
PhysicalOperator childPOP = child.getPhysicalOperator(creator);
- childPOP = PrelUtil.removeSvIfRequired(childPOP, SelectionVectorMode.NONE);
- Sort g = new Sort(childPOP, PrelUtil.getOrdering(this.collation, getChild().getRowType()), false);
+// childPOP = PrelUtil.removeSvIfRequired(childPOP, SelectionVectorMode.NONE);
+// Sort g = new Sort(childPOP, PrelUtil.getOrdering(this.collation, getChild().getRowType()), false);
-// childPOP = PrelUtil.removeSvIfRequired(childPOP, SelectionVectorMode.NONE, SelectionVectorMode.TWO_BYTE);
-// Sort g = new ExternalSort(childPOP, PrelUtil.getOrdering(this.collation, getChild().getRowType()), false);
+ childPOP = PrelUtil.removeSvIfRequired(childPOP, SelectionVectorMode.NONE, SelectionVectorMode.TWO_BYTE);
+ Sort g = new ExternalSort(childPOP, PrelUtil.getOrdering(this.collation, getChild().getRowType()), false);
return g;
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractRecordBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractRecordBatch.java
index d4f458f..214f81c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractRecordBatch.java
@@ -20,7 +20,11 @@ package org.apache.drill.exec.record;
import java.util.Iterator;
import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.ExecConstants;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.memory.OutOfMemoryException;
import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.OperatorContext;
import org.apache.drill.exec.physical.base.PhysicalOperator;
import org.apache.drill.exec.record.selection.SelectionVector2;
import org.apache.drill.exec.record.selection.SelectionVector4;
@@ -31,11 +35,13 @@ public abstract class AbstractRecordBatch<T extends PhysicalOperator> implements
protected final VectorContainer container = new VectorContainer();
protected final T popConfig;
protected final FragmentContext context;
-
- protected AbstractRecordBatch(T popConfig, FragmentContext context) {
+ protected final OperatorContext oContext;
+
+ protected AbstractRecordBatch(T popConfig, FragmentContext context) throws OutOfMemoryException {
super();
this.context = context;
this.popConfig = popConfig;
+ this.oContext = new OperatorContext(popConfig, context);
}
@Override
@@ -60,13 +66,13 @@ public abstract class AbstractRecordBatch<T extends PhysicalOperator> implements
@Override
public void kill() {
killIncoming();
- cleanup();
}
protected abstract void killIncoming();
public void cleanup(){
container.clear();
+ oContext.close();
}
@Override
@@ -97,6 +103,4 @@ public abstract class AbstractRecordBatch<T extends PhysicalOperator> implements
return batch;
}
-
-
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractSingleRecordBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractSingleRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractSingleRecordBatch.java
index 7b832e4..dd2cfe0 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractSingleRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractSingleRecordBatch.java
@@ -18,16 +18,19 @@
package org.apache.drill.exec.record;
import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.memory.OutOfMemoryException;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.base.PhysicalOperator;
public abstract class AbstractSingleRecordBatch<T extends PhysicalOperator> extends AbstractRecordBatch<T> {
final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(this.getClass());
-
+
protected final RecordBatch incoming;
private boolean first = true;
+ protected boolean outOfMemory = false;
- public AbstractSingleRecordBatch(T popConfig, FragmentContext context, RecordBatch incoming) {
+ public AbstractSingleRecordBatch(T popConfig, FragmentContext context, RecordBatch incoming) throws OutOfMemoryException {
super(popConfig, context);
this.incoming = incoming;
}
@@ -46,7 +49,8 @@ public abstract class AbstractSingleRecordBatch<T extends PhysicalOperator> exte
case NONE:
case NOT_YET:
case STOP:
- cleanup();
+ return upstream;
+ case OUT_OF_MEMORY:
return upstream;
case OK_NEW_SCHEMA:
try{
@@ -60,6 +64,10 @@ public abstract class AbstractSingleRecordBatch<T extends PhysicalOperator> exte
// fall through.
case OK:
doWork();
+ if (outOfMemory) {
+ outOfMemory = false;
+ return IterOutcome.OUT_OF_MEMORY;
+ }
return upstream; // change if upstream changed, otherwise normal.
default:
throw new UnsupportedOperationException();
@@ -69,8 +77,8 @@ public abstract class AbstractSingleRecordBatch<T extends PhysicalOperator> exte
@Override
public void cleanup() {
// logger.debug("Cleaning up.");
- incoming.cleanup();
super.cleanup();
+ incoming.cleanup();
}
protected abstract void setupNewSchema() throws SchemaChangeException;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/record/RawFragmentBatchProvider.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/RawFragmentBatchProvider.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/RawFragmentBatchProvider.java
index 6f5f7a7..acbd8bd 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/record/RawFragmentBatchProvider.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/RawFragmentBatchProvider.java
@@ -25,5 +25,5 @@ public interface RawFragmentBatchProvider {
public RawFragmentBatch getNext() throws IOException;
public void kill(FragmentContext context);
-
+ public void cleanup();
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/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 b77a6a8..31283c6 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
@@ -45,7 +45,8 @@ public interface RecordBatch extends VectorAccessible {
OK_NEW_SCHEMA, // A full collection of records
STOP, // Informs parent nodes that the query has terminated. In this case, a consumer can consume their QueryContext
// to understand the current state of things.
- NOT_YET // used by batches that haven't received incoming data yet.
+ NOT_YET, // used by batches that haven't received incoming data yet.
+ OUT_OF_MEMORY // an upstream operator was unable to allocate memory. A batch receiving this should release memory if it can
}
public static enum SetupOutcome {
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/record/TransferPair.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/TransferPair.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/TransferPair.java
index 96a1c22..ba2c7b2 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/record/TransferPair.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/TransferPair.java
@@ -21,6 +21,7 @@ import org.apache.drill.exec.vector.ValueVector;
public interface TransferPair {
public void transfer();
+ public void splitAndTransfer(int startIndex, int length);
public ValueVector getTo();
public void copyValue(int from, int to);
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/record/WritableBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/WritableBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/WritableBatch.java
index 288aa7f..396834c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/record/WritableBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/WritableBatch.java
@@ -71,6 +71,7 @@ public class WritableBatch {
cbb.addComponent(buf);
}
+
List<FieldMetadata> fields = def.getFieldList();
int bufferOffset = 0;
@@ -83,7 +84,10 @@ public class WritableBatch {
for (VectorWrapper<?> vv : container) {
FieldMetadata fmd = fields.get(vectorIndex);
ValueVector v = vv.getValueVector();
- v.load(fmd, cbb.slice(bufferOffset, fmd.getBufferLength()));
+ ByteBuf bb = cbb.slice(bufferOffset, fmd.getBufferLength());
+// v.load(fmd, cbb.slice(bufferOffset, fmd.getBufferLength()));
+ v.load(fmd, bb);
+ bb.release();
vectorIndex++;
bufferOffset += fmd.getBufferLength();
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/record/selection/SelectionVector2.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/selection/SelectionVector2.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/selection/SelectionVector2.java
index f7b5155..af1e2b6 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/record/selection/SelectionVector2.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/selection/SelectionVector2.java
@@ -77,10 +77,14 @@ public class SelectionVector2 implements Closeable{
public void setIndex(int index, char value){
buffer.setChar(index * RECORD_SIZE, value);
}
-
- public void allocateNew(int size){
+
+ public boolean allocateNew(int size){
clear();
buffer = allocator.buffer(size * RECORD_SIZE);
+ if (buffer == null) {
+ return false;
+ }
+ return true;
}
public SelectionVector2 clone(){
@@ -98,7 +102,7 @@ public class SelectionVector2 implements Closeable{
}
public void clear() {
- if (buffer != DeadBuf.DEAD_BUFFER) {
+ if (buffer != null && buffer != DeadBuf.DEAD_BUFFER) {
buffer.release();
buffer = DeadBuf.DEAD_BUFFER;
recordCount = 0;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/BasicClientWithConnection.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/BasicClientWithConnection.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/BasicClientWithConnection.java
index c665949..d50a64e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/BasicClientWithConnection.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/BasicClientWithConnection.java
@@ -46,7 +46,7 @@ public abstract class BasicClientWithConnection<T extends EnumLite, HANDSHAKE_SE
protected GenericFutureListener<ChannelFuture> getCloseHandler(ServerConnection clientConnection) {
return getCloseHandler(clientConnection.getChannel());
}
-
+
@Override
protected Response handle(ServerConnection connection, int rpcType, ByteBuf pBody, ByteBuf dBody) throws RpcException {
return handleReponse( (ConnectionThrottle) connection, rpcType, pBody, dBody);
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/BasicServer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/BasicServer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/BasicServer.java
index f5e77f1..8f533e3 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/BasicServer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/BasicServer.java
@@ -73,7 +73,7 @@ public abstract class BasicServer<T extends EnumLite, C extends RemoteConnection
ch.closeFuture().addListener(getCloseHandler(connection));
ch.pipeline().addLast( //
- getDecoder(connection.getAllocator()), //
+ getDecoder(connection.getAllocator(), getOutOfMemoryHandler()), //
new RpcDecoder("s-" + rpcConfig.getName()), //
new RpcEncoder("s-" + rpcConfig.getName()), //
getHandshakeHandler(connection), new InboundHandler(connection), //
@@ -85,7 +85,11 @@ public abstract class BasicServer<T extends EnumLite, C extends RemoteConnection
});
}
- public abstract ProtobufLengthDecoder getDecoder(BufferAllocator allocator);
+ public OutOfMemoryHandler getOutOfMemoryHandler() {
+ return OutOfMemoryHandler.DEFAULT_INSTANCE;
+ }
+
+ public abstract ProtobufLengthDecoder getDecoder(BufferAllocator allocator, OutOfMemoryHandler outOfMemoryHandler);
@Override
public boolean isClient() {
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/InboundRpcMessage.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/InboundRpcMessage.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/InboundRpcMessage.java
index 1527e79..8fc446f 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/InboundRpcMessage.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/InboundRpcMessage.java
@@ -41,7 +41,7 @@ public class InboundRpcMessage extends RpcMessage{
}
void release(){
- pBody.release();
+ if (pBody != null) pBody.release();
if(dBody != null) dBody.release();
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/OutOfMemoryHandler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/OutOfMemoryHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/OutOfMemoryHandler.java
new file mode 100644
index 0000000..5d7db47
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/OutOfMemoryHandler.java
@@ -0,0 +1,31 @@
+/**
+ * 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.rpc;
+
+public interface OutOfMemoryHandler {
+
+ public static OutOfMemoryHandler DEFAULT_INSTANCE = new OutOfMemoryHandler() {
+ @Override
+ public void handle() {
+ throw new UnsupportedOperationException();
+ }
+ };
+
+ public void handle();
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/ProtobufLengthDecoder.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/ProtobufLengthDecoder.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/ProtobufLengthDecoder.java
index 23fa46d..473e3e6 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/ProtobufLengthDecoder.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/ProtobufLengthDecoder.java
@@ -18,6 +18,8 @@
package org.apache.drill.exec.rpc;
import io.netty.buffer.ByteBuf;
+import io.netty.buffer.SwappedByteBuf;
+import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.CorruptedFrameException;
@@ -27,6 +29,7 @@ import java.util.List;
import org.apache.drill.exec.memory.BufferAllocator;
import com.google.protobuf.CodedInputStream;
+import org.apache.drill.exec.proto.GeneralRPCProtos.RpcMode;
/**
* Modified version of ProtobufVarint32FrameDecoder that avoids bytebuf copy.
@@ -34,12 +37,13 @@ import com.google.protobuf.CodedInputStream;
public class ProtobufLengthDecoder extends ByteToMessageDecoder {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ProtobufLengthDecoder.class);
-
private BufferAllocator allocator;
+ private OutOfMemoryHandler outOfMemoryHandler;
- public ProtobufLengthDecoder(BufferAllocator allocator) {
+ public ProtobufLengthDecoder(BufferAllocator allocator, OutOfMemoryHandler outOfMemoryHandler) {
super();
this.allocator = allocator;
+ this.outOfMemoryHandler = outOfMemoryHandler;
}
@@ -82,6 +86,7 @@ public class ProtobufLengthDecoder extends ByteToMessageDecoder {
if(outBuf == null){
logger.warn("Failure allocating buffer on incoming stream due to memory limits. Current Allocation: {}.", allocator.getAllocatedMemory());
in.resetReaderIndex();
+ outOfMemoryHandler.handle();
return;
}
outBuf.writeBytes(in, in.readerIndex(), length);
@@ -104,4 +109,9 @@ public class ProtobufLengthDecoder extends ByteToMessageDecoder {
}
+ @Override
+ public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
+ ctx.fireChannelReadComplete();
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a2355d42/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcBus.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcBus.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcBus.java
index 1a422ee..30101b2 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcBus.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcBus.java
@@ -149,7 +149,7 @@ public abstract class RpcBus<T extends EnumLite, C extends RemoteConnection> imp
if (!ctx.channel().isOpen()) return;
if (RpcConstants.EXTRA_DEBUGGING) logger.debug("Received message {}", msg);
switch (msg.mode) {
- case REQUEST:
+ case REQUEST: {
// handle message and ack.
Response r = handle(connection, msg.rpcType, msg.pBody, msg.dBody);
msg.release(); // we release our ownership. Handle could have taken over ownership.
@@ -159,6 +159,7 @@ public abstract class RpcBus<T extends EnumLite, C extends RemoteConnection> imp
if (RpcConstants.EXTRA_DEBUGGING) logger.debug("Adding message to outbound buffer. {}", outMessage);
ctx.writeAndFlush(outMessage);
break;
+ }
case RESPONSE:
try{