You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by ar...@apache.org on 2018/01/26 12:44:31 UTC
[07/11] drill git commit: DRILL-5730: Mock testing improvements and
interface improvements
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterBatchCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterBatchCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterBatchCreator.java
index e9b3051..ace4f24 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterBatchCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterBatchCreator.java
@@ -20,22 +20,18 @@ package org.apache.drill.exec.physical.impl.filter;
import java.util.List;
import org.apache.drill.common.exceptions.ExecutionSetupException;
-import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.ExecutorFragmentContext;
import org.apache.drill.exec.physical.config.Filter;
import org.apache.drill.exec.physical.impl.BatchCreator;
import org.apache.drill.exec.record.RecordBatch;
import com.google.common.base.Preconditions;
-public class FilterBatchCreator implements BatchCreator<Filter>{
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(FilterBatchCreator.class);
-
+public class FilterBatchCreator implements BatchCreator<Filter> {
@Override
- public FilterRecordBatch getBatch(FragmentContext context, Filter config, List<RecordBatch> children)
+ public FilterRecordBatch getBatch(ExecutorFragmentContext context, Filter config, List<RecordBatch> children)
throws ExecutionSetupException {
Preconditions.checkArgument(children.size() == 1);
return new FilterRecordBatch(config, children.iterator().next(), context);
}
-
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterRecordBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterRecordBatch.java
index 1bdd097..f0b832a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterRecordBatch.java
@@ -42,9 +42,7 @@ import org.apache.drill.exec.vector.ValueVector;
import com.google.common.collect.Lists;
-public class FilterRecordBatch extends AbstractSingleRecordBatch<Filter>{
- //private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(FilterRecordBatch.class);
-
+public class FilterRecordBatch extends AbstractSingleRecordBatch<Filter> {
private SelectionVector2 sv2;
private SelectionVector4 sv4;
private Filterer filter;
@@ -120,16 +118,7 @@ public class FilterRecordBatch extends AbstractSingleRecordBatch<Filter>{
* logic that handles SV4 + filter should always be pushed beyond sort so disabling
* it in FilterPrel.
*
-
- // set up the multi-batch selection vector
- this.svAllocator = oContext.getAllocator().getNewPreAllocator();
- if (!svAllocator.preAllocate(incoming.getRecordCount()*4))
- throw new SchemaChangeException("Attempted to filter an SV4 which exceeds allowed memory (" +
- incoming.getRecordCount() * 4 + " bytes)");
- sv4 = new SelectionVector4(svAllocator.getAllocation(), incoming.getRecordCount(), Character.MAX_VALUE);
- this.filter = generateSV4Filterer();
- break;
- */
+ */
default:
throw new UnsupportedOperationException();
}
@@ -197,8 +186,6 @@ public class FilterRecordBatch extends AbstractSingleRecordBatch<Filter>{
final TransferPair[] tx = transfers.toArray(new TransferPair[transfers.size()]);
CodeGenerator<Filterer> codeGen = cg.getCodeGenerator();
codeGen.plainJavaCapable(true);
- // Uncomment out this line to debug the generated code.
-// cg.saveCodeForDebugging(true);
final Filterer filter = context.getImplementationClass(codeGen);
filter.setup(context, incoming, this, tx);
return filter;
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterSignature.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterSignature.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterSignature.java
index 74a5d16..db62d36 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterSignature.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterSignature.java
@@ -20,12 +20,12 @@ package org.apache.drill.exec.physical.impl.filter;
import javax.inject.Named;
import org.apache.drill.exec.compile.sig.CodeGeneratorSignature;
-import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.FragmentContextImpl;
import org.apache.drill.exec.record.RecordBatch;
public interface FilterSignature extends CodeGeneratorSignature{
- public void doSetup(@Named("context") FragmentContext context, @Named("incoming") RecordBatch incoming, @Named("outgoing") RecordBatch outgoing);
+ public void doSetup(@Named("context") FragmentContextImpl context, @Named("incoming") RecordBatch incoming, @Named("outgoing") RecordBatch outgoing);
public boolean doEval(@Named("inIndex") int inIndex, @Named("outIndex") int outIndex);
}
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterTemplate2.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterTemplate2.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterTemplate2.java
index d014a2e..52533bd 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterTemplate2.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterTemplate2.java
@@ -27,9 +27,7 @@ import org.apache.drill.exec.record.RecordBatch;
import org.apache.drill.exec.record.TransferPair;
import org.apache.drill.exec.record.selection.SelectionVector2;
-public abstract class FilterTemplate2 implements Filterer{
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(FilterTemplate2.class);
-
+public abstract class FilterTemplate2 implements Filterer {
private SelectionVector2 outgoingSelectionVector;
private SelectionVector2 incomingSelectionVector;
private SelectionVectorMode svMode;
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterTemplate4.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterTemplate4.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterTemplate4.java
index fd1f9e6..4850cff 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterTemplate4.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/FilterTemplate4.java
@@ -26,8 +26,6 @@ import org.apache.drill.exec.record.TransferPair;
import org.apache.drill.exec.record.selection.SelectionVector4;
public abstract class FilterTemplate4 implements Filterer {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(FilterTemplate4.class);
-
private SelectionVector4 outgoingSelectionVector;
private SelectionVector4 incomingSelectionVector;
private TransferPair[] transfers;
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/Filterer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/Filterer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/Filterer.java
index aa45f54..a3d03c2 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/Filterer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/filter/Filterer.java
@@ -24,11 +24,9 @@ import org.apache.drill.exec.record.RecordBatch;
import org.apache.drill.exec.record.TransferPair;
public interface Filterer {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(Filterer.class);
+ TemplateClassDefinition<Filterer> TEMPLATE_DEFINITION2 = new TemplateClassDefinition<Filterer>(Filterer.class, FilterTemplate2.class);
+ TemplateClassDefinition<Filterer> TEMPLATE_DEFINITION4 = new TemplateClassDefinition<Filterer>(Filterer.class, FilterTemplate4.class);
- public void setup(FragmentContext context, RecordBatch incoming, RecordBatch outgoing, TransferPair[] transfers) throws SchemaChangeException;
- public void filterBatch(int recordCount) throws SchemaChangeException;
-
- public static TemplateClassDefinition<Filterer> TEMPLATE_DEFINITION2 = new TemplateClassDefinition<Filterer>(Filterer.class, FilterTemplate2.class);
- public static TemplateClassDefinition<Filterer> TEMPLATE_DEFINITION4 = new TemplateClassDefinition<Filterer>(Filterer.class, FilterTemplate4.class);
+ void setup(FragmentContext context, RecordBatch incoming, RecordBatch outgoing, TransferPair[] transfers) throws SchemaChangeException;
+ void filterBatch(int recordCount) throws SchemaChangeException;
}
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/flatten/FlattenBatchCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/flatten/FlattenBatchCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/flatten/FlattenBatchCreator.java
index 94203d8..bfda4f4 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/flatten/FlattenBatchCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/flatten/FlattenBatchCreator.java
@@ -20,21 +20,18 @@ package org.apache.drill.exec.physical.impl.flatten;
import java.util.List;
import org.apache.drill.common.exceptions.ExecutionSetupException;
-import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.ExecutorFragmentContext;
import org.apache.drill.exec.physical.config.FlattenPOP;
import org.apache.drill.exec.physical.impl.BatchCreator;
import org.apache.drill.exec.record.RecordBatch;
import com.google.common.base.Preconditions;
-public class FlattenBatchCreator implements BatchCreator<FlattenPOP>{
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(FlattenBatchCreator.class);
-
+public class FlattenBatchCreator implements BatchCreator<FlattenPOP> {
@Override
- public FlattenRecordBatch getBatch(FragmentContext context, FlattenPOP config, List<RecordBatch> children)
+ public FlattenRecordBatch getBatch(ExecutorFragmentContext context, FlattenPOP config, List<RecordBatch> children)
throws ExecutionSetupException {
Preconditions.checkArgument(children.size() == 1);
return new FlattenRecordBatch(config, children.iterator().next(), context);
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/flatten/FlattenRecordBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/flatten/FlattenRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/flatten/FlattenRecordBatch.java
index 2aa841b..8be16ad 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/flatten/FlattenRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/flatten/FlattenRecordBatch.java
@@ -303,8 +303,6 @@ public class FlattenRecordBatch extends AbstractSingleRecordBatch<FlattenPOP> {
final ClassGenerator<Flattener> cg = CodeGenerator.getRoot(Flattener.TEMPLATE_DEFINITION, context.getOptions());
cg.getCodeGenerator().plainJavaCapable(true);
- // Uncomment out this line to debug the generated code.
- // cg.getCodeGenerator().saveCodeForDebugging(true);
final IntHashSet transferFieldIds = new IntHashSet();
final NamedExpression flattenExpr = new NamedExpression(popConfig.getColumn(), new FieldReference(popConfig.getColumn()));
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/flatten/Flattener.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/flatten/Flattener.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/flatten/Flattener.java
index 5293060..392757e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/flatten/Flattener.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/flatten/Flattener.java
@@ -27,9 +27,11 @@ import org.apache.drill.exec.record.TransferPair;
import org.apache.drill.exec.vector.complex.RepeatedValueVector;
public interface Flattener {
- public void setup(FragmentContext context, RecordBatch incoming, RecordBatch outgoing, List<TransferPair> transfers) throws SchemaChangeException;
+ TemplateClassDefinition<Flattener> TEMPLATE_DEFINITION = new TemplateClassDefinition<Flattener>(Flattener.class, FlattenTemplate.class);
- public interface Monitor {
+ void setup(FragmentContext context, RecordBatch incoming, RecordBatch outgoing, List<TransferPair> transfers) throws SchemaChangeException;
+
+ interface Monitor {
/**
* Get the required buffer size for the specified number of records.
* {@see ValueVector#getBufferSizeFor(int)} for the meaning of this.
@@ -37,14 +39,14 @@ public interface Flattener {
* @param recordCount the number of records processed so far
* @return the buffer size the vectors report as being in use
*/
- public int getBufferSizeFor(int recordCount);
- };
+ int getBufferSizeFor(int recordCount);
+ }
+
+ int flattenRecords(int recordCount, int firstOutputIndex, Monitor monitor);
- public int flattenRecords(int recordCount, int firstOutputIndex, Monitor monitor);
+ void setFlattenField(RepeatedValueVector repeatedColumn);
- public void setFlattenField(RepeatedValueVector repeatedColumn);
- public RepeatedValueVector getFlattenField();
- public void resetGroupIndex();
+ RepeatedValueVector getFlattenField();
- public static final TemplateClassDefinition<Flattener> TEMPLATE_DEFINITION = new TemplateClassDefinition<Flattener>(Flattener.class, FlattenTemplate.class);
+ void resetGroupIndex();
}
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/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 7b679c0..e087bc8 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
@@ -256,12 +256,9 @@ public class HashJoinBatch extends AbstractBinaryRecordBatch<HashJoinPOP> {
// No more output records, clean up and return
state = BatchState.DONE;
- // if (first) {
- // return IterOutcome.OK_NEW_SCHEMA;
- // }
return IterOutcome.NONE;
} catch (ClassTransformationException | SchemaChangeException | IOException e) {
- context.fail(e);
+ context.getExecutorState().fail(e);
killIncoming(false);
return IterOutcome.STOP;
}
@@ -405,8 +402,6 @@ public class HashJoinBatch extends AbstractBinaryRecordBatch<HashJoinPOP> {
public HashJoinProbe setupHashJoinProbe() throws ClassTransformationException, IOException {
final CodeGenerator<HashJoinProbe> cg = CodeGenerator.get(HashJoinProbe.TEMPLATE_DEFINITION, context.getOptions());
cg.plainJavaCapable(true);
- // Uncomment out this line to debug the generated code.
- // cg.saveCodeForDebugging(true);
final ClassGenerator<HashJoinProbe> g = cg.getRoot();
// Generate the code to project build side records
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinBatchCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinBatchCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinBatchCreator.java
index 1402769..a005559 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinBatchCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinBatchCreator.java
@@ -20,7 +20,7 @@ package org.apache.drill.exec.physical.impl.join;
import java.util.List;
import org.apache.drill.common.exceptions.ExecutionSetupException;
-import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.ExecutorFragmentContext;
import org.apache.drill.exec.physical.config.HashJoinPOP;
import org.apache.drill.exec.physical.impl.BatchCreator;
import org.apache.drill.exec.record.RecordBatch;
@@ -30,7 +30,7 @@ import com.google.common.base.Preconditions;
public class HashJoinBatchCreator implements BatchCreator<HashJoinPOP> {
@Override
- public HashJoinBatch getBatch(FragmentContext context, HashJoinPOP config, List<RecordBatch> children)
+ public HashJoinBatch getBatch(ExecutorFragmentContext context, HashJoinPOP config, List<RecordBatch> children)
throws ExecutionSetupException {
Preconditions.checkArgument(children.size() == 2);
return new HashJoinBatch(config, context, children.get(0), children.get(1));
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinProbe.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinProbe.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinProbe.java
index cc6bd55..4ef28e6 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinProbe.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinProbe.java
@@ -30,7 +30,7 @@ import org.apache.drill.exec.record.VectorContainer;
import org.apache.calcite.rel.core.JoinRelType;
public interface HashJoinProbe {
- public static TemplateClassDefinition<HashJoinProbe> TEMPLATE_DEFINITION = new TemplateClassDefinition<HashJoinProbe>(HashJoinProbe.class, HashJoinProbeTemplate.class);
+ TemplateClassDefinition<HashJoinProbe> TEMPLATE_DEFINITION = new TemplateClassDefinition<HashJoinProbe>(HashJoinProbe.class, HashJoinProbeTemplate.class);
/* The probe side of the hash join can be in the following two states
* 1. PROBE_PROJECT: Inner join case, we probe our hash table to see if we have a
@@ -40,15 +40,15 @@ public interface HashJoinProbe {
* case we handle it internally by projecting the record if there isn't a match on the build side
* 3. DONE: Once we have projected all possible records we are done
*/
- public static enum ProbeState {
+ enum ProbeState {
PROBE_PROJECT, PROJECT_RIGHT, DONE
}
- public abstract void setupHashJoinProbe(FragmentContext context, VectorContainer buildBatch, RecordBatch probeBatch,
- int probeRecordCount, HashJoinBatch outgoing, HashTable hashTable, HashJoinHelper hjHelper,
- JoinRelType joinRelType);
- public abstract void doSetup(FragmentContext context, VectorContainer buildBatch, RecordBatch probeBatch, RecordBatch outgoing);
- public abstract int probeAndProject() throws SchemaChangeException, ClassTransformationException, IOException;
- public abstract void projectBuildRecord(int buildIndex, int outIndex);
- public abstract void projectProbeRecord(int probeIndex, int outIndex);
+ void setupHashJoinProbe(FragmentContext context, VectorContainer buildBatch, RecordBatch probeBatch,
+ int probeRecordCount, HashJoinBatch outgoing, HashTable hashTable, HashJoinHelper hjHelper,
+ JoinRelType joinRelType);
+ void doSetup(FragmentContext context, VectorContainer buildBatch, RecordBatch probeBatch, RecordBatch outgoing);
+ int probeAndProject() throws SchemaChangeException, ClassTransformationException, IOException;
+ void projectBuildRecord(int buildIndex, int outIndex);
+ void projectProbeRecord(int probeIndex, int outIndex);
}
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinWorker.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinWorker.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinWorker.java
index 55322f8..95f7c3d 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinWorker.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinWorker.java
@@ -24,15 +24,12 @@ import org.apache.drill.exec.record.VectorContainer;
public interface JoinWorker {
+ TemplateClassDefinition<JoinWorker> TEMPLATE_DEFINITION = new TemplateClassDefinition<>(JoinWorker.class, JoinTemplate.class);
- public static enum JoinOutcome {
+ enum JoinOutcome {
NO_MORE_DATA, BATCH_RETURNED, SCHEMA_CHANGED, WAITING, FAILURE;
}
- public void setupJoin(FragmentContext context, JoinStatus status, VectorContainer outgoing) throws SchemaChangeException;
- public boolean doJoin(JoinStatus status);
-
- public static TemplateClassDefinition<JoinWorker> TEMPLATE_DEFINITION = new TemplateClassDefinition<>(
- JoinWorker.class, JoinTemplate.class);
-
+ void setupJoin(FragmentContext context, JoinStatus status, VectorContainer outgoing) throws SchemaChangeException;
+ boolean doJoin(JoinStatus status);
}
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/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 8ad3f84..1ed4722 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
@@ -199,7 +199,7 @@ public class MergeJoinBatch extends AbstractRecordBatch<MergeJoinPOP> {
this.worker = generateNewWorker();
first = true;
} catch (ClassTransformationException | IOException | SchemaChangeException e) {
- context.fail(new SchemaChangeException(e));
+ context.getExecutorState().fail(new SchemaChangeException(e));
kill(false);
return IterOutcome.STOP;
} finally {
@@ -269,12 +269,9 @@ public class MergeJoinBatch extends AbstractRecordBatch<MergeJoinPOP> {
right.kill(sendUpstream);
}
- private JoinWorker generateNewWorker() throws ClassTransformationException, IOException, SchemaChangeException{
-
+ private JoinWorker generateNewWorker() throws ClassTransformationException, IOException, SchemaChangeException {
final ClassGenerator<JoinWorker> cg = CodeGenerator.getRoot(JoinWorker.TEMPLATE_DEFINITION, context.getOptions());
cg.getCodeGenerator().plainJavaCapable(true);
- // Uncomment out this line to debug the generated code.
- // cg.getCodeGenerator().saveCodeForDebugging(true);
final ErrorCollector collector = new ErrorCollectorImpl();
// Generate members and initialization code
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/MergeJoinCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/MergeJoinCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/MergeJoinCreator.java
index 24f5533..b24624e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/MergeJoinCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/MergeJoinCreator.java
@@ -20,7 +20,7 @@ package org.apache.drill.exec.physical.impl.join;
import java.util.List;
import org.apache.drill.common.exceptions.ExecutionSetupException;
-import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.ExecutorFragmentContext;
import org.apache.drill.exec.physical.config.MergeJoinPOP;
import org.apache.drill.exec.physical.impl.BatchCreator;
import org.apache.drill.exec.record.RecordBatch;
@@ -29,10 +29,8 @@ import org.apache.calcite.rel.core.JoinRelType;
import com.google.common.base.Preconditions;
public class MergeJoinCreator implements BatchCreator<MergeJoinPOP> {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(MergeJoinCreator.class);
-
@Override
- public MergeJoinBatch getBatch(FragmentContext context, MergeJoinPOP config, List<RecordBatch> children)
+ public MergeJoinBatch getBatch(ExecutorFragmentContext context, MergeJoinPOP config, List<RecordBatch> children)
throws ExecutionSetupException {
Preconditions.checkArgument(children.size() == 2);
if(config.getJoinType() == JoinRelType.RIGHT){
@@ -40,6 +38,5 @@ public class MergeJoinCreator implements BatchCreator<MergeJoinPOP> {
}else{
return new MergeJoinBatch(config, context, children.get(0), children.get(1));
}
-
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/NestedLoopJoinBatchCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/NestedLoopJoinBatchCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/NestedLoopJoinBatchCreator.java
index 2e708a6..ef4cab1 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/NestedLoopJoinBatchCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/NestedLoopJoinBatchCreator.java
@@ -20,14 +20,14 @@ package org.apache.drill.exec.physical.impl.join;
import java.util.List;
import org.apache.drill.common.exceptions.ExecutionSetupException;
-import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.ExecutorFragmentContext;
import org.apache.drill.exec.physical.config.NestedLoopJoinPOP;
import org.apache.drill.exec.physical.impl.BatchCreator;
import org.apache.drill.exec.record.RecordBatch;
public class NestedLoopJoinBatchCreator implements BatchCreator<NestedLoopJoinPOP> {
@Override
- public NestedLoopJoinBatch getBatch(FragmentContext context, NestedLoopJoinPOP config, List<RecordBatch> children)
+ public NestedLoopJoinBatch getBatch(ExecutorFragmentContext context, NestedLoopJoinPOP config, List<RecordBatch> children)
throws ExecutionSetupException {
return new NestedLoopJoinBatch(config, context, children.get(0), children.get(1));
}
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/limit/LimitBatchCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/limit/LimitBatchCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/limit/LimitBatchCreator.java
index f954e72..15e5275 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/limit/LimitBatchCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/limit/LimitBatchCreator.java
@@ -20,7 +20,7 @@ package org.apache.drill.exec.physical.impl.limit;
import java.util.List;
import org.apache.drill.common.exceptions.ExecutionSetupException;
-import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.ExecutorFragmentContext;
import org.apache.drill.exec.physical.config.Limit;
import org.apache.drill.exec.physical.impl.BatchCreator;
import org.apache.drill.exec.record.RecordBatch;
@@ -29,7 +29,7 @@ import com.google.common.collect.Iterables;
public class LimitBatchCreator implements BatchCreator<Limit> {
@Override
- public LimitRecordBatch getBatch(FragmentContext context, Limit config, List<RecordBatch> children)
+ public LimitRecordBatch getBatch(ExecutorFragmentContext context, Limit config, List<RecordBatch> children)
throws ExecutionSetupException {
return new LimitRecordBatch(config, context, Iterables.getOnlyElement(children));
}
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/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 f9ceff2..7e5ff21 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
@@ -41,6 +41,7 @@ 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.ops.ExchangeFragmentContext;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.ops.MetricDef;
import org.apache.drill.exec.physical.MinorFragmentEndpoint;
@@ -96,7 +97,7 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
private RecordBatchLoader[] batchLoaders;
private final RawFragmentBatchProvider[] fragProviders;
- private final FragmentContext context;
+ private final ExchangeFragmentContext context;
private VectorContainer outgoingContainer;
private MergingReceiverGeneratorBase merger;
private final MergingReceiverPOP config;
@@ -109,12 +110,11 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
private RawFragmentBatch[] incomingBatches;
private int[] batchOffsets;
private PriorityQueue <Node> pqueue;
- private RawFragmentBatch emptyBatch = null;
private RawFragmentBatch[] tempBatchHolder;
private long[] inputCounts;
private long[] outputCounts;
- public static enum Metric implements MetricDef{
+ public enum Metric implements MetricDef {
BYTES_RECEIVED,
NUM_SENDERS,
NEXT_WAIT_NANOS;
@@ -125,7 +125,7 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
}
}
- public MergingRecordBatch(final FragmentContext context,
+ public MergingRecordBatch(final ExchangeFragmentContext context,
final MergingReceiverPOP config,
final RawFragmentBatchProvider[] fragProviders) throws OutOfMemoryException {
super(config, context, true, context.newOperatorContext(config));
@@ -210,11 +210,11 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
try {
rawBatch = getNext(p);
} catch (final IOException e) {
- context.fail(e);
+ context.getExecutorState().fail(e);
return IterOutcome.STOP;
}
}
- if (rawBatch == null && !context.shouldContinue()) {
+ if (rawBatch == null && !context.getExecutorState().shouldContinue()) {
clearBatches(rawBatches);
return IterOutcome.STOP;
}
@@ -241,12 +241,12 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
while ((rawBatch = getNext(p)) != null && rawBatch.getHeader().getDef().getRecordCount() == 0) {
// Do nothing
}
- if (rawBatch == null && !context.shouldContinue()) {
+ if (rawBatch == null && !context.getExecutorState().shouldContinue()) {
clearBatches(rawBatches);
return IterOutcome.STOP;
}
} catch (final IOException e) {
- context.fail(e);
+ context.getExecutorState().fail(e);
clearBatches(rawBatches);
return IterOutcome.STOP;
}
@@ -315,7 +315,7 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
// SchemaChangeException, so check/clean catch clause below.
} catch(final SchemaChangeException e) {
logger.error("MergingReceiver failed to load record batch from remote host. {}", e);
- context.fail(e);
+ context.getExecutorState().fail(e);
return IterOutcome.STOP;
}
batch.release();
@@ -328,7 +328,7 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
// Ensure all the incoming batches have the identical schema.
// Note: RecordBatchLoader permutes the columns to obtain the same columns order for all batches.
if (!isSameSchemaAmongBatches(batchLoaders)) {
- context.fail(new SchemaChangeException("Incoming batches for merging receiver have different schemas!"));
+ context.getExecutorState().fail(new SchemaChangeException("Incoming batches for merging receiver have different schemas!"));
return IterOutcome.STOP;
}
@@ -351,7 +351,7 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
merger = createMerger();
} catch (final SchemaChangeException e) {
logger.error("Failed to generate code for MergingReceiver. {}", e);
- context.fail(e);
+ context.getExecutorState().fail(e);
return IterOutcome.STOP;
}
@@ -380,12 +380,12 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
} else {
batchLoaders[b].clear();
batchLoaders[b] = null;
- if (!context.shouldContinue()) {
+ if (!context.getExecutorState().shouldContinue()) {
return IterOutcome.STOP;
}
}
} catch (IOException | SchemaChangeException e) {
- context.fail(e);
+ context.getExecutorState().fail(e);
return IterOutcome.STOP;
}
}
@@ -418,11 +418,11 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
assert nextBatch != null || inputCounts[node.batchId] == outputCounts[node.batchId]
: String.format("Stream %d input count: %d output count %d", node.batchId, inputCounts[node.batchId], outputCounts[node.batchId]);
- if (nextBatch == null && !context.shouldContinue()) {
+ if (nextBatch == null && !context.getExecutorState().shouldContinue()) {
return IterOutcome.STOP;
}
} catch (final IOException e) {
- context.fail(e);
+ context.getExecutorState().fail(e);
return IterOutcome.STOP;
}
@@ -456,7 +456,7 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
// TODO: Clean: DRILL-2933: That load(...) no longer throws
// SchemaChangeException, so check/clean catch clause below.
} catch(final SchemaChangeException ex) {
- context.fail(ex);
+ context.getExecutorState().fail(ex);
return IterOutcome.STOP;
}
incomingBatches[node.batchId].release();
@@ -548,7 +548,7 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
}
final RawFragmentBatch batch = getNext(i);
if (batch == null) {
- if (!context.shouldContinue()) {
+ if (!context.getExecutorState().shouldContinue()) {
state = BatchState.STOP;
} else {
state = BatchState.DONE;
@@ -605,7 +605,9 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
.setReceiver(context.getHandle())
.setSender(sender)
.build();
- context.getControlTunnel(providingEndpoint.getEndpoint()).informReceiverFinished(new OutcomeListener(), finishedReceiver);
+ context.getController()
+ .getTunnel(providingEndpoint.getEndpoint())
+ .informReceiverFinished(new OutcomeListener(), finishedReceiver);
}
}
@@ -624,10 +626,10 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
@Override
public void interrupted(final InterruptedException e) {
- if (context.shouldContinue()) {
+ if (context.getExecutorState().shouldContinue()) {
final String errMsg = "Received an interrupt RPC outcome while sending ReceiverFinished message";
logger.error(errMsg, e);
- context.fail(new RpcException(errMsg, e));
+ context.getExecutorState().fail(new RpcException(errMsg, e));
}
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/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 7f662ae..9e82af8 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
@@ -196,7 +196,7 @@ public class OrderedPartitionRecordBatch extends AbstractRecordBatch<OrderedPart
}
}
VectorContainer sortedSamples = new VectorContainer();
- builder.build(context, sortedSamples);
+ builder.build(sortedSamples);
// Sort the records according the orderings given in the configuration
@@ -262,7 +262,7 @@ public class OrderedPartitionRecordBatch extends AbstractRecordBatch<OrderedPart
Thread.sleep(timeout);
return true;
} catch (final InterruptedException e) {
- if (!context.shouldContinue()) {
+ if (!context.getExecutorState().shouldContinue()) {
return false;
}
}
@@ -329,7 +329,7 @@ public class OrderedPartitionRecordBatch extends AbstractRecordBatch<OrderedPart
} catch (final ClassTransformationException | IOException | SchemaChangeException ex) {
kill(false);
- context.fail(ex);
+ context.getExecutorState().fail(ex);
return false;
// TODO InterruptedException
}
@@ -349,7 +349,7 @@ public class OrderedPartitionRecordBatch extends AbstractRecordBatch<OrderedPart
for (CachedVectorContainer w : mmap.get(mapKey)) {
containerBuilder.add(w.get());
}
- containerBuilder.build(context, allSamplesContainer);
+ containerBuilder.build(allSamplesContainer);
List<Ordering> orderDefs = Lists.newArrayList();
int i = 0;
@@ -390,7 +390,7 @@ public class OrderedPartitionRecordBatch extends AbstractRecordBatch<OrderedPart
candidatePartitionTable.setRecordCount(copier.getOutputRecords());
@SuppressWarnings("resource")
WritableBatch batch = WritableBatch.getBatchNoHVWrap(candidatePartitionTable.getRecordCount(), candidatePartitionTable, false);
- wrap = new CachedVectorContainer(batch, context.getDrillbitContext().getAllocator());
+ wrap = new CachedVectorContainer(batch, context.getAllocator());
tableMap.putIfAbsent(mapKey + "final", wrap, 1, TimeUnit.MINUTES);
} finally {
candidatePartitionTable.clear();
@@ -486,7 +486,7 @@ public class OrderedPartitionRecordBatch extends AbstractRecordBatch<OrderedPart
} catch (SchemaChangeException ex) {
kill(false);
logger.error("Failure during query", ex);
- context.fail(ex);
+ context.getExecutorState().fail(ex);
return IterOutcome.STOP;
}
doWork(vc);
@@ -519,7 +519,7 @@ public class OrderedPartitionRecordBatch extends AbstractRecordBatch<OrderedPart
} catch (SchemaChangeException ex) {
kill(false);
logger.error("Failure during query", ex);
- context.fail(ex);
+ context.getExecutorState().fail(ex);
return IterOutcome.STOP;
}
doWork(vc);
@@ -550,7 +550,7 @@ public class OrderedPartitionRecordBatch extends AbstractRecordBatch<OrderedPart
} catch (SchemaChangeException ex) {
kill(false);
logger.error("Failure during query", ex);
- context.fail(ex);
+ context.getExecutorState().fail(ex);
return IterOutcome.STOP;
}
// fall through.
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionSenderCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionSenderCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionSenderCreator.java
index d2e07e7..5705aca 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionSenderCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionSenderCreator.java
@@ -20,7 +20,7 @@ package org.apache.drill.exec.physical.impl.orderedpartitioner;
import java.util.List;
import org.apache.drill.common.exceptions.ExecutionSetupException;
-import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.ExecutorFragmentContext;
import org.apache.drill.exec.physical.config.HashPartitionSender;
import org.apache.drill.exec.physical.config.OrderedPartitionSender;
import org.apache.drill.exec.physical.impl.RootCreator;
@@ -61,7 +61,7 @@ public class OrderedPartitionSenderCreator implements RootCreator<OrderedPartiti
@SuppressWarnings("resource")
@Override
- public RootExec getRoot(FragmentContext context, OrderedPartitionSender config,
+ public RootExec getRoot(ExecutorFragmentContext context, OrderedPartitionSender config,
List<RecordBatch> children) throws ExecutionSetupException {
Preconditions.checkArgument(children.size() == 1);
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionSenderCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionSenderCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionSenderCreator.java
index 06fd115..e0b7b9a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionSenderCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionSenderCreator.java
@@ -20,7 +20,7 @@ package org.apache.drill.exec.physical.impl.partitionsender;
import java.util.List;
import org.apache.drill.common.exceptions.ExecutionSetupException;
-import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.ExecutorFragmentContext;
import org.apache.drill.exec.physical.config.HashPartitionSender;
import org.apache.drill.exec.physical.impl.RootCreator;
import org.apache.drill.exec.physical.impl.RootExec;
@@ -29,7 +29,7 @@ import org.apache.drill.exec.record.RecordBatch;
public class PartitionSenderCreator implements RootCreator<HashPartitionSender> {
@Override
- public RootExec getRoot(FragmentContext context,
+ public RootExec getRoot(ExecutorFragmentContext context,
HashPartitionSender config,
List<RecordBatch> children) throws ExecutionSetupException {
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/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 108d539..25be50a 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
@@ -33,9 +33,10 @@ 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.ops.AccountingDataTunnel;
-import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.ExchangeFragmentContext;
import org.apache.drill.exec.ops.MetricDef;
import org.apache.drill.exec.ops.OperatorStats;
+import org.apache.drill.exec.ops.RootFragmentContext;
import org.apache.drill.exec.physical.MinorFragmentEndpoint;
import org.apache.drill.exec.physical.config.HashPartitionSender;
import org.apache.drill.exec.physical.impl.BaseRootExec;
@@ -63,7 +64,7 @@ public class PartitionSenderRootExec extends BaseRootExec {
private HashPartitionSender operator;
private PartitionerDecorator partitioner;
- private FragmentContext context;
+ private ExchangeFragmentContext context;
private boolean ok = true;
private final int outGoingBatchCount;
private final HashPartitionSender popConfig;
@@ -98,13 +99,13 @@ public class PartitionSenderRootExec extends BaseRootExec {
}
}
- public PartitionSenderRootExec(FragmentContext context,
+ public PartitionSenderRootExec(RootFragmentContext context,
RecordBatch incoming,
HashPartitionSender operator) throws OutOfMemoryException {
this(context, incoming, operator, false);
}
- public PartitionSenderRootExec(FragmentContext context,
+ public PartitionSenderRootExec(RootFragmentContext context,
RecordBatch incoming,
HashPartitionSender operator,
boolean closeIncoming) throws OutOfMemoryException {
@@ -173,7 +174,7 @@ public class PartitionSenderRootExec extends BaseRootExec {
} catch (IOException e) {
incoming.kill(false);
logger.error("Error while creating partitioning sender or flushing outgoing batches", e);
- context.fail(e);
+ context.getExecutorState().fail(e);
}
return false;
@@ -203,19 +204,19 @@ public class PartitionSenderRootExec extends BaseRootExec {
} catch (IOException e) {
incoming.kill(false);
logger.error("Error while flushing outgoing batches", e);
- context.fail(e);
+ context.getExecutorState().fail(e);
return false;
} catch (SchemaChangeException e) {
incoming.kill(false);
logger.error("Error while setting up partitioner", e);
- context.fail(e);
+ context.getExecutorState().fail(e);
return false;
}
case OK:
try {
partitioner.partitionBatch(incoming);
} catch (IOException e) {
- context.fail(e);
+ context.getExecutorState().fail(e);
incoming.kill(false);
return false;
}
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/Partitioner.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/Partitioner.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/Partitioner.java
index 95a4813..5d1b08c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/Partitioner.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/Partitioner.java
@@ -22,34 +22,33 @@ import java.util.List;
import org.apache.drill.exec.compile.TemplateClassDefinition;
import org.apache.drill.exec.exception.SchemaChangeException;
-import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.ExchangeFragmentContext;
import org.apache.drill.exec.ops.OperatorContext;
import org.apache.drill.exec.ops.OperatorStats;
import org.apache.drill.exec.physical.config.HashPartitionSender;
import org.apache.drill.exec.record.RecordBatch;
public interface Partitioner {
+ void setup(ExchangeFragmentContext context,
+ RecordBatch incoming,
+ HashPartitionSender popConfig,
+ OperatorStats stats,
+ OperatorContext oContext,
+ int start, int count) throws SchemaChangeException;
- public abstract void setup(FragmentContext context,
- RecordBatch incoming,
- HashPartitionSender popConfig,
- OperatorStats stats,
- OperatorContext oContext,
- int start, int count) throws SchemaChangeException;
-
- public abstract void partitionBatch(RecordBatch incoming) throws IOException;
- public abstract void flushOutgoingBatches(boolean isLastBatch, boolean schemaChanged) throws IOException;
- public abstract void initialize();
- public abstract void clear();
- public abstract List<? extends PartitionOutgoingBatch> getOutgoingBatches();
+ void partitionBatch(RecordBatch incoming) throws IOException;
+ void flushOutgoingBatches(boolean isLastBatch, boolean schemaChanged) throws IOException;
+ void initialize();
+ void clear();
+ List<? extends PartitionOutgoingBatch> getOutgoingBatches();
/**
* Method to get PartitionOutgoingBatch based on the fact that there can be > 1 Partitioner
* @param index
* @return PartitionOutgoingBatch that matches index within Partitioner. This method can
* return null if index does not fall within boundary of this Partitioner
*/
- public abstract PartitionOutgoingBatch getOutgoingBatch(int index);
- public abstract OperatorStats getStats();
+ PartitionOutgoingBatch getOutgoingBatch(int index);
+ OperatorStats getStats();
- public static TemplateClassDefinition<Partitioner> TEMPLATE_DEFINITION = new TemplateClassDefinition<>(Partitioner.class, PartitionerTemplate.class);
+ TemplateClassDefinition<Partitioner> TEMPLATE_DEFINITION = new TemplateClassDefinition<>(Partitioner.class, PartitionerTemplate.class);
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionerDecorator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionerDecorator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionerDecorator.java
index 042222a..78b8d03 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionerDecorator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionerDecorator.java
@@ -57,7 +57,7 @@ public class PartitionerDecorator {
this.partitioners = partitioners;
this.stats = stats;
this.context = context;
- this.executor = context.getDrillbitContext().getExecutor();
+ this.executor = context.getExecutor();
this.tName = Thread.currentThread().getName();
this.childThreadPrefix = "Partitioner-" + tName + "-";
}
@@ -177,7 +177,7 @@ public class PartitionerDecorator {
break;
} catch (final InterruptedException e) {
// If the fragment state says we shouldn't continue, cancel or interrupt partitioner threads
- if (!context.shouldContinue()) {
+ if (!context.getExecutorState().shouldContinue()) {
logger.debug("Interrupting partioner threads. Fragment thread {}", tName);
for(Future<?> f : taskFutures) {
f.cancel(true);
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionerTemplate.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionerTemplate.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionerTemplate.java
index aa72c44..0d52b53 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionerTemplate.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionerTemplate.java
@@ -29,6 +29,7 @@ import org.apache.drill.exec.exception.SchemaChangeException;
import org.apache.drill.exec.expr.TypeHelper;
import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.ops.AccountingDataTunnel;
+import org.apache.drill.exec.ops.ExchangeFragmentContext;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.ops.OperatorContext;
import org.apache.drill.exec.ops.OperatorStats;
@@ -81,7 +82,7 @@ public abstract class PartitionerTemplate implements Partitioner {
}
@Override
- public final void setup(FragmentContext context,
+ public final void setup(ExchangeFragmentContext context,
RecordBatch incoming,
HashPartitionSender popConfig,
OperatorStats stats,
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/producer/ProducerConsumerBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/producer/ProducerConsumerBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/producer/ProducerConsumerBatch.java
index 3afa852..bbcb758 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/producer/ProducerConsumerBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/producer/ProducerConsumerBatch.java
@@ -68,8 +68,8 @@ public class ProducerConsumerBatch extends AbstractRecordBatch<ProducerConsumer>
wrapper = queue.take();
logger.debug("Got batch from queue");
} catch (final InterruptedException e) {
- if (context.shouldContinue()) {
- context.fail(e);
+ if (context.getExecutorState().shouldContinue()) {
+ context.getExecutorState().fail(e);
}
return IterOutcome.STOP;
// TODO InterruptedException
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/producer/ProducerConsumerBatchCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/producer/ProducerConsumerBatchCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/producer/ProducerConsumerBatchCreator.java
index 6542576..779728a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/producer/ProducerConsumerBatchCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/producer/ProducerConsumerBatchCreator.java
@@ -20,7 +20,7 @@ package org.apache.drill.exec.physical.impl.producer;
import java.util.List;
import org.apache.drill.common.exceptions.ExecutionSetupException;
-import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.ExecutorFragmentContext;
import org.apache.drill.exec.physical.config.ProducerConsumer;
import org.apache.drill.exec.physical.impl.BatchCreator;
import org.apache.drill.exec.record.RecordBatch;
@@ -29,7 +29,7 @@ import com.google.common.collect.Iterables;
public class ProducerConsumerBatchCreator implements BatchCreator<ProducerConsumer> {
@Override
- public ProducerConsumerBatch getBatch(FragmentContext context, ProducerConsumer config, List<RecordBatch> children)
+ public ProducerConsumerBatch getBatch(ExecutorFragmentContext context, ProducerConsumer config, List<RecordBatch> children)
throws ExecutionSetupException {
return new ProducerConsumerBatch(config, context, Iterables.getOnlyElement(children));
}
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ComplexToJsonBatchCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ComplexToJsonBatchCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ComplexToJsonBatchCreator.java
index f249540..73ab441 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ComplexToJsonBatchCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ComplexToJsonBatchCreator.java
@@ -20,7 +20,7 @@ package org.apache.drill.exec.physical.impl.project;
import java.util.List;
import org.apache.drill.common.exceptions.ExecutionSetupException;
-import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.ExecutorFragmentContext;
import org.apache.drill.exec.physical.config.ComplexToJson;
import org.apache.drill.exec.physical.config.Project;
import org.apache.drill.exec.physical.impl.BatchCreator;
@@ -29,15 +29,12 @@ import org.apache.drill.exec.record.RecordBatch;
import com.google.common.base.Preconditions;
public class ComplexToJsonBatchCreator implements BatchCreator<ComplexToJson> {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ComplexToJsonBatchCreator.class);
-
@Override
- public ProjectRecordBatch getBatch(FragmentContext context, ComplexToJson flatten, List<RecordBatch> children)
+ public ProjectRecordBatch getBatch(ExecutorFragmentContext context, ComplexToJson flatten, List<RecordBatch> children)
throws ExecutionSetupException {
Preconditions.checkArgument(children.size() == 1);
return new ProjectRecordBatch(new Project(null, flatten.getChild()),
children.iterator().next(),
context);
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectBatchCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectBatchCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectBatchCreator.java
index e7a6b05..37753cd 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectBatchCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectBatchCreator.java
@@ -20,7 +20,7 @@ package org.apache.drill.exec.physical.impl.project;
import java.util.List;
import org.apache.drill.common.exceptions.ExecutionSetupException;
-import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.ExecutorFragmentContext;
import org.apache.drill.exec.physical.config.Project;
import org.apache.drill.exec.physical.impl.BatchCreator;
import org.apache.drill.exec.record.RecordBatch;
@@ -28,13 +28,10 @@ import org.apache.drill.exec.record.RecordBatch;
import com.google.common.base.Preconditions;
public class ProjectBatchCreator implements BatchCreator<Project>{
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ProjectBatchCreator.class);
-
@Override
- public ProjectRecordBatch getBatch(FragmentContext context, Project config, List<RecordBatch> children)
+ public ProjectRecordBatch getBatch(ExecutorFragmentContext context, Project config, List<RecordBatch> children)
throws ExecutionSetupException {
Preconditions.checkArgument(children.size() == 1);
return new ProjectRecordBatch(config, children.iterator().next(), context);
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/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 be0f61f..89e0ee9 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
@@ -804,7 +804,7 @@ public class ProjectRecordBatch extends AbstractSingleRecordBatch<Project> {
} catch (SchemaChangeException e) {
kill(false);
logger.error("Failure during query", e);
- context.fail(e);
+ context.getExecutorState().fail(e);
return IterOutcome.STOP;
}
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/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 d711592..f38b62e 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
@@ -132,7 +132,7 @@ public class SortBatch extends AbstractRecordBatch<Sort> {
return IterOutcome.NONE;
}
- builder.build(context, container);
+ builder.build(container);
sorter = createNewSorter();
sorter.setup(context, getSelectionVector4(), this.container);
sorter.sort(getSelectionVector4(), this.container);
@@ -142,7 +142,7 @@ public class SortBatch extends AbstractRecordBatch<Sort> {
} catch(SchemaChangeException | ClassTransformationException | IOException ex) {
kill(false);
logger.error("Failure during query", ex);
- context.fail(ex);
+ context.getExecutorState().fail(ex);
return IterOutcome.STOP;
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/sort/SortBatchCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/sort/SortBatchCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/sort/SortBatchCreator.java
index 559558f..ccd5561 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/sort/SortBatchCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/sort/SortBatchCreator.java
@@ -20,7 +20,7 @@ package org.apache.drill.exec.physical.impl.sort;
import java.util.List;
import org.apache.drill.common.exceptions.ExecutionSetupException;
-import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.ExecutorFragmentContext;
import org.apache.drill.exec.physical.config.Sort;
import org.apache.drill.exec.physical.impl.BatchCreator;
import org.apache.drill.exec.record.RecordBatch;
@@ -28,14 +28,10 @@ import org.apache.drill.exec.record.RecordBatch;
import com.google.common.base.Preconditions;
public class SortBatchCreator implements BatchCreator<Sort>{
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(SortBatchCreator.class);
-
@Override
- public SortBatch getBatch(FragmentContext context, Sort config, List<RecordBatch> children)
+ public SortBatch getBatch(ExecutorFragmentContext context, Sort config, List<RecordBatch> children)
throws ExecutionSetupException {
Preconditions.checkArgument(children.size() == 1);
return new SortBatch(config, context, children.iterator().next());
}
-
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/sort/SortRecordBatchBuilder.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/sort/SortRecordBatchBuilder.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/sort/SortRecordBatchBuilder.java
index 6b3de25..6c66c01 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/sort/SortRecordBatchBuilder.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/sort/SortRecordBatchBuilder.java
@@ -27,7 +27,6 @@ import org.apache.drill.exec.exception.SchemaChangeException;
import org.apache.drill.exec.memory.AllocationReservation;
import org.apache.drill.exec.memory.BaseAllocator;
import org.apache.drill.exec.memory.BufferAllocator;
-import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.record.BatchSchema;
import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
import org.apache.drill.exec.record.MaterializedField;
@@ -136,10 +135,6 @@ public class SortRecordBatchBuilder implements AutoCloseable {
return batches.isEmpty();
}
- public void build(FragmentContext context, VectorContainer outputContainer) throws SchemaChangeException {
- build(outputContainer);
- }
-
@SuppressWarnings("resource")
public void build(VectorContainer outputContainer) throws SchemaChangeException {
outputContainer.clear();
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/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 4304c2c..66fe261 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
@@ -24,7 +24,6 @@ import org.apache.drill.exec.exception.ClassTransformationException;
import org.apache.drill.exec.exception.OutOfMemoryException;
import org.apache.drill.exec.exception.SchemaChangeException;
import org.apache.drill.exec.expr.CodeGenerator;
-import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.config.SelectionVectorRemover;
import org.apache.drill.exec.record.AbstractSingleRecordBatch;
@@ -235,10 +234,11 @@ public class RemovingRecordBatch extends AbstractSingleRecordBatch<SelectionVect
private Copier getGenerated4Copier() throws SchemaChangeException {
Preconditions.checkArgument(incoming.getSchema().getSelectionVectorMode() == SelectionVectorMode.FOUR_BYTE);
- return getGenerated4Copier(incoming, context, oContext.getAllocator(), container, this, callBack);
+ return getGenerated4Copier(incoming, context, container, this, callBack);
}
- public static Copier getGenerated4Copier(RecordBatch batch, FragmentContext context, BufferAllocator allocator, VectorContainer container, RecordBatch outgoing, SchemaChangeCallBack callBack) throws SchemaChangeException{
+ public static Copier getGenerated4Copier(RecordBatch batch, FragmentContext context, VectorContainer container, RecordBatch outgoing,
+ SchemaChangeCallBack callBack) throws SchemaChangeException{
for(VectorWrapper<?> vv : batch){
@SuppressWarnings("resource")
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/SVRemoverCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/SVRemoverCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/SVRemoverCreator.java
index 9ab39a3..4bf5b5c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/SVRemoverCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/svremover/SVRemoverCreator.java
@@ -20,22 +20,18 @@ package org.apache.drill.exec.physical.impl.svremover;
import java.util.List;
import org.apache.drill.common.exceptions.ExecutionSetupException;
-import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.ExecutorFragmentContext;
import org.apache.drill.exec.physical.config.SelectionVectorRemover;
import org.apache.drill.exec.physical.impl.BatchCreator;
import org.apache.drill.exec.record.RecordBatch;
import com.google.common.base.Preconditions;
-public class SVRemoverCreator implements BatchCreator<SelectionVectorRemover>{
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(SVRemoverCreator.class);
-
+public class SVRemoverCreator implements BatchCreator<SelectionVectorRemover> {
@Override
- public RemovingRecordBatch getBatch(FragmentContext context, SelectionVectorRemover config, List<RecordBatch> children)
+ public RemovingRecordBatch getBatch(ExecutorFragmentContext context, SelectionVectorRemover config, List<RecordBatch> children)
throws ExecutionSetupException {
Preconditions.checkArgument(children.size() == 1);
return new RemovingRecordBatch(config, context, children.iterator().next());
}
-
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/trace/TraceBatchCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/trace/TraceBatchCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/trace/TraceBatchCreator.java
index 40ef2bb..dd2f6db 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/trace/TraceBatchCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/trace/TraceBatchCreator.java
@@ -21,19 +21,15 @@ package org.apache.drill.exec.physical.impl.trace;
import java.util.List;
import org.apache.drill.common.exceptions.ExecutionSetupException;
-import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.ExecutorFragmentContext;
import org.apache.drill.exec.physical.config.Trace;
import org.apache.drill.exec.physical.impl.BatchCreator;
import org.apache.drill.exec.record.RecordBatch;
public class TraceBatchCreator implements BatchCreator<Trace> {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TraceBatchCreator.class);
-
@Override
- public TraceRecordBatch getBatch(FragmentContext context, Trace config, List<RecordBatch> children)
+ public TraceRecordBatch getBatch(ExecutorFragmentContext context, Trace config, List<RecordBatch> children)
throws ExecutionSetupException {
- // Preconditions.checkArgument(children.size() == 1);
return new TraceRecordBatch(config, children.iterator().next(), context);
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/union/UnionAllBatchCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/union/UnionAllBatchCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/union/UnionAllBatchCreator.java
index 1ef3142..bdc1a3d 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/union/UnionAllBatchCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/union/UnionAllBatchCreator.java
@@ -20,22 +20,18 @@ package org.apache.drill.exec.physical.impl.union;
import java.util.List;
import org.apache.drill.common.exceptions.ExecutionSetupException;
-import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.ExecutorFragmentContext;
import org.apache.drill.exec.physical.config.UnionAll;
import org.apache.drill.exec.physical.impl.BatchCreator;
import org.apache.drill.exec.record.RecordBatch;
import com.google.common.base.Preconditions;
-public class UnionAllBatchCreator implements BatchCreator<UnionAll>{
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(UnionAllBatchCreator.class);
-
+public class UnionAllBatchCreator implements BatchCreator<UnionAll> {
@Override
- public UnionAllRecordBatch getBatch(FragmentContext context, UnionAll config, List<RecordBatch> children)
+ public UnionAllRecordBatch getBatch(ExecutorFragmentContext context, UnionAll config, List<RecordBatch> children)
throws ExecutionSetupException {
Preconditions.checkArgument(children.size() >= 1);
return new UnionAllRecordBatch(config, children, context);
}
-
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/union/UnionAllRecordBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/union/UnionAllRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/union/UnionAllRecordBatch.java
index 1d1ecb0..b4d0e77 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/union/UnionAllRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/union/UnionAllRecordBatch.java
@@ -129,7 +129,7 @@ public class UnionAllRecordBatch extends AbstractBinaryRecordBatch<UnionAll> {
}
}
} catch (ClassTransformationException | IOException | SchemaChangeException ex) {
- context.fail(ex);
+ context.getExecutorState().fail(ex);
killIncoming(false);
return IterOutcome.STOP;
}
@@ -168,8 +168,6 @@ public class UnionAllRecordBatch extends AbstractBinaryRecordBatch<UnionAll> {
final ClassGenerator<UnionAller> cg = CodeGenerator.getRoot(UnionAller.TEMPLATE_DEFINITION, context.getOptions());
cg.getCodeGenerator().plainJavaCapable(true);
- // Uncomment out this line to debug the generated code.
- // cg.getCodeGenerator().saveCodeForDebugging(true);
int index = 0;
for(VectorWrapper<?> vw : inputBatch) {
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverBatch.java
index cfdc06d..9da8a4b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverBatch.java
@@ -25,6 +25,7 @@ import java.util.Iterator;
import org.apache.drill.common.expression.SchemaPath;
import org.apache.drill.exec.exception.OutOfMemoryException;
import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.ops.ExchangeFragmentContext;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.ops.MetricDef;
import org.apache.drill.exec.ops.OperatorContext;
@@ -57,7 +58,7 @@ public class UnorderedReceiverBatch implements CloseableRecordBatch {
private final RecordBatchLoader batchLoader;
private final RawFragmentBatchProvider fragProvider;
- private final FragmentContext context;
+ private final ExchangeFragmentContext context;
private BatchSchema schema;
private final OperatorStats stats;
private boolean first = true;
@@ -74,7 +75,7 @@ public class UnorderedReceiverBatch implements CloseableRecordBatch {
}
}
- public UnorderedReceiverBatch(final FragmentContext context, final RawFragmentBatchProvider fragProvider, final UnorderedReceiver config) throws OutOfMemoryException {
+ public UnorderedReceiverBatch(final ExchangeFragmentContext context, final RawFragmentBatchProvider fragProvider, final UnorderedReceiver config) throws OutOfMemoryException {
this.fragProvider = fragProvider;
this.context = context;
// In normal case, batchLoader does not require an allocator. However, in case of splitAndTransfer of a value vector,
@@ -171,13 +172,13 @@ public class UnorderedReceiverBatch implements CloseableRecordBatch {
if (batch == null) {
batchLoader.clear();
- if (!context.shouldContinue()) {
+ if (!context.getExecutorState().shouldContinue()) {
return IterOutcome.STOP;
}
return IterOutcome.NONE;
}
- if (context.isOverMemoryLimit()) {
+ if (context.getAllocator().isOverLimit()) {
return IterOutcome.OUT_OF_MEMORY;
}
@@ -197,7 +198,7 @@ public class UnorderedReceiverBatch implements CloseableRecordBatch {
return IterOutcome.OK;
}
} catch(SchemaChangeException | IOException ex) {
- context.fail(ex);
+ context.getExecutorState().fail(ex);
return IterOutcome.STOP;
} finally {
stats.stopProcessing();
@@ -233,7 +234,9 @@ public class UnorderedReceiverBatch implements CloseableRecordBatch {
.setReceiver(context.getHandle())
.setSender(sender)
.build();
- context.getControlTunnel(providingEndpoint.getEndpoint()).informReceiverFinished(new OutcomeListener(), finishedReceiver);
+ context.getController()
+ .getTunnel(providingEndpoint.getEndpoint())
+ .informReceiverFinished(new OutcomeListener(), finishedReceiver);
}
}
@@ -252,12 +255,11 @@ public class UnorderedReceiverBatch implements CloseableRecordBatch {
@Override
public void interrupted(final InterruptedException e) {
- if (context.shouldContinue()) {
+ if (context.getExecutorState().shouldContinue()) {
final String errMsg = "Received an interrupt RPC outcome while sending ReceiverFinished message";
logger.error(errMsg, e);
- context.fail(new RpcException(errMsg, e));
+ context.getExecutorState().fail(new RpcException(errMsg, e));
}
}
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverCreator.java
index 6d4f1d7..01a4588 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverCreator.java
@@ -20,18 +20,18 @@ package org.apache.drill.exec.physical.impl.unorderedreceiver;
import java.util.List;
import org.apache.drill.common.exceptions.ExecutionSetupException;
-import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.ExecutorFragmentContext;
import org.apache.drill.exec.physical.config.UnorderedReceiver;
import org.apache.drill.exec.physical.impl.BatchCreator;
import org.apache.drill.exec.record.RecordBatch;
import org.apache.drill.exec.work.batch.IncomingBuffers;
import org.apache.drill.exec.work.batch.RawBatchBuffer;
-public class UnorderedReceiverCreator implements BatchCreator<UnorderedReceiver>{
+public class UnorderedReceiverCreator implements BatchCreator<UnorderedReceiver> {
@SuppressWarnings("resource")
@Override
- public UnorderedReceiverBatch getBatch(FragmentContext context, UnorderedReceiver receiver, List<RecordBatch> children)
+ public UnorderedReceiverBatch getBatch(ExecutorFragmentContext context, UnorderedReceiver receiver, List<RecordBatch> children)
throws ExecutionSetupException {
assert children == null || children.isEmpty();
IncomingBuffers bufHolder = context.getBuffers();
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/validate/IteratorValidatorCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/validate/IteratorValidatorCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/validate/IteratorValidatorCreator.java
index 4199191..e27f881 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/validate/IteratorValidatorCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/validate/IteratorValidatorCreator.java
@@ -21,7 +21,7 @@ import java.util.List;
import org.apache.drill.common.exceptions.ExecutionSetupException;
import org.apache.drill.exec.ExecConstants;
-import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.ExecutorFragmentContext;
import org.apache.drill.exec.physical.config.IteratorValidator;
import org.apache.drill.exec.physical.impl.BatchCreator;
import org.apache.drill.exec.record.RecordBatch;
@@ -32,8 +32,8 @@ public class IteratorValidatorCreator implements BatchCreator<IteratorValidator>
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(IteratorValidatorCreator.class);
@Override
- public IteratorValidatorBatchIterator getBatch(FragmentContext context, IteratorValidator config,
- List<RecordBatch> children)
+ public IteratorValidatorBatchIterator getBatch(ExecutorFragmentContext context, IteratorValidator config,
+ List<RecordBatch> children)
throws ExecutionSetupException {
Preconditions.checkArgument(children.size() == 1);
RecordBatch child = children.iterator().next();
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/values/ValuesBatchCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/values/ValuesBatchCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/values/ValuesBatchCreator.java
index a8eddbc..c2bcab0 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/values/ValuesBatchCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/values/ValuesBatchCreator.java
@@ -22,7 +22,7 @@ import java.util.List;
import org.apache.drill.common.exceptions.ExecutionSetupException;
import org.apache.drill.common.expression.SchemaPath;
-import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.ExecutorFragmentContext;
import org.apache.drill.exec.physical.config.Values;
import org.apache.drill.exec.physical.impl.BatchCreator;
import org.apache.drill.exec.physical.impl.ScanBatch;
@@ -30,11 +30,9 @@ import org.apache.drill.exec.record.RecordBatch;
import org.apache.drill.exec.store.RecordReader;
import org.apache.drill.exec.store.easy.json.JSONRecordReader;
-import com.google.common.collect.Iterators;
-
public class ValuesBatchCreator implements BatchCreator<Values> {
@Override
- public ScanBatch getBatch(FragmentContext context, Values config, List<RecordBatch> children)
+ public ScanBatch getBatch(ExecutorFragmentContext context, Values config, List<RecordBatch> children)
throws ExecutionSetupException {
assert children.isEmpty();
http://git-wip-us.apache.org/repos/asf/drill/blob/186536d5/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/window/WindowFrameBatchCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/window/WindowFrameBatchCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/window/WindowFrameBatchCreator.java
index 59bc115..6ca9652 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/window/WindowFrameBatchCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/window/WindowFrameBatchCreator.java
@@ -21,7 +21,7 @@ package org.apache.drill.exec.physical.impl.window;
import java.util.List;
import org.apache.drill.common.exceptions.ExecutionSetupException;
-import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.ExecutorFragmentContext;
import org.apache.drill.exec.physical.config.WindowPOP;
import org.apache.drill.exec.physical.impl.BatchCreator;
import org.apache.drill.exec.record.RecordBatch;
@@ -29,9 +29,8 @@ import org.apache.drill.exec.record.RecordBatch;
import com.google.common.base.Preconditions;
public class WindowFrameBatchCreator implements BatchCreator<WindowPOP> {
-
@Override
- public WindowFrameRecordBatch getBatch(FragmentContext context, WindowPOP config, List<RecordBatch> children)
+ public WindowFrameRecordBatch getBatch(ExecutorFragmentContext context, WindowPOP config, List<RecordBatch> children)
throws ExecutionSetupException {
Preconditions.checkArgument(children.size() == 1);
return new WindowFrameRecordBatch(config, context, children.iterator().next());