You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@drill.apache.org by ja...@apache.org on 2013/10/30 23:37:24 UTC
[5/7] git commit: DRILL-230: Build a sampling range partitioner
DRILL-230: Build a sampling range partitioner
Project: http://git-wip-us.apache.org/repos/asf/incubator-drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-drill/commit/5ca503c1
Tree: http://git-wip-us.apache.org/repos/asf/incubator-drill/tree/5ca503c1
Diff: http://git-wip-us.apache.org/repos/asf/incubator-drill/diff/5ca503c1
Branch: refs/heads/master
Commit: 5ca503c141f76d8c01c89d0e3a58e1c117ef051f
Parents: 135b0db
Author: Steven Phillips <sp...@maprtech.com>
Authored: Thu Sep 12 20:13:15 2013 -0700
Committer: Steven Phillips <sp...@maprtech.com>
Committed: Wed Oct 30 15:26:22 2013 -0700
----------------------------------------------------------------------
.../common/expression/OutputTypeDeterminer.java | 3 +-
exec/java-exec/pom.xml | 4 +
.../org/apache/drill/exec/cache/HazelCache.java | 23 +-
.../org/apache/drill/exec/cache/VectorWrap.java | 106 +++++
.../apache/drill/exec/client/DrillClient.java | 3 +
.../drill/exec/client/QuerySubmitter.java | 3 +-
.../exec/expr/ExpressionTreeMaterializer.java | 7 +-
.../physical/base/AbstractPhysicalVisitor.java | 5 +
.../exec/physical/base/PhysicalVisitor.java | 1 +
.../config/OrderedPartitionExchange.java | 93 +++++
.../physical/config/OrderedPartitionSender.java | 93 +++++
.../drill/exec/physical/impl/ImplCreator.java | 12 +
.../drill/exec/physical/impl/ScanBatch.java | 3 +-
.../exec/physical/impl/WireRecordBatch.java | 2 +
.../OrderedPartitionBatchCreator.java | 39 ++
.../OrderedPartitionProjector.java | 40 ++
.../OrderedPartitionProjectorTemplate.java | 101 +++++
.../OrderedPartitionRecordBatch.java | 403 +++++++++++++++++++
.../impl/orderedpartitioner/SampleCopier.java | 33 ++
.../SampleCopierTemplate.java | 63 +++
.../orderedpartitioner/SampleSortTemplate.java | 61 +++
.../impl/orderedpartitioner/SampleSorter.java | 33 ++
.../SortContainerBuilder.java | 101 +++++
.../partitionsender/OutgoingRecordBatch.java | 7 +-
.../PartitionSenderRootExec.java | 11 +-
.../impl/project/ProjectRecordBatch.java | 2 +-
.../physical/impl/sort/RecordBatchData.java | 16 +
.../exec/physical/impl/sort/SortBatch.java | 32 +-
.../impl/sort/SortRecordBatchBuilder.java | 19 +-
.../exec/physical/impl/sort/SortTemplate.java | 8 +-
.../drill/exec/physical/impl/sort/Sorter.java | 2 +-
.../planner/fragment/SimpleParallelizer.java | 2 +-
.../drill/exec/record/AbstractRecordBatch.java | 2 +-
.../apache/drill/exec/record/RecordBatch.java | 3 +-
.../drill/exec/record/RecordBatchLoader.java | 6 +-
.../drill/exec/record/VectorAccessible.java | 35 ++
.../drill/exec/record/VectorContainer.java | 21 +-
.../apache/drill/exec/util/BatchPrinter.java | 90 +++++
.../drill/exec/cache/TestVectorCache.java | 97 +++++
.../drill/exec/fn/impl/GeneratorFunctions.java | 110 +++++
.../physical/impl/TestHashToRandomExchange.java | 2 +
.../TestOrderedPartitionExchange.java | 124 ++++++
.../apache/drill/exec/pop/PopUnitTestBase.java | 4 +-
.../test/resources/sender/hash_exchange.json | 4 +-
.../test/resources/sender/hash_exchange2.json | 47 +++
.../test/resources/sender/ordered_exchange.json | 74 ++++
46 files changed, 1894 insertions(+), 56 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/common/src/main/java/org/apache/drill/common/expression/OutputTypeDeterminer.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/drill/common/expression/OutputTypeDeterminer.java b/common/src/main/java/org/apache/drill/common/expression/OutputTypeDeterminer.java
index ed227ec..2f583d5 100644
--- a/common/src/main/java/org/apache/drill/common/expression/OutputTypeDeterminer.java
+++ b/common/src/main/java/org/apache/drill/common/expression/OutputTypeDeterminer.java
@@ -29,7 +29,8 @@ public interface OutputTypeDeterminer {
public static OutputTypeDeterminer FIXED_BIT = new FixedType(Types.required(MinorType.BIT));
public static OutputTypeDeterminer FIXED_INT = new FixedType(Types.required(MinorType.INT));
public static OutputTypeDeterminer FIXED_BIGINT = new FixedType(Types.required(MinorType.BIGINT));
-
+ public static OutputTypeDeterminer FIXED_FLOAT8 = new FixedType(Types.required(MinorType.FLOAT8));
+
public MajorType getOutputType(List<LogicalExpression> expressions);
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/exec/java-exec/pom.xml
----------------------------------------------------------------------
diff --git a/exec/java-exec/pom.xml b/exec/java-exec/pom.xml
index f6543f9..063e60e 100644
--- a/exec/java-exec/pom.xml
+++ b/exec/java-exec/pom.xml
@@ -268,6 +268,10 @@
<filtering>true</filtering>
</resource>
<resource>
+ <directory>src/test/java</directory>
+ <filtering>true</filtering>
+ </resource>
+ <resource>
<directory>target/generated-sources</directory>
<!-- <include>*/org</include> -->
<filtering>true</filtering>
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/exec/java-exec/src/main/java/org/apache/drill/exec/cache/HazelCache.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/HazelCache.java b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/HazelCache.java
index 22435bd..41eca9b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/HazelCache.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/HazelCache.java
@@ -20,6 +20,8 @@ package org.apache.drill.exec.cache;
import java.io.IOException;
import java.util.List;
+import com.hazelcast.core.*;
+import com.hazelcast.nio.DataSerializable;
import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.cache.ProtoBufImpl.HWorkQueueStatus;
@@ -33,12 +35,6 @@ import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.protobuf.InvalidProtocolBufferException;
import com.hazelcast.config.Config;
-import com.hazelcast.core.Hazelcast;
-import com.hazelcast.core.HazelcastInstance;
-import com.hazelcast.core.IMap;
-import com.hazelcast.core.ITopic;
-import com.hazelcast.core.Message;
-import com.hazelcast.core.MessageListener;
public class HazelCache implements DistributedCache {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(HazelCache.class);
@@ -78,7 +74,11 @@ public class HazelCache implements DistributedCache {
if (instance.getName().equals(this.instanceName))
return instance;
}
+ try {
return Hazelcast.newHazelcastInstance(c);
+ } catch (DuplicateInstanceNameException e) {
+ return getInstanceOrCreateNew(c);
+ }
}
// @Override
@@ -108,6 +108,17 @@ public class HazelCache implements DistributedCache {
}
+ public <K,V extends DataSerializable> MultiMap<K,V> getMultiMap(String name) {
+ return this.instance.getMultiMap(name);
+ }
+
+ public <K,V extends DataSerializable> IMap<K,V> getMap(String name) {
+ return this.instance.getMap(name);
+ }
+
+ public AtomicNumber getAtomicNumber(String name) {
+ return this.instance.getAtomicNumber(name);
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/exec/java-exec/src/main/java/org/apache/drill/exec/cache/VectorWrap.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/VectorWrap.java b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/VectorWrap.java
new file mode 100644
index 0000000..4811ea5
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/VectorWrap.java
@@ -0,0 +1,106 @@
+/**
+ * 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.cache;
+
+import com.beust.jcommander.internal.Lists;
+import com.google.protobuf.MessageLite;
+import com.google.protobuf.Parser;
+import com.hazelcast.nio.DataSerializable;
+import io.netty.buffer.ByteBuf;
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.exec.expr.TypeHelper;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.proto.UserBitShared;
+import org.apache.drill.exec.record.MaterializedField;
+import org.apache.drill.exec.server.BootStrapContext;
+import org.apache.drill.exec.server.DrillbitContext;
+import org.apache.drill.exec.vector.BaseDataValueVector;
+import org.apache.drill.exec.vector.ValueVector;
+import org.apache.drill.exec.proto.UserBitShared.FieldMetadata;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.List;
+
+public class VectorWrap implements DataSerializable{
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(VectorWrap.class);
+
+ List<ValueVector> vectorList;
+ BootStrapContext context;
+ final Parser<FieldMetadata> parser = FieldMetadata.PARSER;
+
+ public VectorWrap(List<ValueVector> vectorList){
+ this.vectorList = vectorList;
+ this.context = new BootStrapContext(DrillConfig.create());
+ }
+
+ public VectorWrap() {
+ this.vectorList = Lists.newArrayList();
+ this.context = new BootStrapContext(DrillConfig.create());
+ }
+
+ @Override
+ public void readData(DataInput arg0) throws IOException {
+ int size = arg0.readInt();
+ for (int i = 0; i < size; i++) {
+ int metaLength = arg0.readInt();
+ byte[] meta = new byte[metaLength];
+ arg0.readFully(meta);
+ FieldMetadata metaData = parser.parseFrom(meta);
+ int dataLength = metaData.getBufferLength();
+ byte[] bytes = new byte[dataLength];
+ arg0.readFully(bytes);
+ MaterializedField field = MaterializedField.create(metaData.getDef());
+ ByteBuf buf = context.getAllocator().buffer(dataLength);
+ buf.setBytes(0, bytes);
+ ValueVector vector = TypeHelper.getNewVector(field, context.getAllocator());
+ vector.load(metaData, buf);
+ vectorList.add((BaseDataValueVector) vector);
+ }
+ }
+
+ @Override
+ public void writeData(DataOutput arg0) throws IOException {
+ int size = vectorList.size();
+ arg0.writeInt(size);
+ for (ValueVector vector : vectorList) {
+ byte[] meta = vector.getMetadata().toByteArray();
+ int length = vector.getBufferSize();
+ byte[] bytes = new byte[length];
+ ByteBuf[] data = vector.getBuffers();
+ int pos = 0;
+ for (ByteBuf bb : data) {
+ bb.getBytes(0, bytes, pos, bb.readableBytes());
+ pos += bb.readableBytes();
+ }
+ arg0.writeInt(meta.length);
+ arg0.write(meta);
+ arg0.write(bytes);
+ }
+ }
+
+ public List<ValueVector> get() {
+ return vectorList;
+ }
+
+ public void set(List<ValueVector> vectorList) {
+ this.vectorList = vectorList;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/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 3dadb0c..4d6088d 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
@@ -204,12 +204,15 @@ public class DrillClient implements Closeable{
} else {
fail(ex);
}
+ } else {
+ fail(ex);
}
}
private void fail(Exception ex) {
logger.debug("Submission failed.", ex);
future.setException(ex);
+ future.set(results);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/exec/java-exec/src/main/java/org/apache/drill/exec/client/QuerySubmitter.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/client/QuerySubmitter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/client/QuerySubmitter.java
index 835b9b5..160ef7f 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/client/QuerySubmitter.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/client/QuerySubmitter.java
@@ -149,6 +149,7 @@ public class QuerySubmitter {
@Override
public void resultArrived(QueryResultBatch result) {
int rows = result.getHeader().getRowCount();
+ count.addAndGet(rows);
if (result.getData() != null) {
try {
loader.load(result.getHeader().getDef(), result.getData());
@@ -173,11 +174,9 @@ public class QuerySubmitter {
Object o = vw.getValueVector().getAccessor().getObject(row);
if (o instanceof byte[]) {
String value = new String((byte[]) o);
- count.addAndGet(1);
System.out.printf("| %-15s",value.length() <= 15 ? value : value.substring(0, 14));
} else {
String value = o.toString();
- count.addAndGet(1);
System.out.printf("| %-15s",value.length() <= 15 ? value : value.substring(0,14));
}
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/exec/java-exec/src/main/java/org/apache/drill/exec/expr/ExpressionTreeMaterializer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/ExpressionTreeMaterializer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/ExpressionTreeMaterializer.java
index bb9fc25..5d576b1 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/ExpressionTreeMaterializer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/ExpressionTreeMaterializer.java
@@ -32,6 +32,7 @@ import org.apache.drill.exec.record.RecordBatch;
import org.apache.drill.exec.record.TypedFieldId;
import com.google.common.collect.Lists;
+import org.apache.drill.exec.record.VectorAccessible;
public class ExpressionTreeMaterializer {
@@ -42,16 +43,16 @@ public class ExpressionTreeMaterializer {
private ExpressionTreeMaterializer() {
};
- public static LogicalExpression materialize(LogicalExpression expr, RecordBatch batch, ErrorCollector errorCollector) {
+ public static LogicalExpression materialize(LogicalExpression expr, VectorAccessible batch, ErrorCollector errorCollector) {
return expr.accept(new MaterializeVisitor(batch, errorCollector), null);
}
private static class MaterializeVisitor extends SimpleExprVisitor<LogicalExpression> {
private final ErrorCollector errorCollector;
- private final RecordBatch batch;
+ private final VectorAccessible batch;
private ExpressionValidator validator = new ExpressionValidator();
- public MaterializeVisitor(RecordBatch batch, ErrorCollector errorCollector) {
+ public MaterializeVisitor(VectorAccessible batch, ErrorCollector errorCollector) {
this.batch = batch;
this.errorCollector = errorCollector;
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractPhysicalVisitor.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractPhysicalVisitor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractPhysicalVisitor.java
index 6c087a1..bf6c68c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractPhysicalVisitor.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractPhysicalVisitor.java
@@ -101,6 +101,11 @@ public abstract class AbstractPhysicalVisitor<T, X, E extends Throwable> impleme
}
@Override
+ public T visitOrderedPartitionSender(OrderedPartitionSender op, X value) throws E {
+ return visitSender(op, value);
+ }
+
+ @Override
public T visitRandomReceiver(RandomReceiver op, X value) throws E {
return visitReceiver(op, value);
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/PhysicalVisitor.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/PhysicalVisitor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/PhysicalVisitor.java
index aef9d78..5692b9f 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/PhysicalVisitor.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/PhysicalVisitor.java
@@ -47,6 +47,7 @@ public interface PhysicalVisitor<RETURN, EXTRA, EXCEP extends Throwable> {
public RETURN visitHashPartitionSender(HashPartitionSender op, EXTRA value) throws EXCEP;
+ public RETURN visitOrderedPartitionSender(OrderedPartitionSender op, EXTRA value) throws EXCEP;
public RETURN visitRandomReceiver(RandomReceiver op, EXTRA value) throws EXCEP;
public RETURN visitHashPartitionSender(HashToRandomExchange op, EXTRA value) throws EXCEP;
public RETURN visitRangeSender(RangeSender op, EXTRA value) throws EXCEP;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/OrderedPartitionExchange.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/OrderedPartitionExchange.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/OrderedPartitionExchange.java
new file mode 100644
index 0000000..2ae0bb5
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/OrderedPartitionExchange.java
@@ -0,0 +1,93 @@
+/**
+ * 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.physical.config;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import org.apache.drill.common.defs.OrderDef;
+import org.apache.drill.common.expression.FieldReference;
+import org.apache.drill.common.expression.LogicalExpression;
+import org.apache.drill.exec.physical.base.AbstractExchange;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.physical.base.Receiver;
+import org.apache.drill.exec.physical.base.Sender;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+
+import java.util.List;
+
+@JsonTypeName("ordered-partition-exchange")
+public class OrderedPartitionExchange extends AbstractExchange {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(OrderedPartitionExchange.class);
+
+
+ private final List<OrderDef> orderings;
+ private final FieldReference ref;
+
+ //ephemeral for setup tasks.
+ private List<DrillbitEndpoint> senderLocations;
+ private List<DrillbitEndpoint> receiverLocations;
+
+ @JsonCreator
+ public OrderedPartitionExchange(@JsonProperty("orderings") List<OrderDef> orderings, @JsonProperty("ref") FieldReference ref, @JsonProperty("child") PhysicalOperator child) {
+ super(child);
+ this.orderings = orderings;
+ this.ref = ref;
+ }
+
+ @Override
+ public int getMaxSendWidth() {
+ return Integer.MAX_VALUE;
+ }
+
+
+ @Override
+ protected void setupSenders(List<DrillbitEndpoint> senderLocations) {
+ this.senderLocations = senderLocations;
+ }
+
+ @Override
+ protected void setupReceivers(List<DrillbitEndpoint> receiverLocations) {
+ this.receiverLocations = receiverLocations;
+ }
+
+ @Override
+ public Sender getSender(int minorFragmentId, PhysicalOperator child) {
+ return new OrderedPartitionSender(orderings, ref, child, receiverLocations, receiverMajorFragmentId, senderLocations.size());
+ }
+
+ @Override
+ public Receiver getReceiver(int minorFragmentId) {
+ return new RandomReceiver(senderMajorFragmentId, senderLocations);
+ }
+
+ @Override
+ protected PhysicalOperator getNewWithChild(PhysicalOperator child) {
+ return new OrderedPartitionExchange(orderings, ref, child);
+ }
+
+ @Override
+ public boolean supportsSelectionVector() {
+ return true;
+ }
+
+
+
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/OrderedPartitionSender.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/OrderedPartitionSender.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/OrderedPartitionSender.java
new file mode 100644
index 0000000..f2d7e58
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/OrderedPartitionSender.java
@@ -0,0 +1,93 @@
+/**
+ * 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.physical.config;
+
+import com.beust.jcommander.internal.Lists;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import org.apache.drill.common.defs.OrderDef;
+import org.apache.drill.common.expression.FieldReference;
+import org.apache.drill.exec.physical.OperatorCost;
+import org.apache.drill.exec.physical.base.*;
+import org.apache.drill.exec.proto.CoordinationProtos;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+
+import java.util.List;
+
+@JsonTypeName("OrderedPartitionSender")
+public class OrderedPartitionSender extends AbstractSender {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(OrderedPartitionSender.class);
+
+ private final List<OrderDef> orderings;
+ private final FieldReference ref;
+ private final List<DrillbitEndpoint> endpoints;
+ private final int sendingWidth;
+
+ @JsonCreator
+ public OrderedPartitionSender(@JsonProperty("orderings") List<OrderDef> orderings, @JsonProperty("ref") FieldReference ref, @JsonProperty("child") PhysicalOperator child,
+ @JsonProperty("destinations") List<DrillbitEndpoint> endpoints, @JsonProperty("receiver-major-fragment") int oppositeMajorFragmentId,
+ @JsonProperty("sending-fragment-width") int sendingWidth) {
+ super(oppositeMajorFragmentId, child);
+ if (orderings == null) {
+ this.orderings = Lists.newArrayList();
+ } else {
+ this.orderings = orderings;
+ }
+ this.ref = ref;
+ this.endpoints = endpoints;
+ this.sendingWidth = sendingWidth;
+ }
+
+ public int getSendingWidth() {
+ return sendingWidth;
+ }
+
+ public List<DrillbitEndpoint> getDestinations() {
+ return endpoints;
+ }
+
+ public List<OrderDef> getOrderings() {
+ return orderings;
+ }
+
+ public FieldReference getRef() {
+ return ref;
+ }
+
+ @Override
+ public <T, X, E extends Throwable> T accept(PhysicalVisitor<T, X, E> physicalVisitor, X value) throws E{
+ return physicalVisitor.visitOrderedPartitionSender(this, value);
+ }
+
+ @Override
+ public OperatorCost getCost() {
+ return new OperatorCost(0, 0, 1000, child.getSize().getRecordCount());
+ }
+
+ @Override
+ public Size getSize() {
+ //TODO: This should really change the row width...
+ return child.getSize();
+ }
+
+ @Override
+ protected PhysicalOperator getNewWithChild(PhysicalOperator child) {
+ return new OrderedPartitionSender(orderings, ref, child, endpoints, oppositeMajorFragmentId, sendingWidth);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ImplCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ImplCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ImplCreator.java
index 35ef8ac..844b3a7 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ImplCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ImplCreator.java
@@ -32,6 +32,7 @@ import org.apache.drill.exec.physical.config.Union;
import org.apache.drill.exec.physical.impl.filter.FilterBatchCreator;
import org.apache.drill.exec.physical.impl.join.MergeJoinCreator;
import org.apache.drill.exec.physical.impl.limit.LimitBatchCreator;
+import org.apache.drill.exec.physical.impl.orderedpartitioner.OrderedPartitionBatchCreator;
import org.apache.drill.exec.physical.impl.partitionsender.PartitionSenderCreator;
import org.apache.drill.exec.physical.impl.project.ProjectBatchCreator;
import org.apache.drill.exec.physical.impl.sort.SortBatchCreator;
@@ -60,8 +61,10 @@ public class ImplCreator extends AbstractPhysicalVisitor<RecordBatch, FragmentCo
private ScreenCreator sc = new ScreenCreator();
private RandomReceiverCreator rrc = new RandomReceiverCreator();
private PartitionSenderCreator hsc = new PartitionSenderCreator();
+ private OrderedPartitionBatchCreator opc = new OrderedPartitionBatchCreator();
private SingleSenderCreator ssc = new SingleSenderCreator();
private ProjectBatchCreator pbc = new ProjectBatchCreator();
+ private OrderedPartitionBatchCreator smplbc = new OrderedPartitionBatchCreator();
private FilterBatchCreator fbc = new FilterBatchCreator();
private LimitBatchCreator lbc = new LimitBatchCreator();
private UnionBatchCreator unionbc = new UnionBatchCreator();
@@ -135,6 +138,15 @@ public class ImplCreator extends AbstractPhysicalVisitor<RecordBatch, FragmentCo
root = hsc.getRoot(context, op, getChildren(op, context));
return null;
}
+
+ @Override
+ public RecordBatch visitOrderedPartitionSender(OrderedPartitionSender op, FragmentContext context) throws ExecutionSetupException {
+ List<RecordBatch> children = Lists.newArrayList();
+ children.add(opc.getBatch(context, op, getChildren(op, context)));
+ HashPartitionSender config = new HashPartitionSender(op.getOppositeMajorFragmentId(), op, op.getRef(),op.getDestinations());
+ root = hsc.getRoot(context, config, children);
+ return null;
+ }
@Override
public RecordBatch visitFilter(Filter filter, FragmentContext context) throws ExecutionSetupException {
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ScanBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ScanBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ScanBatch.java
index 10d595a..ce0e15f 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ScanBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ScanBatch.java
@@ -28,7 +28,6 @@ import org.apache.drill.exec.record.BatchSchema;
import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
import org.apache.drill.exec.record.MaterializedField;
import org.apache.drill.exec.record.RecordBatch;
-import org.apache.drill.exec.record.SchemaBuilder;
import org.apache.drill.exec.record.TypedFieldId;
import org.apache.drill.exec.record.VectorContainer;
import org.apache.drill.exec.record.VectorWrapper;
@@ -129,7 +128,7 @@ public class ScanBatch implements RecordBatch {
@Override
public TypedFieldId getValueVectorId(SchemaPath path) {
- return container.getValueVector(path);
+ return container.getValueVectorId(path);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/WireRecordBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/WireRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/WireRecordBatch.java
index 5f8b416..465bb3d 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/WireRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/WireRecordBatch.java
@@ -33,6 +33,7 @@ import org.apache.drill.exec.record.VectorWrapper;
import org.apache.drill.exec.record.WritableBatch;
import org.apache.drill.exec.record.selection.SelectionVector2;
import org.apache.drill.exec.record.selection.SelectionVector4;
+import org.apache.drill.exec.util.BatchPrinter;
import org.apache.drill.exec.vector.ValueVector;
public class WireRecordBatch implements RecordBatch{
@@ -105,6 +106,7 @@ public class WireRecordBatch implements RecordBatch{
RecordBatchDef rbd = batch.getHeader().getDef();
boolean schemaChanged = batchLoader.load(rbd, batch.getBody());
+// System.out.println(rbd.getRecordCount());
batch.release();
if(schemaChanged){
this.schema = batchLoader.getSchema();
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionBatchCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionBatchCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionBatchCreator.java
new file mode 100644
index 0000000..615cf21
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionBatchCreator.java
@@ -0,0 +1,39 @@
+/**
+ * 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.physical.impl.orderedpartitioner;
+
+import com.google.common.base.Preconditions;
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.physical.config.OrderedPartitionSender;
+import org.apache.drill.exec.physical.impl.BatchCreator;
+import org.apache.drill.exec.record.RecordBatch;
+
+import java.util.List;
+
+public class OrderedPartitionBatchCreator implements BatchCreator<OrderedPartitionSender>{
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(OrderedPartitionBatchCreator.class);
+
+ @Override
+ public RecordBatch getBatch(FragmentContext context, OrderedPartitionSender config, List<RecordBatch> children) throws ExecutionSetupException {
+ Preconditions.checkArgument(children.size() == 1);
+ return new OrderedPartitionRecordBatch(config, children.iterator().next(), context);
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionProjector.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionProjector.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionProjector.java
new file mode 100644
index 0000000..2265150
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionProjector.java
@@ -0,0 +1,40 @@
+/**
+ * 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.physical.impl.orderedpartitioner;
+
+import org.apache.drill.common.expression.SchemaPath;
+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.record.RecordBatch;
+import org.apache.drill.exec.record.TransferPair;
+import org.apache.drill.exec.record.VectorAccessible;
+import org.apache.drill.exec.record.VectorContainer;
+
+import java.util.List;
+
+public interface OrderedPartitionProjector {
+
+ public abstract void setup(FragmentContext context, VectorAccessible incoming, RecordBatch outgoing, List<TransferPair> transfers,
+ VectorContainer partitionVectors, int partitions, SchemaPath outputField)
+ throws SchemaChangeException;
+ public abstract int projectRecords(int recordCount, int firstOutputIndex);
+
+ public static TemplateClassDefinition<OrderedPartitionProjector> TEMPLATE_DEFINITION = new TemplateClassDefinition<OrderedPartitionProjector>(OrderedPartitionProjector.class, OrderedPartitionProjectorTemplate.class);
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionProjectorTemplate.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionProjectorTemplate.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionProjectorTemplate.java
new file mode 100644
index 0000000..e22450e
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionProjectorTemplate.java
@@ -0,0 +1,101 @@
+/**
+ * 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.physical.impl.orderedpartitioner;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
+import org.apache.drill.exec.record.RecordBatch;
+import org.apache.drill.exec.record.TransferPair;
+import org.apache.drill.exec.record.VectorAccessible;
+import org.apache.drill.exec.record.VectorContainer;
+import org.apache.drill.exec.record.selection.SelectionVector2;
+import org.apache.drill.exec.record.selection.SelectionVector4;
+import org.apache.drill.exec.vector.IntVector;
+
+import javax.inject.Named;
+import java.util.List;
+
+public abstract class OrderedPartitionProjectorTemplate implements OrderedPartitionProjector {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(OrderedPartitionProjectorTemplate.class);
+
+ private ImmutableList<TransferPair> transfers;
+ private VectorContainer partitionVectors;
+ private int partitions;
+ private SelectionVector2 vector2;
+ private SelectionVector4 vector4;
+ private SelectionVectorMode svMode;
+ private RecordBatch outBatch;
+ private SchemaPath outputField;
+
+ public OrderedPartitionProjectorTemplate() throws SchemaChangeException{
+ }
+
+ private int getPartition(int index) {
+ //TODO replace this with binary search
+ int partitionIndex = 0;
+ while (partitionIndex < partitions - 1 && doEval(index, partitionIndex) >= 0) {
+ partitionIndex++;
+ }
+ return partitionIndex;
+ }
+
+ @Override
+ public final int projectRecords(final int recordCount, int firstOutputIndex) {
+ final int countN = recordCount;
+ IntVector partitionValues = (IntVector) outBatch.getValueAccessorById(outBatch.getValueVectorId(outputField).getFieldId(), IntVector.class).getValueVector();
+ int counter = 0;
+ for (int i = 0; i < countN; i++, firstOutputIndex++) {
+ int partition = getPartition(i);
+ partitionValues.getMutator().set(i, partition);
+ counter++;
+ }
+ for(TransferPair t : transfers){
+ t.transfer();
+ }
+ return counter;
+ }
+
+ @Override
+ public final void setup(FragmentContext context, VectorAccessible incoming, RecordBatch outgoing, List<TransferPair> transfers,
+ VectorContainer partitionVectors, int partitions, SchemaPath outputField) throws SchemaChangeException{
+
+ this.svMode = incoming.getSchema().getSelectionVectorMode();
+ this.outBatch = outgoing;
+ this.outputField = outputField;
+ switch(svMode){
+ case FOUR_BYTE:
+ case TWO_BYTE:
+ throw new UnsupportedOperationException("Selection vector not supported");
+ }
+ this.transfers = ImmutableList.copyOf(transfers);
+ this.partitions = partitions;
+ doSetup(context, incoming, outgoing, partitionVectors);
+ }
+
+ public abstract void doSetup(@Named("context") FragmentContext context, @Named("incoming") VectorAccessible incoming,
+ @Named("outgoing") RecordBatch outgoing, @Named("partitionVectors") VectorContainer partitionVectors);
+ public abstract int doEval(@Named("inIndex") int inIndex, @Named("partitionIndex") int partitionIndex);
+
+
+
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/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
new file mode 100644
index 0000000..af0f1ff
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/OrderedPartitionRecordBatch.java
@@ -0,0 +1,403 @@
+/**
+ * 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.physical.impl.orderedpartitioner;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+import com.hazelcast.core.AtomicNumber;
+import com.hazelcast.core.IMap;
+import com.hazelcast.core.MultiMap;
+import com.sun.codemodel.JConditional;
+import com.sun.codemodel.JExpr;
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.common.defs.OrderDef;
+import org.apache.drill.common.expression.*;
+import org.apache.drill.common.logical.data.NamedExpression;
+import org.apache.drill.common.logical.data.Order;
+import org.apache.drill.common.types.TypeProtos;
+import org.apache.drill.common.types.Types;
+import org.apache.drill.exec.cache.HazelCache;
+import org.apache.drill.exec.cache.VectorWrap;
+import org.apache.drill.exec.compile.sig.MappingSet;
+import org.apache.drill.exec.exception.ClassTransformationException;
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.expr.*;
+import org.apache.drill.exec.expr.fn.impl.ComparatorFunctions;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.physical.config.OrderedPartitionSender;
+import org.apache.drill.exec.physical.impl.sort.SortBatch;
+import org.apache.drill.exec.physical.impl.sort.SortRecordBatchBuilder;
+import org.apache.drill.exec.physical.impl.sort.Sorter;
+import org.apache.drill.exec.record.*;
+import org.apache.drill.exec.record.selection.SelectionVector4;
+import org.apache.drill.exec.util.BatchPrinter;
+import org.apache.drill.exec.vector.*;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+public class OrderedPartitionRecordBatch extends AbstractRecordBatch<OrderedPartitionSender>{
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(OrderedPartitionRecordBatch.class);
+
+ public static final MappingSet MAIN_MAPPING = new MappingSet( (String) null, null, CodeGenerator.DEFAULT_SCALAR_MAP, CodeGenerator.DEFAULT_SCALAR_MAP);
+ public static final MappingSet INCOMING_MAPPING = new MappingSet("inIndex", null, "incoming", null, CodeGenerator.DEFAULT_SCALAR_MAP, CodeGenerator.DEFAULT_SCALAR_MAP);
+ public static final MappingSet PARTITION_MAPPING = new MappingSet("partitionIndex", null, "partitionVectors", null,
+ CodeGenerator.DEFAULT_SCALAR_MAP, CodeGenerator.DEFAULT_SCALAR_MAP);
+
+ private static long MAX_SORT_BYTES = 8l * 1024 * 1024 * 1024;
+ private static int RECORDS_TO_SAMPLE = 5000;
+ private static int SAMPLING_FACTOR = 10;
+ private static float COMPLETION_FACTOR = .75f;
+ private static String SAMPLE_MAP_NAME = "sampleMap";
+ private static String TABLE_MAP_NAME = "TableMap";
+ protected final RecordBatch incoming;
+ private boolean first = true;
+ private OrderedPartitionProjector projector;
+ private List<ValueVector> allocationVectors;
+ private VectorContainer partitionVectors = new VectorContainer();
+ private int partitions;
+ private Queue<VectorContainer> batchQueue;
+ private SortRecordBatchBuilder builder;
+ private int recordsSampled;
+ private int sendingMajorFragmentWidth;
+ private boolean startedUnsampledBatches = false;
+ private boolean upstreamNone = false;
+ private int recordCount;
+
+ public OrderedPartitionRecordBatch(OrderedPartitionSender pop, RecordBatch incoming, FragmentContext context){
+ super(pop, context);
+ this.incoming = incoming;
+ this.partitions = pop.getDestinations().size();
+ this.sendingMajorFragmentWidth = pop.getSendingWidth();
+ }
+
+ private boolean getPartitionVectors() {
+ VectorContainer sampleContainer = new VectorContainer();
+ recordsSampled = 0;
+ IterOutcome upstream;
+ builder = new SortRecordBatchBuilder(context.getAllocator(), MAX_SORT_BYTES, sampleContainer);
+ builder.add(incoming);
+ recordsSampled += incoming.getRecordCount();
+ try {
+ outer: while (recordsSampled < RECORDS_TO_SAMPLE) {
+ upstream = incoming.next();
+ switch(upstream) {
+ case NONE:
+ case NOT_YET:
+ case STOP:
+ upstreamNone = true;
+ break outer;
+ }
+ builder.add(incoming);
+ recordsSampled += incoming.getRecordCount();
+ if (upstream == IterOutcome.NONE) break;
+ //TODO handle upstream cases
+ }
+ builder.build(context);
+ Sorter sorter = SortBatch.createNewSorter(context, popConfig.getOrderings(), sampleContainer);
+ SelectionVector4 sv4 = builder.getSv4();
+ sorter.setup(context, sv4, sampleContainer);
+ sorter.sort(sv4, sampleContainer);
+
+ VectorContainer containerToCache = new VectorContainer();
+ SampleCopier copier = getCopier(sv4, sampleContainer, containerToCache, popConfig.getOrderings());
+ copier.copyRecords(recordsSampled/(SAMPLING_FACTOR * partitions), 0, SAMPLING_FACTOR * partitions);
+
+ for (VectorWrapper vw : containerToCache) {
+ vw.getValueVector().getMutator().setValueCount(copier.getOutputRecords());
+ }
+
+// BatchPrinter.printBatch(containerToCache);
+
+ HazelCache cache = new HazelCache(DrillConfig.create());
+ cache.run();
+ String mapKey = String.format("%s_%d", context.getHandle().getQueryId(), context.getHandle().getMajorFragmentId());
+ MultiMap<String, VectorWrap> mmap = cache.getMultiMap(SAMPLE_MAP_NAME);
+
+ List<ValueVector> vectorList = Lists.newArrayList();
+ for (VectorWrapper vw : containerToCache) {
+ vectorList.add(vw.getValueVector());
+ }
+
+ VectorWrap wrap = new VectorWrap(vectorList);
+
+ mmap.put(mapKey, wrap);
+
+ AtomicNumber minorFragmentSampleCount = cache.getAtomicNumber(mapKey);
+
+ long val = minorFragmentSampleCount.incrementAndGet();
+ logger.debug("Incremented mfsc, got {}", val);
+
+ for (int i = 0; i < 10 && minorFragmentSampleCount.get() < sendingMajorFragmentWidth * COMPLETION_FACTOR; i++) {
+ Thread.sleep(10);
+ }
+
+ Collection<VectorWrap> allSamplesWrap = mmap.get(mapKey);
+ VectorContainer allSamplesContainer = new VectorContainer();
+ int orderSize = popConfig.getOrderings().size();
+ SortContainerBuilder containerBuilder = new SortContainerBuilder(context.getAllocator(), MAX_SORT_BYTES, allSamplesContainer, orderSize);
+ for (VectorWrap w : allSamplesWrap) {
+ containerBuilder.add(w.get());
+ }
+ containerBuilder.build(context);
+
+// BatchPrinter.printHyperBatch(allSamplesContainer);
+
+ List<OrderDef> orderDefs = Lists.newArrayList();
+ int i = 0;
+ for (OrderDef od : popConfig.getOrderings()) {
+ SchemaPath sp = new SchemaPath("f" + i++, ExpressionPosition.UNKNOWN);
+ orderDefs.add(new OrderDef(od.getDirection(), new FieldReference(sp)));
+ }
+
+ SelectionVector4 newSv4 = containerBuilder.getSv4();
+ Sorter sorter2 = SortBatch.createNewSorter(context, orderDefs, allSamplesContainer);
+ sorter2.setup(context, newSv4, allSamplesContainer);
+ sorter2.sort(newSv4, allSamplesContainer);
+
+ VectorContainer candidatePartitionTable = new VectorContainer();
+ SampleCopier copier2 = getCopier(newSv4, allSamplesContainer, candidatePartitionTable, orderDefs);
+ int skipRecords = containerBuilder.getSv4().getTotalCount() / partitions;
+ copier2.copyRecords(skipRecords, skipRecords, partitions - 1);
+ assert copier2.getOutputRecords() == partitions - 1 : String.format("output records: %d partitions: %d", copier2.getOutputRecords(), partitions);
+ for (VectorWrapper vw : candidatePartitionTable) {
+ vw.getValueVector().getMutator().setValueCount(copier2.getOutputRecords());
+ }
+
+// BatchPrinter.printBatch(candidatePartitionTable);
+
+ vectorList = Lists.newArrayList();
+ for (VectorWrapper vw: candidatePartitionTable) {
+ vectorList.add(vw.getValueVector());
+ }
+ wrap = new VectorWrap(vectorList);
+
+ IMap<String, VectorWrap> tableMap = cache.getMap(TABLE_MAP_NAME);
+ tableMap.putIfAbsent(mapKey + "final", wrap, 1, TimeUnit.MINUTES);
+ wrap = tableMap.get(mapKey + "final");
+ Preconditions.checkState(wrap != null);
+ for (ValueVector vv : wrap.get()) {
+ partitionVectors.add(vv);
+ }
+ } catch (ClassTransformationException | IOException | SchemaChangeException | InterruptedException ex) {
+ kill();
+ logger.error("Failure during query", ex);
+ context.fail(ex);
+ return false;
+ }
+ return true;
+ }
+
+ private SampleCopier getCopier(SelectionVector4 sv4, VectorContainer incoming, VectorContainer outgoing, List<OrderDef> orderings) throws SchemaChangeException{
+ List<ValueVector> localAllocationVectors = Lists.newArrayList();
+ final ErrorCollector collector = new ErrorCollectorImpl();
+ final CodeGenerator<SampleCopier> cg = new CodeGenerator<SampleCopier>(SampleCopier.TEMPLATE_DEFINITION, context.getFunctionRegistry());
+
+ int i = 0;
+ for(OrderDef od : orderings) {
+ final LogicalExpression expr = ExpressionTreeMaterializer.materialize(od.getExpr(), incoming, collector);
+ SchemaPath schemaPath = new SchemaPath("f" + i++, ExpressionPosition.UNKNOWN);
+ TypeProtos.MajorType.Builder builder = TypeProtos.MajorType.newBuilder().mergeFrom(expr.getMajorType()).clearMode().setMode(TypeProtos.DataMode.REQUIRED);
+ TypeProtos.MajorType newType = builder.build();
+ MaterializedField outputField = MaterializedField.create(schemaPath, newType);
+ if(collector.hasErrors()){
+ throw new SchemaChangeException(String.format("Failure while trying to materialize incoming schema. Errors:\n %s.", collector.toErrorString()));
+ }
+
+ ValueVector vector = TypeHelper.getNewVector(outputField, context.getAllocator());
+ localAllocationVectors.add(vector);
+ TypedFieldId fid = outgoing.add(vector);
+ ValueVectorWriteExpression write = new ValueVectorWriteExpression(fid, expr);
+ cg.addExpr(write);
+ logger.debug("Added eval.");
+ }
+ for (ValueVector vv : localAllocationVectors) {
+ AllocationHelper.allocate(vv, SAMPLING_FACTOR * partitions, 50);
+ }
+// outgoing.addCollection(allocationVectors);
+ outgoing.buildSchema(BatchSchema.SelectionVectorMode.NONE);
+ try {
+ SampleCopier sampleCopier = context.getImplementationClass(cg);
+ sampleCopier.setupCopier(context, sv4, incoming, outgoing);
+ return sampleCopier;
+ } catch (ClassTransformationException | IOException e) {
+ throw new SchemaChangeException(e);
+ }
+ }
+
+ @Override
+ protected void killIncoming() {
+ incoming.kill();
+ }
+
+ @Override
+ public IterOutcome next() {
+ if (upstreamNone && (batchQueue == null || batchQueue.size() == 0)) return IterOutcome.NONE;
+ if (batchQueue != null && batchQueue.size() > 0) {
+ VectorContainer vc = batchQueue.poll();
+ recordCount = vc.getRecordCount();
+ try{
+ setupNewSchema(vc);
+ }catch(SchemaChangeException ex){
+ kill();
+ logger.error("Failure during query", ex);
+ context.fail(ex);
+ return IterOutcome.STOP;
+ }
+ doWork(vc);
+ return IterOutcome.OK_NEW_SCHEMA;
+ }
+ IterOutcome upstream = incoming.next();
+ recordCount = incoming.getRecordCount();
+ if(this.first && upstream == IterOutcome.OK) {
+ upstream = IterOutcome.OK_NEW_SCHEMA;
+ }
+ if(this.first && upstream == IterOutcome.OK_NEW_SCHEMA) {
+ if (!getPartitionVectors()) return IterOutcome.STOP;
+ batchQueue = new LinkedBlockingQueue<>(builder.getContainers());
+ first = false;
+ VectorContainer vc = batchQueue.poll();
+ try{
+ setupNewSchema(vc);
+ }catch(SchemaChangeException ex){
+ kill();
+ logger.error("Failure during query", ex);
+ context.fail(ex);
+ return IterOutcome.STOP;
+ }
+ doWork(vc);
+ return IterOutcome.OK_NEW_SCHEMA;
+ }
+ if (this.startedUnsampledBatches == false) {
+ this.startedUnsampledBatches = true;
+ if (upstream == IterOutcome.OK) upstream = IterOutcome.OK_NEW_SCHEMA;
+ }
+ switch(upstream){
+ case NONE:
+ case NOT_YET:
+ case STOP:
+ container.zeroVectors();
+ return upstream;
+ case OK_NEW_SCHEMA:
+ try{
+ setupNewSchema(incoming);
+ }catch(SchemaChangeException ex){
+ kill();
+ logger.error("Failure during query", ex);
+ context.fail(ex);
+ return IterOutcome.STOP;
+ }
+ // fall through.
+ case OK:
+ doWork(incoming);
+ return upstream; // change if upstream changed, otherwise normal.
+ default:
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ @Override
+ public int getRecordCount() {
+ return recordCount;
+ }
+
+ protected void doWork(VectorAccessible batch) {
+ int recordCount = batch.getRecordCount();
+ for(ValueVector v : this.allocationVectors){
+ AllocationHelper.allocate(v, recordCount, 50);
+ }
+ projector.projectRecords(recordCount, 0);
+ for(VectorWrapper<?> v : container){
+ ValueVector.Mutator m = v.getValueVector().getMutator();
+ m.setValueCount(recordCount);
+ }
+ }
+
+ /** hack to make ref and full work together... need to figure out if this is still necessary. **/
+ private FieldReference getRef(NamedExpression e){
+ FieldReference ref = e.getRef();
+ PathSegment seg = ref.getRootSegment();
+ if(seg.isNamed() && "output".contentEquals(seg.getNameSegment().getPath())){
+ return new FieldReference(ref.getPath().toString().subSequence(7, ref.getPath().length()), ref.getPosition());
+ }
+ return ref;
+ }
+
+ protected void setupNewSchema(VectorAccessible batch) throws SchemaChangeException{
+ this.allocationVectors = Lists.newArrayList();
+ container.clear();
+ final ErrorCollector collector = new ErrorCollectorImpl();
+ final List<TransferPair> transfers = Lists.newArrayList();
+
+ final CodeGenerator<OrderedPartitionProjector> cg = new CodeGenerator<OrderedPartitionProjector>(OrderedPartitionProjector.TEMPLATE_DEFINITION, context.getFunctionRegistry());
+
+ for (VectorWrapper vw : batch) {
+ TransferPair tp = vw.getValueVector().getTransferPair();
+ transfers.add(tp);
+ container.add(tp.getTo());
+ }
+
+ cg.setMappingSet(MAIN_MAPPING);
+
+ int count = 0;
+ for(OrderDef od : popConfig.getOrderings()){
+ // first, we rewrite the evaluation stack for each side of the comparison.
+ final LogicalExpression expr = ExpressionTreeMaterializer.materialize(od.getExpr(), batch, collector);
+ if(collector.hasErrors()) throw new SchemaChangeException("Failure while materializing expression. " + collector.toErrorString());
+ cg.setMappingSet(INCOMING_MAPPING);
+ CodeGenerator.HoldingContainer left = cg.addExpr(expr, false);
+ cg.setMappingSet(PARTITION_MAPPING);
+ CodeGenerator.HoldingContainer right = cg.addExpr(new ValueVectorReadExpression(new TypedFieldId(expr.getMajorType(), count++)), false);
+ cg.setMappingSet(MAIN_MAPPING);
+
+ // next we wrap the two comparison sides and add the expression block for the comparison.
+ FunctionCall f = new FunctionCall(ComparatorFunctions.COMPARE_TO, ImmutableList.of((LogicalExpression) new HoldingContainerExpression(left), new HoldingContainerExpression(right)), ExpressionPosition.UNKNOWN);
+ CodeGenerator.HoldingContainer out = cg.addExpr(f, false);
+ JConditional jc = cg.getEvalBlock()._if(out.getValue().ne(JExpr.lit(0)));
+
+ if(od.getDirection() == Order.Direction.ASC){
+ jc._then()._return(out.getValue());
+ }else{
+ jc._then()._return(out.getValue().minus());
+ }
+ }
+
+ cg.getEvalBlock()._return(JExpr.lit(0));
+
+ SchemaPath outputPath = new SchemaPath(popConfig.getRef().getPath(), ExpressionPosition.UNKNOWN);
+ MaterializedField outputField = MaterializedField.create(outputPath, Types.required(TypeProtos.MinorType.INT));
+ ValueVector v = TypeHelper.getNewVector(outputField, context.getAllocator());
+ container.add(v);
+ this.allocationVectors.add(v);
+ container.buildSchema(batch.getSchema().getSelectionVectorMode());
+
+ try {
+ this.projector = context.getImplementationClass(cg);
+ projector.setup(context, batch, this, transfers, partitionVectors, partitions, outputPath);
+ } catch (ClassTransformationException | IOException e) {
+ throw new SchemaChangeException("Failure while attempting to load generated class", e);
+ }
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/SampleCopier.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/SampleCopier.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/SampleCopier.java
new file mode 100644
index 0000000..ddb605b
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/SampleCopier.java
@@ -0,0 +1,33 @@
+/**
+ * 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.physical.impl.orderedpartitioner;
+
+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.record.VectorAccessible;
+import org.apache.drill.exec.record.selection.SelectionVector4;
+
+public interface SampleCopier {
+ public static TemplateClassDefinition<SampleCopier> TEMPLATE_DEFINITION = new TemplateClassDefinition<SampleCopier>(SampleCopier.class, SampleCopierTemplate.class);
+
+ public void setupCopier(FragmentContext context, SelectionVector4 sv4, VectorAccessible incoming, VectorAccessible outgoing) throws SchemaChangeException;
+ public abstract void copyRecords(int skip, int start, int total);
+ public int getOutputRecords();
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/SampleCopierTemplate.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/SampleCopierTemplate.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/SampleCopierTemplate.java
new file mode 100644
index 0000000..ddb56c1
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/SampleCopierTemplate.java
@@ -0,0 +1,63 @@
+/**
+ * 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.physical.impl.orderedpartitioner;
+
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.record.VectorAccessible;
+import org.apache.drill.exec.record.selection.SelectionVector4;
+
+import javax.inject.Named;
+
+public abstract class SampleCopierTemplate implements SampleCopier {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(SampleCopierTemplate.class);
+
+ private SelectionVector4 sv4;
+ private int outputRecords = 0;
+
+ @Override
+ public void setupCopier(FragmentContext context, SelectionVector4 sv4, VectorAccessible incoming, VectorAccessible outgoing)
+ throws SchemaChangeException{
+ this.sv4 = sv4;
+ doSetup(context, incoming, outgoing);
+ }
+
+ @Override
+ public int getOutputRecords() {
+ return outputRecords;
+ }
+
+
+ @Override
+ public void copyRecords(int skip, int start, int total) {
+ final int recordCount = sv4.getCount();
+ int outgoingPosition = 0;
+ int increment = skip > 0 ? skip : 1;
+ for(int svIndex = start; svIndex < sv4.getCount() && outputRecords < total; svIndex += increment, outgoingPosition++){
+ int deRefIndex = sv4.get(svIndex);
+ doEval(deRefIndex, outgoingPosition);
+ outputRecords++;
+ }
+ }
+
+ public abstract void doSetup(@Named("context") FragmentContext context, @Named("incoming") VectorAccessible incoming, @Named("outgoing") VectorAccessible outgoing);
+ public abstract void doEval(@Named("inIndex") int inIndex, @Named("outIndex") int outIndex);
+
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/SampleSortTemplate.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/SampleSortTemplate.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/SampleSortTemplate.java
new file mode 100644
index 0000000..5a83541
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/SampleSortTemplate.java
@@ -0,0 +1,61 @@
+/**
+ * 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.physical.impl.orderedpartitioner;
+
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.record.RecordBatch;
+import org.apache.drill.exec.record.VectorContainer;
+import org.apache.drill.exec.record.selection.SelectionVector2;
+import org.apache.hadoop.util.IndexedSortable;
+import org.apache.hadoop.util.QuickSort;
+
+import javax.inject.Named;
+
+public abstract class SampleSortTemplate implements SampleSorter, IndexedSortable{
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(SampleSortTemplate.class);
+
+ SelectionVector2 vector2;
+
+ public void setup(FragmentContext context, VectorContainer sampleBatch, SelectionVector2 vector2) throws SchemaChangeException{
+ this.vector2 = vector2;
+ doSetup(context, sampleBatch, null);
+ }
+
+ @Override
+ public void sort(SelectionVector2 vector2, VectorContainer container){
+ QuickSort qs = new QuickSort();
+ qs.sort(this, 0, vector2.getCount());
+ }
+
+ @Override
+ public void swap(int sv0, int sv1) {
+ char tmp = vector2.getIndex(sv0);
+ vector2.setIndex(sv0, vector2.getIndex(sv1));
+ vector2.setIndex(sv1, tmp);
+ }
+
+ @Override
+ public int compare(int leftIndex, int rightIndex) {
+ return doEval(leftIndex, rightIndex);
+ }
+
+ public abstract void doSetup(@Named("context") FragmentContext context, @Named("incoming") VectorContainer incoming, @Named("outgoing") RecordBatch outgoing);
+ public abstract int doEval(@Named("leftIndex") int leftIndex, @Named("rightIndex") int rightIndex);
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/SampleSorter.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/SampleSorter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/SampleSorter.java
new file mode 100644
index 0000000..b58f15e
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/SampleSorter.java
@@ -0,0 +1,33 @@
+/**
+ * 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.physical.impl.orderedpartitioner;
+
+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.record.RecordBatch;
+import org.apache.drill.exec.record.VectorContainer;
+import org.apache.drill.exec.record.selection.SelectionVector2;
+
+public interface SampleSorter {
+ public void setup(FragmentContext context, RecordBatch hyperBatch) throws SchemaChangeException;
+ public void sort(SelectionVector2 vector2, VectorContainer container);
+
+ public static TemplateClassDefinition<SampleSorter> TEMPLATE_DEFINITION = new TemplateClassDefinition<SampleSorter>(SampleSorter.class, SampleSortTemplate.class);
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/SortContainerBuilder.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/SortContainerBuilder.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/SortContainerBuilder.java
new file mode 100644
index 0000000..d479108
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/orderedpartitioner/SortContainerBuilder.java
@@ -0,0 +1,101 @@
+/**
+ * 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.physical.impl.orderedpartitioner;
+
+import com.beust.jcommander.internal.Lists;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ArrayListMultimap;
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.memory.BufferAllocator.PreAllocator;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.physical.impl.sort.RecordBatchData;
+import org.apache.drill.exec.record.*;
+import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
+import org.apache.drill.exec.record.selection.SelectionVector4;
+import org.apache.drill.exec.vector.ValueVector;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class SortContainerBuilder {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(SortContainerBuilder.class);
+
+ private final ArrayListMultimap<BatchSchema, RecordBatchData> batches = ArrayListMultimap.create();
+ private final VectorContainer container;
+
+ private int recordCount;
+ private SelectionVector4 sv4;
+ final PreAllocator svAllocator;
+ private List<List<ValueVector>> hyperVectors;
+
+ public SortContainerBuilder(BufferAllocator a, long maxBytes, VectorContainer container, int fields){
+ this.svAllocator = a.getPreAllocator();
+ this.container = container;
+ this.hyperVectors = Lists.newArrayList();
+ for (int i = 0; i < fields; i++) {
+ List<ValueVector> list = Lists.newArrayList();
+ hyperVectors.add(list);
+ }
+ }
+
+ private long getSize(RecordBatch batch){
+ long bytes = 0;
+ for(VectorWrapper<?> v : batch){
+ bytes += v.getValueVector().getBufferSize();
+ }
+ return bytes;
+ }
+
+ public boolean add(List<ValueVector> vectors){
+ Preconditions.checkArgument(vectors.size() == hyperVectors.size());
+ int i = 0;
+ for (ValueVector vv : vectors) {
+ if (i == 0) recordCount += vv.getMetadata().getValueCount();
+ hyperVectors.get(i++).add(vv);
+ }
+ return true;
+ }
+
+ public void build(FragmentContext context) throws SchemaChangeException{
+ container.clear();
+ svAllocator.preAllocate(recordCount * 4);
+ sv4 = new SelectionVector4(svAllocator.getAllocation(), recordCount, Character.MAX_VALUE);
+
+ // now we're going to generate the sv4 pointers
+ int index = 0;
+ int recordBatchId = 0;
+ for(ValueVector vv : hyperVectors.get(0)){
+ for(int i = 0; i < vv.getMetadata().getValueCount(); i++, index++){
+ sv4.set(index, recordBatchId, i);
+ }
+ recordBatchId++;
+ }
+
+ for (List<ValueVector> hyperVector : hyperVectors) {
+ container.addHyperList(hyperVector);
+ }
+
+ container.buildSchema(SelectionVectorMode.FOUR_BYTE);
+ }
+
+ public SelectionVector4 getSv4() {
+ return sv4;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/OutgoingRecordBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/OutgoingRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/OutgoingRecordBatch.java
index 29ed795..9f38c77 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/OutgoingRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/OutgoingRecordBatch.java
@@ -39,6 +39,7 @@ import org.apache.drill.exec.record.selection.SelectionVector4;
import org.apache.drill.exec.rpc.BaseRpcOutcomeListener;
import org.apache.drill.exec.rpc.RpcException;
import org.apache.drill.exec.rpc.bit.BitTunnel;
+import org.apache.drill.exec.util.BatchPrinter;
import org.apache.drill.exec.vector.ValueVector;
import org.apache.drill.exec.vector.allocator.VectorAllocator;
import org.apache.drill.exec.work.foreman.ErrorHelper;
@@ -103,7 +104,9 @@ public class OutgoingRecordBatch implements RecordBatch {
for(VectorWrapper<?> w : vectorContainer){
w.getValueVector().getMutator().setValueCount(recordCount);
}
-
+
+// BatchPrinter.printBatch(vectorContainer);
+
FragmentWritableBatch writableBatch = new FragmentWritableBatch(isLast,
handle.getQueryId(),
handle.getMajorFragmentId(),
@@ -222,7 +225,7 @@ public class OutgoingRecordBatch implements RecordBatch {
@Override
public TypedFieldId getValueVectorId(SchemaPath path) {
- return vectorContainer.getValueVector(path);
+ return vectorContainer.getValueVectorId(path);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/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 b26b817..e26ea1b 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
@@ -35,6 +35,7 @@ import org.apache.drill.exec.proto.CoordinationProtos;
import org.apache.drill.exec.record.RecordBatch;
import org.apache.drill.exec.record.TypedFieldId;
import org.apache.drill.exec.record.VectorWrapper;
+import org.apache.drill.exec.util.BatchPrinter;
import org.apache.drill.exec.vector.ValueVector;
import com.sun.codemodel.JArray;
@@ -68,13 +69,9 @@ class PartitionSenderRootExec implements RootExec {
outgoing[fieldId] = new OutgoingRecordBatch(operator,
context.getCommunicator().getTunnel(endpoint),
incoming,
- context);
- try {
- createPartitioner();
- } catch (SchemaChangeException e) {
- ok = false;
- logger.error("Failed to create partitioning sender during query ", e);
- context.fail(e);
+ context,
+ fieldId);
+ fieldId++;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/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 2386dc2..41a4c4d 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
@@ -96,7 +96,7 @@ public class ProjectRecordBatch extends AbstractSingleRecordBatch<Project>{
final CodeGenerator<Projector> cg = new CodeGenerator<Projector>(Projector.TEMPLATE_DEFINITION, context.getFunctionRegistry());
- for(int i =0; i < exprs.size(); i++){
+ for(int i = 0; i < exprs.size(); i++){
final NamedExpression namedExpression = exprs.get(i);
final LogicalExpression expr = ExpressionTreeMaterializer.materialize(namedExpression.getExpr(), incoming, collector);
final MaterializedField outputField = MaterializedField.create(getRef(namedExpression), expr.getMajorType());
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/sort/RecordBatchData.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/sort/RecordBatchData.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/sort/RecordBatchData.java
index 1226e85..7300d9e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/sort/RecordBatchData.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/sort/RecordBatchData.java
@@ -22,6 +22,7 @@ import java.util.List;
import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
import org.apache.drill.exec.record.RecordBatch;
import org.apache.drill.exec.record.TransferPair;
+import org.apache.drill.exec.record.VectorContainer;
import org.apache.drill.exec.record.VectorWrapper;
import org.apache.drill.exec.record.selection.SelectionVector2;
import org.apache.drill.exec.vector.ValueVector;
@@ -37,6 +38,7 @@ public class RecordBatchData {
final List<ValueVector> vectors = Lists.newArrayList();
final SelectionVector2 sv2;
final int recordCount;
+ VectorContainer container;
public RecordBatchData(RecordBatch batch){
this.sv2 = batch.getSchema().getSelectionVectorMode() == SelectionVectorMode.TWO_BYTE ? batch.getSelectionVector2().clone() : null;
@@ -61,4 +63,18 @@ public class RecordBatchData {
public SelectionVector2 getSv2() {
return sv2;
}
+
+ public VectorContainer getContainer() {
+ if (this.container == null) buildContainer();
+ return this.container;
+ }
+
+ private void buildContainer() {
+ assert container == null;
+ container = new VectorContainer();
+ for (ValueVector vv : vectors) {
+ container.add(vv);
+ }
+ container.buildSchema(sv2 == null ? SelectionVectorMode.NONE : SelectionVectorMode.TWO_BYTE);
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/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 0bfd6ca..4be5065 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
@@ -18,6 +18,7 @@
package org.apache.drill.exec.physical.impl.sort;
import java.io.IOException;
+import java.util.List;
import org.apache.drill.common.defs.OrderDef;
import org.apache.drill.common.expression.ErrorCollector;
@@ -35,11 +36,9 @@ import org.apache.drill.exec.expr.ExpressionTreeMaterializer;
import org.apache.drill.exec.expr.HoldingContainerExpression;
import org.apache.drill.exec.expr.fn.impl.ComparatorFunctions;
import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
import org.apache.drill.exec.physical.config.Sort;
-import org.apache.drill.exec.record.AbstractRecordBatch;
-import org.apache.drill.exec.record.BatchSchema;
-import org.apache.drill.exec.record.RecordBatch;
-import org.apache.drill.exec.record.WritableBatch;
+import org.apache.drill.exec.record.*;
import org.apache.drill.exec.record.selection.SelectionVector2;
import org.apache.drill.exec.record.selection.SelectionVector4;
@@ -139,7 +138,7 @@ public class SortBatch extends AbstractRecordBatch<Sort> {
sv4 = builder.getSv4();
sorter = createNewSorter();
- sorter.setup(context, this);
+ sorter.setup(context, this.getSelectionVector4(), this.container);
long t1 = System.nanoTime();
sorter.sort(sv4, container);
logger.debug("Sorted {} records in {} micros.", sv4.getTotalCount(), (System.nanoTime() - t1)/1000);
@@ -154,22 +153,29 @@ public class SortBatch extends AbstractRecordBatch<Sort> {
}
}
+ private Sorter createNewSorter() throws ClassTransformationException, IOException, SchemaChangeException {
+ return createNewSorter(this.context, this.popConfig.getOrderings(), this, MAIN_MAPPING, LEFT_MAPPING, RIGHT_MAPPING);
+ }
+
+ public static Sorter createNewSorter(FragmentContext context, List<OrderDef> orderings, VectorAccessible batch) throws ClassTransformationException, IOException, SchemaChangeException {
+ return createNewSorter(context, orderings, batch, MAIN_MAPPING, LEFT_MAPPING, RIGHT_MAPPING);
+ }
-
- private Sorter createNewSorter() throws ClassTransformationException, IOException, SchemaChangeException{
+ public static Sorter createNewSorter(FragmentContext context, List<OrderDef> orderings, VectorAccessible batch, MappingSet mainMapping, MappingSet leftMapping, MappingSet rightMapping)
+ throws ClassTransformationException, IOException, SchemaChangeException{
CodeGenerator<Sorter> g = new CodeGenerator<Sorter>(Sorter.TEMPLATE_DEFINITION, context.getFunctionRegistry());
- g.setMappingSet(MAIN_MAPPING);
+ g.setMappingSet(mainMapping);
- for(OrderDef od : popConfig.getOrderings()){
+ for(OrderDef od : orderings){
// first, we rewrite the evaluation stack for each side of the comparison.
ErrorCollector collector = new ErrorCollectorImpl();
- final LogicalExpression expr = ExpressionTreeMaterializer.materialize(od.getExpr(), this, collector);
+ final LogicalExpression expr = ExpressionTreeMaterializer.materialize(od.getExpr(), batch, collector);
if(collector.hasErrors()) throw new SchemaChangeException("Failure while materializing expression. " + collector.toErrorString());
- g.setMappingSet(LEFT_MAPPING);
+ g.setMappingSet(leftMapping);
HoldingContainer left = g.addExpr(expr, false);
- g.setMappingSet(RIGHT_MAPPING);
+ g.setMappingSet(rightMapping);
HoldingContainer right = g.addExpr(expr, false);
- g.setMappingSet(MAIN_MAPPING);
+ g.setMappingSet(mainMapping);
// next we wrap the two comparison sides and add the expression block for the comparison.
FunctionCall f = new FunctionCall(ComparatorFunctions.COMPARE_TO, ImmutableList.of((LogicalExpression) new HoldingContainerExpression(left), new HoldingContainerExpression(right)), ExpressionPosition.UNKNOWN);
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/5ca503c1/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 8873792..4024d57 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
@@ -17,8 +17,10 @@
*/
package org.apache.drill.exec.physical.impl.sort;
+import com.google.common.collect.Lists;
import io.netty.buffer.ByteBuf;
+import java.util.ArrayList;
import java.util.List;
import org.apache.drill.exec.exception.SchemaChangeException;
@@ -74,11 +76,13 @@ public class SortRecordBatchBuilder {
if (batch.getRecordCount() == 0) return true; // skip over empty record batches.
long batchBytes = getSize(batch);
+ if (batchBytes == 0) {return true;}
if(batchBytes + runningBytes > maxBytes) return false; // enough data memory.
if(runningBatches+1 > Character.MAX_VALUE) return false; // allowed in batch.
if(!svAllocator.preAllocate(batch.getRecordCount()*4)) return false; // sv allocation available.
-
+
+ if (batch.getRecordCount() == 0) return true;
RecordBatchData bd = new RecordBatchData(batch);
runningBytes += batchBytes;
batches.put(batch.getSchema(), bd);
@@ -143,5 +147,18 @@ public class SortRecordBatchBuilder {
public SelectionVector4 getSv4() {
return sv4;
}
+
+ public List<VectorContainer> getContainers() {
+ ArrayList containerList = Lists.newArrayList();
+ int recordCount = 0;
+ for (BatchSchema bs : batches.keySet()) {
+ for (RecordBatchData bd : batches.get(bs)) {
+ VectorContainer c = bd.getContainer();
+ c.setRecordCount(bd.getRecordCount());
+ containerList.add(c);
+ }
+ }
+ return containerList;
+ }
}