You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by ja...@apache.org on 2014/05/24 18:14:00 UTC
[11/21] status changes
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/669efeb3/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggBatch.java
index 88bada5..5b61a82 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggBatch.java
@@ -79,64 +79,69 @@ public class StreamingAggBatch extends AbstractRecordBatch<StreamingAggregate> {
@Override
public IterOutcome next() {
-
- // this is only called on the first batch. Beyond this, the aggregator manages batches.
- if (aggregator == null) {
- IterOutcome outcome = incoming.next();
- logger.debug("Next outcome of {}", outcome);
- switch (outcome) {
- case NONE:
- case NOT_YET:
- case STOP:
- return outcome;
- case OK_NEW_SCHEMA:
- if (!createAggregator()){
- done = true;
- return IterOutcome.STOP;
+ stats.startProcessing();
+ try{
+ // this is only called on the first batch. Beyond this, the aggregator manages batches.
+ if (aggregator == null) {
+ IterOutcome outcome = next(incoming);
+ logger.debug("Next outcome of {}", outcome);
+ switch (outcome) {
+ case NONE:
+ case NOT_YET:
+ case STOP:
+ return outcome;
+ case OK_NEW_SCHEMA:
+ if (!createAggregator()){
+ done = true;
+ return IterOutcome.STOP;
+ }
+ break;
+ case OK:
+ throw new IllegalStateException("You should never get a first batch without a new schema");
+ default:
+ throw new IllegalStateException(String.format("unknown outcome %s", outcome));
}
- break;
- case OK:
- throw new IllegalStateException("You should never get a first batch without a new schema");
- default:
- throw new IllegalStateException(String.format("unknown outcome %s", outcome));
}
- }
- while(true){
- AggOutcome out = aggregator.doWork();
- logger.debug("Aggregator response {}, records {}", out, aggregator.getOutputCount());
- switch(out){
- case CLEANUP_AND_RETURN:
- container.clear();
- done = true;
- return aggregator.getOutcome();
- case RETURN_OUTCOME:
- return aggregator.getOutcome();
- case UPDATE_AGGREGATOR:
- aggregator = null;
- if(!createAggregator()){
- return IterOutcome.STOP;
+ while(true){
+ AggOutcome out = aggregator.doWork();
+ logger.debug("Aggregator response {}, records {}", out, aggregator.getOutputCount());
+ switch(out){
+ case CLEANUP_AND_RETURN:
+ container.clear();
+ done = true;
+ return aggregator.getOutcome();
+ case RETURN_OUTCOME:
+ return aggregator.getOutcome();
+ case UPDATE_AGGREGATOR:
+ aggregator = null;
+ if(!createAggregator()){
+ return IterOutcome.STOP;
+ }
+ continue;
+ default:
+ throw new IllegalStateException(String.format("Unknown state %s.", out));
}
- continue;
- default:
- throw new IllegalStateException(String.format("Unknown state %s.", out));
}
+ }finally{
+ stats.stopProcessing();
}
-
}
-
-
+
+
/**
* Creates a new Aggregator based on the current schema. If setup fails, this method is responsible for cleaning up
* and informing the context of the failure state, as well is informing the upstream operators.
- *
+ *
* @return true if the aggregator was setup successfully. false if there was a failure.
*/
private boolean createAggregator() {
logger.debug("Creating new aggregator.");
try{
+ stats.startSetup();
this.aggregator = createAggregatorInternal();
+ stats.stopSetup();
return true;
}catch(SchemaChangeException | ClassTransformationException | IOException ex){
context.fail(ex);
@@ -153,13 +158,13 @@ public class StreamingAggBatch extends AbstractRecordBatch<StreamingAggregate> {
ClassGenerator<StreamingAggregator> cg = CodeGenerator.getRoot(StreamingAggTemplate.TEMPLATE_DEFINITION, context.getFunctionRegistry());
container.clear();
List<VectorAllocator> allocators = Lists.newArrayList();
-
+
LogicalExpression[] keyExprs = new LogicalExpression[popConfig.getKeys().length];
LogicalExpression[] valueExprs = new LogicalExpression[popConfig.getExprs().length];
TypedFieldId[] keyOutputIds = new TypedFieldId[popConfig.getKeys().length];
-
+
ErrorCollector collector = new ErrorCollectorImpl();
-
+
for(int i =0; i < keyExprs.length; i++){
NamedExpression ne = popConfig.getKeys()[i];
final LogicalExpression expr = ExpressionTreeMaterializer.materialize(ne.getExpr(), incoming, collector,context.getFunctionRegistry() );
@@ -170,38 +175,38 @@ public class StreamingAggBatch extends AbstractRecordBatch<StreamingAggregate> {
allocators.add(VectorAllocator.getAllocator(vector, 50));
keyOutputIds[i] = container.add(vector);
}
-
+
for(int i =0; i < valueExprs.length; i++){
NamedExpression ne = popConfig.getExprs()[i];
final LogicalExpression expr = ExpressionTreeMaterializer.materialize(ne.getExpr(), incoming, collector, context.getFunctionRegistry());
if(expr == null) continue;
-
+
final MaterializedField outputField = MaterializedField.create(ne.getRef(), expr.getMajorType());
ValueVector vector = TypeHelper.getNewVector(outputField, oContext.getAllocator());
allocators.add(VectorAllocator.getAllocator(vector, 50));
TypedFieldId id = container.add(vector);
valueExprs[i] = new ValueVectorWriteExpression(id, expr, true);
}
-
+
if(collector.hasErrors()) throw new SchemaChangeException("Failure while materializing expression. " + collector.toErrorString());
-
+
setupIsSame(cg, keyExprs);
setupIsSameApart(cg, keyExprs);
addRecordValues(cg, valueExprs);
outputRecordKeys(cg, keyOutputIds, keyExprs);
outputRecordKeysPrev(cg, keyOutputIds, keyExprs);
-
+
cg.getBlock("resetValues")._return(JExpr.TRUE);
getIndex(cg);
-
+
container.buildSchema(SelectionVectorMode.NONE);
StreamingAggregator agg = context.getImplementationClass(cg);
agg.setup(context, incoming, this, allocators.toArray(new VectorAllocator[allocators.size()]));
return agg;
}
-
-
-
+
+
+
private final GeneratorMapping IS_SAME = GeneratorMapping.create("setupInterior", "isSame", null, null);
private final MappingSet IS_SAME_I1 = new MappingSet("index1", null, IS_SAME, IS_SAME);
private final MappingSet IS_SAME_I2 = new MappingSet("index2", null, IS_SAME, IS_SAME);
@@ -214,19 +219,19 @@ public class StreamingAggBatch extends AbstractRecordBatch<StreamingAggregate> {
HoldingContainer first = cg.addExpr(expr, false);
cg.setMappingSet(IS_SAME_I2);
HoldingContainer second = cg.addExpr(expr, false);
-
+
LogicalExpression fh = FunctionGenerationHelper.getComparator(first, second, context.getFunctionRegistry());
HoldingContainer out = cg.addExpr(fh, false);
cg.getEvalBlock()._if(out.getValue().ne(JExpr.lit(0)))._then()._return(JExpr.FALSE);
}
cg.getEvalBlock()._return(JExpr.TRUE);
}
-
+
private final GeneratorMapping IS_SAME_PREV_INTERNAL_BATCH_READ = GeneratorMapping.create("isSamePrev", "isSamePrev", null, null); // the internal batch changes each time so we need to redo setup.
private final GeneratorMapping IS_SAME_PREV = GeneratorMapping.create("setupInterior", "isSamePrev", null, null);
private final MappingSet ISA_B1 = new MappingSet("b1Index", null, "b1", null, IS_SAME_PREV_INTERNAL_BATCH_READ, IS_SAME_PREV_INTERNAL_BATCH_READ);
private final MappingSet ISA_B2 = new MappingSet("b2Index", null, "incoming", null, IS_SAME_PREV, IS_SAME_PREV);
-
+
private void setupIsSameApart(ClassGenerator<StreamingAggregator> cg, LogicalExpression[] keyExprs){
cg.setMappingSet(ISA_B1);
for(LogicalExpression expr : keyExprs){
@@ -242,11 +247,11 @@ public class StreamingAggBatch extends AbstractRecordBatch<StreamingAggregate> {
}
cg.getEvalBlock()._return(JExpr.TRUE);
}
-
+
private final GeneratorMapping EVAL_INSIDE = GeneratorMapping.create("setupInterior", "addRecord", null, null);
private final GeneratorMapping EVAL_OUTSIDE = GeneratorMapping.create("setupInterior", "outputRecordValues", "resetValues", "cleanup");
private final MappingSet EVAL = new MappingSet("index", "outIndex", "incoming", "outgoing", EVAL_INSIDE, EVAL_OUTSIDE, EVAL_INSIDE);
-
+
private void addRecordValues(ClassGenerator<StreamingAggregator> cg, LogicalExpression[] valueExprs){
cg.setMappingSet(EVAL);
for(LogicalExpression ex : valueExprs){
@@ -255,9 +260,9 @@ public class StreamingAggBatch extends AbstractRecordBatch<StreamingAggregate> {
}
cg.getBlock(BlockType.EVAL)._return(JExpr.TRUE);
}
-
+
private final MappingSet RECORD_KEYS = new MappingSet(GeneratorMapping.create("setupInterior", "outputRecordKeys", null, null));
-
+
private void outputRecordKeys(ClassGenerator<StreamingAggregator> cg, TypedFieldId[] keyOutputIds, LogicalExpression[] keyExprs){
cg.setMappingSet(RECORD_KEYS);
for(int i =0; i < keyExprs.length; i++){
@@ -266,13 +271,13 @@ public class StreamingAggBatch extends AbstractRecordBatch<StreamingAggregate> {
}
cg.getBlock(BlockType.EVAL)._return(JExpr.TRUE);
}
-
+
private final GeneratorMapping PREVIOUS_KEYS_OUT = GeneratorMapping.create("setupInterior", "outputRecordKeysPrev", null, null);
private final MappingSet RECORD_KEYS_PREV_OUT = new MappingSet("previousIndex", "outIndex", "previous", "outgoing", PREVIOUS_KEYS_OUT, PREVIOUS_KEYS_OUT);
private final GeneratorMapping PREVIOUS_KEYS = GeneratorMapping.create("outputRecordKeysPrev", "outputRecordKeysPrev", null, null);
private final MappingSet RECORD_KEYS_PREV = new MappingSet("previousIndex", "outIndex", "previous", null, PREVIOUS_KEYS, PREVIOUS_KEYS);
-
+
private void outputRecordKeysPrev(ClassGenerator<StreamingAggregator> cg, TypedFieldId[] keyOutputIds, LogicalExpression[] keyExprs){
cg.setMappingSet(RECORD_KEYS_PREV);
@@ -285,11 +290,11 @@ public class StreamingAggBatch extends AbstractRecordBatch<StreamingAggregate> {
cg.setMappingSet(RECORD_KEYS_PREV_OUT);
HoldingContainer outerExpression = cg.addExpr(new ValueVectorWriteExpression(keyOutputIds[i], new HoldingContainerExpression(innerExpression), true), false);
cg.getBlock(BlockType.EVAL)._if(outerExpression.getValue().eq(JExpr.lit(0)))._then()._return(JExpr.FALSE);
-
+
}
cg.getBlock(BlockType.EVAL)._return(JExpr.TRUE);
}
-
+
private void getIndex(ClassGenerator<StreamingAggregator> g){
switch(incoming.getSchema().getSelectionVectorMode()){
case FOUR_BYTE: {
@@ -308,12 +313,12 @@ public class StreamingAggBatch extends AbstractRecordBatch<StreamingAggregate> {
g.getBlock("getVectorIndex")._return(var.invoke("getIndex").arg(JExpr.direct("recordIndex")));;
return;
}
-
+
default:
throw new IllegalStateException();
-
+
}
-
+
}
@Override
@@ -321,7 +326,7 @@ public class StreamingAggBatch extends AbstractRecordBatch<StreamingAggregate> {
super.cleanup();
incoming.cleanup();
}
-
+
@Override
protected void killIncoming() {
incoming.kill();
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/669efeb3/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinBatch.java
index 9afc033..6b768c9 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinBatch.java
@@ -17,25 +17,16 @@
*/
package org.apache.drill.exec.physical.impl.join;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
import org.apache.drill.common.expression.FieldReference;
import org.apache.drill.common.logical.data.JoinCondition;
import org.apache.drill.common.logical.data.NamedExpression;
import org.apache.drill.common.types.TypeProtos.DataMode;
import org.apache.drill.common.types.TypeProtos.MajorType;
import org.apache.drill.common.types.Types;
-import org.apache.drill.exec.memory.BufferAllocator;
-import org.apache.drill.exec.memory.OutOfMemoryException;
-import org.apache.drill.exec.record.*;
-import org.eigenbase.rel.JoinRelType;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-
-import com.sun.codemodel.JExpression;
-import com.sun.codemodel.JVar;
-import com.sun.codemodel.JExpr;
-
import org.apache.drill.exec.compile.sig.GeneratorMapping;
import org.apache.drill.exec.compile.sig.MappingSet;
import org.apache.drill.exec.exception.ClassTransformationException;
@@ -44,6 +35,7 @@ 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.expr.holders.IntHolder;
+import org.apache.drill.exec.memory.OutOfMemoryException;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.config.HashJoinPOP;
import org.apache.drill.exec.physical.impl.common.ChainedHashTable;
@@ -51,405 +43,421 @@ import org.apache.drill.exec.physical.impl.common.HashTable;
import org.apache.drill.exec.physical.impl.common.HashTableConfig;
import org.apache.drill.exec.physical.impl.sort.RecordBatchData;
import org.apache.drill.exec.physical.impl.svremover.RemovingRecordBatch;
+import org.apache.drill.exec.record.AbstractRecordBatch;
+import org.apache.drill.exec.record.BatchSchema;
+import org.apache.drill.exec.record.ExpandableHyperContainer;
+import org.apache.drill.exec.record.MaterializedField;
+import org.apache.drill.exec.record.RecordBatch;
+import org.apache.drill.exec.record.TypedFieldId;
+import org.apache.drill.exec.record.VectorWrapper;
import org.apache.drill.exec.vector.ValueVector;
import org.apache.drill.exec.vector.allocator.VectorAllocator;
+import org.eigenbase.rel.JoinRelType;
+
+import com.sun.codemodel.JExpr;
+import com.sun.codemodel.JExpression;
+import com.sun.codemodel.JVar;
public class HashJoinBatch extends AbstractRecordBatch<HashJoinPOP> {
public static final long ALLOCATOR_INITIAL_RESERVATION = 1*1024*1024;
public static final long ALLOCATOR_MAX_RESERVATION = 20L*1000*1000*1000;
- // Probe side record batch
- private final RecordBatch left;
+ private static final int LEFT_INPUT = 0;
+ private static final int RIGHT_INPUT = 1;
- // Build side record batch
- private final RecordBatch right;
+ // Probe side record batch
+ private final RecordBatch left;
- // Join type, INNER, LEFT, RIGHT or OUTER
- private final JoinRelType joinType;
+ // Build side record batch
+ private final RecordBatch right;
- // Join conditions
- private final List<JoinCondition> conditions;
+ // Join type, INNER, LEFT, RIGHT or OUTER
+ private final JoinRelType joinType;
- // Runtime generated class implementing HashJoinProbe interface
- private HashJoinProbe hashJoinProbe = null;
+ // Join conditions
+ private final List<JoinCondition> conditions;
- /* Helper class
- * Maintains linked list of build side records with the same key
- * Keeps information about which build records have a corresponding
- * matching key in the probe side (for outer, right joins)
- */
- private HashJoinHelper hjHelper = null;
+ // Runtime generated class implementing HashJoinProbe interface
+ private HashJoinProbe hashJoinProbe = null;
- // Underlying hashtable used by the hash join
- private HashTable hashTable = null;
+ /* Helper class
+ * Maintains linked list of build side records with the same key
+ * Keeps information about which build records have a corresponding
+ * matching key in the probe side (for outer, right joins)
+ */
+ private HashJoinHelper hjHelper = null;
- /* Hyper container to store all build side record batches.
- * Records are retrieved from this container when there is a matching record
- * on the probe side
- */
- private ExpandableHyperContainer hyperContainer;
+ // Underlying hashtable used by the hash join
+ private HashTable hashTable = null;
- // Number of records in the output container
- private int outputRecords;
+ /* Hyper container to store all build side record batches.
+ * Records are retrieved from this container when there is a matching record
+ * on the probe side
+ */
+ private ExpandableHyperContainer hyperContainer;
- // Current batch index on the build side
- private int buildBatchIndex = 0;
+ // Number of records in the output container
+ private int outputRecords;
- // List of vector allocators
- private List<VectorAllocator> allocators = null;
+ // Current batch index on the build side
+ private int buildBatchIndex = 0;
- // Schema of the build side
- private BatchSchema rightSchema = null;
+ // List of vector allocators
+ private List<VectorAllocator> allocators = null;
- // Generator mapping for the build side
- private static final GeneratorMapping PROJECT_BUILD = GeneratorMapping.create("doSetup"/* setup method */,
- "projectBuildRecord" /* eval method */,
- null /* reset */, null /* cleanup */);
+ // Schema of the build side
+ private BatchSchema rightSchema = null;
- // Generator mapping for the probe side
- private static final GeneratorMapping PROJECT_PROBE = GeneratorMapping.create("doSetup" /* setup method */,
- "projectProbeRecord" /* eval method */,
- null /* reset */, null /* cleanup */);
+ // Generator mapping for the build side
+ private static final GeneratorMapping PROJECT_BUILD = GeneratorMapping.create("doSetup"/* setup method */,
+ "projectBuildRecord" /* eval method */,
+ null /* reset */, null /* cleanup */);
- // Mapping set for the build side
- private final MappingSet projectBuildMapping = new MappingSet("buildIndex" /* read index */, "outIndex" /* write index */,
- "buildBatch" /* read container */,
- "outgoing" /* write container */,
- PROJECT_BUILD, PROJECT_BUILD);
+ // Generator mapping for the probe side
+ private static final GeneratorMapping PROJECT_PROBE = GeneratorMapping.create("doSetup" /* setup method */,
+ "projectProbeRecord" /* eval method */,
+ null /* reset */, null /* cleanup */);
- // Mapping set for the probe side
- private final MappingSet projectProbeMapping = new MappingSet("probeIndex" /* read index */, "outIndex" /* write index */,
- "probeBatch" /* read container */,
- "outgoing" /* write container */,
- PROJECT_PROBE, PROJECT_PROBE);
+ // Mapping set for the build side
+ private final MappingSet projectBuildMapping = new MappingSet("buildIndex" /* read index */, "outIndex" /* write index */,
+ "buildBatch" /* read container */,
+ "outgoing" /* write container */,
+ PROJECT_BUILD, PROJECT_BUILD);
- // indicates if we have previously returned an output batch
- boolean firstOutputBatch = true;
+ // Mapping set for the probe side
+ private final MappingSet projectProbeMapping = new MappingSet("probeIndex" /* read index */, "outIndex" /* write index */,
+ "probeBatch" /* read container */,
+ "outgoing" /* write container */,
+ PROJECT_PROBE, PROJECT_PROBE);
- IterOutcome leftUpstream = IterOutcome.NONE;
+ // indicates if we have previously returned an output batch
+ boolean firstOutputBatch = true;
- @Override
- public int getRecordCount() {
- return outputRecords;
- }
+ IterOutcome leftUpstream = IterOutcome.NONE;
+ @Override
+ public int getRecordCount() {
+ return outputRecords;
+ }
- @Override
- public IterOutcome next() {
- try {
- /* If we are here for the first time, execute the build phase of the
- * hash join and setup the run time generated class for the probe side
- */
- if (hashJoinProbe == null) {
+ @Override
+ public IterOutcome next() {
+ stats.startProcessing();
+ try {
+ /* If we are here for the first time, execute the build phase of the
+ * hash join and setup the run time generated class for the probe side
+ */
+ if (hashJoinProbe == null) {
- // Initialize the hash join helper context
- hjHelper = new HashJoinHelper(context, oContext.getAllocator());
+ // Initialize the hash join helper context
+ hjHelper = new HashJoinHelper(context, oContext.getAllocator());
- /* Build phase requires setting up the hash table. Hash table will
- * materialize both the build and probe side expressions while
- * creating the hash table. So we need to invoke next() on our probe batch
- * as well, for the materialization to be successful. This batch will not be used
- * till we complete the build phase.
- */
- leftUpstream = left.next();
+ /* Build phase requires setting up the hash table. Hash table will
+ * materialize both the build and probe side expressions while
+ * creating the hash table. So we need to invoke next() on our probe batch
+ * as well, for the materialization to be successful. This batch will not be used
+ * till we complete the build phase.
+ */
+ leftUpstream = next(LEFT_INPUT, left);
- // Build the hash table, using the build side record batches.
- executeBuildPhase();
+ // Build the hash table, using the build side record batches.
+ executeBuildPhase();
- // Create the run time generated code needed to probe and project
- hashJoinProbe = setupHashJoinProbe();
- }
+ // Create the run time generated code needed to probe and project
+ hashJoinProbe = setupHashJoinProbe();
+ }
- // Store the number of records projected
- if (hashTable != null) {
+ // Store the number of records projected
+ if (hashTable != null) {
+
+ // Allocate the memory for the vectors in the output container
+ allocateVectors();
+
+ outputRecords = hashJoinProbe.probeAndProject();
+
+ /* We are here because of one the following
+ * 1. Completed processing of all the records and we are done
+ * 2. We've filled up the outgoing batch to the maximum and we need to return upstream
+ * Either case build the output container's schema and return
+ */
+ if (outputRecords > 0) {
+
+ // Build the container schema and set the counts
+ container.buildSchema(BatchSchema.SelectionVectorMode.NONE);
+ container.setRecordCount(outputRecords);
+
+ for (VectorWrapper<?> v : container) {
+ v.getValueVector().getMutator().setValueCount(outputRecords);
+ }
+
+ // First output batch, return OK_NEW_SCHEMA
+ if (firstOutputBatch == true) {
+ firstOutputBatch = false;
+ return IterOutcome.OK_NEW_SCHEMA;
+ }
+
+ // Not the first output batch
+ return IterOutcome.OK;
+ }
+ } else {
+ // Our build side is empty, we won't have any matches, clear the probe side
+ if (leftUpstream == IterOutcome.OK_NEW_SCHEMA || leftUpstream == IterOutcome.OK) {
+ for (VectorWrapper<?> wrapper : left) {
+ wrapper.getValueVector().clear();
+ }
+ leftUpstream = next(LEFT_INPUT, left);
+ while (leftUpstream == IterOutcome.OK_NEW_SCHEMA || leftUpstream == IterOutcome.OK) {
+ for (VectorWrapper<?> wrapper : left) {
+ wrapper.getValueVector().clear();
+ }
+ leftUpstream = next(LEFT_INPUT, left);
+ }
+ }
+ }
- // Allocate the memory for the vectors in the output container
- allocateVectors();
+ // No more output records, clean up and return
+ return IterOutcome.NONE;
- outputRecords = hashJoinProbe.probeAndProject();
+ } catch (ClassTransformationException | SchemaChangeException | IOException e) {
+ stats.stopProcessing();
+ context.fail(e);
+ killIncoming();
+ return IterOutcome.STOP;
+ }
+ }
- /* We are here because of one the following
- * 1. Completed processing of all the records and we are done
- * 2. We've filled up the outgoing batch to the maximum and we need to return upstream
- * Either case build the output container's schema and return
- */
- if (outputRecords > 0) {
+ public void setupHashTable() throws IOException, SchemaChangeException, ClassTransformationException {
- // Build the container schema and set the counts
- container.buildSchema(BatchSchema.SelectionVectorMode.NONE);
- container.setRecordCount(outputRecords);
+ // Setup the hash table configuration object
+ int conditionsSize = conditions.size();
- for (VectorWrapper<?> v : container) {
- v.getValueVector().getMutator().setValueCount(outputRecords);
- }
+ NamedExpression rightExpr[] = new NamedExpression[conditionsSize];
+ NamedExpression leftExpr[] = new NamedExpression[conditionsSize];
- // First output batch, return OK_NEW_SCHEMA
- if (firstOutputBatch == true) {
- firstOutputBatch = false;
- return IterOutcome.OK_NEW_SCHEMA;
- }
+ // Create named expressions from the conditions
+ for (int i = 0; i < conditionsSize; i++) {
+ rightExpr[i] = new NamedExpression(conditions.get(i).getRight(), new FieldReference("build_side_" + i ));
+ leftExpr[i] = new NamedExpression(conditions.get(i).getLeft(), new FieldReference("probe_side_" + i));
- // Not the first output batch
- return IterOutcome.OK;
+ // Hash join only supports equality currently.
+ assert conditions.get(i).getRelationship().equals("==");
}
- } else {
- // Our build side is empty, we won't have any matches, clear the probe side
- if (leftUpstream == IterOutcome.OK_NEW_SCHEMA || leftUpstream == IterOutcome.OK) {
- for (VectorWrapper<?> wrapper : left) {
- wrapper.getValueVector().clear();
- }
- leftUpstream = left.next();
- while (leftUpstream == IterOutcome.OK_NEW_SCHEMA || leftUpstream == IterOutcome.OK) {
- for (VectorWrapper<?> wrapper : left) {
- wrapper.getValueVector().clear();
- }
- leftUpstream = left.next();
+
+ // Set the left named expression to be null if the probe batch is empty.
+ if (leftUpstream != IterOutcome.OK_NEW_SCHEMA && leftUpstream != IterOutcome.OK) {
+ leftExpr = null;
+ } else {
+ if (left.getSchema().getSelectionVectorMode() != BatchSchema.SelectionVectorMode.NONE) {
+ throw new SchemaChangeException("Hash join does not support probe batch with selection vectors");
}
}
- }
-
- // No more output records, clean up and return
- return IterOutcome.NONE;
-
- } catch (ClassTransformationException | SchemaChangeException | IOException e) {
- context.fail(e);
- killIncoming();
- return IterOutcome.STOP;
- }
- }
-
- public void setupHashTable() throws IOException, SchemaChangeException, ClassTransformationException {
-
- // Setup the hash table configuration object
- int conditionsSize = conditions.size();
- NamedExpression rightExpr[] = new NamedExpression[conditionsSize];
- NamedExpression leftExpr[] = new NamedExpression[conditionsSize];
+ HashTableConfig htConfig = new HashTableConfig(HashTable.DEFAULT_INITIAL_CAPACITY, HashTable.DEFAULT_LOAD_FACTOR, rightExpr, leftExpr);
- // Create named expressions from the conditions
- for (int i = 0; i < conditionsSize; i++) {
- rightExpr[i] = new NamedExpression(conditions.get(i).getRight(), new FieldReference("build_side_" + i ));
- leftExpr[i] = new NamedExpression(conditions.get(i).getLeft(), new FieldReference("probe_side_" + i));
-
- // Hash join only supports equality currently.
- assert conditions.get(i).getRelationship().equals("==");
+ // Create the chained hash table
+ ChainedHashTable ht = new ChainedHashTable(htConfig, context, oContext.getAllocator(), this.right, this.left, null);
+ hashTable = ht.createAndSetupHashTable(null);
}
- // Set the left named expression to be null if the probe batch is empty.
- if (leftUpstream != IterOutcome.OK_NEW_SCHEMA && leftUpstream != IterOutcome.OK) {
- leftExpr = null;
- } else {
- if (left.getSchema().getSelectionVectorMode() != BatchSchema.SelectionVectorMode.NONE) {
- throw new SchemaChangeException("Hash join does not support probe batch with selection vectors");
- }
+ public void executeBuildPhase() throws SchemaChangeException, ClassTransformationException, IOException {
+
+ //Setup the underlying hash table
+ IterOutcome rightUpstream = next(RIGHT_INPUT, right);
+
+ boolean moreData = true;
+
+ while (moreData) {
+
+ switch (rightUpstream) {
+
+ case NONE:
+ case NOT_YET:
+ case STOP:
+ moreData = false;
+ continue;
+
+ case OK_NEW_SCHEMA:
+ if (rightSchema == null) {
+ rightSchema = right.getSchema();
+
+ if (rightSchema.getSelectionVectorMode() != BatchSchema.SelectionVectorMode.NONE) {
+ throw new SchemaChangeException("Hash join does not support build batch with selection vectors");
+ }
+ setupHashTable();
+ } else {
+ throw new SchemaChangeException("Hash join does not support schema changes");
+ }
+ // Fall through
+ case OK:
+ int currentRecordCount = right.getRecordCount();
+
+ /* For every new build batch, we store some state in the helper context
+ * Add new state to the helper context
+ */
+ hjHelper.addNewBatch(currentRecordCount);
+
+ // Holder contains the global index where the key is hashed into using the hash table
+ IntHolder htIndex = new IntHolder();
+
+ // For every record in the build batch , hash the key columns
+ for (int i = 0; i < currentRecordCount; i++) {
+
+ HashTable.PutStatus status = hashTable.put(i, htIndex);
+
+ if (status != HashTable.PutStatus.PUT_FAILED) {
+ /* Use the global index returned by the hash table, to store
+ * the current record index and batch index. This will be used
+ * later when we probe and find a match.
+ */
+ hjHelper.setCurrentIndex(htIndex.value, buildBatchIndex, i);
+ }
+ }
+
+ /* Completed hashing all records in this batch. Transfer the batch
+ * to the hyper vector container. Will be used when we want to retrieve
+ * records that have matching keys on the probe side.
+ */
+ RecordBatchData nextBatch = new RecordBatchData(right);
+ if (hyperContainer == null) {
+ hyperContainer = new ExpandableHyperContainer(nextBatch.getContainer());
+ } else {
+ hyperContainer.addBatch(nextBatch.getContainer());
+ }
+
+ // completed processing a batch, increment batch index
+ buildBatchIndex++;
+ break;
+ }
+ // Get the next record batch
+ rightUpstream = next(RIGHT_INPUT, right);
+ }
}
- HashTableConfig htConfig = new HashTableConfig(HashTable.DEFAULT_INITIAL_CAPACITY, HashTable.DEFAULT_LOAD_FACTOR, rightExpr, leftExpr);
-
- // Create the chained hash table
- ChainedHashTable ht = new ChainedHashTable(htConfig, context, oContext.getAllocator(), this.right, this.left, null);
- hashTable = ht.createAndSetupHashTable(null);
- }
+ public HashJoinProbe setupHashJoinProbe() throws ClassTransformationException, IOException {
- public void executeBuildPhase() throws SchemaChangeException, ClassTransformationException, IOException {
+ allocators = new ArrayList<>();
- //Setup the underlying hash table
- IterOutcome rightUpstream = right.next();
+ final CodeGenerator<HashJoinProbe> cg = CodeGenerator.get(HashJoinProbe.TEMPLATE_DEFINITION, context.getFunctionRegistry());
+ ClassGenerator<HashJoinProbe> g = cg.getRoot();
- boolean moreData = true;
+ // Generate the code to project build side records
+ g.setMappingSet(projectBuildMapping);
- while (moreData) {
- switch (rightUpstream) {
+ int fieldId = 0;
+ JExpression buildIndex = JExpr.direct("buildIndex");
+ JExpression outIndex = JExpr.direct("outIndex");
+ g.rotateBlock();
- case NONE:
- case NOT_YET:
- case STOP:
- moreData = false;
- continue;
+ if (hyperContainer != null) {
+ for(VectorWrapper<?> vv : hyperContainer) {
- case OK_NEW_SCHEMA:
- if (rightSchema == null) {
- rightSchema = right.getSchema();
+ MajorType inputType = vv.getField().getType();
+ MajorType outputType;
+ if (joinType == JoinRelType.LEFT && inputType.getMode() == DataMode.REQUIRED) {
+ outputType = Types.overrideMode(inputType, DataMode.OPTIONAL);
+ } else {
+ outputType = inputType;
+ }
- if (rightSchema.getSelectionVectorMode() != BatchSchema.SelectionVectorMode.NONE) {
- throw new SchemaChangeException("Hash join does not support build batch with selection vectors");
- }
- setupHashTable();
- } else {
- throw new SchemaChangeException("Hash join does not support schema changes");
- }
- // Fall through
- case OK:
- int currentRecordCount = right.getRecordCount();
+ // Add the vector to our output container
+ ValueVector v = TypeHelper.getNewVector(MaterializedField.create(vv.getField().getPath(), outputType), context.getAllocator());
+ container.add(v);
+ allocators.add(RemovingRecordBatch.getAllocator4(v));
- /* For every new build batch, we store some state in the helper context
- * Add new state to the helper context
- */
- hjHelper.addNewBatch(currentRecordCount);
+ JVar inVV = g.declareVectorValueSetupAndMember("buildBatch", new TypedFieldId(vv.getField().getType(), true, fieldId));
+ JVar outVV = g.declareVectorValueSetupAndMember("outgoing", new TypedFieldId(outputType, false, fieldId));
+ g.getEvalBlock()._if(outVV.invoke("copyFromSafe")
+ .arg(buildIndex.band(JExpr.lit((int) Character.MAX_VALUE)))
+ .arg(outIndex)
+ .arg(inVV.component(buildIndex.shrz(JExpr.lit(16)))).not())._then()._return(JExpr.FALSE);
- // Holder contains the global index where the key is hashed into using the hash table
- IntHolder htIndex = new IntHolder();
-
- // For every record in the build batch , hash the key columns
- for (int i = 0; i < currentRecordCount; i++) {
-
- HashTable.PutStatus status = hashTable.put(i, htIndex);
-
- if (status != HashTable.PutStatus.PUT_FAILED) {
- /* Use the global index returned by the hash table, to store
- * the current record index and batch index. This will be used
- * later when we probe and find a match.
- */
- hjHelper.setCurrentIndex(htIndex.value, buildBatchIndex, i);
+ fieldId++;
}
- }
-
- /* Completed hashing all records in this batch. Transfer the batch
- * to the hyper vector container. Will be used when we want to retrieve
- * records that have matching keys on the probe side.
- */
- RecordBatchData nextBatch = new RecordBatchData(right);
- if (hyperContainer == null) {
- hyperContainer = new ExpandableHyperContainer(nextBatch.getContainer());
- } else {
- hyperContainer.addBatch(nextBatch.getContainer());
- }
+ }
+ g.rotateBlock();
+ g.getEvalBlock()._return(JExpr.TRUE);
- // completed processing a batch, increment batch index
- buildBatchIndex++;
- break;
- }
- // Get the next record batch
- rightUpstream = right.next();
- }
- }
+ // Generate the code to project probe side records
+ g.setMappingSet(projectProbeMapping);
- public HashJoinProbe setupHashJoinProbe() throws ClassTransformationException, IOException {
+ int outputFieldId = fieldId;
+ fieldId = 0;
+ JExpression probeIndex = JExpr.direct("probeIndex");
+ int recordCount = 0;
- allocators = new ArrayList<>();
+ if (leftUpstream == IterOutcome.OK || leftUpstream == IterOutcome.OK_NEW_SCHEMA) {
+ for (VectorWrapper<?> vv : left) {
- final CodeGenerator<HashJoinProbe> cg = CodeGenerator.get(HashJoinProbe.TEMPLATE_DEFINITION, context.getFunctionRegistry());
- ClassGenerator<HashJoinProbe> g = cg.getRoot();
+ MajorType inputType = vv.getField().getType();
+ MajorType outputType;
+ if (joinType == JoinRelType.RIGHT && inputType.getMode() == DataMode.REQUIRED) {
+ outputType = Types.overrideMode(inputType, DataMode.OPTIONAL);
+ } else {
+ outputType = inputType;
+ }
- // Generate the code to project build side records
- g.setMappingSet(projectBuildMapping);
+ ValueVector v = TypeHelper.getNewVector(MaterializedField.create(vv.getField().getPath(), outputType), oContext.getAllocator());
+ container.add(v);
+ allocators.add(RemovingRecordBatch.getAllocator4(v));
+ JVar inVV = g.declareVectorValueSetupAndMember("probeBatch", new TypedFieldId(inputType, false, fieldId));
+ JVar outVV = g.declareVectorValueSetupAndMember("outgoing", new TypedFieldId(outputType, false, outputFieldId));
- int fieldId = 0;
- JExpression buildIndex = JExpr.direct("buildIndex");
- JExpression outIndex = JExpr.direct("outIndex");
- g.rotateBlock();
+ g.getEvalBlock()._if(outVV.invoke("copyFromSafe").arg(probeIndex).arg(outIndex).arg(inVV).not())._then()._return(JExpr.FALSE);
- if (hyperContainer != null) {
- for(VectorWrapper<?> vv : hyperContainer) {
+ fieldId++;
+ outputFieldId++;
+ }
+ g.rotateBlock();
+ g.getEvalBlock()._return(JExpr.TRUE);
- MajorType inputType = vv.getField().getType();
- MajorType outputType;
- if (joinType == JoinRelType.LEFT && inputType.getMode() == DataMode.REQUIRED) {
- outputType = Types.overrideMode(inputType, DataMode.OPTIONAL);
- } else {
- outputType = inputType;
+ recordCount = left.getRecordCount();
}
- // Add the vector to our output container
- ValueVector v = TypeHelper.getNewVector(MaterializedField.create(vv.getField().getPath(), outputType), context.getAllocator());
- container.add(v);
- allocators.add(RemovingRecordBatch.getAllocator4(v));
- JVar inVV = g.declareVectorValueSetupAndMember("buildBatch", new TypedFieldId(vv.getField().getType(), true, fieldId));
- JVar outVV = g.declareVectorValueSetupAndMember("outgoing", new TypedFieldId(outputType, false, fieldId));
- g.getEvalBlock()._if(outVV.invoke("copyFromSafe")
- .arg(buildIndex.band(JExpr.lit((int) Character.MAX_VALUE)))
- .arg(outIndex)
- .arg(inVV.component(buildIndex.shrz(JExpr.lit(16)))).not())._then()._return(JExpr.FALSE);
+ HashJoinProbe hj = context.getImplementationClass(cg);
- fieldId++;
- }
+ hj.setupHashJoinProbe(context, hyperContainer, left, recordCount, this, hashTable, hjHelper, joinType);
+ return hj;
}
- g.rotateBlock();
- g.getEvalBlock()._return(JExpr.TRUE);
-
- // Generate the code to project probe side records
- g.setMappingSet(projectProbeMapping);
-
- int outputFieldId = fieldId;
- fieldId = 0;
- JExpression probeIndex = JExpr.direct("probeIndex");
- int recordCount = 0;
-
- if (leftUpstream == IterOutcome.OK || leftUpstream == IterOutcome.OK_NEW_SCHEMA) {
- for (VectorWrapper<?> vv : left) {
- MajorType inputType = vv.getField().getType();
- MajorType outputType;
- if (joinType == JoinRelType.RIGHT && inputType.getMode() == DataMode.REQUIRED) {
- outputType = Types.overrideMode(inputType, DataMode.OPTIONAL);
- } else {
- outputType = inputType;
+ private void allocateVectors(){
+ for(VectorAllocator a : allocators){
+ a.alloc(RecordBatch.MAX_BATCH_SIZE);
}
-
- ValueVector v = TypeHelper.getNewVector(MaterializedField.create(vv.getField().getPath(), outputType), oContext.getAllocator());
- container.add(v);
- allocators.add(RemovingRecordBatch.getAllocator4(v));
-
- JVar inVV = g.declareVectorValueSetupAndMember("probeBatch", new TypedFieldId(inputType, false, fieldId));
- JVar outVV = g.declareVectorValueSetupAndMember("outgoing", new TypedFieldId(outputType, false, outputFieldId));
-
- g.getEvalBlock()._if(outVV.invoke("copyFromSafe").arg(probeIndex).arg(outIndex).arg(inVV).not())._then()._return(JExpr.FALSE);
-
- fieldId++;
- outputFieldId++;
- }
- g.rotateBlock();
- g.getEvalBlock()._return(JExpr.TRUE);
-
- recordCount = left.getRecordCount();
}
+ public HashJoinBatch(HashJoinPOP popConfig, FragmentContext context, RecordBatch left, RecordBatch right) throws OutOfMemoryException {
+ super(popConfig, context);
+ this.left = left;
+ this.right = right;
+ this.joinType = popConfig.getJoinType();
+ this.conditions = popConfig.getConditions();
+ }
- HashJoinProbe hj = context.getImplementationClass(cg);
+ @Override
+ public void killIncoming() {
+ this.left.kill();
+ this.right.kill();
+ }
- hj.setupHashJoinProbe(context, hyperContainer, left, recordCount, this, hashTable, hjHelper, joinType);
- return hj;
- }
+ @Override
+ public void cleanup() {
+ hjHelper.clear();
- private void allocateVectors(){
- for(VectorAllocator a : allocators){
- a.alloc(RecordBatch.MAX_BATCH_SIZE);
- }
- }
-
- public HashJoinBatch(HashJoinPOP popConfig, FragmentContext context, RecordBatch left, RecordBatch right) throws OutOfMemoryException {
- super(popConfig, context);
- this.left = left;
- this.right = right;
- this.joinType = popConfig.getJoinType();
- this.conditions = popConfig.getConditions();
- }
-
- @Override
- public void killIncoming() {
- this.left.kill();
- this.right.kill();
- }
-
- @Override
- public void cleanup() {
- hjHelper.clear();
-
- // If we didn't receive any data, hyperContainer may be null, check before clearing
- if (hyperContainer != null) {
- hyperContainer.clear();
- }
+ // If we didn't receive any data, hyperContainer may be null, check before clearing
+ if (hyperContainer != null) {
+ hyperContainer.clear();
+ }
- if (hashTable != null) {
- hashTable.clear();
+ if (hashTable != null) {
+ hashTable.clear();
+ }
+ super.cleanup();
+ left.cleanup();
+ right.cleanup();
}
- super.cleanup();
- left.cleanup();
- right.cleanup();
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/669efeb3/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 c07878a..faca32a 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
@@ -35,6 +35,9 @@ public final class JoinStatus {
INCOMING, SV4;
}
+ private static final int LEFT_INPUT = 0;
+ private static final int RIGHT_INPUT = 1;
+
public final RecordBatch left;
private int leftPosition;
private IterOutcome lastLeft;
@@ -63,10 +66,18 @@ public final class JoinStatus {
this.joinType = output.getJoinType();
}
+ private final IterOutcome nextLeft(){
+ return outputBatch.next(LEFT_INPUT, left);
+ }
+
+ private final IterOutcome nextRight(){
+ return outputBatch.next(RIGHT_INPUT, right);
+ }
+
public final void ensureInitial(){
if(!initialSet){
- this.lastLeft = left.next();
- this.lastRight = right.next();
+ this.lastLeft = nextLeft();
+ this.lastRight = nextRight();
initialSet = true;
}
}
@@ -148,7 +159,7 @@ public final class JoinStatus {
if (!isLeftPositionInCurrentBatch()) {
leftPosition = 0;
releaseData(left);
- lastLeft = left.next();
+ lastLeft = nextLeft();
return lastLeft == IterOutcome.OK;
}
lastLeft = IterOutcome.OK;
@@ -167,7 +178,7 @@ public final class JoinStatus {
if (!isRightPositionInCurrentBatch()) {
rightPosition = 0;
releaseData(right);
- lastRight = right.next();
+ lastRight = nextRight();
return lastRight == IterOutcome.OK;
}
lastRight = IterOutcome.OK;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/669efeb3/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 48b7fea..46dea64 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
@@ -66,9 +66,9 @@ public class MergeJoinBatch extends AbstractRecordBatch<MergeJoinPOP> {
public static final long ALLOCATOR_INITIAL_RESERVATION = 1*1024*1024;
public static final long ALLOCATOR_MAX_RESERVATION = 20L*1000*1000*1000;
-
+
public final MappingSet setupMapping =
- new MappingSet("null", "null",
+ new MappingSet("null", "null",
GM("doSetup", "doSetup", null, null),
GM("doSetup", "doSetup", null, null));
public final MappingSet copyLeftMapping =
@@ -96,7 +96,7 @@ public class MergeJoinBatch extends AbstractRecordBatch<MergeJoinPOP> {
GM("doSetup", "doCompareNextLeftKey", null, null),
GM("doSetup", "doCompareNextLeftKey", null, null));
-
+
private final RecordBatch left;
private final RecordBatch right;
private final JoinStatus status;
@@ -104,7 +104,7 @@ public class MergeJoinBatch extends AbstractRecordBatch<MergeJoinPOP> {
private final JoinRelType joinType;
private JoinWorker worker;
public MergeJoinBatchBuilder batchBuilder;
-
+
protected MergeJoinBatch(MergeJoinPOP popConfig, FragmentContext context, RecordBatch left, RecordBatch right) throws OutOfMemoryException {
super(popConfig, context);
@@ -130,10 +130,10 @@ public class MergeJoinBatch extends AbstractRecordBatch<MergeJoinPOP> {
@Override
public IterOutcome next() {
-
+
// we do this in the here instead of the constructor because don't necessary want to start consuming on construction.
status.ensureInitial();
-
+
// loop so we can start over again if we find a new batch was created.
while(true){
@@ -153,14 +153,17 @@ public class MergeJoinBatch extends AbstractRecordBatch<MergeJoinPOP> {
logger.debug("NO MORE DATA; returning {} NONE");
return IterOutcome.NONE;
}
-
+
boolean first = false;
if(worker == null){
try {
logger.debug("Creating New Worker");
+ stats.startSetup();
this.worker = generateNewWorker();
first = true;
+ stats.stopSetup();
} catch (ClassTransformationException | IOException | SchemaChangeException e) {
+ stats.stopSetup();
context.fail(new SchemaChangeException(e));
kill();
return IterOutcome.STOP;
@@ -222,13 +225,13 @@ public class MergeJoinBatch extends AbstractRecordBatch<MergeJoinPOP> {
left.cleanup();
right.cleanup();
}
-
- private void generateDoCompareNextLeft(ClassGenerator<JoinWorker> cg, JVar incomingRecordBatch,
+
+ private void generateDoCompareNextLeft(ClassGenerator<JoinWorker> cg, JVar incomingRecordBatch,
JVar incomingLeftRecordBatch, JVar joinStatus, ErrorCollector collector) throws ClassTransformationException {
boolean nextLeftIndexDeclared = false;
cg.setMappingSet(compareLeftMapping);
-
+
for (JoinCondition condition : conditions) {
final LogicalExpression leftFieldExpr = condition.getLeft();
@@ -242,52 +245,52 @@ public class MergeJoinBatch extends AbstractRecordBatch<MergeJoinPOP> {
////////////////////////////////
cg.setMappingSet(compareLeftMapping);
cg.getSetupBlock().assign(JExpr._this().ref(incomingRecordBatch), JExpr._this().ref(incomingLeftRecordBatch));
-
+
if (!nextLeftIndexDeclared) {
// int nextLeftIndex = leftIndex + 1;
cg.getEvalBlock().decl(JType.parse(cg.getModel(), "int"), "nextLeftIndex", JExpr.direct("leftIndex").plus(JExpr.lit(1)));
nextLeftIndexDeclared = true;
- }
+ }
// check if the next key is in this batch
cg.getEvalBlock()._if(joinStatus.invoke("isNextLeftPositionInCurrentBatch").eq(JExpr.lit(false)))
._then()
._return(JExpr.lit(-1));
-
+
// generate VV read expressions
ClassGenerator.HoldingContainer compareThisLeftExprHolder = cg.addExpr(materializedLeftExpr, false);
cg.setMappingSet(compareNextLeftMapping); // change mapping from 'leftIndex' to 'nextLeftIndex'
ClassGenerator.HoldingContainer compareNextLeftExprHolder = cg.addExpr(materializedLeftExpr, false);
-
+
if (compareThisLeftExprHolder.isOptional()) {
// handle null == null
cg.getEvalBlock()._if(compareThisLeftExprHolder.getIsSet().eq(JExpr.lit(0))
.cand(compareNextLeftExprHolder.getIsSet().eq(JExpr.lit(0))))
._then()
._return(JExpr.lit(0));
-
+
// handle null == !null
cg.getEvalBlock()._if(compareThisLeftExprHolder.getIsSet().eq(JExpr.lit(0))
.cor(compareNextLeftExprHolder.getIsSet().eq(JExpr.lit(0))))
._then()
._return(JExpr.lit(1));
}
-
+
// check value equality
-
+
LogicalExpression gh = FunctionGenerationHelper.getComparator(compareThisLeftExprHolder,
compareNextLeftExprHolder,
context.getFunctionRegistry());
HoldingContainer out = cg.addExpr(gh, false);
-
- //If not 0, it means not equal. We return this out value.
+
+ //If not 0, it means not equal. We return this out value.
JConditional jc = cg.getEvalBlock()._if(out.getValue().ne(JExpr.lit(0)));
jc._then()._return(out.getValue());
}
-
+
//Pass the equality check for all the join conditions. Finally, return 0.
cg.getEvalBlock()._return(JExpr.lit(0));
- }
-
+ }
+
private JoinWorker generateNewWorker() throws ClassTransformationException, IOException, SchemaChangeException{
final ClassGenerator<JoinWorker> cg = CodeGenerator.getRoot(JoinWorker.TEMPLATE_DEFINITION, context.getFunctionRegistry());
@@ -322,11 +325,11 @@ public class MergeJoinBatch extends AbstractRecordBatch<MergeJoinPOP> {
//generate doCompare() method
/////////////////////////////////////////
generateDoCompare(cg, incomingRecordBatch, incomingLeftRecordBatch, incomingRightRecordBatch, collector);
-
+
//generate doCompareNextLeftKey() method
/////////////////////////////////////////
generateDoCompareNextLeft(cg, incomingRecordBatch, incomingLeftRecordBatch, joinStatus, collector);
-
+
// generate copyLeft()
//////////////////////
cg.setMappingSet(copyLeftMapping);
@@ -394,12 +397,12 @@ public class MergeJoinBatch extends AbstractRecordBatch<MergeJoinPOP> {
private void allocateBatch() {
// allocate new batch space.
container.clear();
-
+
//estimation of joinBatchSize : max of left/right size, expanded by a factor of 16, which is then bounded by MAX_BATCH_SIZE.
int leftCount = status.isLeftPositionAllowed() ? left.getRecordCount() : 0;
int rightCount = status.isRightPositionAllowed() ? right.getRecordCount() : 0;
int joinBatchSize = Math.min(Math.max(leftCount, rightCount) * 16, MAX_BATCH_SIZE);
-
+
// add fields from both batches
if (leftCount > 0) {
@@ -436,11 +439,11 @@ public class MergeJoinBatch extends AbstractRecordBatch<MergeJoinPOP> {
logger.debug("Built joined schema: {}", container.getSchema());
}
- private void generateDoCompare(ClassGenerator<JoinWorker> cg, JVar incomingRecordBatch,
+ private void generateDoCompare(ClassGenerator<JoinWorker> cg, JVar incomingRecordBatch,
JVar incomingLeftRecordBatch, JVar incomingRightRecordBatch, ErrorCollector collector) throws ClassTransformationException {
-
+
cg.setMappingSet(compareMapping);
-
+
for (JoinCondition condition : conditions) {
final LogicalExpression leftFieldExpr = condition.getLeft();
final LogicalExpression rightFieldExpr = condition.getRight();
@@ -482,37 +485,37 @@ public class MergeJoinBatch extends AbstractRecordBatch<MergeJoinPOP> {
.cand(compareRightExprHolder.getIsSet().eq(JExpr.lit(0))))
._then()
._return(JExpr.lit(0));
-
+
// handle null == !null
cg.getEvalBlock()._if(compareLeftExprHolder.getIsSet().eq(JExpr.lit(0))
.cor(compareRightExprHolder.getIsSet().eq(JExpr.lit(0))))
._then()
._return(JExpr.lit(1));
-
+
} else if (compareLeftExprHolder.isOptional()) {
// handle null == required (null is less than any value)
cg.getEvalBlock()._if(compareLeftExprHolder.getIsSet().eq(JExpr.lit(0)))
._then()
._return(JExpr.lit(-1));
-
+
} else if (compareRightExprHolder.isOptional()) {
// handle required == null (null is less than any value)
cg.getEvalBlock()._if(compareRightExprHolder.getIsSet().eq(JExpr.lit(0)))
._then()
._return(JExpr.lit(1));
}
-
+
LogicalExpression fh = FunctionGenerationHelper.getComparator(compareLeftExprHolder,
compareRightExprHolder,
- context.getFunctionRegistry());
+ context.getFunctionRegistry());
HoldingContainer out = cg.addExpr(fh, false);
-
- //If not 0, it means not equal. We return this out value.
+
+ //If not 0, it means not equal. We return this out value.
JConditional jc = cg.getEvalBlock()._if(out.getValue().ne(JExpr.lit(0)));
jc._then()._return(out.getValue());
}
-
- //Pass the equality check for all the join conditions. Finally, return 0.
- cg.getEvalBlock()._return(JExpr.lit(0));
+
+ //Pass the equality check for all the join conditions. Finally, return 0.
+ cg.getEvalBlock()._return(JExpr.lit(0));
}
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/669efeb3/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/mergereceiver/MergingRecordBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/mergereceiver/MergingRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/mergereceiver/MergingRecordBatch.java
index 3d496d3..cc38cbe 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/mergereceiver/MergingRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/mergereceiver/MergingRecordBatch.java
@@ -27,8 +27,6 @@ import java.util.PriorityQueue;
import org.apache.drill.common.expression.ErrorCollector;
import org.apache.drill.common.expression.ErrorCollectorImpl;
-import org.apache.drill.common.expression.ExpressionPosition;
-import org.apache.drill.common.expression.FunctionCall;
import org.apache.drill.common.expression.LogicalExpression;
import org.apache.drill.common.expression.SchemaPath;
import org.apache.drill.common.logical.data.Order.Ordering;
@@ -41,19 +39,28 @@ 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.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.ops.MetricDef;
import org.apache.drill.exec.physical.config.MergingReceiverPOP;
import org.apache.drill.exec.proto.UserBitShared;
-import org.apache.drill.exec.record.*;
+import org.apache.drill.exec.record.AbstractRecordBatch;
+import org.apache.drill.exec.record.BatchSchema;
+import org.apache.drill.exec.record.RawFragmentBatch;
+import org.apache.drill.exec.record.RawFragmentBatchProvider;
+import org.apache.drill.exec.record.RecordBatch;
+import org.apache.drill.exec.record.RecordBatchLoader;
+import org.apache.drill.exec.record.SchemaBuilder;
+import org.apache.drill.exec.record.TypedFieldId;
+import org.apache.drill.exec.record.VectorContainer;
+import org.apache.drill.exec.record.VectorWrapper;
+import org.apache.drill.exec.record.WritableBatch;
import org.apache.drill.exec.record.selection.SelectionVector2;
import org.apache.drill.exec.record.selection.SelectionVector4;
import org.apache.drill.exec.vector.ValueVector;
import org.apache.drill.exec.vector.allocator.VectorAllocator;
import org.eigenbase.rel.RelFieldCollation.Direction;
-import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import com.sun.codemodel.JArray;
import com.sun.codemodel.JClass;
@@ -92,6 +99,15 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
private List<VectorAllocator> allocators;
private MergingReceiverPOP config;
+ public static enum Metric implements MetricDef{
+ NEXT_WAIT_NANOS;
+
+ @Override
+ public int metricId() {
+ return ordinal();
+ }
+ }
+
public MergingRecordBatch(FragmentContext context,
MergingReceiverPOP config,
RawFragmentBatchProvider[] fragProviders) throws OutOfMemoryException {
@@ -104,8 +120,21 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
this.outgoingContainer = new VectorContainer();
}
+ private RawFragmentBatch getNext(RawFragmentBatchProvider provider) throws IOException{
+ long startNext = System.nanoTime();
+ RawFragmentBatch b = provider.getNext();
+ if(b != null){
+ stats.batchReceived(0, b.getHeader().getDef().getRecordCount(), false);
+ }
+ stats.addLongStat(Metric.NEXT_WAIT_NANOS, System.nanoTime() - startNext);
+ return b;
+ }
+
@Override
public IterOutcome next() {
+ stats.startProcessing();
+ try{
+
if (fragProviders.length == 0) return IterOutcome.NONE;
boolean schemaChanged = false;
@@ -131,7 +160,7 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
for (RawFragmentBatchProvider provider : fragProviders) {
RawFragmentBatch rawBatch = null;
try {
- rawBatch = provider.getNext();
+ rawBatch = getNext(provider);
} catch (IOException e) {
context.fail(e);
return IterOutcome.STOP;
@@ -238,10 +267,10 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
// reached the end of an incoming record batch
RawFragmentBatch nextBatch = null;
try {
- nextBatch = fragProviders[node.batchId].getNext();
+ nextBatch = getNext(fragProviders[node.batchId]);
while (nextBatch != null && nextBatch.getHeader().getDef().getRecordCount() == 0) {
- nextBatch = fragProviders[node.batchId].getNext();
+ nextBatch = getNext(fragProviders[node.batchId]);
}
} catch (IOException e) {
context.fail(e);
@@ -301,6 +330,11 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
return IterOutcome.OK_NEW_SCHEMA;
else
return IterOutcome.OK;
+
+ }finally{
+ stats.stopProcessing();
+ }
+
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/669efeb3/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionRecordBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionRecordBatch.java
index 4641de6..f105363 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionRecordBatch.java
@@ -166,7 +166,7 @@ public class OrderedPartitionRecordBatch extends AbstractRecordBatch<OrderedPart
recordsSampled += incoming.getRecordCount();
outer: while (recordsSampled < recordsToSample) {
- upstream = incoming.next();
+ upstream = next(incoming);
switch (upstream) {
case NONE:
case NOT_YET:
@@ -414,97 +414,102 @@ public class OrderedPartitionRecordBatch extends AbstractRecordBatch<OrderedPart
@Override
public IterOutcome next() {
- container.zeroVectors();
-
- // if we got IterOutcome.NONE while getting partition vectors, and there are no batches on the queue, then we are
- // done
- if (upstreamNone && (batchQueue == null || batchQueue.size() == 0))
- return IterOutcome.NONE;
-
- // if there are batches on the queue, process them first, rather than calling incoming.next()
- if (batchQueue != null && batchQueue.size() > 0) {
- VectorContainer vc = batchQueue.poll();
- recordCount = vc.getRecordCount();
- try {
-
- // Must set up a new schema each time, because ValueVectors are not reused between containers in queue
- setupNewSchema(vc);
- } catch (SchemaChangeException ex) {
- kill();
- logger.error("Failure during query", ex);
- context.fail(ex);
- return IterOutcome.STOP;
+ stats.startProcessing();
+ try{
+ container.zeroVectors();
+
+ // if we got IterOutcome.NONE while getting partition vectors, and there are no batches on the queue, then we are
+ // done
+ if (upstreamNone && (batchQueue == null || batchQueue.size() == 0))
+ return IterOutcome.NONE;
+
+ // if there are batches on the queue, process them first, rather than calling incoming.next()
+ if (batchQueue != null && batchQueue.size() > 0) {
+ VectorContainer vc = batchQueue.poll();
+ recordCount = vc.getRecordCount();
+ try {
+
+ // Must set up a new schema each time, because ValueVectors are not reused between containers in queue
+ setupNewSchema(vc);
+ } catch (SchemaChangeException ex) {
+ kill();
+ logger.error("Failure during query", ex);
+ context.fail(ex);
+ return IterOutcome.STOP;
+ }
+ doWork(vc);
+ vc.zeroVectors();
+ return IterOutcome.OK_NEW_SCHEMA;
}
- doWork(vc);
- vc.zeroVectors();
- return IterOutcome.OK_NEW_SCHEMA;
- }
- // Reaching this point, either this is the first iteration, or there are no batches left on the queue and there are
- // more incoming
- IterOutcome upstream = incoming.next();
+ // Reaching this point, either this is the first iteration, or there are no batches left on the queue and there are
+ // more incoming
+ IterOutcome upstream = next(incoming);
- if (this.first && upstream == IterOutcome.OK) {
- throw new RuntimeException("Invalid state: First batch should have OK_NEW_SCHEMA");
- }
-
- // If this is the first iteration, we need to generate the partition vectors before we can proceed
- if (this.first && upstream == IterOutcome.OK_NEW_SCHEMA) {
- if (!getPartitionVectors()){
- cleanup();
- return IterOutcome.STOP;
+ if (this.first && upstream == IterOutcome.OK) {
+ throw new RuntimeException("Invalid state: First batch should have OK_NEW_SCHEMA");
}
- batchQueue = new LinkedBlockingQueue<>(this.sampledIncomingBatches);
- first = false;
-
- // Now that we have the partition vectors, we immediately process the first batch on the queue
- VectorContainer vc = batchQueue.poll();
- try {
- setupNewSchema(vc);
- } catch (SchemaChangeException ex) {
- kill();
- logger.error("Failure during query", ex);
- context.fail(ex);
- return IterOutcome.STOP;
+ // If this is the first iteration, we need to generate the partition vectors before we can proceed
+ if (this.first && upstream == IterOutcome.OK_NEW_SCHEMA) {
+ if (!getPartitionVectors()){
+ cleanup();
+ return IterOutcome.STOP;
+ }
+
+ batchQueue = new LinkedBlockingQueue<>(this.sampledIncomingBatches);
+ first = false;
+
+ // Now that we have the partition vectors, we immediately process the first batch on the queue
+ VectorContainer vc = batchQueue.poll();
+ try {
+ setupNewSchema(vc);
+ } catch (SchemaChangeException ex) {
+ kill();
+ logger.error("Failure during query", ex);
+ context.fail(ex);
+ return IterOutcome.STOP;
+ }
+ doWork(vc);
+ vc.zeroVectors();
+ recordCount = vc.getRecordCount();
+ return IterOutcome.OK_NEW_SCHEMA;
}
- doWork(vc);
- vc.zeroVectors();
- recordCount = vc.getRecordCount();
- return IterOutcome.OK_NEW_SCHEMA;
- }
- // if this now that all the batches on the queue are processed, we begin processing the incoming batches. For the
- // first one
- // we need to generate a new schema, even if the outcome is IterOutcome.OK After that we can reuse the schema.
- if (this.startedUnsampledBatches == false) {
- this.startedUnsampledBatches = true;
- if (upstream == IterOutcome.OK)
- upstream = IterOutcome.OK_NEW_SCHEMA;
- }
- switch (upstream) {
- case NONE:
- case NOT_YET:
- case STOP:
- cleanup();
- recordCount = 0;
- return upstream;
- case OK_NEW_SCHEMA:
- try {
- setupNewSchema(incoming);
- } catch (SchemaChangeException ex) {
- kill();
- logger.error("Failure during query", ex);
- context.fail(ex);
- return IterOutcome.STOP;
+ // if this now that all the batches on the queue are processed, we begin processing the incoming batches. For the
+ // first one
+ // we need to generate a new schema, even if the outcome is IterOutcome.OK After that we can reuse the schema.
+ if (this.startedUnsampledBatches == false) {
+ this.startedUnsampledBatches = true;
+ if (upstream == IterOutcome.OK)
+ upstream = IterOutcome.OK_NEW_SCHEMA;
+ }
+ switch (upstream) {
+ case NONE:
+ case NOT_YET:
+ case STOP:
+ cleanup();
+ recordCount = 0;
+ return upstream;
+ case OK_NEW_SCHEMA:
+ try {
+ setupNewSchema(incoming);
+ } catch (SchemaChangeException ex) {
+ kill();
+ logger.error("Failure during query", ex);
+ context.fail(ex);
+ return IterOutcome.STOP;
+ }
+ // fall through.
+ case OK:
+ doWork(incoming);
+ recordCount = incoming.getRecordCount();
+ return upstream; // change if upstream changed, otherwise normal.
+ default:
+ throw new UnsupportedOperationException();
}
- // fall through.
- case OK:
- doWork(incoming);
- recordCount = incoming.getRecordCount();
- return upstream; // change if upstream changed, otherwise normal.
- default:
- throw new UnsupportedOperationException();
+ }finally{
+ stats.stopProcessing();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/669efeb3/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 844d6db..b6b4c33 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,11 +20,10 @@ 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.ops.OperatorStats;
import org.apache.drill.exec.physical.base.PhysicalOperator;
import org.apache.drill.exec.record.selection.SelectionVector2;
import org.apache.drill.exec.record.selection.SelectionVector4;
@@ -36,12 +35,14 @@ public abstract class AbstractRecordBatch<T extends PhysicalOperator> implements
protected final T popConfig;
protected final FragmentContext context;
protected final OperatorContext oContext;
+ protected final OperatorStats stats;
protected AbstractRecordBatch(T popConfig, FragmentContext context) throws OutOfMemoryException {
super();
this.context = context;
this.popConfig = popConfig;
this.oContext = new OperatorContext(popConfig, context);
+ this.stats = oContext.getStats();
}
@Override
@@ -58,6 +59,27 @@ public abstract class AbstractRecordBatch<T extends PhysicalOperator> implements
return popConfig;
}
+ public final IterOutcome next(RecordBatch b){
+ return next(0, b);
+ }
+
+ public final IterOutcome next(int inputIndex, RecordBatch b){
+ stats.stopProcessing();
+ IterOutcome next = b.next();
+
+ switch(next){
+ case OK_NEW_SCHEMA:
+ stats.batchReceived(inputIndex, b.getRecordCount(), true);
+ break;
+ case OK:
+ stats.batchReceived(inputIndex, b.getRecordCount(), false);
+ break;
+ }
+
+ stats.startProcessing();
+ return next;
+ }
+
@Override
public BatchSchema getSchema() {
return container.getSchema();
@@ -75,6 +97,7 @@ public abstract class AbstractRecordBatch<T extends PhysicalOperator> implements
oContext.close();
}
+
@Override
public SelectionVector2 getSelectionVector2() {
throw new UnsupportedOperationException();
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/669efeb3/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 dd2cfe0..13e4ac8 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
@@ -29,7 +29,7 @@ public abstract class AbstractSingleRecordBatch<T extends PhysicalOperator> exte
protected final RecordBatch incoming;
private boolean first = true;
protected boolean outOfMemory = false;
-
+
public AbstractSingleRecordBatch(T popConfig, FragmentContext context, RecordBatch incoming) throws OutOfMemoryException {
super(popConfig, context);
this.incoming = incoming;
@@ -42,36 +42,46 @@ public abstract class AbstractSingleRecordBatch<T extends PhysicalOperator> exte
@Override
public IterOutcome next() {
- IterOutcome upstream = incoming.next();
- if(first && upstream == IterOutcome.OK) upstream = IterOutcome.OK_NEW_SCHEMA;
- first = false;
- switch(upstream){
- case NONE:
- case NOT_YET:
- case STOP:
- return upstream;
- case OUT_OF_MEMORY:
- return upstream;
- case OK_NEW_SCHEMA:
- try{
- setupNewSchema();
- }catch(SchemaChangeException ex){
- kill();
- logger.error("Failure during query", ex);
- context.fail(ex);
- return IterOutcome.STOP;
- }
- // fall through.
- case OK:
- doWork();
- if (outOfMemory) {
- outOfMemory = false;
- return IterOutcome.OUT_OF_MEMORY;
+ try{
+ stats.startProcessing();
+ IterOutcome upstream = next(incoming);
+ if(first && upstream == IterOutcome.OK) upstream = IterOutcome.OK_NEW_SCHEMA;
+ first = false;
+ switch(upstream){
+ case NONE:
+ case NOT_YET:
+ case STOP:
+ return upstream;
+ case OUT_OF_MEMORY:
+ return upstream;
+ case OK_NEW_SCHEMA:
+ try{
+ stats.startSetup();
+ setupNewSchema();
+ stats.stopSetup();
+ }catch(SchemaChangeException ex){
+ stats.stopSetup();
+ kill();
+ logger.error("Failure during query", ex);
+ context.fail(ex);
+ return IterOutcome.STOP;
+ }
+ // 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();
}
- return upstream; // change if upstream changed, otherwise normal.
- default:
- throw new UnsupportedOperationException();
+ }finally{
+ stats.stopProcessing();
}
+
+
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/669efeb3/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java
index 7f607a3..7297dc3 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java
@@ -22,7 +22,6 @@ import java.io.Closeable;
import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.cache.DistributedCache;
-import org.apache.drill.exec.cache.DistributedMultiMap;
import org.apache.drill.exec.cache.HazelCache;
import org.apache.drill.exec.coord.ClusterCoordinator;
import org.apache.drill.exec.coord.ClusterCoordinator.RegistrationHandle;
@@ -31,6 +30,8 @@ import org.apache.drill.exec.exception.DrillbitStartupException;
import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
import org.apache.drill.exec.service.ServiceEngine;
import org.apache.drill.exec.work.WorkManager;
+import org.eclipse.jetty.server.Server;
+import org.eclipse.jetty.servlet.ServletHandler;
import com.google.common.io.Closeables;
@@ -71,6 +72,7 @@ public class Drillbit implements Closeable{
final DistributedCache cache;
final WorkManager manager;
final BootStrapContext context;
+ final Server embeddedJetty;
private volatile RegistrationHandle handle;
@@ -89,8 +91,17 @@ public class Drillbit implements Closeable{
this.engine = new ServiceEngine(manager.getControlMessageHandler(), manager.getUserWorker(), context, manager.getWorkBus(), manager.getDataHandler());
this.cache = new HazelCache(config, context.getAllocator());
}
+ this.embeddedJetty = new Server(474747);
}
+ private void setupJetty(){
+ ServletHandler handler = new ServletHandler();
+ embeddedJetty.setHandler(handler);
+
+ }
+
+
+
public void run() throws Exception {
coord.start(10000);
DrillbitEndpoint md = engine.start();
@@ -99,6 +110,7 @@ public class Drillbit implements Closeable{
manager.getContext().getStorage().init();
manager.getContext().getOptionManager().init();
handle = coord.register(md);
+ embeddedJetty.start();
}
public void close() {
@@ -109,7 +121,11 @@ public class Drillbit implements Closeable{
} catch (InterruptedException e) {
logger.warn("Interrupted while sleeping during coordination deregistration.");
}
-
+ try {
+ embeddedJetty.stop();
+ } catch (Exception e) {
+ logger.warn("Failure while shutting down embedded jetty server.");
+ }
Closeables.closeQuietly(engine);
Closeables.closeQuietly(coord);
Closeables.closeQuietly(manager);
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/669efeb3/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/easy/EasyFormatPlugin.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/easy/EasyFormatPlugin.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/easy/EasyFormatPlugin.java
index 2965e79..718da23 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/easy/EasyFormatPlugin.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/easy/EasyFormatPlugin.java
@@ -35,6 +35,7 @@ import org.apache.drill.exec.physical.base.AbstractWriter;
import org.apache.drill.exec.physical.base.PhysicalOperator;
import org.apache.drill.exec.physical.impl.ScanBatch;
import org.apache.drill.exec.physical.impl.WriterRecordBatch;
+import org.apache.drill.exec.proto.UserBitShared.CoreOperatorType;
import org.apache.drill.exec.record.RecordBatch;
import org.apache.drill.exec.server.DrillbitContext;
import org.apache.drill.exec.store.RecordReader;
@@ -65,7 +66,7 @@ public abstract class EasyFormatPlugin<T extends FormatPluginConfig> implements
private final String name;
protected final CompressionCodecFactory codecFactory;
private final boolean compressible;
-
+
protected EasyFormatPlugin(String name, DrillbitContext context, DrillFileSystem fs, StoragePluginConfig storageConfig,
T formatConfig, boolean readable, boolean writable, boolean blockSplittable, boolean compressible, List<String> extensions, String defaultName){
this.matcher = new BasicFormatMatcher(this, fs, extensions, compressible);
@@ -80,7 +81,7 @@ public abstract class EasyFormatPlugin<T extends FormatPluginConfig> implements
this.name = name == null ? defaultName : name;
this.codecFactory = new CompressionCodecFactory(new Configuration(fs.getUnderlying().getConf()));
}
-
+
@Override
public DrillFileSystem getFileSystem() {
return fs;
@@ -90,7 +91,7 @@ public abstract class EasyFormatPlugin<T extends FormatPluginConfig> implements
public DrillbitContext getContext() {
return context;
}
-
+
@Override
public String getName() {
return name;
@@ -99,7 +100,7 @@ public abstract class EasyFormatPlugin<T extends FormatPluginConfig> implements
/**
* Whether or not you can split the format based on blocks within file boundaries. If not, the simple format engine will
* only split on file boundaries.
- *
+ *
* @return True if splittable.
*/
public boolean isBlockSplittable() {
@@ -184,7 +185,7 @@ public abstract class EasyFormatPlugin<T extends FormatPluginConfig> implements
public AbstractGroupScan getGroupScan(FileSelection selection, List<SchemaPath> columns) throws IOException {
return new EasyGroupScan(selection, this, columns, selection.selectionRoot);
}
-
+
@Override
public FormatPluginConfig getConfig() {
return formatConfig;
@@ -214,5 +215,8 @@ public abstract class EasyFormatPlugin<T extends FormatPluginConfig> implements
public Set<StoragePluginOptimizerRule> getOptimizerRules() {
return ImmutableSet.of();
}
-
+
+ public abstract int getReaderOperatorType();
+ public abstract int getWriterOperatorType();
+
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/669efeb3/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/easy/EasySubScan.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/easy/EasySubScan.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/easy/EasySubScan.java
index 0b3fe0f..5f9226e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/easy/EasySubScan.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/easy/EasySubScan.java
@@ -26,6 +26,7 @@ import org.apache.drill.common.expression.SchemaPath;
import org.apache.drill.common.logical.FormatPluginConfig;
import org.apache.drill.common.logical.StoragePluginConfig;
import org.apache.drill.exec.physical.base.AbstractSubScan;
+import org.apache.drill.exec.proto.UserBitShared.CoreOperatorType;
import org.apache.drill.exec.store.StoragePluginRegistry;
import org.apache.drill.exec.store.dfs.NamedFormatPluginConfig;
import org.apache.drill.exec.store.schedule.CompleteFileWork.FileWorkImpl;
@@ -45,13 +46,13 @@ public class EasySubScan extends AbstractSubScan{
private final EasyFormatPlugin<?> formatPlugin;
private final List<SchemaPath> columns;
private String selectionRoot;
-
+
@JsonCreator
public EasySubScan(
@JsonProperty("files") List<FileWorkImpl> files, //
@JsonProperty("storage") StoragePluginConfig storageConfig, //
@JsonProperty("format") FormatPluginConfig formatConfig, //
- @JacksonInject StoragePluginRegistry engineRegistry, //
+ @JacksonInject StoragePluginRegistry engineRegistry, //
@JsonProperty("columns") List<SchemaPath> columns, //
@JsonProperty("selectionRoot") String selectionRoot
) throws IOException, ExecutionSetupException {
@@ -62,7 +63,7 @@ public class EasySubScan extends AbstractSubScan{
this.columns = columns;
this.selectionRoot = selectionRoot;
}
-
+
public EasySubScan(List<FileWorkImpl> files, EasyFormatPlugin<?> plugin, List<SchemaPath> columns, String selectionRoot){
this.formatPlugin = plugin;
this.files = files;
@@ -74,7 +75,7 @@ public class EasySubScan extends AbstractSubScan{
public String getSelectionRoot() {
return selectionRoot;
}
-
+
@JsonIgnore
public EasyFormatPlugin<?> getFormatPlugin(){
return formatPlugin;
@@ -100,11 +101,15 @@ public class EasySubScan extends AbstractSubScan{
return formatPlugin.getConfig();
}
}
-
+
@JsonProperty("columns")
public List<SchemaPath> getColumns(){
return columns;
}
-
+ @Override
+ public int getOperatorType() {
+ return formatPlugin.getReaderOperatorType();
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/669efeb3/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/easy/EasyWriter.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/easy/EasyWriter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/easy/EasyWriter.java
index 864ae48..5ca781b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/easy/EasyWriter.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/easy/EasyWriter.java
@@ -94,4 +94,9 @@ public class EasyWriter extends AbstractWriter {
// TODO:
return new OperatorCost(1,1,1,1);
}
+
+ @Override
+ public int getOperatorType() {
+ return formatPlugin.getReaderOperatorType();
+ }
}