You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by am...@apache.org on 2014/12/13 02:46:32 UTC
drill git commit: DRILL-1846: Use max receiver width during stats
collection for parallelism planning. SingleMergeExchange and UnionExchange
have a max receive width of 1. Others can go higher.
Repository: drill
Updated Branches:
refs/heads/0.7.0 d925eab56 -> 285ea8147
DRILL-1846: Use max receiver width during stats collection for parallelism planning. SingleMergeExchange and UnionExchange have a max receive width of 1. Others can go higher.
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/285ea814
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/285ea814
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/285ea814
Branch: refs/heads/0.7.0
Commit: 285ea81473470d777cc65648aefc5682d3723471
Parents: d925eab
Author: Aman Sinha <as...@maprtech.com>
Authored: Fri Dec 12 08:14:21 2014 -0800
Committer: Aman Sinha <as...@maprtech.com>
Committed: Fri Dec 12 12:09:17 2014 -0800
----------------------------------------------------------------------
.../drill/exec/physical/base/AbstractExchange.java | 5 +++++
.../apache/drill/exec/physical/base/Exchange.java | 8 ++++++++
.../exec/physical/config/SingleMergeExchange.java | 5 +++++
.../drill/exec/physical/config/UnionExchange.java | 5 +++++
.../drill/exec/planner/fragment/StatsCollector.java | 2 ++
.../java/org/apache/drill/TestExampleQueries.java | 15 +++++++++++++++
6 files changed, 40 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/285ea814/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractExchange.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractExchange.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractExchange.java
index af5b0c2..73280ea 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractExchange.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractExchange.java
@@ -41,6 +41,11 @@ public abstract class AbstractExchange extends AbstractSingle implements Exchang
return false;
}
+ @Override
+ public int getMaxReceiveWidth() {
+ return Integer.MAX_VALUE;
+ }
+
protected abstract void setupSenders(List<DrillbitEndpoint> senderLocations) throws PhysicalOperatorSetupException ;
protected abstract void setupReceivers(List<DrillbitEndpoint> senderLocations) throws PhysicalOperatorSetupException ;
http://git-wip-us.apache.org/repos/asf/drill/blob/285ea814/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/Exchange.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/Exchange.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/Exchange.java
index f2fa4dc..7be7f20 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/Exchange.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/Exchange.java
@@ -73,6 +73,14 @@ public interface Exchange extends PhysicalOperator {
public abstract int getMaxSendWidth();
/**
+ * The widest width this receiver can receive(max receive parallelization). Default is Integer.MAX_VALUE.
+ *
+ * @return
+ */
+ @JsonIgnore
+ public abstract int getMaxReceiveWidth();
+
+ /**
* Return the feeding child of this operator node.
*
* @return
http://git-wip-us.apache.org/repos/asf/drill/blob/285ea814/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/SingleMergeExchange.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/SingleMergeExchange.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/SingleMergeExchange.java
index 6470734..2914112 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/SingleMergeExchange.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/SingleMergeExchange.java
@@ -55,6 +55,11 @@ public class SingleMergeExchange extends AbstractExchange {
}
@Override
+ public int getMaxReceiveWidth() {
+ return 1;
+ }
+
+ @Override
protected void setupSenders(List<CoordinationProtos.DrillbitEndpoint> senderLocations) {
this.senderLocations = senderLocations;
}
http://git-wip-us.apache.org/repos/asf/drill/blob/285ea814/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/UnionExchange.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/UnionExchange.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/UnionExchange.java
index d58048c..cfc21ac 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/UnionExchange.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/UnionExchange.java
@@ -70,6 +70,11 @@ public class UnionExchange extends AbstractExchange{
}
@Override
+ public int getMaxReceiveWidth() {
+ return 1;
+ }
+
+ @Override
protected PhysicalOperator getNewWithChild(PhysicalOperator child) {
return new UnionExchange(child);
}
http://git-wip-us.apache.org/repos/asf/drill/blob/285ea814/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/StatsCollector.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/StatsCollector.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/StatsCollector.java
index a2685ad..41ff678 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/StatsCollector.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/StatsCollector.java
@@ -71,6 +71,8 @@ public class StatsCollector {
@Override
public Void visitReceivingExchange(Exchange exchange, Wrapper wrapper) throws RuntimeException {
+ Stats stats = wrapper.getStats();
+ stats.addMaxWidth(exchange.getMaxReceiveWidth()) ;
// no traversal since it would cross fragment boundary.
return null;
}
http://git-wip-us.apache.org/repos/asf/drill/blob/285ea814/exec/java-exec/src/test/java/org/apache/drill/TestExampleQueries.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/TestExampleQueries.java b/exec/java-exec/src/test/java/org/apache/drill/TestExampleQueries.java
index 987c070..2427d2c 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/TestExampleQueries.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/TestExampleQueries.java
@@ -20,6 +20,7 @@ package org.apache.drill;
import static org.junit.Assert.assertEquals;
import org.apache.drill.common.util.FileUtils;
+import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.rpc.RpcException;
import org.junit.Ignore;
import org.junit.Test;
@@ -588,4 +589,18 @@ public class TestExampleQueries extends BaseTestQuery{
}
+ @Test // DRILL-1846 (this tests issue with SimpleMergeExchange)
+ public void testOrderByDiffColumnsInSubqAndOuter() throws Exception {
+ String query = "select n.n_nationkey from (select n_nationkey, n_regionkey from cp.`tpch/nation.parquet` order by n_regionkey) n order by n.n_nationkey";
+ // set slice_target = 1 to force exchanges
+ test("alter session set `planner.slice_target` = 1; " + query);
+ }
+
+ @Test // DRILL-1846 (this tests issue with UnionExchange)
+ public void testLimitInSubqAndOrderByOuter() throws Exception {
+ String query = "select t2.n_nationkey from (select n_nationkey, n_regionkey from cp.`tpch/nation.parquet` t1 group by n_nationkey, n_regionkey limit 10) t2 order by t2.n_nationkey";
+ // set slice_target = 1 to force exchanges
+ test("alter session set `planner.slice_target` = 1; " + query);
+ }
+
}