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:04 UTC
[15/21] git commit: Fix for project column ordering is wrong.
Fix for project column ordering is wrong.
Project: http://git-wip-us.apache.org/repos/asf/incubator-drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-drill/commit/c56f1a66
Tree: http://git-wip-us.apache.org/repos/asf/incubator-drill/tree/c56f1a66
Diff: http://git-wip-us.apache.org/repos/asf/incubator-drill/diff/c56f1a66
Branch: refs/heads/master
Commit: c56f1a66736f496c593626c5bf8058b25a7e8ccb
Parents: 3355d08
Author: Jacques Nadeau <ja...@apache.org>
Authored: Tue May 20 16:39:48 2014 -0700
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Fri May 23 10:02:53 2014 -0700
----------------------------------------------------------------------
.../org/apache/drill/exec/ExecConstants.java | 3 +
.../drill/exec/cache/CachedVectorContainer.java | 24 +++-
.../drill/exec/cache/local/LocalCache.java | 7 +-
.../apache/drill/exec/client/DrillClient.java | 2 +-
.../drill/exec/expr/EvaluationVisitor.java | 3 +-
.../exec/physical/impl/join/MergeJoinBatch.java | 132 ++++++++++---------
.../impl/project/ProjectRecordBatch.java | 7 +-
.../exec/planner/common/DrillWriterRelBase.java | 15 +++
.../exec/planner/logical/DrillWriterRel.java | 1 +
.../exec/planner/physical/ExchangePrel.java | 1 +
.../drill/exec/planner/physical/FilterPrel.java | 1 +
.../exec/planner/physical/HashAggPrel.java | 1 +
.../drill/exec/planner/physical/JoinPrel.java | 1 +
.../planner/physical/JoinPrelRenameVisitor.java | 75 -----------
.../drill/exec/planner/physical/LimitPrel.java | 1 +
.../drill/exec/planner/physical/Prel.java | 1 +
.../exec/planner/physical/PrelVisitor.java | 30 -----
.../exec/planner/physical/ProjectPrel.java | 1 +
.../drill/exec/planner/physical/ScanPrel.java | 1 +
.../drill/exec/planner/physical/ScreenPrel.java | 5 +-
.../physical/SelectionVectorPrelVisitor.java | 66 ----------
.../drill/exec/planner/physical/SinglePrel.java | 1 +
.../drill/exec/planner/physical/SortPrel.java | 1 +
.../exec/planner/physical/StreamAggPrel.java | 12 ++
.../drill/exec/planner/physical/WriterPrel.java | 15 ++-
.../planner/physical/explain/PrelSequencer.java | 9 +-
.../physical/visitor/BasePrelVisitor.java | 53 ++++++++
.../physical/visitor/FinalColumnReorderer.java | 88 +++++++++++++
.../physical/visitor/JoinPrelRenameVisitor.java | 73 ++++++++++
.../planner/physical/visitor/PrelVisitor.java | 38 ++++++
.../visitor/SelectionVectorPrelVisitor.java | 58 ++++++++
.../drill/exec/planner/sql/DrillSqlWorker.java | 2 +-
.../planner/sql/handlers/DefaultSqlHandler.java | 18 ++-
.../planner/sql/handlers/ExplainHandler.java | 3 +-
.../org/apache/drill/exec/rpc/BasicServer.java | 5 +-
.../drill/exec/rpc/control/ControllerImpl.java | 6 +-
.../exec/rpc/data/DataConnectionCreator.java | 6 +-
.../org/apache/drill/exec/server/Drillbit.java | 17 ++-
.../drill/exec/service/ServiceEngine.java | 18 +--
.../drill/exec/store/sys/SystemTableScan.java | 6 +
.../src/main/resources/drill-module.conf | 4 +
.../apache/drill/exec/server/TestBitRpc.java | 2 +-
.../exec/store/json/JsonRecordReader2Test.java | 4 +-
exec/java-exec/src/test/resources/logback.xml | 2 +-
pom.xml | 2 +-
45 files changed, 538 insertions(+), 283 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java b/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java
index 238fae9..d9e0833 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java
@@ -71,5 +71,8 @@ public interface ExecConstants {
public static final OptionValidator OUTPUT_FORMAT_VALIDATOR = new StringValidator(OUTPUT_FORMAT_OPTION, "parquet");
public static final String PARQUET_BLOCK_SIZE = "parquet.block.size";
public static final OptionValidator PARQUET_BLOCK_SIZE_VALIDATOR = new LongValidator(PARQUET_BLOCK_SIZE, 512*1024*1024);
+ public static final String HTTP_ENABLE = "drill.exec.http.enabled";
+ public static final String HTTP_PORT = "drill.exec.http.port";
+
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/cache/CachedVectorContainer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/CachedVectorContainer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/CachedVectorContainer.java
index 1447e28..da0b186 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/CachedVectorContainer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/CachedVectorContainer.java
@@ -18,6 +18,8 @@
package org.apache.drill.exec.cache;
import java.io.ByteArrayInputStream;
+import java.io.DataInput;
+import java.io.DataOutput;
import java.io.IOException;
import org.apache.commons.io.output.ByteArrayOutputStream;
@@ -29,7 +31,7 @@ import org.apache.drill.exec.record.WritableBatch;
public class CachedVectorContainer extends LoopedAbstractDrillSerializable {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(CachedVectorContainer.class);
- private final byte[] data;
+ private byte[] data;
private final BufferAllocator allocator;
private VectorContainer container;
@@ -42,6 +44,10 @@ public class CachedVectorContainer extends LoopedAbstractDrillSerializable {
va.clear();
}
+ public CachedVectorContainer(BufferAllocator allocator) {
+ this.allocator = allocator;
+ }
+
public CachedVectorContainer(byte[] data, BufferAllocator allocator) {
this.data = data;
this.allocator = allocator;
@@ -58,6 +64,20 @@ public class CachedVectorContainer extends LoopedAbstractDrillSerializable {
}
+
+ @Override
+ public void read(DataInput input) throws IOException {
+ int len = input.readInt();
+ this.data = new byte[len];
+ input.readFully(data);
+ }
+
+ @Override
+ public void write(DataOutput output) throws IOException {
+ output.writeInt(data.length);
+ output.write(data);
+ }
+
public VectorAccessible get() {
if (container == null) {
construct();
@@ -66,7 +86,7 @@ public class CachedVectorContainer extends LoopedAbstractDrillSerializable {
}
public void clear() {
- container.clear();
+ if(container != null) container.clear();
container = null;
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/cache/local/LocalCache.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/local/LocalCache.java b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/local/LocalCache.java
index 942e09e..1b44c6b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/local/LocalCache.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/local/LocalCache.java
@@ -17,6 +17,7 @@
*/
package org.apache.drill.exec.cache.local;
+import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
@@ -31,7 +32,6 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.drill.common.config.DrillConfig;
-import org.apache.drill.common.util.DataInputInputStream;
import org.apache.drill.common.util.DataOutputOutputStream;
import org.apache.drill.exec.cache.Counter;
import org.apache.drill.exec.cache.DistributedCache;
@@ -171,8 +171,7 @@ public class LocalCache implements DistributedCache {
}
}
- ByteArrayDataInput in = ByteStreams.newDataInput(bytes);
- InputStream inputStream = DataInputInputStream.constructInputStream(in);
+ InputStream inputStream = new ByteArrayInputStream(bytes);
try {
V obj = clazz.getConstructor(BufferAllocator.class).newInstance(allocator);
obj.readFromStream(inputStream);
@@ -220,7 +219,7 @@ public class LocalCache implements DistributedCache {
if (m.get(key) == null) return null;
ByteArrayDataOutput b = m.get(key);
byte[] bytes = b.toByteArray();
- return (V) deserialize(m.get(key).toByteArray(), this.clazz);
+ return (V) deserialize(bytes, this.clazz);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java b/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java
index 3b87dc4..92097e7 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java
@@ -188,7 +188,7 @@ public class DrillClient implements Closeable, ConnectionThrottle{
* Closes this client's connection to the server
*/
public void close(){
- this.client.close();
+ if(this.client != null) this.client.close();
if(ownsZkConnection){
try {
this.clusterCoordinator.close();
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/expr/EvaluationVisitor.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/EvaluationVisitor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/EvaluationVisitor.java
index 6a34b9c..dcc273c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/EvaluationVisitor.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/EvaluationVisitor.java
@@ -386,7 +386,8 @@ public class EvaluationVisitor {
PathSegment seg = e.getReadPath();
int listNum = 0;
boolean lastWasArray = false;
- while(true){
+
+ while(seg != null){
if(seg.isArray()){
lastWasArray = true;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/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 46dea64..121cfec 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
@@ -130,77 +130,83 @@ public class MergeJoinBatch extends AbstractRecordBatch<MergeJoinPOP> {
@Override
public IterOutcome next() {
+ stats.startProcessing();
+
+ try{
+ // 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){
+
+ JoinOutcome outcome = status.getOutcome();
+ // if the previous outcome was a change in schema or we sent a batch, we have to set up a new batch.
+ if (outcome == JoinOutcome.BATCH_RETURNED ||
+ outcome == JoinOutcome.SCHEMA_CHANGED)
+ allocateBatch();
+
+ // reset the output position to zero after our parent iterates this RecordBatch
+ if (outcome == JoinOutcome.BATCH_RETURNED ||
+ outcome == JoinOutcome.SCHEMA_CHANGED ||
+ outcome == JoinOutcome.NO_MORE_DATA)
+ status.resetOutputPos();
+
+ if (outcome == JoinOutcome.NO_MORE_DATA) {
+ logger.debug("NO MORE DATA; returning {} NONE");
+ return IterOutcome.NONE;
+ }
- // 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){
-
- JoinOutcome outcome = status.getOutcome();
- // if the previous outcome was a change in schema or we sent a batch, we have to set up a new batch.
- if (outcome == JoinOutcome.BATCH_RETURNED ||
- outcome == JoinOutcome.SCHEMA_CHANGED)
- allocateBatch();
-
- // reset the output position to zero after our parent iterates this RecordBatch
- if (outcome == JoinOutcome.BATCH_RETURNED ||
- outcome == JoinOutcome.SCHEMA_CHANGED ||
- outcome == JoinOutcome.NO_MORE_DATA)
- status.resetOutputPos();
+ 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;
+ }
+ }
- if (outcome == JoinOutcome.NO_MORE_DATA) {
- logger.debug("NO MORE DATA; returning {} NONE");
- return IterOutcome.NONE;
- }
+ // join until we have a complete outgoing batch
+ if (!worker.doJoin(status))
+ worker = null;
- 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));
+ // get the outcome of the join.
+ switch(status.getOutcome()){
+ case BATCH_RETURNED:
+ // only return new schema if new worker has been setup.
+ logger.debug("BATCH RETURNED; returning {}", (first ? "OK_NEW_SCHEMA" : "OK"));
+ return first ? IterOutcome.OK_NEW_SCHEMA : IterOutcome.OK;
+ case FAILURE:
kill();
return IterOutcome.STOP;
+ case NO_MORE_DATA:
+ logger.debug("NO MORE DATA; returning {}", (status.getOutPosition() > 0 ? (first ? "OK_NEW_SCHEMA" : "OK") : "NONE"));
+ return status.getOutPosition() > 0 ? (first ? IterOutcome.OK_NEW_SCHEMA : IterOutcome.OK): IterOutcome.NONE;
+ case SCHEMA_CHANGED:
+ worker = null;
+ if(status.getOutPosition() > 0){
+ // if we have current data, let's return that.
+ logger.debug("SCHEMA CHANGED; returning {} ", (first ? "OK_NEW_SCHEMA" : "OK"));
+ return first ? IterOutcome.OK_NEW_SCHEMA : IterOutcome.OK;
+ }else{
+ // loop again to rebuild worker.
+ continue;
+ }
+ case WAITING:
+ return IterOutcome.NOT_YET;
+ default:
+ throw new IllegalStateException();
}
}
- // join until we have a complete outgoing batch
- if (!worker.doJoin(status))
- worker = null;
-
- // get the outcome of the join.
- switch(status.getOutcome()){
- case BATCH_RETURNED:
- // only return new schema if new worker has been setup.
- logger.debug("BATCH RETURNED; returning {}", (first ? "OK_NEW_SCHEMA" : "OK"));
- return first ? IterOutcome.OK_NEW_SCHEMA : IterOutcome.OK;
- case FAILURE:
- kill();
- return IterOutcome.STOP;
- case NO_MORE_DATA:
- logger.debug("NO MORE DATA; returning {}", (status.getOutPosition() > 0 ? (first ? "OK_NEW_SCHEMA" : "OK") : "NONE"));
- return status.getOutPosition() > 0 ? (first ? IterOutcome.OK_NEW_SCHEMA : IterOutcome.OK): IterOutcome.NONE;
- case SCHEMA_CHANGED:
- worker = null;
- if(status.getOutPosition() > 0){
- // if we have current data, let's return that.
- logger.debug("SCHEMA CHANGED; returning {} ", (first ? "OK_NEW_SCHEMA" : "OK"));
- return first ? IterOutcome.OK_NEW_SCHEMA : IterOutcome.OK;
- }else{
- // loop again to rebuild worker.
- continue;
- }
- case WAITING:
- return IterOutcome.NOT_YET;
- default:
- throw new IllegalStateException();
- }
+ }finally{
+ stats.stopProcessing();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/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 efdd0ee..05a6724 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
@@ -181,10 +181,15 @@ public class ProjectRecordBatch extends AbstractSingleRecordBatch<Project>{
boolean isAnyWildcard = isAnyWildcard(exprs);
if(isAnyWildcard){
+
+ // add this until we have sv2 project on wildcard working correctly.
+ if(incoming.getSchema().getSelectionVectorMode() != SelectionVectorMode.NONE){
+ throw new UnsupportedOperationException("Drill doesn't yet wildcard projects where there is a sv2, patch coming shortly.");
+ }
for(VectorWrapper<?> wrapper : incoming){
ValueVector vvIn = wrapper.getValueVector();
- String name = vvIn.getField().getPath().getLastSegment().getNameSegment().getPath();
+ String name = vvIn.getField().getPath().getRootSegment().getPath();
FieldReference ref = new FieldReference(name);
TransferPair tp = wrapper.getValueVector().getTransferPair(ref);
transfers.add(tp);
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/planner/common/DrillWriterRelBase.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/common/DrillWriterRelBase.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/common/DrillWriterRelBase.java
index 357cb2e..03431d6 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/common/DrillWriterRelBase.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/common/DrillWriterRelBase.java
@@ -17,18 +17,33 @@
*/
package org.apache.drill.exec.planner.common;
+import java.util.List;
+
import org.apache.drill.exec.planner.logical.CreateTableEntry;
import org.eigenbase.rel.RelNode;
import org.eigenbase.rel.SingleRel;
import org.eigenbase.relopt.Convention;
import org.eigenbase.relopt.RelOptCluster;
import org.eigenbase.relopt.RelTraitSet;
+import org.eigenbase.reltype.RelDataType;
+import org.eigenbase.sql.type.SqlTypeName;
+
+import com.google.common.collect.ImmutableList;
+import com.google.hive12.common.collect.Lists;
/** Base class for logical and physical Writer implemented in Drill. */
public abstract class DrillWriterRelBase extends SingleRel implements DrillRelNode {
+ private static final List<String> FIELD_NAMES = ImmutableList.of("Fragment", "Number of records written");
private final CreateTableEntry createTableEntry;
+ protected void setRowType(){
+ List<RelDataType> fields = Lists.newArrayList();
+ fields.add(this.getCluster().getTypeFactory().createSqlType(SqlTypeName.VARCHAR, 255));
+ fields.add(this.getCluster().getTypeFactory().createSqlType(SqlTypeName.BIGINT));
+ this.rowType = this.getCluster().getTypeFactory().createStructType(fields, FIELD_NAMES);
+ }
+
public DrillWriterRelBase(Convention convention, RelOptCluster cluster, RelTraitSet traitSet, RelNode input,
CreateTableEntry createTableEntry) {
super(cluster, traitSet, input);
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillWriterRel.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillWriterRel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillWriterRel.java
index f212026..04dd133 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillWriterRel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillWriterRel.java
@@ -30,6 +30,7 @@ public class DrillWriterRel extends DrillWriterRelBase implements DrillRel {
public DrillWriterRel(RelOptCluster cluster, RelTraitSet traitSet, RelNode input, CreateTableEntry createTableEntry) {
super(DRILL_LOGICAL, cluster, traitSet, input, createTableEntry);
+ setRowType();
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ExchangePrel.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ExchangePrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ExchangePrel.java
index 21bf3b1..5a75adb 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ExchangePrel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ExchangePrel.java
@@ -18,6 +18,7 @@
package org.apache.drill.exec.planner.physical;
+import org.apache.drill.exec.planner.physical.visitor.PrelVisitor;
import org.eigenbase.rel.RelNode;
import org.eigenbase.relopt.RelOptCluster;
import org.eigenbase.relopt.RelTraitSet;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/FilterPrel.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/FilterPrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/FilterPrel.java
index 9632911..1b13dcd 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/FilterPrel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/FilterPrel.java
@@ -28,6 +28,7 @@ import org.apache.drill.exec.planner.common.DrillFilterRelBase;
import org.apache.drill.exec.planner.cost.DrillCostBase;
import org.apache.drill.exec.planner.cost.DrillCostBase.DrillCostFactory;
import org.apache.drill.exec.planner.logical.DrillParseContext;
+import org.apache.drill.exec.planner.physical.visitor.PrelVisitor;
import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
import org.eigenbase.rel.RelNode;
import org.eigenbase.rel.metadata.RelMetadataQuery;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/HashAggPrel.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/HashAggPrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/HashAggPrel.java
index b9d897e..d4a72f6 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/HashAggPrel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/HashAggPrel.java
@@ -37,6 +37,7 @@ import org.apache.drill.exec.physical.config.HashAggregate;
import org.apache.drill.exec.planner.cost.DrillCostBase;
import org.apache.drill.exec.planner.cost.DrillCostBase.DrillCostFactory;
import org.apache.drill.exec.planner.logical.DrillParseContext;
+import org.apache.drill.exec.planner.physical.visitor.PrelVisitor;
import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
import org.eigenbase.rel.AggregateCall;
import org.eigenbase.rel.AggregateRelBase;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/JoinPrel.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/JoinPrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/JoinPrel.java
index 9e1f6fb..0db9a97 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/JoinPrel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/JoinPrel.java
@@ -22,6 +22,7 @@ import java.util.Iterator;
import java.util.List;
import org.apache.drill.exec.planner.common.DrillJoinRelBase;
+import org.apache.drill.exec.planner.physical.visitor.PrelVisitor;
import org.eigenbase.rel.InvalidRelException;
import org.eigenbase.rel.JoinRelType;
import org.eigenbase.rel.RelNode;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/JoinPrelRenameVisitor.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/JoinPrelRenameVisitor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/JoinPrelRenameVisitor.java
deleted file mode 100644
index ed4b8f8..0000000
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/JoinPrelRenameVisitor.java
+++ /dev/null
@@ -1,75 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.drill.exec.planner.physical;
-
-import java.util.List;
-
-import org.eigenbase.rel.RelNode;
-
-import com.google.common.collect.Lists;
-
-public class JoinPrelRenameVisitor implements PrelVisitor<Prel, Void, RuntimeException>{
-
- private static JoinPrelRenameVisitor INSTANCE = new JoinPrelRenameVisitor();
-
- public static Prel insertRenameProject(Prel prel){
- return prel.accept(INSTANCE, null);
- }
-
- @Override
- public Prel visitExchange(ExchangePrel prel, Void value) throws RuntimeException {
- return visitPrel(prel, value);
- }
-
- @Override
- public Prel visitPrel(Prel prel, Void value) throws RuntimeException {
- List<RelNode> children = Lists.newArrayList();
- for(Prel child : prel){
- child = child.accept(this, null);
- children.add(child);
- }
-
- return (Prel) prel.copy(prel.getTraitSet(), children);
-
- }
-
- @Override
- public Prel visitJoin(JoinPrel prel, Void value) throws RuntimeException {
-
- List<RelNode> children = Lists.newArrayList();
-
- for(Prel child : prel){
- child = child.accept(this, null);
- children.add(child);
- }
-
- final int leftCount = children.get(0).getRowType().getFieldCount();
-
- List<RelNode> reNamedChildren = Lists.newArrayList();
-
- RelNode left = prel.getJoinInput(0, children.get(0));
- RelNode right = prel.getJoinInput(leftCount, children.get(1));
-
- reNamedChildren.add(left);
- reNamedChildren.add(right);
-
- return (Prel) prel.copy(prel.getTraitSet(), reNamedChildren);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/LimitPrel.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/LimitPrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/LimitPrel.java
index 794593a..376f334 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/LimitPrel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/LimitPrel.java
@@ -30,6 +30,7 @@ import org.apache.drill.exec.planner.common.DrillLimitRelBase;
import org.apache.drill.exec.planner.logical.DrillImplementor;
import org.apache.drill.exec.planner.logical.DrillParseContext;
import org.apache.drill.exec.planner.logical.DrillRel;
+import org.apache.drill.exec.planner.physical.visitor.PrelVisitor;
import org.apache.drill.exec.planner.torel.ConversionContext;
import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
import org.eigenbase.rel.InvalidRelException;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/Prel.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/Prel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/Prel.java
index a6c6b7c..1b4eb38 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/Prel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/Prel.java
@@ -21,6 +21,7 @@ import java.io.IOException;
import org.apache.drill.exec.physical.base.PhysicalOperator;
import org.apache.drill.exec.planner.common.DrillRelNode;
+import org.apache.drill.exec.planner.physical.visitor.PrelVisitor;
import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
import org.eigenbase.relopt.Convention;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PrelVisitor.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PrelVisitor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PrelVisitor.java
deleted file mode 100644
index 93f95a8..0000000
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PrelVisitor.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.drill.exec.planner.physical;
-
-
-public interface PrelVisitor<RETURN, EXTRA, EXCEP extends Throwable> {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(PrelVisitor.class);
-
- public RETURN visitExchange(ExchangePrel prel, EXTRA value) throws EXCEP;
-
- public RETURN visitJoin(JoinPrel prel, EXTRA value) throws EXCEP;
-
- public RETURN visitPrel(Prel prel, EXTRA value) throws EXCEP;
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ProjectPrel.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ProjectPrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ProjectPrel.java
index 70dca25..55f9f32 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ProjectPrel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ProjectPrel.java
@@ -28,6 +28,7 @@ import org.apache.drill.exec.physical.config.Project;
import org.apache.drill.exec.physical.config.SelectionVectorRemover;
import org.apache.drill.exec.planner.common.DrillProjectRelBase;
import org.apache.drill.exec.planner.logical.DrillParseContext;
+import org.apache.drill.exec.planner.physical.visitor.PrelVisitor;
import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
import org.eigenbase.rel.ProjectRelBase;
import org.eigenbase.rel.RelNode;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ScanPrel.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ScanPrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ScanPrel.java
index 8461e24..445ecd5 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ScanPrel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ScanPrel.java
@@ -30,6 +30,7 @@ import org.eigenbase.rel.AbstractRelNode;
import org.apache.drill.exec.planner.common.DrillScanRelBase;
import org.apache.drill.exec.planner.cost.DrillCostBase;
import org.apache.drill.exec.planner.cost.DrillCostBase.DrillCostFactory;
+import org.apache.drill.exec.planner.physical.visitor.PrelVisitor;
import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
import org.apache.drill.exec.store.StoragePlugin;
import org.eigenbase.rel.RelNode;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ScreenPrel.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ScreenPrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ScreenPrel.java
index d02ed44..37823ab 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ScreenPrel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ScreenPrel.java
@@ -25,6 +25,7 @@ import org.apache.drill.exec.physical.base.PhysicalOperator;
import org.apache.drill.exec.physical.config.Screen;
import org.apache.drill.exec.physical.config.SelectionVectorRemover;
import org.apache.drill.exec.planner.common.DrillScreenRelBase;
+import org.apache.drill.exec.planner.physical.visitor.PrelVisitor;
import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
import org.eigenbase.rel.RelNode;
import org.eigenbase.relopt.RelOptCluster;
@@ -40,7 +41,7 @@ public class ScreenPrel extends DrillScreenRelBase implements Prel {
}
@Override
- public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
+ public ScreenPrel copy(RelTraitSet traitSet, List<RelNode> inputs) {
return new ScreenPrel(getCluster(), traitSet, sole(inputs));
}
@@ -63,7 +64,7 @@ public class ScreenPrel extends DrillScreenRelBase implements Prel {
@Override
public <T, X, E extends Throwable> T accept(PrelVisitor<T, X, E> logicalVisitor, X value) throws E {
- return logicalVisitor.visitPrel(this, value);
+ return logicalVisitor.visitScreen(this, value);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SelectionVectorPrelVisitor.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SelectionVectorPrelVisitor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SelectionVectorPrelVisitor.java
deleted file mode 100644
index 8f2ecd4..0000000
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SelectionVectorPrelVisitor.java
+++ /dev/null
@@ -1,66 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.drill.exec.planner.physical;
-
-import java.util.List;
-
-import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
-import org.eigenbase.rel.RelNode;
-
-import com.google.common.collect.Lists;
-
-
-public class SelectionVectorPrelVisitor implements PrelVisitor<Prel, Void, RuntimeException>{
-
- private static SelectionVectorPrelVisitor INSTANCE = new SelectionVectorPrelVisitor();
-
- public static Prel addSelectionRemoversWhereNecessary(Prel prel){
- return prel.accept(INSTANCE, null);
- }
-
- @Override
- public Prel visitExchange(ExchangePrel prel, Void value) throws RuntimeException {
- return visitPrel(prel, value);
- }
-
- @Override
- public Prel visitJoin(JoinPrel prel, Void value) throws RuntimeException {
- return visitPrel(prel, value);
- }
-
- @Override
- public Prel visitPrel(Prel prel, Void value) throws RuntimeException {
- SelectionVectorMode[] encodings = prel.getSupportedEncodings();
- List<RelNode> children = Lists.newArrayList();
- for(Prel child : prel){
- child = child.accept(this, null);
- children.add(convert(encodings, child));
- }
-
- return (Prel) prel.copy(prel.getTraitSet(), children);
- }
-
- private Prel convert(SelectionVectorMode[] encodings, Prel prel){
- for(SelectionVectorMode m : encodings){
- if(prel.getEncoding() == m) return prel;
- }
- return new SelectionVectorRemoverPrel(prel);
- }
-
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SinglePrel.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SinglePrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SinglePrel.java
index e47dc7f..c23e2a1 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SinglePrel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SinglePrel.java
@@ -20,6 +20,7 @@ package org.apache.drill.exec.planner.physical;
import java.util.Collections;
import java.util.Iterator;
+import org.apache.drill.exec.planner.physical.visitor.PrelVisitor;
import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
import org.eigenbase.rel.RelNode;
import org.eigenbase.rel.SingleRel;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SortPrel.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SortPrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SortPrel.java
index fa5e900..464e1bb 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SortPrel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/SortPrel.java
@@ -26,6 +26,7 @@ import org.apache.drill.exec.physical.config.SingleMergeExchange;
import org.apache.drill.exec.physical.config.Sort;
import org.apache.drill.exec.planner.cost.DrillCostBase;
import org.apache.drill.exec.planner.cost.DrillCostBase.DrillCostFactory;
+import org.apache.drill.exec.planner.physical.visitor.PrelVisitor;
import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
import org.eigenbase.rel.RelCollation;
import org.eigenbase.rel.RelNode;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/StreamAggPrel.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/StreamAggPrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/StreamAggPrel.java
index abd60c5..889e6c1 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/StreamAggPrel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/StreamAggPrel.java
@@ -26,6 +26,8 @@ import org.apache.drill.exec.physical.base.PhysicalOperator;
import org.apache.drill.exec.physical.config.StreamingAggregate;
import org.apache.drill.exec.planner.cost.DrillCostBase;
import org.apache.drill.exec.planner.cost.DrillCostBase.DrillCostFactory;
+import org.apache.drill.exec.planner.logical.DrillParseContext;
+import org.apache.drill.exec.planner.physical.visitor.PrelVisitor;
import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
import org.eigenbase.rel.AggregateCall;
import org.eigenbase.rel.AggregateRelBase;
@@ -88,6 +90,16 @@ public class StreamAggPrel extends AggPrelBase implements Prel{
}
@Override
+ public Iterator<Prel> iterator() {
+ return PrelUtil.iter(getChild());
+ }
+
+ @Override
+ public <T, X, E extends Throwable> T accept(PrelVisitor<T, X, E> logicalVisitor, X value) throws E {
+ return logicalVisitor.visitPrel(this, value);
+ }
+
+ @Override
public SelectionVectorMode[] getSupportedEncodings() {
return SelectionVectorMode.ALL;
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/WriterPrel.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/WriterPrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/WriterPrel.java
index e948125..233b20b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/WriterPrel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/WriterPrel.java
@@ -24,19 +24,28 @@ import java.util.List;
import org.apache.drill.exec.physical.base.PhysicalOperator;
import org.apache.drill.exec.planner.common.DrillWriterRelBase;
import org.apache.drill.exec.planner.logical.CreateTableEntry;
+import org.apache.drill.exec.planner.physical.visitor.PrelVisitor;
import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
import org.eigenbase.rel.RelNode;
import org.eigenbase.relopt.RelOptCluster;
import org.eigenbase.relopt.RelTraitSet;
+import org.eigenbase.reltype.RelDataType;
+import org.eigenbase.sql.type.SqlTypeName;
+
+import com.google.common.collect.ImmutableList;
+import com.google.hive12.common.collect.Lists;
public class WriterPrel extends DrillWriterRelBase implements Prel {
+
+
public WriterPrel(RelOptCluster cluster, RelTraitSet traits, RelNode child, CreateTableEntry createTableEntry) {
super(Prel.DRILL_PHYSICAL, cluster, traits, child, createTableEntry);
+ setRowType();
}
@Override
- public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
+ public WriterPrel copy(RelTraitSet traitSet, List<RelNode> inputs) {
return new WriterPrel(getCluster(), traitSet, sole(inputs), getCreateTableEntry());
}
@@ -49,6 +58,7 @@ public class WriterPrel extends DrillWriterRelBase implements Prel {
return g;
}
+
@Override
public Iterator<Prel> iterator() {
return PrelUtil.iter(getChild());
@@ -56,7 +66,7 @@ public class WriterPrel extends DrillWriterRelBase implements Prel {
@Override
public <T, X, E extends Throwable> T accept(PrelVisitor<T, X, E> logicalVisitor, X value) throws E {
- return logicalVisitor.visitPrel(this, value);
+ return logicalVisitor.visitWriter(this, value);
}
@Override
@@ -64,6 +74,7 @@ public class WriterPrel extends DrillWriterRelBase implements Prel {
return SelectionVectorMode.DEFAULT;
}
+
@Override
public SelectionVectorMode getEncoding() {
return SelectionVectorMode.NONE;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/explain/PrelSequencer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/explain/PrelSequencer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/explain/PrelSequencer.java
index 771546a..ebff287 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/explain/PrelSequencer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/explain/PrelSequencer.java
@@ -27,14 +27,14 @@ import java.util.Queue;
import org.apache.drill.exec.planner.physical.ExchangePrel;
import org.apache.drill.exec.planner.physical.JoinPrel;
import org.apache.drill.exec.planner.physical.Prel;
-import org.apache.drill.exec.planner.physical.PrelVisitor;
+import org.apache.drill.exec.planner.physical.visitor.BasePrelVisitor;
import org.eigenbase.rel.RelWriter;
import org.eigenbase.sql.SqlExplainLevel;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
-public class PrelSequencer implements PrelVisitor<Void, PrelSequencer.Frag, RuntimeException>{
+public class PrelSequencer extends BasePrelVisitor<Void, PrelSequencer.Frag, RuntimeException>{
private List<Frag> frags = Lists.newLinkedList();
@@ -235,11 +235,6 @@ public class PrelSequencer implements PrelVisitor<Void, PrelSequencer.Frag, Runt
}
@Override
- public Void visitJoin(JoinPrel prel, Frag value) throws RuntimeException {
- return visitPrel(prel, value);
- }
-
- @Override
public Void visitPrel(Prel prel, Frag value) throws RuntimeException {
for(Prel children : prel){
children.accept(this, value);
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/BasePrelVisitor.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/BasePrelVisitor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/BasePrelVisitor.java
new file mode 100644
index 0000000..e3cfecc
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/BasePrelVisitor.java
@@ -0,0 +1,53 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.exec.planner.physical.visitor;
+
+import org.apache.drill.exec.planner.physical.ExchangePrel;
+import org.apache.drill.exec.planner.physical.JoinPrel;
+import org.apache.drill.exec.planner.physical.Prel;
+import org.apache.drill.exec.planner.physical.ScreenPrel;
+import org.apache.drill.exec.planner.physical.WriterPrel;
+
+public class BasePrelVisitor<RETURN, EXTRA, EXCEP extends Throwable> implements PrelVisitor<RETURN, EXTRA, EXCEP> {
+
+ @Override
+ public RETURN visitExchange(ExchangePrel prel, EXTRA value) throws EXCEP {
+ return visitPrel(prel, value);
+ }
+
+ @Override
+ public RETURN visitJoin(JoinPrel prel, EXTRA value) throws EXCEP {
+ return visitPrel(prel, value);
+ }
+
+ @Override
+ public RETURN visitScreen(ScreenPrel prel, EXTRA value) throws EXCEP {
+ return visitPrel(prel, value);
+ }
+
+ @Override
+ public RETURN visitWriter(WriterPrel prel, EXTRA value) throws EXCEP {
+ return visitPrel(prel, value);
+ }
+
+ @Override
+ public RETURN visitPrel(Prel prel, EXTRA value) throws EXCEP {
+ throw new UnsupportedOperationException(String.format("No visit method defined for prel %s in visitor %s.", prel, this.getClass().getName()));
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/FinalColumnReorderer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/FinalColumnReorderer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/FinalColumnReorderer.java
new file mode 100644
index 0000000..4ea82cd
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/FinalColumnReorderer.java
@@ -0,0 +1,88 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.exec.planner.physical.visitor;
+
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.drill.exec.planner.physical.Prel;
+import org.apache.drill.exec.planner.physical.ProjectPrel;
+import org.apache.drill.exec.planner.physical.ScreenPrel;
+import org.apache.drill.exec.planner.physical.WriterPrel;
+import org.eigenbase.rel.RelNode;
+import org.eigenbase.reltype.RelDataType;
+import org.eigenbase.rex.RexBuilder;
+import org.eigenbase.rex.RexNode;
+
+import com.google.common.collect.Lists;
+
+public class FinalColumnReorderer extends BasePrelVisitor<Prel, Void, RuntimeException>{
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(FinalColumnReorderer.class);
+
+ private static FinalColumnReorderer INSTANCE = new FinalColumnReorderer();
+
+ public static Prel addFinalColumnOrdering(Prel prel){
+ return prel.accept(INSTANCE, null);
+ }
+
+ @Override
+ public Prel visitScreen(ScreenPrel prel, Void value) throws RuntimeException {
+ Prel newChild = ((Prel) prel.getChild()).accept(this, value);
+ return prel.copy(prel.getTraitSet(), Collections.singletonList( (RelNode) addTrivialOrderedProjectPrel( newChild )));
+ }
+
+ private Prel addTrivialOrderedProjectPrel(Prel prel){
+ RelDataType t = prel.getRowType();
+
+ RexBuilder b = prel.getCluster().getRexBuilder();
+ List<RexNode> projections = Lists.newArrayList();
+ int projectCount = t.getFieldList().size();
+
+ // no point in reordering if we only have one column
+ if(projectCount < 2) return prel;
+
+ for(int i =0; i < projectCount; i++){
+ projections.add(b.makeInputRef(prel, i));
+ }
+ return new ProjectPrel(prel.getCluster(), prel.getTraitSet(), prel, projections, prel.getRowType());
+ }
+
+ @Override
+ public Prel visitWriter(WriterPrel prel, Void value) throws RuntimeException {
+ Prel newChild = ((Prel) prel.getChild()).accept(this, null);
+ return prel.copy(prel.getTraitSet(), Collections.singletonList( (RelNode) addTrivialOrderedProjectPrel( newChild )));
+ }
+
+ @Override
+ public Prel visitPrel(Prel prel, Void value) throws RuntimeException {
+ List<RelNode> children = Lists.newArrayList();
+ boolean changed = false;
+ for(Prel p : prel){
+ Prel newP = p.accept(this, null);
+ if(newP != p) changed = true;
+ children.add(newP);
+ }
+ if(changed){
+ return (Prel) prel.copy(prel.getTraitSet(), children);
+ }else{
+ return prel;
+ }
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/JoinPrelRenameVisitor.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/JoinPrelRenameVisitor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/JoinPrelRenameVisitor.java
new file mode 100644
index 0000000..3d38484
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/JoinPrelRenameVisitor.java
@@ -0,0 +1,73 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.drill.exec.planner.physical.visitor;
+
+import java.util.List;
+
+import org.apache.drill.exec.planner.physical.ExchangePrel;
+import org.apache.drill.exec.planner.physical.JoinPrel;
+import org.apache.drill.exec.planner.physical.Prel;
+import org.eigenbase.rel.RelNode;
+
+import com.google.common.collect.Lists;
+
+public class JoinPrelRenameVisitor extends BasePrelVisitor<Prel, Void, RuntimeException>{
+
+ private static JoinPrelRenameVisitor INSTANCE = new JoinPrelRenameVisitor();
+
+ public static Prel insertRenameProject(Prel prel){
+ return prel.accept(INSTANCE, null);
+ }
+
+ @Override
+ public Prel visitPrel(Prel prel, Void value) throws RuntimeException {
+ List<RelNode> children = Lists.newArrayList();
+ for(Prel child : prel){
+ child = child.accept(this, null);
+ children.add(child);
+ }
+
+ return (Prel) prel.copy(prel.getTraitSet(), children);
+
+ }
+
+ @Override
+ public Prel visitJoin(JoinPrel prel, Void value) throws RuntimeException {
+
+ List<RelNode> children = Lists.newArrayList();
+
+ for(Prel child : prel){
+ child = child.accept(this, null);
+ children.add(child);
+ }
+
+ final int leftCount = children.get(0).getRowType().getFieldCount();
+
+ List<RelNode> reNamedChildren = Lists.newArrayList();
+
+ RelNode left = prel.getJoinInput(0, children.get(0));
+ RelNode right = prel.getJoinInput(leftCount, children.get(1));
+
+ reNamedChildren.add(left);
+ reNamedChildren.add(right);
+
+ return (Prel) prel.copy(prel.getTraitSet(), reNamedChildren);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/PrelVisitor.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/PrelVisitor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/PrelVisitor.java
new file mode 100644
index 0000000..b834464
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/PrelVisitor.java
@@ -0,0 +1,38 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.exec.planner.physical.visitor;
+
+import org.apache.drill.exec.planner.physical.ExchangePrel;
+import org.apache.drill.exec.planner.physical.JoinPrel;
+import org.apache.drill.exec.planner.physical.Prel;
+import org.apache.drill.exec.planner.physical.ScreenPrel;
+import org.apache.drill.exec.planner.physical.WriterPrel;
+
+
+public interface PrelVisitor<RETURN, EXTRA, EXCEP extends Throwable> {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(PrelVisitor.class);
+
+ public RETURN visitExchange(ExchangePrel prel, EXTRA value) throws EXCEP;
+ public RETURN visitScreen(ScreenPrel prel, EXTRA value) throws EXCEP;
+ public RETURN visitWriter(WriterPrel prel, EXTRA value) throws EXCEP;
+
+ public RETURN visitJoin(JoinPrel prel, EXTRA value) throws EXCEP;
+
+ public RETURN visitPrel(Prel prel, EXTRA value) throws EXCEP;
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/SelectionVectorPrelVisitor.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/SelectionVectorPrelVisitor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/SelectionVectorPrelVisitor.java
new file mode 100644
index 0000000..8d0b155
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/SelectionVectorPrelVisitor.java
@@ -0,0 +1,58 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.exec.planner.physical.visitor;
+
+import java.util.List;
+
+import org.apache.drill.exec.planner.physical.Prel;
+import org.apache.drill.exec.planner.physical.SelectionVectorRemoverPrel;
+import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
+import org.eigenbase.rel.RelNode;
+
+import com.google.common.collect.Lists;
+
+
+public class SelectionVectorPrelVisitor extends BasePrelVisitor<Prel, Void, RuntimeException>{
+
+ private static SelectionVectorPrelVisitor INSTANCE = new SelectionVectorPrelVisitor();
+
+ public static Prel addSelectionRemoversWhereNecessary(Prel prel){
+ return prel.accept(INSTANCE, null);
+ }
+
+ @Override
+ public Prel visitPrel(Prel prel, Void value) throws RuntimeException {
+ SelectionVectorMode[] encodings = prel.getSupportedEncodings();
+ List<RelNode> children = Lists.newArrayList();
+ for(Prel child : prel){
+ child = child.accept(this, null);
+ children.add(convert(encodings, child));
+ }
+
+ return (Prel) prel.copy(prel.getTraitSet(), children);
+ }
+
+ private Prel convert(SelectionVectorMode[] encodings, Prel prel){
+ for(SelectionVectorMode m : encodings){
+ if(prel.getEncoding() == m) return prel;
+ }
+ return new SelectionVectorRemoverPrel(prel);
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillSqlWorker.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillSqlWorker.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillSqlWorker.java
index df66dcf..eb2c891 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillSqlWorker.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillSqlWorker.java
@@ -100,7 +100,7 @@ public class DrillSqlWorker {
}
public PhysicalPlan getPlan(String sql) throws SqlParseException, ValidationException, RelConversionException, IOException{
- return getPlan(null);
+ return getPlan(sql, null);
}
public PhysicalPlan getPlan(String sql, Pointer<String> textPlan) throws SqlParseException, ValidationException, RelConversionException, IOException{
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/DefaultSqlHandler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/DefaultSqlHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/DefaultSqlHandler.java
index e8bd837..29ed1ec 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/DefaultSqlHandler.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/DefaultSqlHandler.java
@@ -38,11 +38,12 @@ import org.apache.drill.exec.planner.logical.DrillRel;
import org.apache.drill.exec.planner.logical.DrillScreenRel;
import org.apache.drill.exec.planner.logical.DrillStoreRel;
import org.apache.drill.exec.planner.physical.DrillDistributionTrait;
-import org.apache.drill.exec.planner.physical.JoinPrelRenameVisitor;
import org.apache.drill.exec.planner.physical.PhysicalPlanCreator;
import org.apache.drill.exec.planner.physical.Prel;
-import org.apache.drill.exec.planner.physical.SelectionVectorPrelVisitor;
import org.apache.drill.exec.planner.physical.explain.PrelSequencer;
+import org.apache.drill.exec.planner.physical.visitor.FinalColumnReorderer;
+import org.apache.drill.exec.planner.physical.visitor.JoinPrelRenameVisitor;
+import org.apache.drill.exec.planner.physical.visitor.SelectionVectorPrelVisitor;
import org.apache.drill.exec.planner.sql.DrillSqlWorker;
import org.apache.drill.exec.util.Pointer;
import org.eigenbase.rel.RelNode;
@@ -80,9 +81,13 @@ public class DefaultSqlHandler extends AbstractSqlHandler {
}
protected void log(String name, Prel node) {
- if(textPlan != null) textPlan.value = PrelSequencer.printWithIds(node, SqlExplainLevel.ALL_ATTRIBUTES);
+ String plan = PrelSequencer.printWithIds(node, SqlExplainLevel.ALL_ATTRIBUTES);;
+ if(textPlan != null){
+ textPlan.value = plan;
+ }
+
if (logger.isDebugEnabled()) {
- logger.debug(name + " : \n" + textPlan.value);
+ logger.debug(name + " : \n" + plan);
}
}
@@ -138,6 +143,11 @@ public class DefaultSqlHandler extends AbstractSqlHandler {
// In such case, we have to insert Project to rename the conflicting names.
phyRelNode = JoinPrelRenameVisitor.insertRenameProject(phyRelNode);
+ // Since our operators work via names rather than indices, we have to make to reorder any output
+ // before we return data to the user as we may have accindentally shuffled things. This adds
+ // a trivial project to reorder columns prior to output.
+ phyRelNode = FinalColumnReorderer.addFinalColumnOrdering(phyRelNode);
+
// the last thing we do is add any required selection vector removers given the supported encodings of each
// operator. This will ultimately move to a new trait but we're managing here for now to avoid introducing new
// issues in planning before the next release
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/ExplainHandler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/ExplainHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/ExplainHandler.java
index 86ce6c5..63db153 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/ExplainHandler.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/ExplainHandler.java
@@ -32,6 +32,7 @@ import org.apache.drill.exec.planner.logical.DrillImplementor;
import org.apache.drill.exec.planner.logical.DrillParseContext;
import org.apache.drill.exec.planner.logical.DrillRel;
import org.apache.drill.exec.planner.physical.Prel;
+import org.apache.drill.exec.planner.physical.explain.PrelSequencer;
import org.apache.drill.exec.planner.sql.DirectPlan;
import org.eigenbase.rel.RelNode;
import org.eigenbase.relopt.RelOptUtil;
@@ -111,7 +112,7 @@ public class ExplainHandler extends DefaultSqlHandler{
public String json;
public PhysicalExplain(RelNode node, PhysicalPlan plan, SqlExplainLevel level, QueryContext context){
- this.text = RelOptUtil.toString(node, level);
+ this.text = PrelSequencer.printWithIds((Prel) node, level);
this.json = plan.unparse(context.getConfig().getMapper().writer());
}
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/BasicServer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/BasicServer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/BasicServer.java
index b1ebd04..599f0fd 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/BasicServer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/BasicServer.java
@@ -143,15 +143,16 @@ public abstract class BasicServer<T extends EnumLite, C extends RemoteConnection
return null;
}
- public int bind(final int initialPort) throws InterruptedException, DrillbitStartupException {
+ public int bind(final int initialPort, boolean allowPortHunting) throws InterruptedException, DrillbitStartupException {
int port = initialPort - 1;
while (true) {
try {
b.bind(++port).sync();
break;
} catch (Exception e) {
- if (e instanceof BindException)
+ if (e instanceof BindException && allowPortHunting){
continue;
+ }
throw new DrillbitStartupException("Could not bind Drillbit", e);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/ControllerImpl.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/ControllerImpl.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/ControllerImpl.java
index de8caf6..1cacc4f 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/ControllerImpl.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/ControllerImpl.java
@@ -39,19 +39,21 @@ public class ControllerImpl implements Controller {
private final ControlMessageHandler handler;
private final BootStrapContext context;
private final ConnectionManagerRegistry connectionRegistry;
+ private final boolean allowPortHunting;
- public ControllerImpl(BootStrapContext context, ControlMessageHandler handler) {
+ public ControllerImpl(BootStrapContext context, ControlMessageHandler handler, boolean allowPortHunting) {
super();
this.handler = handler;
this.context = context;
this.connectionRegistry = new ConnectionManagerRegistry(handler, context);
+ this.allowPortHunting = allowPortHunting;
}
@Override
public DrillbitEndpoint start(DrillbitEndpoint partialEndpoint) throws InterruptedException, DrillbitStartupException {
server = new ControlServer(handler, context, connectionRegistry);
int port = context.getConfig().getInt(ExecConstants.INITIAL_BIT_PORT);
- port = server.bind(port);
+ port = server.bind(port, allowPortHunting);
DrillbitEndpoint completeEndpoint = partialEndpoint.toBuilder().setControlPort(port).build();
connectionRegistry.setEndpoint(completeEndpoint);
return completeEndpoint;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataConnectionCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataConnectionCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataConnectionCreator.java
index f15494f..9c2ef5b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataConnectionCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataConnectionCreator.java
@@ -37,17 +37,19 @@ public class DataConnectionCreator implements Closeable {
private final BootStrapContext context;
private final WorkEventBus workBus;
private final DataResponseHandler dataHandler;
+ private final boolean allowPortHunting;
- public DataConnectionCreator(BootStrapContext context, WorkEventBus workBus, DataResponseHandler dataHandler) {
+ public DataConnectionCreator(BootStrapContext context, WorkEventBus workBus, DataResponseHandler dataHandler, boolean allowPortHunting) {
super();
this.context = context;
this.workBus = workBus;
this.dataHandler = dataHandler;
+ this.allowPortHunting = allowPortHunting;
}
public DrillbitEndpoint start(DrillbitEndpoint partialEndpoint) throws InterruptedException, DrillbitStartupException {
server = new DataServer(context, workBus, dataHandler);
- int port = server.bind(partialEndpoint.getControlPort() + 1);
+ int port = server.bind(partialEndpoint.getControlPort() + 1, allowPortHunting);
DrillbitEndpoint completeEndpoint = partialEndpoint.toBuilder().setDataPort(port).build();
return completeEndpoint;
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/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 eba6e92..fb499b6 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
@@ -81,11 +81,18 @@ public class Drillbit implements Closeable{
private volatile RegistrationHandle handle;
public Drillbit(DrillConfig config, RemoteServiceSet serviceSet) throws Exception {
-
+ boolean allowPortHunting = serviceSet != null;
+ boolean enableHttp = config.getBoolean(ExecConstants.HTTP_ENABLE);
this.context = new BootStrapContext(config);
this.manager = new WorkManager(context);
- this.engine = new ServiceEngine(manager.getControlMessageHandler(), manager.getUserWorker(), context, manager.getWorkBus(), manager.getDataHandler());
- this.embeddedJetty = new Server(8047);
+ this.engine = new ServiceEngine(manager.getControlMessageHandler(), manager.getUserWorker(), context, manager.getWorkBus(), manager.getDataHandler(), allowPortHunting);
+
+ if(enableHttp){
+ this.embeddedJetty = new Server(config.getInt(ExecConstants.HTTP_PORT));
+ }else{
+ this.embeddedJetty = null;
+ }
+
if(serviceSet != null){
this.coord = serviceSet.getCoordinator();
@@ -99,6 +106,8 @@ public class Drillbit implements Closeable{
}
private void startJetty() throws Exception{
+ if(embeddedJetty == null) return;
+
ServletContextHandler context = new ServletContextHandler(ServletContextHandler.NO_SESSIONS);
context.setContextPath("/");
embeddedJetty.setHandler(context);
@@ -131,7 +140,7 @@ public class Drillbit implements Closeable{
logger.warn("Interrupted while sleeping during coordination deregistration.");
}
try {
- embeddedJetty.stop();
+ if(embeddedJetty != null) embeddedJetty.stop();
} catch (Exception e) {
logger.warn("Failure while shutting down embedded jetty server.");
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/service/ServiceEngine.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/service/ServiceEngine.java b/exec/java-exec/src/main/java/org/apache/drill/exec/service/ServiceEngine.java
index cfbde73..bd745d1 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/service/ServiceEngine.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/service/ServiceEngine.java
@@ -43,23 +43,25 @@ import com.google.common.io.Closeables;
public class ServiceEngine implements Closeable{
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ServiceEngine.class);
-
+
private final UserServer userServer;
private final Controller controller;
private final DataConnectionCreator dataPool;
private final DrillConfig config;
boolean useIP = false;
-
- public ServiceEngine(ControlMessageHandler controlMessageHandler, UserWorker userWorker, BootStrapContext context, WorkEventBus workBus, DataResponseHandler dataHandler){
+ private final boolean allowPortHunting;
+
+ public ServiceEngine(ControlMessageHandler controlMessageHandler, UserWorker userWorker, BootStrapContext context, WorkEventBus workBus, DataResponseHandler dataHandler, boolean allowPortHunting){
this.userServer = new UserServer(context.getAllocator(), new NioEventLoopGroup(context.getConfig().getInt(ExecConstants.USER_SERVER_RPC_THREADS),
new NamedThreadFactory("UserServer-")), userWorker);
- this.controller = new ControllerImpl(context, controlMessageHandler);
- this.dataPool = new DataConnectionCreator(context, workBus, dataHandler);
+ this.controller = new ControllerImpl(context, controlMessageHandler, allowPortHunting);
+ this.dataPool = new DataConnectionCreator(context, workBus, dataHandler, allowPortHunting);
this.config = context.getConfig();
+ this.allowPortHunting = allowPortHunting;
}
-
+
public DrillbitEndpoint start() throws DrillbitStartupException, InterruptedException, UnknownHostException{
- int userPort = userServer.bind(config.getInt(ExecConstants.INITIAL_USER_PORT));
+ int userPort = userServer.bind(config.getInt(ExecConstants.INITIAL_USER_PORT), allowPortHunting);
String address = useIP ? InetAddress.getLocalHost().getHostAddress() : InetAddress.getLocalHost().getCanonicalHostName();
DrillbitEndpoint partialEndpoint = DrillbitEndpoint.newBuilder()
.setAddress(address)
@@ -74,7 +76,7 @@ public class ServiceEngine implements Closeable{
public DataConnectionCreator getDataConnectionCreator(){
return dataPool;
}
-
+
public Controller getController() {
return controller;
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/SystemTableScan.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/SystemTableScan.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/SystemTableScan.java
index 9a745ac..b0133f3 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/SystemTableScan.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/SystemTableScan.java
@@ -32,6 +32,7 @@ import org.apache.drill.exec.physical.base.PhysicalOperator;
import org.apache.drill.exec.physical.base.Size;
import org.apache.drill.exec.physical.base.SubScan;
import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.proto.UserBitShared.CoreOperatorType;
import org.apache.drill.exec.store.StoragePluginRegistry;
import parquet.org.codehaus.jackson.annotate.JsonCreator;
@@ -112,6 +113,11 @@ public class SystemTableScan extends AbstractGroupScan implements SubScan{
@Override
+ public int getOperatorType() {
+ return CoreOperatorType.SYSTEM_TABLE_SCAN_VALUE;
+ }
+
+ @Override
public GroupScan clone(List<SchemaPath> columns) {
return this;
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/main/resources/drill-module.conf
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/resources/drill-module.conf b/exec/java-exec/src/main/resources/drill-module.conf
index 9a180fd..26205bd 100644
--- a/exec/java-exec/src/main/resources/drill-module.conf
+++ b/exec/java-exec/src/main/resources/drill-module.conf
@@ -65,6 +65,10 @@ drill.exec: {
delay: 500
}
},
+ http: {
+ enabled: true,
+ port: 8047
+ },
functions: ["org.apache.drill.expr.fn.impl"],
network: {
start: 35000
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/test/java/org/apache/drill/exec/server/TestBitRpc.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/server/TestBitRpc.java b/exec/java-exec/src/test/java/org/apache/drill/exec/server/TestBitRpc.java
index 74f5ba9..f579448 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/server/TestBitRpc.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/server/TestBitRpc.java
@@ -74,7 +74,7 @@ public class TestBitRpc extends ExecTest {
DataResponseHandler drp = new BitComTestHandler();
DataServer server = new DataServer(c, workBus, drp);
- port = server.bind(port);
+ port = server.bind(port, false);
DrillbitEndpoint ep = DrillbitEndpoint.newBuilder().setAddress("localhost").setDataPort(port).build();
DataConnectionManager manager = new DataConnectionManager(FragmentHandle.getDefaultInstance(), ep, c2);
DataTunnel tunnel = new DataTunnel(manager);
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/test/java/org/apache/drill/exec/store/json/JsonRecordReader2Test.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/store/json/JsonRecordReader2Test.java b/exec/java-exec/src/test/java/org/apache/drill/exec/store/json/JsonRecordReader2Test.java
index 0abdbd3..34bcb5e 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/store/json/JsonRecordReader2Test.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/store/json/JsonRecordReader2Test.java
@@ -35,14 +35,14 @@ public class JsonRecordReader2Test extends BaseTestQuery{
}
@Test
- public void z() throws Exception{
+ public void testComplexMultipleTimes() throws Exception{
for(int i =0 ; i < 5; i++){
test("select * from cp.`join/merge_join.json`");
}
}
@Test
- public void y() throws Exception{
+ public void trySimpleQueryWithLimit() throws Exception{
test("select * from cp.`limit/test1.json` limit 10");
}
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/exec/java-exec/src/test/resources/logback.xml
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/resources/logback.xml b/exec/java-exec/src/test/resources/logback.xml
index cd3d971..b8e0ca2 100644
--- a/exec/java-exec/src/test/resources/logback.xml
+++ b/exec/java-exec/src/test/resources/logback.xml
@@ -42,7 +42,7 @@
<!-- </logger> -->
<root>
- <level value="info" />
+ <level value="error" />
<appender-ref ref="STDOUT" />
</root>
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c56f1a66/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index d4c077a..5d976e8 100644
--- a/pom.xml
+++ b/pom.xml
@@ -260,7 +260,7 @@
<artifactId>maven-surefire-plugin</artifactId>
<version>2.17</version>
<configuration>
- <argLine>-Xms512m -Xmx1g -XX:MaxDirectMemorySize=2096M -XX:+CMSClassUnloadingEnabled</argLine>
+ <argLine>-Xms512m -Xmx1g -Ddrill.exec.http.enabled=false -XX:MaxPermSize=256M -XX:MaxDirectMemorySize=2096M -XX:+CMSClassUnloadingEnabled</argLine>
<forkCount>4</forkCount>
<reuseForks>true</reuseForks>
<additionalClasspathElements>