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 2014/04/20 06:08:09 UTC
[01/17] git commit: DRILL-536: Check for number of aggregate
functions.
Repository: incubator-drill
Updated Branches:
refs/heads/master ecaa838fe -> a88102bfa
DRILL-536: Check for number of aggregate functions.
Project: http://git-wip-us.apache.org/repos/asf/incubator-drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-drill/commit/8296be55
Tree: http://git-wip-us.apache.org/repos/asf/incubator-drill/tree/8296be55
Diff: http://git-wip-us.apache.org/repos/asf/incubator-drill/diff/8296be55
Branch: refs/heads/master
Commit: 8296be5594fa98da5a1d37a5fde871b6de4bca00
Parents: ecaa838
Author: Aman Sinha <as...@maprtech.com>
Authored: Wed Apr 16 19:10:21 2014 -0700
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Sat Apr 19 21:06:56 2014 -0700
----------------------------------------------------------------------
.../physical/impl/aggregate/HashAggBatch.java | 12 +++--
.../impl/aggregate/HashAggTemplate.java | 21 +++++---
.../exec/physical/impl/agg/TestHashAggr.java | 5 ++
.../src/test/resources/agg/hashagg/q7_3.json | 56 ++++++++++++++++++++
4 files changed, 82 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/8296be55/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggBatch.java
index 9add544..a75aac9 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggBatch.java
@@ -181,15 +181,17 @@ public class HashAggBatch extends AbstractRecordBatch<HashAggregate> {
List<VectorAllocator> keyAllocators = Lists.newArrayList();
List<VectorAllocator> valueAllocators = Lists.newArrayList();
- aggrExprs = new LogicalExpression[popConfig.getAggrExprs().length];
- groupByOutFieldIds = new TypedFieldId[popConfig.getGroupByExprs().length];
- aggrOutFieldIds = new TypedFieldId[popConfig.getAggrExprs().length];
+ int numGroupByExprs = (popConfig.getGroupByExprs() != null) ? popConfig.getGroupByExprs().length : 0;
+ int numAggrExprs = (popConfig.getAggrExprs() != null) ? popConfig.getAggrExprs().length : 0;
+ aggrExprs = new LogicalExpression[numAggrExprs];
+ groupByOutFieldIds = new TypedFieldId[numGroupByExprs];
+ aggrOutFieldIds = new TypedFieldId[numAggrExprs];
ErrorCollector collector = new ErrorCollectorImpl();
int i;
- for(i = 0; i < popConfig.getGroupByExprs().length; i++) {
+ for(i = 0; i < numGroupByExprs; i++) {
NamedExpression ne = popConfig.getGroupByExprs()[i];
final LogicalExpression expr = ExpressionTreeMaterializer.materialize(ne.getExpr(), incoming, collector, context.getFunctionRegistry() );
if(expr == null) continue;
@@ -202,7 +204,7 @@ public class HashAggBatch extends AbstractRecordBatch<HashAggregate> {
groupByOutFieldIds[i] = container.add(vv);
}
- for(i = 0; i < aggrExprs.length; i++){
+ for(i = 0; i < numAggrExprs; i++){
NamedExpression ne = popConfig.getAggrExprs()[i];
final LogicalExpression expr = ExpressionTreeMaterializer.materialize(ne.getExpr(), incoming, collector, context.getFunctionRegistry() );
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/8296be55/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggTemplate.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggTemplate.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggTemplate.java
index 21c0c7d..b0f81ef 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggTemplate.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/HashAggTemplate.java
@@ -155,9 +155,14 @@ public abstract class HashAggTemplate implements HashAggregator {
TypedFieldId[] groupByOutFieldIds,
VectorAllocator[] keyAllocators, VectorAllocator[] valueAllocators)
throws SchemaChangeException, ClassTransformationException, IOException {
-
- if (valueFieldIds.size() < valueExprs.length) throw new IllegalArgumentException("Wrong number of workspace variables.");
-
+
+ if (valueExprs == null || valueFieldIds == null) {
+ throw new IllegalArgumentException("Invalid aggr value exprs or workspace variables.");
+ }
+ if (valueFieldIds.size() < valueExprs.length) {
+ throw new IllegalArgumentException("Wrong number of workspace variables.");
+ }
+
this.context = context;
this.incoming = incoming;
this.schema = incoming.getSchema();
@@ -180,10 +185,12 @@ public abstract class HashAggTemplate implements HashAggregator {
this.htIdxHolder = new IntHolder();
materializedValueFields = new MaterializedField[valueFieldIds.size()];
- int i = 0;
- FieldReference ref = new FieldReference("dummy", ExpressionPosition.UNKNOWN, valueFieldIds.get(0).getType());
- for (TypedFieldId id : valueFieldIds) {
- materializedValueFields[i++] = MaterializedField.create(ref, id.getType());
+ if (valueFieldIds.size() > 0) {
+ int i = 0;
+ FieldReference ref = new FieldReference("dummy", ExpressionPosition.UNKNOWN, valueFieldIds.get(0).getType());
+ for (TypedFieldId id : valueFieldIds) {
+ materializedValueFields[i++] = MaterializedField.create(ref, id.getType());
+ }
}
ChainedHashTable ht = new ChainedHashTable(hashAggrConfig.getHtConfig(), context, incoming, null /* no incoming probe */, outgoing) ;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/8296be55/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/agg/TestHashAggr.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/agg/TestHashAggr.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/agg/TestHashAggr.java
index 1ab7248..9047208 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/agg/TestHashAggr.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/agg/TestHashAggr.java
@@ -39,6 +39,11 @@ public class TestHashAggr extends BaseTestQuery{
public void testQ7_2() throws Exception{
testPhysicalFromFile("agg/hashagg/q7_2.json");
}
+
+ @Test
+ public void testQ7_3() throws Exception{
+ testPhysicalFromFile("agg/hashagg/q7_3.json");
+ }
@Ignore // ignore temporarily since this shows memory leak in ParquetRecordReader (DRILL-443)
@Test
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/8296be55/exec/java-exec/src/test/resources/agg/hashagg/q7_3.json
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/resources/agg/hashagg/q7_3.json b/exec/java-exec/src/test/resources/agg/hashagg/q7_3.json
new file mode 100644
index 0000000..11d2665
--- /dev/null
+++ b/exec/java-exec/src/test/resources/agg/hashagg/q7_3.json
@@ -0,0 +1,56 @@
+{
+ head : {
+ version : 1,
+ generator : {
+ type : "optiq",
+ info : "na"
+ },
+ type : "APACHE_DRILL_PHYSICAL"
+ },
+ graph : [ {
+ "pop" : "parquet-scan",
+ "@id" : 1,
+ "entries" : [ {
+ "path" : "tpch/nation.parquet"
+ } ],
+ "storage" : {
+ "type" : "file",
+ "connection" : "classpath:///"
+ },
+ "format" : {
+ "type" : "parquet"
+ }
+ }, {
+ pop : "project",
+ @id : 2,
+ exprs : [ {
+ ref : "output.$f0",
+ expr : "N_REGIONKEY"
+ }, {
+ ref : "output.$f1",
+ expr : "N_NATIONKEY"
+ }, {
+ ref : "output.$f2",
+ expr : "N_NAME"
+ } ],
+ child : 1
+ }, {
+ pop : "hash-aggregate",
+ @id : 3,
+ child : 2,
+ keys : [ {
+ ref : "$f0",
+ expr : "$f0"
+ }, {
+ ref : "$f1",
+ expr : "$f1"
+ }, {
+ ref : "$f2",
+ expr : "$f2 "
+ } ]
+ }, {
+ pop : "screen",
+ @id : 4,
+ child : 3
+ } ]
+}
[04/17] git commit: DRILL-481: libthrift-0.8.0 causes drillbit to
fail while querying Hive tables
Posted by ja...@apache.org.
DRILL-481: libthrift-0.8.0 causes drillbit to fail while querying Hive tables
exclude lib thrift in hbase pom
Project: http://git-wip-us.apache.org/repos/asf/incubator-drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-drill/commit/41dcf204
Tree: http://git-wip-us.apache.org/repos/asf/incubator-drill/tree/41dcf204
Diff: http://git-wip-us.apache.org/repos/asf/incubator-drill/diff/41dcf204
Branch: refs/heads/master
Commit: 41dcf20499ad62601b84cce16baf39e8887b7194
Parents: 05710a6
Author: Steven Phillips <sp...@maprtech.com>
Authored: Wed Apr 2 18:19:03 2014 -0700
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Sat Apr 19 21:07:00 2014 -0700
----------------------------------------------------------------------
contrib/storage-hbase/pom.xml | 4 ++++
1 file changed, 4 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/41dcf204/contrib/storage-hbase/pom.xml
----------------------------------------------------------------------
diff --git a/contrib/storage-hbase/pom.xml b/contrib/storage-hbase/pom.xml
index b656669..f5a4c40 100644
--- a/contrib/storage-hbase/pom.xml
+++ b/contrib/storage-hbase/pom.xml
@@ -55,6 +55,10 @@
<artifactId>asm</artifactId>
<groupId>asm</groupId>
</exclusion>
+ <exclusion>
+ <artifactId>libthrift</artifactId>
+ <groupId>org.apache.thrift</groupId>
+ </exclusion>
</exclusions>
</dependency>
[05/17] git commit: Add tests for POSTFIX operators added in
097fc00b375b8a171a53e8062d03f06b350ae14e
Posted by ja...@apache.org.
Add tests for POSTFIX operators added in 097fc00b375b8a171a53e8062d03f06b350ae14e
Project: http://git-wip-us.apache.org/repos/asf/incubator-drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-drill/commit/ed5d7c91
Tree: http://git-wip-us.apache.org/repos/asf/incubator-drill/tree/ed5d7c91
Diff: http://git-wip-us.apache.org/repos/asf/incubator-drill/diff/ed5d7c91
Branch: refs/heads/master
Commit: ed5d7c91a6eeded6a5b0aec713e58bd2c654a87b
Parents: 8296be5
Author: vkorukanti <ve...@gmail.com>
Authored: Wed Apr 16 11:28:22 2014 -0700
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Sat Apr 19 21:07:00 2014 -0700
----------------------------------------------------------------------
.../exec/physical/impl/TestSimpleFunctions.java | 56 ----------------
.../test/resources/functions/testIsNotNull.json | 35 ----------
.../test/resources/functions/testIsNull.json | 35 ----------
.../apache/drill/jdbc/test/TestJdbcQuery.java | 68 ++++++++++++++++++++
sqlparser/src/test/resources/test_null_op.json | 10 +++
.../src/test/resources/test_true_false_op.json | 11 ++++
6 files changed, 89 insertions(+), 126 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/ed5d7c91/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestSimpleFunctions.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestSimpleFunctions.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestSimpleFunctions.java
index 4a5f470..2206243 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestSimpleFunctions.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestSimpleFunctions.java
@@ -144,62 +144,6 @@ public class TestSimpleFunctions {
}
@Test
- public void testIsNull(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable {
-
- new NonStrictExpectations(){{
- bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
- bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
- }};
-
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/functions/testIsNull.json"), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
-
- while(exec.next()){
- assertEquals(50, exec.getSelectionVector2().getCount());
- }
-
- if(context.getFailureCause() != null){
- throw context.getFailureCause();
- }
-
- assertTrue(!context.isFailed());
-
- }
-
- @Test
- public void testIsNotNull(@Injectable final DrillbitContext bitContext,
- @Injectable UserServer.UserClientConnection connection) throws Throwable{
-
- new NonStrictExpectations(){{
- bitContext.getMetrics(); result = new MetricRegistry();
- bitContext.getAllocator(); result = new TopLevelAllocator();
- bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
- }};
-
- PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
- PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile("/functions/testIsNotNull.json"), Charsets.UTF_8));
- FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
- FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
- SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
-
- while(exec.next()){
- assertEquals(50, exec.getSelectionVector2().getCount());
- }
-
- if(context.getFailureCause() != null){
- throw context.getFailureCause();
- }
-
- assertTrue(!context.isFailed());
-
- }
-
- @Test
public void testSubstring(@Injectable final DrillbitContext bitContext,
@Injectable UserServer.UserClientConnection connection) throws Throwable{
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/ed5d7c91/exec/java-exec/src/test/resources/functions/testIsNotNull.json
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/resources/functions/testIsNotNull.json b/exec/java-exec/src/test/resources/functions/testIsNotNull.json
deleted file mode 100644
index 713d58f..0000000
--- a/exec/java-exec/src/test/resources/functions/testIsNotNull.json
+++ /dev/null
@@ -1,35 +0,0 @@
-{
- head:{
- type:"APACHE_DRILL_PHYSICAL",
- version:"1",
- generator:{
- type:"manual"
- }
- },
- graph:[
- {
- @id:1,
- pop:"mock-sub-scan",
- url: "http://apache.org",
- entries:[
- {records: 100, types: [
- {name: "blue", type: "INT", mode: "REQUIRED"},
- {name: "red", type: "BIGINT", mode: "REQUIRED"},
- {name: "yellow", type: "FLOAT8", mode: "OPTIONAL"},
- {name: "green", type: "INT", mode: "REQUIRED"}
- ]}
- ]
- },
- {
- @id:2,
- child: 1,
- pop:"filter",
- expr: "isnotnull(yellow)"
- },
- {
- @id: 3,
- child: 2,
- pop: "screen"
- }
- ]
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/ed5d7c91/exec/java-exec/src/test/resources/functions/testIsNull.json
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/resources/functions/testIsNull.json b/exec/java-exec/src/test/resources/functions/testIsNull.json
deleted file mode 100644
index e173adc..0000000
--- a/exec/java-exec/src/test/resources/functions/testIsNull.json
+++ /dev/null
@@ -1,35 +0,0 @@
-{
- head:{
- type:"APACHE_DRILL_PHYSICAL",
- version:"1",
- generator:{
- type:"manual"
- }
- },
- graph:[
- {
- @id:1,
- pop:"mock-sub-scan",
- url: "http://apache.org",
- entries:[
- {records: 100, types: [
- {name: "blue", type: "INT", mode: "REQUIRED"},
- {name: "red", type: "BIGINT", mode: "REQUIRED"},
- {name: "yellow", type: "FLOAT8", mode: "OPTIONAL"},
- {name: "green", type: "INT", mode: "REQUIRED"}
- ]}
- ]
- },
- {
- @id:2,
- child: 1,
- pop:"filter",
- expr: "isnull(yellow)"
- },
- {
- @id: 3,
- child: 2,
- pop: "screen"
- }
- ]
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/ed5d7c91/sqlparser/src/test/java/org/apache/drill/jdbc/test/TestJdbcQuery.java
----------------------------------------------------------------------
diff --git a/sqlparser/src/test/java/org/apache/drill/jdbc/test/TestJdbcQuery.java b/sqlparser/src/test/java/org/apache/drill/jdbc/test/TestJdbcQuery.java
index e2666f8..7031283 100644
--- a/sqlparser/src/test/java/org/apache/drill/jdbc/test/TestJdbcQuery.java
+++ b/sqlparser/src/test/java/org/apache/drill/jdbc/test/TestJdbcQuery.java
@@ -194,4 +194,72 @@ public class TestJdbcQuery {
.sql("select substring('asd' from 1 for 2) from INFORMATION_SCHEMA.`TABLES` limit 1")
.returns("EXPR$0=as\n");
}
+
+ @Test
+ public void testNullOpForNullableType() throws Exception{
+ JdbcAssert.withNoDefaultSchema()
+ .sql("SELECT * FROM cp.`test_null_op.json` WHERE intType IS NULL AND varCharType IS NOT NULL")
+ .returns("intType=null; varCharType=val2");
+ }
+
+ @Test
+ public void testNullOpForNonNullableType() throws Exception{
+ // output of (intType IS NULL) is a non-nullable type
+ JdbcAssert.withNoDefaultSchema()
+ .sql("SELECT * FROM cp.`test_null_op.json` "+
+ "WHERE (intType IS NULL) IS NULL AND (varCharType IS NOT NULL) IS NOT NULL")
+ .returns("");
+ }
+
+ @Test
+ public void testTrueOpForNullableType() throws Exception{
+ JdbcAssert.withNoDefaultSchema()
+ .sql("SELECT data FROM cp.`test_true_false_op.json` WHERE booleanType IS TRUE")
+ .returns("data=set to true");
+
+ JdbcAssert.withNoDefaultSchema()
+ .sql("SELECT data FROM cp.`test_true_false_op.json` WHERE booleanType IS FALSE")
+ .returns("data=set to false");
+
+ JdbcAssert.withNoDefaultSchema()
+ .sql("SELECT data FROM cp.`test_true_false_op.json` WHERE booleanType IS NOT TRUE")
+ .returns(
+ "data=set to false\n" +
+ "data=not set"
+ );
+
+ JdbcAssert.withNoDefaultSchema()
+ .sql("SELECT data FROM cp.`test_true_false_op.json` WHERE booleanType IS NOT FALSE")
+ .returns(
+ "data=set to true\n" +
+ "data=not set"
+ );
+ }
+
+
+ @Test
+ public void testTrueOpForNonNullableType() throws Exception{
+ // Output of IS TRUE (and others) is a Non-nullable type
+ JdbcAssert.withNoDefaultSchema()
+ .sql("SELECT data FROM cp.`test_true_false_op.json` WHERE (booleanType IS TRUE) IS TRUE")
+ .returns("data=set to true");
+
+ JdbcAssert.withNoDefaultSchema()
+ .sql("SELECT data FROM cp.`test_true_false_op.json` WHERE (booleanType IS FALSE) IS FALSE")
+ .returns(
+ "data=set to true\n" +
+ "data=not set"
+ );
+
+ JdbcAssert.withNoDefaultSchema()
+ .sql("SELECT data FROM cp.`test_true_false_op.json` WHERE (booleanType IS NOT TRUE) IS NOT TRUE")
+ .returns("data=set to true");
+
+ JdbcAssert.withNoDefaultSchema()
+ .sql("SELECT data FROM cp.`test_true_false_op.json` WHERE (booleanType IS NOT FALSE) IS NOT FALSE")
+ .returns(
+ "data=set to true\n" +
+ "data=not set"
+ );
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/ed5d7c91/sqlparser/src/test/resources/test_null_op.json
----------------------------------------------------------------------
diff --git a/sqlparser/src/test/resources/test_null_op.json b/sqlparser/src/test/resources/test_null_op.json
new file mode 100644
index 0000000..01acead
--- /dev/null
+++ b/sqlparser/src/test/resources/test_null_op.json
@@ -0,0 +1,10 @@
+{
+ "intType": 1,
+ "varCharType": "val1"
+}
+{
+ "varCharType": "val2"
+}
+{
+ "intType": 2
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/ed5d7c91/sqlparser/src/test/resources/test_true_false_op.json
----------------------------------------------------------------------
diff --git a/sqlparser/src/test/resources/test_true_false_op.json b/sqlparser/src/test/resources/test_true_false_op.json
new file mode 100644
index 0000000..9e29ddc
--- /dev/null
+++ b/sqlparser/src/test/resources/test_true_false_op.json
@@ -0,0 +1,11 @@
+{
+ "booleanType": false,
+ "data" : "set to false"
+}
+{
+ "booleanType": true,
+ "data" : "set to true"
+}
+{
+ "data" : "not set"
+}
[17/17] git commit: DRILL-505: Hash Join
Posted by ja...@apache.org.
DRILL-505: Hash Join
Support for left outer, right outer and full joins
Support for multiple join conditions
Add following tests
- Multiple condition join
- Join on JSON scan
- Multi batch join
- Simple equality join
Project: http://git-wip-us.apache.org/repos/asf/incubator-drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-drill/commit/1fc7b982
Tree: http://git-wip-us.apache.org/repos/asf/incubator-drill/tree/1fc7b982
Diff: http://git-wip-us.apache.org/repos/asf/incubator-drill/diff/1fc7b982
Branch: refs/heads/master
Commit: 1fc7b982414bc0dcd29b1d31e312d2207971933a
Parents: c7cb7ba
Author: Mehant Baid <me...@gmail.com>
Authored: Tue Feb 25 01:09:06 2014 -0800
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Sat Apr 19 21:07:29 2014 -0700
----------------------------------------------------------------------
.../physical/base/AbstractPhysicalVisitor.java | 5 +
.../exec/physical/base/PhysicalVisitor.java | 1 +
.../drill/exec/physical/config/HashJoinPOP.java | 143 ++++++++
.../physical/impl/common/ChainedHashTable.java | 40 ++-
.../exec/physical/impl/common/HashTable.java | 2 +-
.../physical/impl/common/HashTableTemplate.java | 11 +-
.../exec/physical/impl/join/HashJoinBatch.java | 355 +++++++++++++++++++
.../impl/join/HashJoinBatchCreator.java | 37 ++
.../exec/physical/impl/join/HashJoinHelper.java | 222 ++++++++++++
.../exec/physical/impl/join/HashJoinProbe.java | 56 +++
.../impl/join/HashJoinProbeTemplate.java | 226 ++++++++++++
.../exec/physical/impl/join/TestHashJoin.java | 216 +++++++++++
.../apache/drill/exec/pop/PopUnitTestBase.java | 2 +-
.../src/test/resources/build_side_input.json | 24 ++
.../src/test/resources/join/hash_join.json | 63 ++++
.../resources/join/hash_join_multi_batch.json | 47 +++
.../join/hj_left_outer_multi_batch.json | 48 +++
.../resources/join/hj_multi_condition_join.json | 66 ++++
.../join/hj_right_outer_multi_batch.json | 48 +++
.../src/test/resources/probe_side_input.json | 28 ++
20 files changed, 1618 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/1fc7b982/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 5f37487..abafc42 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
@@ -106,6 +106,11 @@ public abstract class AbstractPhysicalVisitor<T, X, E extends Throwable> impleme
}
@Override
+ public T visitHashJoin(HashJoinPOP join, X value) throws E {
+ return visitOp(join, value);
+ }
+
+ @Override
public T visitHashPartitionSender(HashPartitionSender op, X value) throws E {
return visitSender(op, value);
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/1fc7b982/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 712fafe..208dab9 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
@@ -41,6 +41,7 @@ public interface PhysicalVisitor<RETURN, EXTRA, EXCEP extends Throwable> {
public RETURN visitSort(Sort sort, EXTRA value) throws EXCEP;
public RETURN visitLimit(Limit limit, EXTRA value) throws EXCEP;
public RETURN visitMergeJoin(MergeJoinPOP join, EXTRA value) throws EXCEP;
+ public RETURN visitHashJoin(HashJoinPOP join, EXTRA value) throws EXCEP;
public RETURN visitSender(Sender sender, EXTRA value) throws EXCEP;
public RETURN visitReceiver(Receiver receiver, EXTRA value) throws EXCEP;
public RETURN visitStreamingAggregate(StreamingAggregate agg, EXTRA value) throws EXCEP;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/1fc7b982/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/HashJoinPOP.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/HashJoinPOP.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/HashJoinPOP.java
new file mode 100644
index 0000000..f4a1fc7
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/HashJoinPOP.java
@@ -0,0 +1,143 @@
+/**
+ * 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.JsonTypeName;
+
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.drill.common.expression.FieldReference;
+import org.apache.drill.common.logical.data.JoinCondition;
+import org.apache.drill.common.logical.data.NamedExpression;
+import org.apache.drill.exec.physical.OperatorCost;
+import org.apache.drill.exec.physical.base.AbstractBase;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.physical.base.PhysicalVisitor;
+import org.apache.drill.exec.physical.base.Size;
+
+import com.beust.jcommander.internal.Lists;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Iterators;
+import org.apache.drill.exec.physical.impl.common.HashTable;
+import org.apache.drill.exec.physical.impl.common.HashTableConfig;
+
+import org.eigenbase.rel.JoinRelType;
+
+@JsonTypeName("hash-join")
+public class HashJoinPOP extends AbstractBase {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(HashJoinPOP.class);
+
+
+ private final PhysicalOperator left;
+ private final PhysicalOperator right;
+ private final List<JoinCondition> conditions;
+ private final JoinRelType joinType;
+ private final HashTableConfig htConfig;
+
+ @Override
+ public OperatorCost getCost() {
+ return new OperatorCost(0,0,0,0);
+ }
+
+ @JsonCreator
+ public HashJoinPOP(
+ @JsonProperty("left") PhysicalOperator left,
+ @JsonProperty("right") PhysicalOperator right,
+ @JsonProperty("join-conditions") List<JoinCondition> conditions,
+ @JsonProperty("join-type") JoinRelType joinType
+ ) {
+ this.left = left;
+ this.right = right;
+ this.conditions = conditions;
+ this.joinType = joinType;
+
+ int conditionsSize = conditions.size();
+
+ NamedExpression rightExpr[] = new NamedExpression[conditionsSize];
+ NamedExpression leftExpr[] = new NamedExpression[conditionsSize];
+
+ for (int i = 0; i < conditionsSize; i++) {
+ rightExpr[i] = new NamedExpression(conditions.get(i).getRight(), new FieldReference("build_side_" + i ));
+ leftExpr[i] = new NamedExpression(conditions.get(i).getLeft(), new FieldReference("probe_side_" + i));
+
+ // Hash join only supports equality currently.
+ assert conditions.get(i).getRelationship().equals("==");
+ }
+
+ this.htConfig = new HashTableConfig(HashTable.DEFAULT_INITIAL_CAPACITY,
+ HashTable.DEFAULT_LOAD_FACTOR,
+ rightExpr, leftExpr);
+ }
+
+ @Override
+ public Size getSize() {
+ return left.getSize().add(right.getSize());
+ }
+
+ @Override
+ public <T, X, E extends Throwable> T accept(PhysicalVisitor<T, X, E> physicalVisitor, X value) throws E {
+ return physicalVisitor.visitHashJoin(this, value);
+ }
+
+ @Override
+ public PhysicalOperator getNewWithChildren(List<PhysicalOperator> children) {
+ Preconditions.checkArgument(children.size() == 2);
+ return new HashJoinPOP(children.get(0), children.get(1), conditions, joinType);
+ }
+
+ @Override
+ public Iterator<PhysicalOperator> iterator() {
+ return Iterators.forArray(left, right);
+ }
+
+ public PhysicalOperator getLeft() {
+ return left;
+ }
+
+ public PhysicalOperator getRight() {
+ return right;
+ }
+
+ public JoinRelType getJoinType() {
+ return joinType;
+ }
+
+ public List<JoinCondition> getConditions() {
+ return conditions;
+ }
+
+ public HashTableConfig getHtConfig() {
+ return htConfig;
+ }
+
+ public HashJoinPOP flipIfRight(){
+ if(joinType == JoinRelType.RIGHT){
+ List<JoinCondition> flippedConditions = Lists.newArrayList(conditions.size());
+ for(JoinCondition c : conditions){
+ flippedConditions.add(c.flip());
+ }
+ return new HashJoinPOP(right, left, flippedConditions, JoinRelType.LEFT);
+ }else{
+ return this;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/1fc7b982/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/ChainedHashTable.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/ChainedHashTable.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/ChainedHashTable.java
index 507be33..ec579fc 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/ChainedHashTable.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/ChainedHashTable.java
@@ -84,6 +84,7 @@ public class ChainedHashTable {
private final MappingSet KeyMatchIncomingBuildMapping = new MappingSet("incomingRowIdx", null, "incomingBuild", null, KEY_MATCH_BUILD, KEY_MATCH_BUILD);
private final MappingSet KeyMatchIncomingProbeMapping = new MappingSet("incomingRowIdx", null, "incomingProbe", null, KEY_MATCH_PROBE, KEY_MATCH_PROBE);
private final MappingSet KeyMatchHtableMapping = new MappingSet("htRowIdx", null, "htContainer", null, KEY_MATCH_BUILD, KEY_MATCH_BUILD);
+ private final MappingSet KeyMatchHtableProbeMapping = new MappingSet("htRowIdx", null, "htContainer", null, KEY_MATCH_PROBE, KEY_MATCH_PROBE);
private final MappingSet GetHashIncomingBuildMapping = new MappingSet("incomingRowIdx", null, "incomingBuild", null, GET_HASH_BUILD, GET_HASH_BUILD);
private final MappingSet GetHashIncomingProbeMapping = new MappingSet("incomingRowIdx", null, "incomingProbe", null, GET_HASH_PROBE, GET_HASH_PROBE);
private final MappingSet SetValueMapping = new MappingSet("incomingRowIdx" /* read index */, "htRowIdx" /* write index */, "incomingBuild" /* read container */, "htContainer" /* write container */, SET_VALUE, SET_VALUE);
@@ -114,10 +115,6 @@ public class ChainedHashTable {
ClassGenerator<HashTable> cg = top.getRoot();
ClassGenerator<HashTable> cgInner = cg.getInnerGenerator("BatchHolder");
- if (outKeyFieldIds.length > htConfig.getKeyExprsBuild().length) {
- throw new IllegalArgumentException("Mismatched number of output key fields.");
- }
-
LogicalExpression[] keyExprsBuild = new LogicalExpression[htConfig.getKeyExprsBuild().length];
LogicalExpression[] keyExprsProbe = null;
boolean isProbe = (htConfig.getKeyExprsProbe() != null) ;
@@ -146,27 +143,34 @@ public class ChainedHashTable {
i++;
}
- if (isProbe) {
+ if (isProbe) {
+ i = 0;
for (NamedExpression ne : htConfig.getKeyExprsProbe()) {
final LogicalExpression expr = ExpressionTreeMaterializer.materialize(ne.getExpr(), incomingProbe, collector, context.getFunctionRegistry());
if(collector.hasErrors()) throw new SchemaChangeException("Failure while materializing expression. " + collector.toErrorString());
if (expr == null) continue;
keyExprsProbe[i] = expr;
+ i++;
}
}
// generate code for isKeyMatch(), setValue(), getHash() and outputRecordKeys()
-
setupIsKeyMatchInternal(cgInner, KeyMatchIncomingBuildMapping, KeyMatchHtableMapping, keyExprsBuild, htKeyFieldIds);
- setupIsKeyMatchInternal(cgInner, KeyMatchIncomingProbeMapping, KeyMatchHtableMapping, keyExprsProbe, htKeyFieldIds) ;
+ setupIsKeyMatchInternal(cgInner, KeyMatchIncomingProbeMapping, KeyMatchHtableProbeMapping, keyExprsProbe, htKeyFieldIds) ;
setupSetValue(cgInner, keyExprsBuild, htKeyFieldIds);
- setupOutputRecordKeys(cgInner, htKeyFieldIds, outKeyFieldIds);
+ if (outgoing != null) {
+
+ if (outKeyFieldIds.length > htConfig.getKeyExprsBuild().length) {
+ throw new IllegalArgumentException("Mismatched number of output key fields.");
+ }
+ }
+ setupOutputRecordKeys(cgInner, htKeyFieldIds, outKeyFieldIds);
setupGetHash(cg /* use top level code generator for getHash */, GetHashIncomingBuildMapping, keyExprsBuild);
setupGetHash(cg /* use top level code generator for getHash */, GetHashIncomingProbeMapping, keyExprsProbe);
- HashTable ht = context.getImplementationClass(top);
+ HashTable ht = context.getImplementationClass(top);
ht.setup(htConfig, context, incomingBuild, incomingProbe, outgoing, htContainerOrig);
return ht;
@@ -227,14 +231,18 @@ public class ChainedHashTable {
cg.setMappingSet(OutputRecordKeysMapping);
- for (int i = 0; i < outKeyFieldIds.length; i++) {
- ValueVectorReadExpression vvrExpr = new ValueVectorReadExpression(htKeyFieldIds[i]);
- ValueVectorWriteExpression vvwExpr = new ValueVectorWriteExpression(outKeyFieldIds[i], vvrExpr, true);
- HoldingContainer hc = cg.addExpr(vvwExpr);
- cg.getEvalBlock()._if(hc.getValue().eq(JExpr.lit(0)))._then()._return(JExpr.FALSE);
- }
+ if (outKeyFieldIds != null) {
+ for (int i = 0; i < outKeyFieldIds.length; i++) {
+ ValueVectorReadExpression vvrExpr = new ValueVectorReadExpression(htKeyFieldIds[i]);
+ ValueVectorWriteExpression vvwExpr = new ValueVectorWriteExpression(outKeyFieldIds[i], vvrExpr, true);
+ HoldingContainer hc = cg.addExpr(vvwExpr);
+ cg.getEvalBlock()._if(hc.getValue().eq(JExpr.lit(0)))._then()._return(JExpr.FALSE);
+ }
- cg.getEvalBlock()._return(JExpr.TRUE);
+ cg.getEvalBlock()._return(JExpr.TRUE);
+ } else {
+ cg.getEvalBlock()._return(JExpr.FALSE);
+ }
}
private void setupGetHash(ClassGenerator<HashTable> cg, MappingSet incomingMapping, LogicalExpression[] keyExprs) throws SchemaChangeException {
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/1fc7b982/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/HashTable.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/HashTable.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/HashTable.java
index d9321b9..2f1172a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/HashTable.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/HashTable.java
@@ -49,7 +49,7 @@ public interface HashTable {
public PutStatus put(int incomingRowIdx, IntHolder htIdxHolder);
- public int containsKey(int incomingRowIdx);
+ public int containsKey(int incomingRowIdx, boolean isProbe);
public int size();
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/1fc7b982/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/HashTableTemplate.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/HashTableTemplate.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/HashTableTemplate.java
index 775766d..f67939e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/HashTableTemplate.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/HashTableTemplate.java
@@ -371,11 +371,13 @@ public abstract class HashTableTemplate implements HashTable {
}
public void clear() {
- for (BatchHolder bh : batchHolders) {
- bh.clear();
+ if (batchHolders != null) {
+ for (BatchHolder bh : batchHolders) {
+ bh.clear();
+ }
+ batchHolders.clear();
+ batchHolders = null;
}
- batchHolders.clear();
- batchHolders = null;
startIndices.clear();
currentIdxHolder = null;
numEntries = 0;
@@ -486,6 +488,7 @@ public abstract class HashTableTemplate implements HashTable {
}
// Return -1 if key is not found in the hash table. Otherwise, return the global index of the key
+ @Override
public int containsKey(int incomingRowIdx, boolean isProbe) {
int hash = isProbe ? getHashProbe(incomingRowIdx) : getHashBuild(incomingRowIdx);
int i = getBucketIndex(hash, numBuckets());
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/1fc7b982/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinBatch.java
new file mode 100644
index 0000000..7ada651
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinBatch.java
@@ -0,0 +1,355 @@
+/**
+ * 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.join;
+
+import org.apache.drill.exec.record.*;
+import org.eigenbase.rel.JoinRelType;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import com.sun.codemodel.JExpression;
+import com.sun.codemodel.JVar;
+import com.sun.codemodel.JExpr;
+
+import org.apache.drill.exec.compile.sig.GeneratorMapping;
+import org.apache.drill.exec.compile.sig.MappingSet;
+import org.apache.drill.exec.exception.ClassTransformationException;
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.expr.ClassGenerator;
+import org.apache.drill.exec.expr.CodeGenerator;
+import org.apache.drill.exec.expr.TypeHelper;
+import org.apache.drill.exec.expr.holders.IntHolder;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.physical.config.HashJoinPOP;
+import org.apache.drill.exec.physical.impl.common.ChainedHashTable;
+import org.apache.drill.exec.physical.impl.common.HashTable;
+import org.apache.drill.exec.physical.impl.common.HashTableConfig;
+import org.apache.drill.exec.physical.impl.sort.RecordBatchData;
+import org.apache.drill.exec.physical.impl.svremover.RemovingRecordBatch;
+import org.apache.drill.exec.vector.ValueVector;
+import org.apache.drill.exec.vector.allocator.VectorAllocator;
+
+public class HashJoinBatch extends AbstractRecordBatch<HashJoinPOP> {
+ // Probe side record batch
+ private final RecordBatch left;
+
+ // Build side record batch
+ private final RecordBatch right;
+
+ // Join type, INNER, LEFT, RIGHT or OUTER
+ private final JoinRelType joinType;
+
+ // hash table configuration, created in HashJoinPOP
+ private HashTableConfig htConfig;
+
+ // Runtime generated class implementing HashJoinProbe interface
+ private HashJoinProbe hashJoinProbe = null;
+
+ /* Helper class
+ * Maintains linked list of build side records with the same key
+ * Keeps information about which build records have a corresponding
+ * matching key in the probe side (for outer, right joins)
+ */
+ private HashJoinHelper hjHelper = null;
+
+ // Underlying hashtable used by the hash join
+ private HashTable hashTable = null;
+
+ /* Hyper container to store all build side record batches.
+ * Records are retrieved from this container when there is a matching record
+ * on the probe side
+ */
+ private ExpandableHyperContainer hyperContainer;
+
+ // Number of records in the output container
+ private int outputRecords;
+
+ // Current batch index on the build side
+ private int buildBatchIndex = 0;
+
+ // List of vector allocators
+ private List<VectorAllocator> allocators = null;
+
+ // Schema of the build side
+ private BatchSchema rightSchema = null;
+
+ // Generator mapping for the build side
+ private static final GeneratorMapping PROJECT_BUILD = GeneratorMapping.create("doSetup"/* setup method */,
+ "projectBuildRecord" /* eval method */,
+ null /* reset */, null /* cleanup */);
+
+ // Generator mapping for the probe side
+ private static final GeneratorMapping PROJECT_PROBE = GeneratorMapping.create("doSetup" /* setup method */,
+ "projectProbeRecord" /* eval method */,
+ null /* reset */, null /* cleanup */);
+
+ // Mapping set for the build side
+ private final MappingSet projectBuildMapping = new MappingSet("buildIndex" /* read index */, "outIndex" /* write index */,
+ "buildBatch" /* read container */,
+ "outgoing" /* write container */,
+ PROJECT_BUILD, PROJECT_BUILD);
+
+ // Mapping set for the probe side
+ private final MappingSet projectProbeMapping = new MappingSet("probeIndex" /* read index */, "outIndex" /* write index */,
+ "probeBatch" /* read container */,
+ "outgoing" /* write container */,
+ PROJECT_PROBE, PROJECT_PROBE);
+
+ @Override
+ public int getRecordCount() {
+ return outputRecords;
+ }
+
+
+ @Override
+ public IterOutcome next() {
+
+ try {
+ /* If we are here for the first time, execute the build phase of the
+ * hash join and setup the run time generated class for the probe side
+ */
+ if (hashJoinProbe == null) {
+
+ // Initialize the hash join helper context
+ hjHelper = new HashJoinHelper(context);
+
+ /* Build phase requires setting up the hash table. Hash table will
+ * materialize both the build and probe side expressions while
+ * creating the hash table. So we need to invoke next() on our probe batch
+ * as well, for the materialization to be successful. This batch will not be used
+ * till we complete the build phase.
+ */
+ left.next();
+
+ // Build the hash table, using the build side record batches.
+ executeBuildPhase();
+
+ // Create the run time generated code needed to probe and project
+ hashJoinProbe = setupHashJoinProbe();
+ }
+
+ // Allocate the memory for the vectors in the output container
+ allocateVectors();
+
+ // Store the number of records projected
+ outputRecords = hashJoinProbe.probeAndProject();
+
+ /* We are here because of one the following
+ * 1. Completed processing of all the records and we are done
+ * 2. We've filled up the outgoing batch to the maximum and we need to return upstream
+ * Either case build the output container's schema and return
+ */
+ if (outputRecords > 0) {
+
+ // Build the container schema and set the counts
+ container.buildSchema(BatchSchema.SelectionVectorMode.NONE);
+ container.setRecordCount(outputRecords);
+
+ for (VectorWrapper<?> v : container) {
+ v.getValueVector().getMutator().setValueCount(outputRecords);
+ }
+
+ return IterOutcome.OK_NEW_SCHEMA;
+ }
+
+ // No more output records, clean up and return
+ cleanup();
+ return IterOutcome.NONE;
+
+ } catch (ClassTransformationException | SchemaChangeException | IOException e) {
+ context.fail(e);
+ killIncoming();
+ cleanup();
+ return IterOutcome.STOP;
+ }
+ }
+
+ public void setupHashTable() throws IOException, SchemaChangeException, ClassTransformationException {
+
+ // Shouldn't be recreating the hash table, this should be done only once
+ assert hashTable == null;
+
+ ChainedHashTable ht = new ChainedHashTable(htConfig, context, this.right, this.left, null);
+ hashTable = ht.createAndSetupHashTable(null);
+
+ }
+
+ public void executeBuildPhase() throws SchemaChangeException, ClassTransformationException, IOException {
+
+ //Setup the underlying hash table
+ IterOutcome rightUpstream = right.next();
+
+ boolean moreData = true;
+
+ setupHashTable();
+
+ while (moreData) {
+
+ switch (rightUpstream) {
+
+ case NONE:
+ case NOT_YET:
+ case STOP:
+ moreData = false;
+ continue;
+
+ case OK_NEW_SCHEMA:
+ if (rightSchema == null) {
+ rightSchema = right.getSchema();
+ } else {
+ throw new SchemaChangeException("Hash join does not support schema changes");
+ }
+ // Fall through
+ case OK:
+ int currentRecordCount = right.getRecordCount();
+
+ /* For every new build batch, we store some state in the helper context
+ * Add new state to the helper context
+ */
+ hjHelper.addNewBatch(currentRecordCount);
+
+ // Holder contains the global index where the key is hashed into using the hash table
+ IntHolder htIndex = new IntHolder();
+
+ // For every record in the build batch , hash the key columns
+ for (int i = 0; i < currentRecordCount; i++) {
+
+ HashTable.PutStatus status = hashTable.put(i, htIndex);
+
+ if (status != HashTable.PutStatus.PUT_FAILED) {
+ /* Use the global index returned by the hash table, to store
+ * the current record index and batch index. This will be used
+ * later when we probe and find a match.
+ */
+ hjHelper.setCurrentIndex(htIndex.value, buildBatchIndex, i);
+ }
+ }
+
+ /* Completed hashing all records in this batch. Transfer the batch
+ * to the hyper vector container. Will be used when we want to retrieve
+ * records that have matching keys on the probe side.
+ */
+ RecordBatchData nextBatch = new RecordBatchData(right);
+ if (hyperContainer == null) {
+ hyperContainer = new ExpandableHyperContainer(nextBatch.getContainer());
+ } else {
+ hyperContainer.addBatch(nextBatch.getContainer());
+ }
+
+ // completed processing a batch, increment batch index
+ buildBatchIndex++;
+ break;
+ }
+ // Get the next record batch
+ rightUpstream = right.next();
+ }
+ }
+
+ public HashJoinProbe setupHashJoinProbe() throws ClassTransformationException, IOException {
+
+ allocators = new ArrayList<>();
+
+ final CodeGenerator<HashJoinProbe> cg = CodeGenerator.get(HashJoinProbe.TEMPLATE_DEFINITION, context.getFunctionRegistry());
+ ClassGenerator<HashJoinProbe> g = cg.getRoot();
+
+ // Generate the code to project build side records
+ g.setMappingSet(projectBuildMapping);
+
+
+ int fieldId = 0;
+ JExpression buildIndex = JExpr.direct("buildIndex");
+ JExpression outIndex = JExpr.direct("outIndex");
+ g.rotateBlock();
+ for(VectorWrapper<?> vv : hyperContainer) {
+
+ // Add the vector to our output container
+ ValueVector v = TypeHelper.getNewVector(vv.getField(), context.getAllocator());
+ container.add(v);
+ allocators.add(RemovingRecordBatch.getAllocator4(v));
+
+ JVar inVV = g.declareVectorValueSetupAndMember("buildBatch", new TypedFieldId(vv.getField().getType(), fieldId, true));
+ JVar outVV = g.declareVectorValueSetupAndMember("outgoing", new TypedFieldId(vv.getField().getType(), fieldId, false));
+
+ g.getEvalBlock().add(outVV.invoke("copyFrom")
+ .arg(buildIndex.band(JExpr.lit((int) Character.MAX_VALUE)))
+ .arg(outIndex)
+ .arg(inVV.component(buildIndex.shrz(JExpr.lit(16)))));
+
+ fieldId++;
+ }
+
+ // Generate the code to project probe side records
+ g.setMappingSet(projectProbeMapping);
+
+ int outputFieldId = fieldId;
+ fieldId = 0;
+ JExpression probeIndex = JExpr.direct("probeIndex");
+ for (VectorWrapper<?> vv : left) {
+
+ ValueVector v = TypeHelper.getNewVector(vv.getField(), context.getAllocator());
+ container.add(v);
+ allocators.add(RemovingRecordBatch.getAllocator4(v));
+
+ JVar inVV = g.declareVectorValueSetupAndMember("probeBatch", new TypedFieldId(vv.getField().getType(), fieldId, false));
+ JVar outVV = g.declareVectorValueSetupAndMember("outgoing", new TypedFieldId(vv.getField().getType(), outputFieldId, false));
+
+ g.getEvalBlock().add(outVV.invoke("copyFrom").arg(probeIndex).arg(outIndex).arg(inVV));
+
+ fieldId++;
+ outputFieldId++;
+ }
+
+ HashJoinProbe hj = context.getImplementationClass(cg);
+ hj.setupHashJoinProbe(context, hyperContainer, left, this, hashTable, hjHelper, joinType);
+ return hj;
+ }
+
+ private void allocateVectors(){
+ for(VectorAllocator a : allocators){
+ a.alloc(RecordBatch.MAX_BATCH_SIZE);
+ }
+ }
+
+ public HashJoinBatch(HashJoinPOP popConfig, FragmentContext context, RecordBatch left, RecordBatch right) {
+ super(popConfig, context);
+ this.left = left;
+ this.right = right;
+ this.joinType = popConfig.getJoinType();
+ this.htConfig = popConfig.getHtConfig();
+ }
+
+ @Override
+ public void killIncoming() {
+ this.left.kill();
+ this.right.kill();
+ cleanup();
+ }
+
+ @Override
+ public void cleanup() {
+ left.cleanup();
+ right.cleanup();
+ hyperContainer.clear();
+ hjHelper.clear();
+ container.clear();
+ hashTable.clear();
+ super.cleanup();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/1fc7b982/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinBatchCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinBatchCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinBatchCreator.java
new file mode 100644
index 0000000..19a4a29
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinBatchCreator.java
@@ -0,0 +1,37 @@
+/**
+ * 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.join;
+
+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.HashJoinPOP;
+import org.apache.drill.exec.physical.impl.BatchCreator;
+import org.apache.drill.exec.record.RecordBatch;
+
+import java.util.List;
+
+public class HashJoinBatchCreator implements BatchCreator<HashJoinPOP> {
+
+ @Override
+ public RecordBatch getBatch(FragmentContext context, HashJoinPOP config, List<RecordBatch> children) throws ExecutionSetupException {
+ Preconditions.checkArgument(children.size() == 2);
+ return new HashJoinBatch(config, context, children.get(0), children.get(1));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/1fc7b982/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinHelper.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinHelper.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinHelper.java
new file mode 100644
index 0000000..e0098b1
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinHelper.java
@@ -0,0 +1,222 @@
+/**
+ * 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.join;
+
+import java.util.ArrayList;
+import java.util.BitSet;
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.record.selection.SelectionVector4;
+import org.apache.drill.exec.physical.impl.common.HashTable;
+
+
+/*
+ * Helper class for hash join. Keeps track of information about the build side batches.
+ *
+ * Hash join is a blocking operator, so we consume all the batches on the build side and
+ * store them in a hyper container. The way we can retrieve records from the hyper container
+ * is by providing the record index and batch index in the hyper container. When we invoke put()
+ * for a given row, hash table returns a global index. We store the current row's record index
+ * and batch index in this global index of the startIndices structure.
+ *
+ * Since there can be many rows with the same key on the build side, we store the first
+ * index in the startIndices list and the remaining are stored as a logical linked list using
+ * the 'links' field in the BuildInfo structures.
+ *
+ * Apart from the indexes into the hyper container, this class also stores information about
+ * which records of the build side had a matching record on the probe side. Stored in a bitvector
+ * keyMatchBitVector, it is used to retrieve all records that did not match a record on probe side
+ * for right outer and full outer joins
+ */
+public class HashJoinHelper {
+
+ /* List of start indexes. Stores the record and batch index of the first record
+ * with a give key.
+ */
+ List<SelectionVector4> startIndices = new ArrayList<>();
+
+ // List of BuildInfo structures. Used to maintain auxiliary information about the build batches
+ List<BuildInfo> buildInfoList = new ArrayList<>();
+
+ // Fragment context
+ FragmentContext context;
+
+ // Constant to indicate index is empty.
+ static final int INDEX_EMPTY = -1;
+
+ // bits to shift while obtaining batch index from SV4
+ static final int SHIFT_SIZE = 16;
+
+ public HashJoinHelper(FragmentContext context) {
+ this.context = context;
+ }
+
+ public void addStartIndexBatch() throws SchemaChangeException {
+ startIndices.add(getNewSV4(HashTable.BATCH_SIZE));
+ }
+
+ public class BuildInfo {
+ // List of links. Logically it helps maintain a linked list of records with the same key value
+ private SelectionVector4 links;
+
+ // List of bitvectors. Keeps track of records on the build side that matched a record on the probe side
+ private BitSet keyMatchBitVector;
+
+ // number of records in this batch
+ private int recordCount;
+
+ public BuildInfo(SelectionVector4 links, BitSet keyMatchBitVector, int recordCount) {
+ this.links = links;
+ this.keyMatchBitVector = keyMatchBitVector;
+ this.recordCount = recordCount;
+ }
+
+ public SelectionVector4 getLinks() {
+ return links;
+ }
+
+ public BitSet getKeyMatchBitVector() {
+ return keyMatchBitVector;
+ }
+ }
+
+ public SelectionVector4 getNewSV4(int recordCount) throws SchemaChangeException {
+
+ ByteBuf vector = context.getAllocator().buffer((recordCount * 4));
+
+ SelectionVector4 sv4 = new SelectionVector4(vector, recordCount, recordCount);
+
+ // Initialize the vector
+ for (int i = 0; i < recordCount; i++) {
+ sv4.set(i, INDEX_EMPTY);
+ }
+
+ return sv4;
+ }
+
+ public void addNewBatch(int recordCount) throws SchemaChangeException {
+ // Add a node to the list of BuildInfo's
+ BuildInfo info = new BuildInfo(getNewSV4(recordCount), new BitSet(recordCount), recordCount);
+ buildInfoList.add(info);
+ }
+
+ public int getStartIndex(int keyIndex) {
+ int batchIdx = keyIndex / HashTable.BATCH_SIZE;
+ int offsetIdx = keyIndex % HashTable.BATCH_SIZE;
+
+ assert batchIdx < startIndices.size();
+
+ SelectionVector4 sv4 = startIndices.get(batchIdx);
+
+ return sv4.get(offsetIdx);
+ }
+
+ public int getNextIndex(int currentIdx) {
+ // Get to the links field of the current index to get the next index
+ int batchIdx = currentIdx >>> SHIFT_SIZE;
+ int recordIdx = currentIdx & HashTable.BATCH_MASK;
+
+ assert batchIdx < buildInfoList.size();
+
+ // Get the corresponding BuildInfo node
+ BuildInfo info = buildInfoList.get(batchIdx);
+ return info.getLinks().get(recordIdx);
+ }
+
+ public List<Integer> getNextUnmatchedIndex() {
+ List<Integer> compositeIndexes = new ArrayList<>();
+
+ for (int i = 0; i < buildInfoList.size(); i++) {
+ BuildInfo info = buildInfoList.get(i);
+ int fromIndex = 0;
+
+ while (((fromIndex = info.getKeyMatchBitVector().nextClearBit(fromIndex)) != -1) && (fromIndex < info.recordCount)) {
+ compositeIndexes.add((i << SHIFT_SIZE) | (fromIndex & HashTable.BATCH_MASK));
+ fromIndex++;
+ }
+ }
+ return compositeIndexes;
+ }
+
+ public void setRecordMatched(int index) {
+ int batchIdx = index >>> SHIFT_SIZE;
+ int recordIdx = index & HashTable.BATCH_MASK;
+
+ // Get the BitVector for the appropriate batch and set the bit to indicate the record matched
+ BuildInfo info = buildInfoList.get(batchIdx);
+ BitSet bitVector = info.getKeyMatchBitVector();
+
+ bitVector.set(recordIdx);
+ }
+
+ public void setCurrentIndex(int keyIndex, int batchIndex, int recordIndex) throws SchemaChangeException {
+
+ /* set the current record batch index and the index
+ * within the batch at the specified keyIndex. The keyIndex
+ * denotes the global index where the key for this record is
+ * stored in the hash table
+ */
+ int batchIdx = keyIndex / HashTable.BATCH_SIZE;
+ int offsetIdx = keyIndex % HashTable.BATCH_SIZE;
+
+ if (keyIndex >= (HashTable.BATCH_SIZE * startIndices.size())) {
+ // allocate a new batch
+ addStartIndexBatch();
+ }
+
+ SelectionVector4 startIndex = startIndices.get(batchIdx);
+ int linkIndex;
+
+ // If its the first value for this key
+ if ((linkIndex = (startIndex.get(offsetIdx))) == INDEX_EMPTY) {
+ startIndex.set(offsetIdx, batchIndex, recordIndex);
+ } else {
+ /* we already have encountered a record with the same key
+ * use links to store this value
+ */
+ SelectionVector4 link;
+ do {
+ //Traverse the links to get an empty slot to insert the current index
+ batchIdx = linkIndex >>> SHIFT_SIZE;
+ offsetIdx = linkIndex & Character.MAX_VALUE;
+
+ // get the next link
+ link = buildInfoList.get(batchIdx).getLinks();
+ } while ((linkIndex = link.get(offsetIdx)) != INDEX_EMPTY);
+
+ // We have the correct batchIdx and offset within the batch to store the next link
+ link.set(offsetIdx, batchIndex, recordIndex);
+ }
+ }
+
+ public void clear() {
+ // Clear the SV4 used for start indices
+ for (SelectionVector4 sv4: startIndices) {
+ sv4.clear();
+ }
+
+ for (BuildInfo info : buildInfoList) {
+ info.getLinks().clear();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/1fc7b982/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinProbe.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinProbe.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinProbe.java
new file mode 100644
index 0000000..c99f2a6
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinProbe.java
@@ -0,0 +1,56 @@
+/**
+ * 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.join;
+
+import org.apache.drill.exec.compile.TemplateClassDefinition;
+import org.apache.drill.exec.exception.ClassTransformationException;
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.physical.impl.common.HashTable;
+import org.apache.drill.exec.physical.impl.common.HashTableConfig;
+import org.apache.drill.exec.record.ExpandableHyperContainer;
+import org.apache.drill.exec.record.RecordBatch;
+import org.apache.drill.exec.record.RecordBatch.IterOutcome;
+import org.apache.drill.exec.record.VectorContainer;
+import org.eigenbase.rel.JoinRelType;
+
+import java.io.IOException;
+
+public interface HashJoinProbe {
+ public static TemplateClassDefinition<HashJoinProbe> TEMPLATE_DEFINITION = new TemplateClassDefinition<HashJoinProbe>(HashJoinProbe.class, HashJoinProbeTemplate.class);
+
+ /* The probe side of the hash join can be in the following two states
+ * 1. PROBE_PROJECT: Inner join case, we probe our hash table to see if we have a
+ * key match and if we do we project the record
+ * 2. PROJECT_RIGHT: Right Outer or Full Outer joins where we are projecting the records
+ * from the build side that did not match any records on the probe side. For Left outer
+ * case we handle it internally by projecting the record if there isn't a match on the build side
+ * 3. DONE: Once we have projected all possible records we are done
+ */
+ public static enum ProbeState {
+ PROBE_PROJECT, PROJECT_RIGHT, DONE
+ }
+
+ public abstract void setupHashJoinProbe(FragmentContext context, VectorContainer buildBatch, RecordBatch probeBatch,
+ RecordBatch outgoing, HashTable hashTable, HashJoinHelper hjHelper, JoinRelType joinRelType);
+ public abstract void doSetup(FragmentContext context, VectorContainer buildBatch, RecordBatch probeBatch, RecordBatch outgoing);
+ public abstract int probeAndProject() throws SchemaChangeException, ClassTransformationException, IOException;
+ public abstract void projectBuildRecord(int buildIndex, int outIndex);
+ public abstract void projectProbeRecord(int probeIndex, int outIndex);
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/1fc7b982/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinProbeTemplate.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinProbeTemplate.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinProbeTemplate.java
new file mode 100644
index 0000000..cc1a257
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/HashJoinProbeTemplate.java
@@ -0,0 +1,226 @@
+/**
+ * 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.join;
+
+import javax.inject.Named;
+
+import org.apache.drill.exec.exception.ClassTransformationException;
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.physical.impl.common.ChainedHashTable;
+import org.apache.drill.exec.physical.impl.common.HashTable;
+import org.apache.drill.exec.physical.impl.common.HashTableConfig;
+import org.apache.drill.exec.physical.impl.sort.RecordBatchData;
+import org.apache.drill.exec.record.BatchSchema;
+import org.apache.drill.exec.record.ExpandableHyperContainer;
+import org.apache.drill.exec.record.RecordBatch;
+import org.apache.drill.exec.record.RecordBatch.IterOutcome;
+import org.apache.drill.exec.record.VectorContainer;
+import org.apache.drill.exec.vector.allocator.VectorAllocator;
+import org.apache.drill.exec.expr.holders.IntHolder;
+import org.eigenbase.rel.JoinRelType;
+import org.eigenbase.sql2rel.StandardConvertletTable;
+
+import java.io.IOException;
+import java.util.List;
+
+public abstract class HashJoinProbeTemplate implements HashJoinProbe {
+
+ // Probe side record batch
+ private RecordBatch probeBatch;
+
+ // Join type, INNER, LEFT, RIGHT or OUTER
+ private JoinRelType joinType;
+
+ /* Helper class
+ * Maintains linked list of build side records with the same key
+ * Keeps information about which build records have a corresponding
+ * matching key in the probe side (for outer, right joins)
+ */
+ private HashJoinHelper hjHelper = null;
+
+ // Underlying hashtable used by the hash join
+ private HashTable hashTable = null;
+
+ // Number of records to process on the probe side
+ private int recordsToProcess = 0;
+
+ // Number of records processed on the probe side
+ private int recordsProcessed = 0;
+
+ // Number of records in the output container
+ private int outputRecords;
+
+ // Indicate if we should drain the next record from the probe side
+ private boolean getNextRecord = true;
+
+ // Contains both batch idx and record idx of the matching record in the build side
+ private int currentCompositeIdx = -1;
+
+ // Current state the hash join algorithm is in
+ private ProbeState probeState = ProbeState.PROBE_PROJECT;
+
+ // For outer or right joins, this is a list of unmatched records that needs to be projected
+ private List<Integer> unmatchedBuildIndexes = null;
+
+ @Override
+ public void setupHashJoinProbe(FragmentContext context, VectorContainer buildBatch, RecordBatch probeBatch,
+ RecordBatch outgoing, HashTable hashTable, HashJoinHelper hjHelper,
+ JoinRelType joinRelType) {
+
+ this.probeBatch = probeBatch;
+ this.joinType = joinRelType;
+ this.recordsToProcess = probeBatch.getRecordCount();
+ this.hashTable = hashTable;
+ this.hjHelper = hjHelper;
+
+ doSetup(context, buildBatch, probeBatch, outgoing);
+ }
+
+ public void executeProjectRightPhase() {
+ while (outputRecords < RecordBatch.MAX_BATCH_SIZE && recordsProcessed < recordsToProcess) {
+ projectBuildRecord(unmatchedBuildIndexes.get(recordsProcessed++), outputRecords++);
+ }
+ }
+
+ public void executeProbePhase() throws SchemaChangeException {
+ while (outputRecords < RecordBatch.MAX_BATCH_SIZE && recordsToProcess > 0) {
+
+ // Check if we have processed all records in this batch we need to invoke next
+ if (recordsProcessed == recordsToProcess) {
+ IterOutcome leftUpstream = probeBatch.next();
+
+ switch (leftUpstream) {
+ case NONE:
+ case NOT_YET:
+ case STOP:
+ recordsProcessed = 0;
+ recordsToProcess = 0;
+ probeState = ProbeState.DONE;
+
+ // We are done with the probe phase. If its a RIGHT or a FULL join get the unmatched indexes from the build side
+ if (joinType == JoinRelType.RIGHT || joinType == JoinRelType.FULL) {
+ probeState = ProbeState.PROJECT_RIGHT;
+ }
+
+ continue;
+
+ case OK_NEW_SCHEMA:
+ throw new SchemaChangeException("Hash join does not support schema changes");
+ case OK:
+ recordsToProcess = probeBatch.getRecordCount();
+ recordsProcessed = 0;
+ }
+ }
+ int probeIndex;
+
+ // Check if we need to drain the next row in the probe side
+ if (getNextRecord) {
+ probeIndex = hashTable.containsKey(recordsProcessed, true);
+
+ if (probeIndex != -1) {
+
+ /* The current probe record has a key that matches. Get the index
+ * of the first row in the build side that matches the current key
+ */
+ currentCompositeIdx = hjHelper.getStartIndex(probeIndex);
+
+ /* Record in the build side at currentCompositeIdx has a matching record in the probe
+ * side. Set the bit corresponding to this index so if we are doing a FULL or RIGHT
+ * join we keep track of which records we need to project at the end
+ */
+ hjHelper.setRecordMatched(currentCompositeIdx);
+
+ projectBuildRecord(currentCompositeIdx, outputRecords);
+ projectProbeRecord(recordsProcessed, outputRecords);
+ outputRecords++;
+
+ /* Projected single row from the build side with matching key but there
+ * may be more rows with the same key. Check if that's the case
+ */
+ currentCompositeIdx = hjHelper.getNextIndex(currentCompositeIdx);
+ if (currentCompositeIdx == -1) {
+ /* We only had one row in the build side that matched the current key
+ * from the probe side. Drain the next row in the probe side.
+ */
+ recordsProcessed++;
+ }
+ else {
+ /* There is more than one row with the same key on the build side
+ * don't drain more records from the probe side till we have projected
+ * all the rows with this key
+ */
+ getNextRecord = false;
+ }
+ }
+ else { // No matching key
+
+ // If we have a left outer join, project the keys
+ if (joinType == JoinRelType.LEFT || joinType == JoinRelType.FULL) {
+ projectProbeRecord(recordsProcessed, outputRecords++);
+ }
+ recordsProcessed++;
+ }
+ }
+ else {
+ hjHelper.setRecordMatched(currentCompositeIdx);
+ projectBuildRecord(currentCompositeIdx, outputRecords);
+ projectProbeRecord(recordsProcessed, outputRecords);
+ outputRecords++;
+
+ currentCompositeIdx = hjHelper.getNextIndex(currentCompositeIdx);
+
+ if (currentCompositeIdx == -1) {
+ // We don't have any more rows matching the current key on the build side, move on to the next probe row
+ getNextRecord = true;
+ recordsProcessed++;
+ }
+ }
+ }
+ }
+
+ public int probeAndProject() throws SchemaChangeException, ClassTransformationException, IOException {
+
+ outputRecords = 0;
+
+ if (probeState == ProbeState.PROBE_PROJECT) {
+ executeProbePhase();
+ }
+
+ if (probeState == ProbeState.PROJECT_RIGHT) {
+
+ // We are here because we have a RIGHT OUTER or a FULL join
+ if (unmatchedBuildIndexes == null) {
+ // Initialize list of build indexes that didn't match a record on the probe side
+ unmatchedBuildIndexes = hjHelper.getNextUnmatchedIndex();
+ recordsToProcess = unmatchedBuildIndexes.size();
+ recordsProcessed = 0;
+ }
+
+ // Project the list of unmatched records on the build side
+ executeProjectRightPhase();
+ }
+
+ return outputRecords;
+ }
+
+ public abstract void doSetup(@Named("context") FragmentContext context, @Named("buildBatch") VectorContainer buildBatch, @Named("probeBatch") RecordBatch probeBatch,
+ @Named("outgoing") RecordBatch outgoing);
+ public abstract void projectBuildRecord(@Named("buildIndex") int buildIndex, @Named("outIndex") int outIndex);
+ public abstract void projectProbeRecord(@Named("probeIndex") int probeIndex, @Named("outIndex") int outIndex);
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/1fc7b982/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestHashJoin.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestHashJoin.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestHashJoin.java
new file mode 100644
index 0000000..529563a
--- /dev/null
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestHashJoin.java
@@ -0,0 +1,216 @@
+/**
+ * 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.join;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import mockit.Injectable;
+import mockit.NonStrictExpectations;
+
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.common.util.FileUtils;
+import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.memory.TopLevelAllocator;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.physical.PhysicalPlan;
+import org.apache.drill.exec.physical.base.FragmentRoot;
+import org.apache.drill.exec.physical.impl.OperatorCreatorRegistry;
+import org.apache.drill.exec.physical.impl.ImplCreator;
+import org.apache.drill.exec.physical.impl.SimpleRootExec;
+import org.apache.drill.exec.planner.PhysicalPlanReader;
+import org.apache.drill.exec.proto.CoordinationProtos;
+import org.apache.drill.exec.proto.ExecProtos;
+import org.apache.drill.exec.proto.BitControl.PlanFragment;
+import org.apache.drill.exec.rpc.user.UserServer;
+import org.apache.drill.exec.rpc.user.UserServer.UserClientConnection;
+import org.apache.drill.exec.server.DrillbitContext;
+import org.apache.drill.exec.store.StoragePluginRegistry;
+import org.apache.drill.exec.vector.ValueVector;
+import org.junit.AfterClass;
+import org.junit.Test;
+
+import com.beust.jcommander.internal.Lists;
+import com.google.common.base.Charsets;
+import com.google.common.io.Files;
+import com.codahale.metrics.MetricRegistry;
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.common.util.FileUtils;
+import org.apache.drill.exec.client.DrillClient;
+import org.apache.drill.exec.pop.PopUnitTestBase;
+import org.apache.drill.exec.proto.UserProtos;
+import org.apache.drill.exec.record.RecordBatchLoader;
+import org.apache.drill.exec.record.VectorWrapper;
+import org.apache.drill.exec.rpc.user.QueryResultBatch;
+import org.apache.drill.exec.server.Drillbit;
+import org.apache.drill.exec.server.RemoteServiceSet;
+import org.apache.drill.exec.vector.ValueVector;
+import org.apache.drill.exec.vector.VarCharVector;
+import org.apache.drill.exec.vector.IntVector;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import com.google.common.base.Charsets;
+import com.google.common.io.Files;
+
+
+public class TestHashJoin extends PopUnitTestBase{
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestMergeJoin.class);
+
+ DrillConfig c = DrillConfig.create();
+
+ private void testHJMockScanCommon(final DrillbitContext bitContext, UserServer.UserClientConnection connection, String physicalPlan, int expectedRows) throws Throwable {
+ new NonStrictExpectations(){{
+ bitContext.getMetrics(); result = new MetricRegistry();
+ bitContext.getAllocator(); result = new TopLevelAllocator();
+ bitContext.getOperatorCreatorRegistry(); result = new OperatorCreatorRegistry(c);
+ }};
+
+ PhysicalPlanReader reader = new PhysicalPlanReader(c, c.getMapper(), CoordinationProtos.DrillbitEndpoint.getDefaultInstance());
+ PhysicalPlan plan = reader.readPhysicalPlan(Files.toString(FileUtils.getResourceAsFile(physicalPlan), Charsets.UTF_8));
+ FunctionImplementationRegistry registry = new FunctionImplementationRegistry(c);
+ FragmentContext context = new FragmentContext(bitContext, PlanFragment.getDefaultInstance(), connection, registry);
+ SimpleRootExec exec = new SimpleRootExec(ImplCreator.getExec(context, (FragmentRoot) plan.getSortedOperators(false).iterator().next()));
+
+ int totalRecordCount = 0;
+ while (exec.next()) {
+ totalRecordCount += exec.getRecordCount();
+ }
+ assertEquals(expectedRows, totalRecordCount);
+ System.out.println("Total Record Count: " + totalRecordCount);
+ if (context.getFailureCause() != null)
+ throw context.getFailureCause();
+ assertTrue(!context.isFailed());
+ }
+
+ @Test
+ public void multiBatchEqualityJoin(@Injectable final DrillbitContext bitContext,
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+
+ testHJMockScanCommon(bitContext, connection, "/join/hash_join_multi_batch.json", 200000);
+ }
+
+ @Test
+ public void multiBatchRightOuterJoin(@Injectable final DrillbitContext bitContext,
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+
+ testHJMockScanCommon(bitContext, connection, "/join/hj_right_outer_multi_batch.json", 100000);
+ }
+
+ @Test
+ public void multiBatchLeftOuterJoin(@Injectable final DrillbitContext bitContext,
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+
+
+ testHJMockScanCommon(bitContext, connection, "/join/hj_left_outer_multi_batch.json", 100000);
+ }
+
+ @Test
+ public void simpleEqualityJoin(@Injectable final DrillbitContext bitContext,
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+
+ // Function checks for casting from Float, Double to Decimal data types
+ try (RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
+ Drillbit bit = new Drillbit(CONFIG, serviceSet);
+ DrillClient client = new DrillClient(CONFIG, serviceSet.getCoordinator())) {
+
+ // run query.
+ bit.run();
+ client.connect();
+ List<QueryResultBatch> results = client.runQuery(UserProtos.QueryType.PHYSICAL,
+ Files.toString(FileUtils.getResourceAsFile("/join/hash_join.json"), Charsets.UTF_8)
+ .replace("#{TEST_FILE_1}", FileUtils.getResourceAsFile("/build_side_input.json").toURI().toString())
+ .replace("#{TEST_FILE_2}", FileUtils.getResourceAsFile("/probe_side_input.json").toURI().toString()));
+
+ RecordBatchLoader batchLoader = new RecordBatchLoader(bit.getContext().getAllocator());
+
+ QueryResultBatch batch = results.get(0);
+ assertTrue(batchLoader.load(batch.getHeader().getDef(), batch.getData()));
+
+ batchLoader.getValueAccessorById(0, IntVector.class);
+
+ Iterator<VectorWrapper<?>> itr = batchLoader.iterator();
+
+ // Just test the join key
+ long colA[] = {1, 1, 2, 2, 1, 1};
+
+ // Check the output of decimal9
+ ValueVector.Accessor intAccessor1 = itr.next().getValueVector().getAccessor();
+
+
+ for (int i = 0; i < intAccessor1.getValueCount(); i++) {
+ assertEquals(intAccessor1.getObject(i), colA[i]);
+ }
+ assertEquals(6, intAccessor1.getValueCount());
+ }
+ }
+
+ @Test
+ public void multipleConditionJoin(@Injectable final DrillbitContext bitContext,
+ @Injectable UserServer.UserClientConnection connection) throws Throwable {
+
+ // Function checks for casting from Float, Double to Decimal data types
+ try (RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
+ Drillbit bit = new Drillbit(CONFIG, serviceSet);
+ DrillClient client = new DrillClient(CONFIG, serviceSet.getCoordinator())) {
+
+ // run query.
+ bit.run();
+ client.connect();
+ List<QueryResultBatch> results = client.runQuery(UserProtos.QueryType.PHYSICAL,
+ Files.toString(FileUtils.getResourceAsFile("/join/hj_multi_condition_join.json"), Charsets.UTF_8)
+ .replace("#{TEST_FILE_1}", FileUtils.getResourceAsFile("/build_side_input.json").toURI().toString())
+ .replace("#{TEST_FILE_2}", FileUtils.getResourceAsFile("/probe_side_input.json").toURI().toString()));
+
+ RecordBatchLoader batchLoader = new RecordBatchLoader(bit.getContext().getAllocator());
+
+ QueryResultBatch batch = results.get(0);
+ assertTrue(batchLoader.load(batch.getHeader().getDef(), batch.getData()));
+
+ batchLoader.getValueAccessorById(0, IntVector.class);
+
+ Iterator<VectorWrapper<?>> itr = batchLoader.iterator();
+
+ // Just test the join key
+ long colA[] = {1, 2, 1};
+ long colC[] = {100, 200, 500};
+
+ // Check the output of decimal9
+ ValueVector.Accessor intAccessor1 = itr.next().getValueVector().getAccessor();
+ ValueVector.Accessor intAccessor2 = itr.next().getValueVector().getAccessor();
+
+
+ for (int i = 0; i < intAccessor1.getValueCount(); i++) {
+ assertEquals(intAccessor1.getObject(i), colA[i]);
+ assertEquals(intAccessor2.getObject(i), colC[i]);
+ }
+ assertEquals(3, intAccessor1.getValueCount());
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/1fc7b982/exec/java-exec/src/test/java/org/apache/drill/exec/pop/PopUnitTestBase.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/pop/PopUnitTestBase.java b/exec/java-exec/src/test/java/org/apache/drill/exec/pop/PopUnitTestBase.java
index 78f7e43..e5cd508 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/pop/PopUnitTestBase.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/pop/PopUnitTestBase.java
@@ -45,7 +45,7 @@ public abstract class PopUnitTestBase {
protected static DrillConfig CONFIG;
// Set a timeout unless we're debugging.
- @Rule public TestRule globalTimeout = IS_DEBUG ? new TestName() : new Timeout(25000);
+ @Rule public TestRule globalTimeout = IS_DEBUG ? new TestName() : new Timeout(500000);
@BeforeClass
public static void setup() {
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/1fc7b982/exec/java-exec/src/test/resources/build_side_input.json
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/resources/build_side_input.json b/exec/java-exec/src/test/resources/build_side_input.json
new file mode 100644
index 0000000..31006a6
--- /dev/null
+++ b/exec/java-exec/src/test/resources/build_side_input.json
@@ -0,0 +1,24 @@
+{
+"A": 1,
+"C": 100
+}
+{
+"A": 1,
+"C": 500
+}
+{
+"A": 2,
+"C": 200
+}
+{
+"A": 3,
+"C": 300
+}
+{
+"A": 4,
+"C": 400
+}
+{
+"A": 5,
+"C": 500
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/1fc7b982/exec/java-exec/src/test/resources/join/hash_join.json
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/resources/join/hash_join.json b/exec/java-exec/src/test/resources/join/hash_join.json
new file mode 100644
index 0000000..60b915b
--- /dev/null
+++ b/exec/java-exec/src/test/resources/join/hash_join.json
@@ -0,0 +1,63 @@
+{
+ "head" : {
+ "version" : 1,
+ "generator" : {
+ "type" : "org.apache.drill.exec.planner.logical.DrillImplementor",
+ "info" : ""
+ },
+ "type" : "APACHE_DRILL_PHYSICAL",
+ "resultMode" : "EXEC"
+ },
+ graph:[
+ {
+ @id:1,
+ pop:"fs-scan",
+ format: {type: "json"},
+ storage:{type: "file", connection: "file:///"},
+ files:["#{TEST_FILE_1}"]
+ },
+ {
+ @id:2,
+ pop:"fs-scan",
+ format: {type: "json"},
+ storage:{type: "file", connection: "file:///"},
+ files:["#{TEST_FILE_2}"]
+ },
+ {
+ "pop" : "project",
+ "@id" : 3,
+ "exprs" : [ {
+ "ref" : "output.A",
+ "expr" : "A"
+ },
+ { "ref" : "output.CCOL", "expr" : "C" }
+ ],
+
+ "child" : 1
+ },
+ {
+ "pop" : "project",
+ "@id" : 4,
+ "exprs" : [ {
+ "ref" : "output.B",
+ "expr" : "B"
+ },
+ { "ref" : "output.DCOL", "expr" : "D" }
+ ],
+
+ "child" : 2
+ },
+ {
+ @id: 5,
+ right: 3,
+ left: 4,
+ pop: "hash-join",
+ join-conditions: [ {relationship: "==", left: "B", right: "A"} ]
+ },
+ {
+ @id: 6,
+ child: 5,
+ pop: "screen"
+ }
+ ]
+ }
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/1fc7b982/exec/java-exec/src/test/resources/join/hash_join_multi_batch.json
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/resources/join/hash_join_multi_batch.json b/exec/java-exec/src/test/resources/join/hash_join_multi_batch.json
new file mode 100644
index 0000000..7e218a4
--- /dev/null
+++ b/exec/java-exec/src/test/resources/join/hash_join_multi_batch.json
@@ -0,0 +1,47 @@
+{
+ head:{
+ type:"APACHE_DRILL_PHYSICAL",
+ version:"1",
+ generator:{
+ type:"manual"
+ }
+ },
+ graph:[
+ {
+ @id:1,
+ pop:"mock-sub-scan",
+ url: "http://source1.apache.org",
+ entries:[
+ {records: 4, types: [
+ {name: "blue", type: "INT", mode: "REQUIRED"},
+ {name: "red", type: "INT", mode: "REQUIRED"},
+ {name: "green", type: "INT", mode: "REQUIRED"}
+ ]}
+ ]
+ },
+ {
+ @id:2,
+ pop:"mock-sub-scan",
+ url: "http://source2.apache.org",
+ entries:[
+ {records: 100000, types: [
+ {name: "blue1", type: "INT", mode: "REQUIRED"},
+ {name: "red1", type: "INT", mode: "REQUIRED"},
+ {name: "green1", type: "INT", mode: "REQUIRED"}
+ ]}
+ ]
+ },
+ {
+ @id: 3,
+ right: 1,
+ left: 2,
+ pop: "hash-join",
+ join-conditions: [ {relationship: "==", left: "blue1", right: "blue"} ]
+ },
+ {
+ @id: 4,
+ child: 3,
+ pop: "screen"
+ }
+ ]
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/1fc7b982/exec/java-exec/src/test/resources/join/hj_left_outer_multi_batch.json
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/resources/join/hj_left_outer_multi_batch.json b/exec/java-exec/src/test/resources/join/hj_left_outer_multi_batch.json
new file mode 100644
index 0000000..0cb5a03
--- /dev/null
+++ b/exec/java-exec/src/test/resources/join/hj_left_outer_multi_batch.json
@@ -0,0 +1,48 @@
+{
+ head:{
+ type:"APACHE_DRILL_PHYSICAL",
+ version:"1",
+ generator:{
+ type:"manual"
+ }
+ },
+ graph:[
+ {
+ @id:1,
+ pop:"mock-sub-scan",
+ url: "http://source1.apache.org",
+ entries:[
+ {records: 1, types: [
+ {name: "blue", type: "INT", mode: "REQUIRED"},
+ {name: "red", type: "INT", mode: "REQUIRED"},
+ {name: "green", type: "INT", mode: "REQUIRED"}
+ ]}
+ ]
+ },
+ {
+ @id:2,
+ pop:"mock-sub-scan",
+ url: "http://source2.apache.org",
+ entries:[
+ {records: 100000, types: [
+ {name: "blue1", type: "INT", mode: "REQUIRED"},
+ {name: "red1", type: "INT", mode: "REQUIRED"},
+ {name: "green1", type: "INT", mode: "REQUIRED"}
+ ]}
+ ]
+ },
+ {
+ @id: 3,
+ right: 1,
+ left: 2,
+ pop: "hash-join",
+ join-type: "LEFT",
+ join-conditions: [ {relationship: "==", left: "blue1", right: "blue"} ]
+ },
+ {
+ @id: 4,
+ child: 3,
+ pop: "screen"
+ }
+ ]
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/1fc7b982/exec/java-exec/src/test/resources/join/hj_multi_condition_join.json
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/resources/join/hj_multi_condition_join.json b/exec/java-exec/src/test/resources/join/hj_multi_condition_join.json
new file mode 100644
index 0000000..fd680d0
--- /dev/null
+++ b/exec/java-exec/src/test/resources/join/hj_multi_condition_join.json
@@ -0,0 +1,66 @@
+{
+ "head" : {
+ "version" : 1,
+ "generator" : {
+ "type" : "org.apache.drill.exec.planner.logical.DrillImplementor",
+ "info" : ""
+ },
+ "type" : "APACHE_DRILL_PHYSICAL",
+ "resultMode" : "EXEC"
+ },
+ graph:[
+ {
+ @id:1,
+ pop:"fs-scan",
+ format: {type: "json"},
+ storage:{type: "file", connection: "file:///"},
+ files:["#{TEST_FILE_1}"]
+ },
+ {
+ @id:2,
+ pop:"fs-scan",
+ format: {type: "json"},
+ storage:{type: "file", connection: "file:///"},
+ files:["#{TEST_FILE_2}"]
+ },
+ {
+ "pop" : "project",
+ "@id" : 3,
+ "exprs" : [ {
+ "ref" : "output.A",
+ "expr" : "A"
+ },
+ { "ref" : "output.CCOL", "expr" : "C" }
+ ],
+
+ "child" : 1
+ },
+ {
+ "pop" : "project",
+ "@id" : 4,
+ "exprs" : [ {
+ "ref" : "output.B",
+ "expr" : "B"
+ },
+ { "ref" : "output.DCOL", "expr" : "D" }
+ ],
+
+ "child" : 2
+ },
+ {
+ @id: 5,
+ right: 3,
+ left: 4,
+ pop: "hash-join",
+ join-conditions: [
+ {relationship: "==", left: "B", right: "A"},
+ {relationship: "==", left: "DCOL", right: "CCOL"}
+ ]
+ },
+ {
+ @id: 6,
+ child: 5,
+ pop: "screen"
+ }
+ ]
+ }
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/1fc7b982/exec/java-exec/src/test/resources/join/hj_right_outer_multi_batch.json
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/resources/join/hj_right_outer_multi_batch.json b/exec/java-exec/src/test/resources/join/hj_right_outer_multi_batch.json
new file mode 100644
index 0000000..b8723aa
--- /dev/null
+++ b/exec/java-exec/src/test/resources/join/hj_right_outer_multi_batch.json
@@ -0,0 +1,48 @@
+{
+ head:{
+ type:"APACHE_DRILL_PHYSICAL",
+ version:"1",
+ generator:{
+ type:"manual"
+ }
+ },
+ graph:[
+ {
+ @id:1,
+ pop:"mock-sub-scan",
+ url: "http://source1.apache.org",
+ entries:[
+ {records: 100000, types: [
+ {name: "blue", type: "INT", mode: "REQUIRED"},
+ {name: "red", type: "INT", mode: "REQUIRED"},
+ {name: "green", type: "INT", mode: "REQUIRED"}
+ ]}
+ ]
+ },
+ {
+ @id:2,
+ pop:"mock-sub-scan",
+ url: "http://source2.apache.org",
+ entries:[
+ {records: 1, types: [
+ {name: "blue1", type: "INT", mode: "REQUIRED"},
+ {name: "red1", type: "INT", mode: "REQUIRED"},
+ {name: "green1", type: "INT", mode: "REQUIRED"}
+ ]}
+ ]
+ },
+ {
+ @id: 3,
+ right: 1,
+ left: 2,
+ pop: "hash-join",
+ join-type: "RIGHT",
+ join-conditions: [ {relationship: "==", left: "blue1", right: "blue"} ]
+ },
+ {
+ @id: 4,
+ child: 3,
+ pop: "screen"
+ }
+ ]
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/1fc7b982/exec/java-exec/src/test/resources/probe_side_input.json
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/resources/probe_side_input.json b/exec/java-exec/src/test/resources/probe_side_input.json
new file mode 100644
index 0000000..d3dbbb3
--- /dev/null
+++ b/exec/java-exec/src/test/resources/probe_side_input.json
@@ -0,0 +1,28 @@
+{
+"B": 1,
+"D": 100
+}
+{
+"B": 2,
+"D": 200
+}
+{
+"B": 2,
+"D": 300
+}
+{
+"B": 9,
+"D": 900
+}
+{
+"B": 1,
+"D": 500
+}
+{
+"B": 10,
+"D": 1000
+}
+{
+"B": 11,
+"D": 1100
+}
\ No newline at end of file
[13/17] git commit: DRILL-475: TestJdbcQuery fails on Windows due to
resource leak
Posted by ja...@apache.org.
DRILL-475: TestJdbcQuery fails on Windows due to resource leak
Project: http://git-wip-us.apache.org/repos/asf/incubator-drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-drill/commit/9c701e13
Tree: http://git-wip-us.apache.org/repos/asf/incubator-drill/tree/9c701e13
Diff: http://git-wip-us.apache.org/repos/asf/incubator-drill/diff/9c701e13
Branch: refs/heads/master
Commit: 9c701e13c5500ab0da77ebf1d1661077720b25ad
Parents: e2bb541
Author: Aditya Kishore <ad...@maprtech.com>
Authored: Wed Apr 2 02:26:30 2014 -0700
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Sat Apr 19 21:07:28 2014 -0700
----------------------------------------------------------------------
sqlparser/pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/9c701e13/sqlparser/pom.xml
----------------------------------------------------------------------
diff --git a/sqlparser/pom.xml b/sqlparser/pom.xml
index 3ba557c..8266dcb 100644
--- a/sqlparser/pom.xml
+++ b/sqlparser/pom.xml
@@ -83,7 +83,7 @@
<artifactId>maven-surefire-plugin</artifactId>
<configuration>
<forkCount>1</forkCount>
- <reuseForks>true</reuseForks>
+ <reuseForks>false</reuseForks>
</configuration>
</plugin>
<plugin>
[09/17] git commit: DRILL-476: Create binary_string() function to
convert encoded binary string to sequence of bytes
Posted by ja...@apache.org.
DRILL-476: Create binary_string() function to convert encoded binary string to sequence of bytes
Project: http://git-wip-us.apache.org/repos/asf/incubator-drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-drill/commit/c7cb7bae
Tree: http://git-wip-us.apache.org/repos/asf/incubator-drill/tree/c7cb7bae
Diff: http://git-wip-us.apache.org/repos/asf/incubator-drill/diff/c7cb7bae
Branch: refs/heads/master
Commit: c7cb7baec66230e117d10494549586f987277510
Parents: 025538c
Author: Aditya Kishore <ad...@maprtech.com>
Authored: Wed Apr 2 00:26:46 2014 -0700
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Sat Apr 19 21:07:28 2014 -0700
----------------------------------------------------------------------
.../exec/expr/fn/impl/StringFunctionUtil.java | 118 +++++++++++++++----
.../exec/expr/fn/impl/StringFunctions.java | 21 ++++
2 files changed, 116 insertions(+), 23 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c7cb7bae/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/StringFunctionUtil.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/StringFunctionUtil.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/StringFunctionUtil.java
index c0dad84..0096a13 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/StringFunctionUtil.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/StringFunctionUtil.java
@@ -20,66 +20,138 @@ package org.apache.drill.exec.expr.fn.impl;
import io.netty.buffer.ByteBuf;
public class StringFunctionUtil {
-
+
/* Decode the input bytebuf using UTF-8, and return the number of characters
*/
public static int getUTF8CharLength(ByteBuf buffer, int start, int end) {
int charCount = 0;
-
+
for (int id = start; id < end; id++) {
byte currentByte = buffer.getByte(id);
-
+
if (currentByte < 0x128 || // 1-byte char. First byte is 0xxxxxxx.
(currentByte & 0xE0) == 0xC0 || // 2-byte char. First byte is 110xxxxx
- (currentByte & 0xF0) == 0xE0 || // 3-byte char. First byte is 1110xxxx
+ (currentByte & 0xF0) == 0xE0 || // 3-byte char. First byte is 1110xxxx
(currentByte & 0xF8) == 0xF0) { //4-byte char. First byte is 11110xxx
charCount ++; //Advance the counter, since we find one char.
- }
+ }
}
return charCount;
}
- /* Decode the input bytebuf using UTF-8. Search in the range of [start, end], find
+ /* Decode the input bytebuf using UTF-8. Search in the range of [start, end], find
* the position of the first byte of next char after we see "charLength" chars.
- *
+ *
*/
public static int getUTF8CharPosition(ByteBuf buffer, int start, int end, int charLength) {
int charCount = 0;
-
+
if (start >=end)
- return -1; //wrong input here.
-
+ return -1; //wrong input here.
+
for (int id = start; id < end; id++) {
-
+
byte currentByte = buffer.getByte(id);
if (currentByte < 0x128 || // 1-byte char. First byte is 0xxxxxxx.
(currentByte & 0xE0) == 0xC0 || // 2-byte char. First byte is 110xxxxx
- (currentByte & 0xF0) == 0xE0 || // 3-byte char. First byte is 1110xxxx
+ (currentByte & 0xF0) == 0xE0 || // 3-byte char. First byte is 1110xxxx
(currentByte & 0xF8) == 0xF0) { //4-byte char. First byte is 11110xxx
- charCount ++; //Advance the counter, since we find one char.
+ charCount ++; //Advance the counter, since we find one char.
if (charCount == charLength + 1) {
return id;
}
- }
+ }
}
- return end;
+ return end;
}
-
- public static int stringLeftMatchUTF8(ByteBuf str, int strStart, int strEnd,
- ByteBuf substr, int subStart, int subEnd) {
+
+ public static int stringLeftMatchUTF8(ByteBuf str, int strStart, int strEnd,
+ ByteBuf substr, int subStart, int subEnd) {
for (int i = strStart; i <= strEnd - (subEnd - subStart); i++) {
int j = subStart;
- for (; j< subEnd; j++) {
+ for (; j< subEnd; j++) {
if (str.getByte(i + j - subStart) != substr.getByte(j))
- break;
+ break;
}
-
- if (j == subEnd && j!= subStart) { // found a matched substr (non-empty) in str.
+
+ if (j == subEnd && j!= subStart) { // found a matched substr (non-empty) in str.
return i; // found a match.
}
}
-
+
return -1;
}
+
+ /**
+ * Return a printable representation of a byte buffer, escaping the non-printable
+ * bytes as '\\xNN' where NN is the hexadecimal representation of such bytes.
+ *
+ * This function does not modify the {@code readerIndex} and {@code writerIndex}
+ * of the byte buffer.
+ */
+ public static String toBinaryString(ByteBuf buf, int strStart, int strEnd) {
+ StringBuilder result = new StringBuilder();
+ for (int i = strStart; i < strEnd ; ++i) {
+ int ch = buf.getByte(i) & 0xFF;
+ if ( (ch >= '0' && ch <= '9')
+ || (ch >= 'A' && ch <= 'Z')
+ || (ch >= 'a' && ch <= 'z')
+ || " `~!@#$%^&*()-_=+[]{}|;:'\",.<>/?".indexOf(ch) >= 0 ) {
+ result.append((char)ch);
+ } else {
+ result.append(String.format("\\x%02X", ch));
+ }
+ }
+ return result.toString();
+ }
+
+ /**
+ * In-place parsing of a hex encoded binary string.
+ *
+ * This function does not modify the {@code readerIndex} and {@code writerIndex}
+ * of the byte buffer.
+ *
+ * @return Index in the byte buffer just after the last written byte.
+ */
+ public static int parseBinaryString(ByteBuf str, int strStart, int strEnd) {
+ int length = (strEnd - strStart);
+ int dstEnd = strStart;
+ for (int i = strStart; i < length ; i++) {
+ byte b = str.getByte(i);
+ if (b == '\\'
+ && length > i+3
+ && (str.getByte(i+1) == 'x' || str.getByte(i+1) == 'X')) {
+ // ok, take next 2 hex digits.
+ byte hd1 = str.getByte(i+2);
+ byte hd2 = str.getByte(i+3);
+ if (isHexDigit(hd1) && isHexDigit(hd2)) { // [a-fA-F0-9]
+ // turn hex ASCII digit -> number
+ b = (byte) ((toBinaryFromHex(hd1) << 4) + toBinaryFromHex(hd2));
+ i += 3; // skip 3
+ }
+ }
+ str.setByte(dstEnd++, b);
+ }
+ return dstEnd;
+ }
+
+ /**
+ * Takes a ASCII digit in the range A-F0-9 and returns
+ * the corresponding integer/ordinal value.
+ * @param ch The hex digit.
+ * @return The converted hex value as a byte.
+ */
+ private static byte toBinaryFromHex(byte ch) {
+ if ( ch >= 'A' && ch <= 'F' )
+ return (byte) ((byte)10 + (byte) (ch - 'A'));
+ else if ( ch >= 'a' && ch <= 'f' )
+ return (byte) ((byte)10 + (byte) (ch - 'a'));
+ return (byte) (ch - '0');
+ }
+
+ private static boolean isHexDigit(byte c) {
+ return (c >= 'a' && c <= 'f') || (c >= 'A' && c <= 'F') || (c >= '0' && c <= '9');
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/c7cb7bae/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/StringFunctions.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/StringFunctions.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/StringFunctions.java
index 5e85012..aca5933 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/StringFunctions.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/StringFunctions.java
@@ -29,6 +29,7 @@ import org.apache.drill.exec.expr.annotations.Param;
import org.apache.drill.exec.expr.annotations.Workspace;
import org.apache.drill.exec.expr.holders.BigIntHolder;
import org.apache.drill.exec.expr.holders.BitHolder;
+import org.apache.drill.exec.expr.holders.VarBinaryHolder;
import org.apache.drill.exec.expr.holders.VarCharHolder;
import org.apache.drill.exec.record.RecordBatch;
@@ -779,4 +780,24 @@ public class StringFunctions{
}
}
+
+ // Converts a hex encoded string into a varbinary type.
+ // "\xca\xfe\xba\xbe" => (byte[]) {(byte)0xca, (byte)0xfe, (byte)0xba, (byte)0xbe}
+ @FunctionTemplate(name = "binary_string", scope = FunctionScope.SIMPLE, nulls = NullHandling.NULL_IF_NULL)
+ public static class BinaryString implements DrillSimpleFunc {
+
+ @Param VarCharHolder in;
+ @Output VarBinaryHolder out;
+
+ public void setup(RecordBatch incoming) { }
+
+ public void eval() {
+ out.buffer = in.buffer;
+ out.start = in.start;
+ out.end = org.apache.drill.exec.expr.fn.impl.StringFunctionUtil.parseBinaryString(in.buffer, in.start, in.end);
+ out.buffer.readerIndex(out.start);
+ out.buffer.writerIndex(out.end);
+ }
+ }
+
}
[02/17] git commit: DRILL-486: Add hbase classpath to drillbit
classpath
Posted by ja...@apache.org.
DRILL-486: Add hbase classpath to drillbit classpath
Project: http://git-wip-us.apache.org/repos/asf/incubator-drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-drill/commit/05710a66
Tree: http://git-wip-us.apache.org/repos/asf/incubator-drill/tree/05710a66
Diff: http://git-wip-us.apache.org/repos/asf/incubator-drill/diff/05710a66
Branch: refs/heads/master
Commit: 05710a66e0249e34442de1cdc684334281842563
Parents: f8f12df
Author: Steven Phillips <sp...@maprtech.com>
Authored: Thu Apr 3 18:44:32 2014 -0700
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Sat Apr 19 21:07:00 2014 -0700
----------------------------------------------------------------------
distribution/src/resources/drill-config.sh | 3 +++
1 file changed, 3 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/05710a66/distribution/src/resources/drill-config.sh
----------------------------------------------------------------------
diff --git a/distribution/src/resources/drill-config.sh b/distribution/src/resources/drill-config.sh
index b406112..c599a7c 100644
--- a/distribution/src/resources/drill-config.sh
+++ b/distribution/src/resources/drill-config.sh
@@ -111,6 +111,9 @@ CP=$DRILL_HOME/contrib/*:$CP
if [ "${HADOOP_CLASSPATH}x" != "x" ]; then
CP=$HADOOP_CLASSPATH:$CP
fi
+if [ "${HBASE_CLASSPATH}x" != "x" ]; then
+ CP=$HBASE_CLASSPATH:$CP
+fi
CP=$DRILL_CONF_DIR:$CP
# Newer versions of glibc use an arena memory allocator that causes virtual
[14/17] git commit: DRILL-405: Fix for not able to query directory
containing json files
Posted by ja...@apache.org.
DRILL-405: Fix for not able to query directory containing json files
Project: http://git-wip-us.apache.org/repos/asf/incubator-drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-drill/commit/fefda25e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-drill/tree/fefda25e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-drill/diff/fefda25e
Branch: refs/heads/master
Commit: fefda25e71da4f764a8559cf511441392b2258a8
Parents: 9d7bb06
Author: Steven Phillips <sp...@maprtech.com>
Authored: Thu Mar 27 14:07:00 2014 -0700
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Sat Apr 19 21:07:28 2014 -0700
----------------------------------------------------------------------
.../org/apache/drill/exec/store/easy/json/JSONRecordReader.java | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/fefda25e/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/json/JSONRecordReader.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/json/JSONRecordReader.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/json/JSONRecordReader.java
index fb23a09..2e8cd2e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/json/JSONRecordReader.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/json/JSONRecordReader.java
@@ -104,6 +104,7 @@ public class JSONRecordReader implements RecordReader {
@Override
public void setup(OutputMutator output) throws ExecutionSetupException {
outputMutator = output;
+ output.removeAllFields();
currentSchema = new ObjectSchema();
diffSchema = new DiffSchema();
removedFields = Lists.newArrayList();
[11/17] git commit: DRILL-358: Fix bug in stream aggregate with sv2
Posted by ja...@apache.org.
DRILL-358: Fix bug in stream aggregate with sv2
Project: http://git-wip-us.apache.org/repos/asf/incubator-drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-drill/commit/d9095509
Tree: http://git-wip-us.apache.org/repos/asf/incubator-drill/tree/d9095509
Diff: http://git-wip-us.apache.org/repos/asf/incubator-drill/diff/d9095509
Branch: refs/heads/master
Commit: d9095509d8624683b25fa1ca6a0c2a8c97faf2fe
Parents: fefda25
Author: Steven Phillips <sp...@maprtech.com>
Authored: Wed Jan 29 16:47:51 2014 -0800
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Sat Apr 19 21:07:28 2014 -0700
----------------------------------------------------------------------
.../drill/exec/physical/impl/aggregate/StreamingAggBatch.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/d9095509/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggBatch.java
index 5eff355..c942dc6 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggBatch.java
@@ -304,7 +304,7 @@ public class StreamingAggBatch extends AbstractRecordBatch<StreamingAggregate> {
case TWO_BYTE: {
JVar var = g.declareClassField("sv2_", g.getModel()._ref(SelectionVector2.class));
g.getBlock("setupInterior").assign(var, JExpr.direct("incoming").invoke("getSelectionVector2"));
- g.getBlock("getVectorIndex")._return(var.invoke("get").arg(JExpr.direct("recordIndex")));;
+ g.getBlock("getVectorIndex")._return(var.invoke("getIndex").arg(JExpr.direct("recordIndex")));;
return;
}
[08/17] git commit: DRILL-498: cast to varchar only returns one row
Posted by ja...@apache.org.
DRILL-498: cast to varchar only returns one row
Project: http://git-wip-us.apache.org/repos/asf/incubator-drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-drill/commit/e2bb5418
Tree: http://git-wip-us.apache.org/repos/asf/incubator-drill/tree/e2bb5418
Diff: http://git-wip-us.apache.org/repos/asf/incubator-drill/diff/e2bb5418
Branch: refs/heads/master
Commit: e2bb54188e060b79bc8bc478390af01a7ab91f10
Parents: 7ec8669
Author: Steven Phillips <sp...@maprtech.com>
Authored: Wed Apr 9 20:03:37 2014 -0700
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Sat Apr 19 21:07:28 2014 -0700
----------------------------------------------------------------------
.../exec/store/parquet/VarLenBinaryReader.java | 51 ++++++++++----------
1 file changed, 25 insertions(+), 26 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/e2bb5418/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/VarLenBinaryReader.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/VarLenBinaryReader.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/VarLenBinaryReader.java
index d9e498e..09d19a8 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/VarLenBinaryReader.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/VarLenBinaryReader.java
@@ -87,20 +87,16 @@ public class VarLenBinaryReader {
NullableVarBinaryVector currNullVec;
// write the first 0 offset
for (ColumnReader columnReader : columns) {
- currVec = (VarBinaryVector) columnReader.valueVecHolder.getValueVector();
- currVec.getAccessor().getOffsetVector().getData().writeInt(0);
columnReader.bytesReadInCurrentPass = 0;
columnReader.valuesReadInCurrentPass = 0;
}
// same for the nullable columns
for (NullableVarLengthColumn columnReader : nullableColumns) {
- currNullVec = (NullableVarBinaryVector) columnReader.valueVecHolder.getValueVector();
- currNullVec.getMutator().getVectorWithValues().getAccessor().getOffsetVector().getData().writeInt(0);
columnReader.bytesReadInCurrentPass = 0;
columnReader.valuesReadInCurrentPass = 0;
columnReader.nullsRead = 0;
}
- do {
+ outer: do {
lengthVarFieldsInCurrentRecord = 0;
for (ColumnReader columnReader : columns) {
if (recordsReadInCurrentPass == columnReader.valueVecHolder.getValueVector().getValueCapacity()){
@@ -121,6 +117,11 @@ public class VarLenBinaryReader {
columnReader.dataTypeLengthInBits = BytesUtils.readIntLittleEndian(bytes,
(int) columnReader.pageReadStatus.readPosInBytes);
lengthVarFieldsInCurrentRecord += columnReader.dataTypeLengthInBits;
+
+ if (columnReader.bytesReadInCurrentPass + columnReader.dataTypeLengthInBits > ((VarBinaryVector) columnReader.valueVecHolder.getValueVector()).getData().capacity()) {
+ break outer;
+ }
+
}
for (NullableVarLengthColumn columnReader : nullableColumns) {
// check to make sure there is capacity for the next value (for nullables this is a check to see if there is
@@ -150,57 +151,55 @@ public class VarLenBinaryReader {
(int) columnReader.pageReadStatus.readPosInBytes);
lengthVarFieldsInCurrentRecord += columnReader.dataTypeLengthInBits;
+ if (columnReader.bytesReadInCurrentPass + columnReader.dataTypeLengthInBits > ((NullableVarBinaryVector) columnReader.valueVecHolder.getValueVector()).getData().capacity()) {
+ break outer;
+ }
}
// check that the next record will fit in the batch
if (rowGroupFinished || (recordsReadInCurrentPass + 1) * parentReader.getBitWidthAllFixedFields() + lengthVarFieldsInCurrentRecord
> parentReader.getBatchSize()){
- break;
- }
- else{
- recordsReadInCurrentPass++;
+ break outer;
}
for (ColumnReader columnReader : columns) {
bytes = columnReader.pageReadStatus.pageDataByteArray;
currVec = (VarBinaryVector) columnReader.valueVecHolder.getValueVector();
// again, I am re-purposing the unused field here, it is a length n BYTES, not bits
- currVec.getAccessor().getOffsetVector().getData().writeInt((int) columnReader.bytesReadInCurrentPass +
- columnReader.dataTypeLengthInBits - 4 * (int) columnReader.valuesReadInCurrentPass);
- currVec.getData().writeBytes(bytes, (int) columnReader.pageReadStatus.readPosInBytes + 4,
- columnReader.dataTypeLengthInBits);
+ boolean success = currVec.getMutator().setSafe(columnReader.valuesReadInCurrentPass, bytes,
+ (int) columnReader.pageReadStatus.readPosInBytes + 4, columnReader.dataTypeLengthInBits);
+ assert success;
columnReader.pageReadStatus.readPosInBytes += columnReader.dataTypeLengthInBits + 4;
columnReader.bytesReadInCurrentPass += columnReader.dataTypeLengthInBits + 4;
columnReader.pageReadStatus.valuesRead++;
columnReader.valuesReadInCurrentPass++;
- currVec.getMutator().setValueCount((int)recordsReadInCurrentPass);
}
for (NullableVarLengthColumn columnReader : nullableColumns) {
bytes = columnReader.pageReadStatus.pageDataByteArray;
currNullVec = (NullableVarBinaryVector) columnReader.valueVecHolder.getValueVector();
// again, I am re-purposing the unused field here, it is a length n BYTES, not bits
- currNullVec.getMutator().getVectorWithValues().getAccessor().getOffsetVector().getData()
- .writeInt(
- (int) columnReader.bytesReadInCurrentPass +
- columnReader.dataTypeLengthInBits - 4 * (columnReader.valuesReadInCurrentPass -
- (columnReader.currentValNull ? Math.max (0, columnReader.nullsRead - 1) : columnReader.nullsRead)));
- columnReader.currentValNull = false;
- if (columnReader.dataTypeLengthInBits > 0){
- currNullVec.getData().writeBytes(bytes, (int) columnReader.pageReadStatus.readPosInBytes + 4,
- columnReader.dataTypeLengthInBits);
- ((NullableVarBinaryVector)columnReader.valueVecHolder.getValueVector()).getMutator().setIndexDefined(columnReader.valuesReadInCurrentPass);
+ if (!columnReader.currentValNull && columnReader.dataTypeLengthInBits > 0){
+ boolean success = currNullVec.getMutator().setSafe(columnReader.valuesReadInCurrentPass, bytes,
+ (int) columnReader.pageReadStatus.readPosInBytes + 4, columnReader.dataTypeLengthInBits);
+ assert success;
}
+ columnReader.currentValNull = false;
if (columnReader.dataTypeLengthInBits > 0){
columnReader.pageReadStatus.readPosInBytes += columnReader.dataTypeLengthInBits + 4;
columnReader.bytesReadInCurrentPass += columnReader.dataTypeLengthInBits + 4;
}
columnReader.pageReadStatus.valuesRead++;
columnReader.valuesReadInCurrentPass++;
- currNullVec.getMutator().setValueCount((int)recordsReadInCurrentPass);
- // reached the end of a page
if ( columnReader.pageReadStatus.valuesRead == columnReader.pageReadStatus.currentPage.getValueCount()) {
columnReader.pageReadStatus.next();
}
}
+ recordsReadInCurrentPass++;
} while (recordsReadInCurrentPass < recordsToReadInThisPass);
+ for (VarLengthColumn columnReader : columns) {
+ columnReader.valueVecHolder.getValueVector().getMutator().setValueCount((int) recordsReadInCurrentPass);
+ }
+ for (NullableVarLengthColumn columnReader : nullableColumns) {
+ columnReader.valueVecHolder.getValueVector().getMutator().setValueCount((int) recordsReadInCurrentPass);
+ }
return recordsReadInCurrentPass;
}
}
\ No newline at end of file
[12/17] git commit: DRILL-477:
o.a.d.exec.vector.ValueHolderHelper.getVarCharHolder() creates a BigEndian
buffer
Posted by ja...@apache.org.
DRILL-477: o.a.d.exec.vector.ValueHolderHelper.getVarCharHolder() creates a BigEndian buffer
Project: http://git-wip-us.apache.org/repos/asf/incubator-drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-drill/commit/025538cb
Tree: http://git-wip-us.apache.org/repos/asf/incubator-drill/tree/025538cb
Diff: http://git-wip-us.apache.org/repos/asf/incubator-drill/diff/025538cb
Branch: refs/heads/master
Commit: 025538cb20b9ec83cc01555938efa5d04d336f24
Parents: 9c701e1
Author: Aditya Kishore <ad...@maprtech.com>
Authored: Wed Apr 2 01:02:43 2014 -0700
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Sat Apr 19 21:07:28 2014 -0700
----------------------------------------------------------------------
.../java/org/apache/drill/exec/vector/ValueHolderHelper.java | 4 +++-
1 file changed, 3 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/025538cb/exec/java-exec/src/main/java/org/apache/drill/exec/vector/ValueHolderHelper.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/vector/ValueHolderHelper.java b/exec/java-exec/src/main/java/org/apache/drill/exec/vector/ValueHolderHelper.java
index 532ba43..55d49d1 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/vector/ValueHolderHelper.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/vector/ValueHolderHelper.java
@@ -17,6 +17,8 @@
*/
package org.apache.drill.exec.vector;
+import java.nio.ByteOrder;
+
import io.netty.buffer.UnpooledByteBufAllocator;
import org.apache.drill.exec.expr.holders.VarCharHolder;
@@ -34,7 +36,7 @@ public class ValueHolderHelper {
byte[] b = s.getBytes(Charsets.UTF_8);
vch.start = 0;
vch.end = b.length;
- vch.buffer = UnpooledByteBufAllocator.DEFAULT.buffer(s.length()); // use the length of input string to allocate buffer.
+ vch.buffer = UnpooledByteBufAllocator.DEFAULT.buffer(s.length()).order(ByteOrder.LITTLE_ENDIAN); // use the length of input string to allocate buffer.
vch.buffer.setBytes(0, b);
return vch;
}
[15/17] git commit: Merge fixes for HashAgg test
Posted by ja...@apache.org.
Merge fixes for HashAgg test
Project: http://git-wip-us.apache.org/repos/asf/incubator-drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-drill/commit/ddb2933e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-drill/tree/ddb2933e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-drill/diff/ddb2933e
Branch: refs/heads/master
Commit: ddb2933e99f821518fa85e7024aba7e5f6957a7b
Parents: d909550
Author: Jacques Nadeau <ja...@apache.org>
Authored: Sat Apr 19 19:07:09 2014 -0700
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Sat Apr 19 21:07:28 2014 -0700
----------------------------------------------------------------------
exec/java-exec/src/test/resources/agg/hashagg/q7_3.json | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/ddb2933e/exec/java-exec/src/test/resources/agg/hashagg/q7_3.json
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/resources/agg/hashagg/q7_3.json b/exec/java-exec/src/test/resources/agg/hashagg/q7_3.json
index 11d2665..6c5fff0 100644
--- a/exec/java-exec/src/test/resources/agg/hashagg/q7_3.json
+++ b/exec/java-exec/src/test/resources/agg/hashagg/q7_3.json
@@ -24,13 +24,13 @@
pop : "project",
@id : 2,
exprs : [ {
- ref : "output.$f0",
+ ref : "$f0",
expr : "N_REGIONKEY"
}, {
- ref : "output.$f1",
+ ref : "$f1",
expr : "N_NATIONKEY"
}, {
- ref : "output.$f2",
+ ref : "$f2",
expr : "N_NAME"
} ],
child : 1
[06/17] git commit: DRILL-456: Remove extra row in the output of
grouped aggregate query.
Posted by ja...@apache.org.
DRILL-456: Remove extra row in the output of grouped aggregate query.
Project: http://git-wip-us.apache.org/repos/asf/incubator-drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-drill/commit/9d7bb062
Tree: http://git-wip-us.apache.org/repos/asf/incubator-drill/tree/9d7bb062
Diff: http://git-wip-us.apache.org/repos/asf/incubator-drill/diff/9d7bb062
Branch: refs/heads/master
Commit: 9d7bb0625b9194014b5b26f3e7cb7b5618dc1a3f
Parents: 41dcf20
Author: Aman Sinha <as...@maprtech.com>
Authored: Mon Apr 14 15:30:09 2014 -0700
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Sat Apr 19 21:07:24 2014 -0700
----------------------------------------------------------------------
.../exec/physical/impl/aggregate/StreamingAggTemplate.java | 8 ++++++--
1 file changed, 6 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/9d7bb062/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggTemplate.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggTemplate.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggTemplate.java
index f0cb0d4..e3eb6fe 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggTemplate.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggTemplate.java
@@ -34,7 +34,7 @@ public abstract class StreamingAggTemplate implements StreamingAggregator {
private IterOutcome lastOutcome = null;
private boolean first = true;
private boolean newSchema = false;
- private int previousIndex = 0;
+ private int previousIndex = -1;
private int underlyingIndex = 0;
private int currentIndex;
private int addedRecordCount = 0;
@@ -123,7 +123,11 @@ public abstract class StreamingAggTemplate implements StreamingAggregator {
// loop through existing records, adding as necessary.
for (; underlyingIndex < incoming.getRecordCount(); incIndex()) {
if(EXTRA_DEBUG) logger.debug("Doing loop with values underlying {}, current {}", underlyingIndex, currentIndex);
- if (isSame( previousIndex, currentIndex )) {
+ if (previousIndex == -1) {
+ if (EXTRA_DEBUG) logger.debug("Adding the initial row's keys and values.");
+ addRecordInc(currentIndex);
+ }
+ else if (isSame( previousIndex, currentIndex )) {
if(EXTRA_DEBUG) logger.debug("Values were found the same, adding.");
addRecordInc(currentIndex);
} else {
[07/17] git commit: DRILL-478: sqlline fails with latest build
Posted by ja...@apache.org.
DRILL-478: sqlline fails with latest build
Exclude jruby-complete in hbase pom
Project: http://git-wip-us.apache.org/repos/asf/incubator-drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-drill/commit/171217d0
Tree: http://git-wip-us.apache.org/repos/asf/incubator-drill/tree/171217d0
Diff: http://git-wip-us.apache.org/repos/asf/incubator-drill/diff/171217d0
Branch: refs/heads/master
Commit: 171217d0ed205cc1f80f90fb3f915bc1c8b16337
Parents: ddb2933
Author: Steven Phillips <sp...@maprtech.com>
Authored: Wed Apr 2 17:48:33 2014 -0700
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Sat Apr 19 21:07:28 2014 -0700
----------------------------------------------------------------------
contrib/storage-hbase/pom.xml | 4 ++++
1 file changed, 4 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/171217d0/contrib/storage-hbase/pom.xml
----------------------------------------------------------------------
diff --git a/contrib/storage-hbase/pom.xml b/contrib/storage-hbase/pom.xml
index f5a4c40..f2a9c62 100644
--- a/contrib/storage-hbase/pom.xml
+++ b/contrib/storage-hbase/pom.xml
@@ -59,6 +59,10 @@
<artifactId>libthrift</artifactId>
<groupId>org.apache.thrift</groupId>
</exclusion>
+ <exclusion>
+ <artifactId>jruby-complete</artifactId>
+ <groupId>org.jruby</groupId>
+ </exclusion>
</exclusions>
</dependency>
[16/17] git commit: Merge fixes for HashJoin
Posted by ja...@apache.org.
Merge fixes for HashJoin
Project: http://git-wip-us.apache.org/repos/asf/incubator-drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-drill/commit/a88102bf
Tree: http://git-wip-us.apache.org/repos/asf/incubator-drill/tree/a88102bf
Diff: http://git-wip-us.apache.org/repos/asf/incubator-drill/diff/a88102bf
Branch: refs/heads/master
Commit: a88102bfa05d3e5eddbb08d6534a41f8221f2690
Parents: 1fc7b98
Author: Jacques Nadeau <ja...@apache.org>
Authored: Sat Apr 19 20:57:10 2014 -0700
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Sat Apr 19 21:07:29 2014 -0700
----------------------------------------------------------------------
exec/java-exec/src/test/resources/join/hash_join.json | 8 ++++----
.../src/test/resources/join/hj_multi_condition_join.json | 8 ++++----
2 files changed, 8 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a88102bf/exec/java-exec/src/test/resources/join/hash_join.json
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/resources/join/hash_join.json b/exec/java-exec/src/test/resources/join/hash_join.json
index 60b915b..6033655 100644
--- a/exec/java-exec/src/test/resources/join/hash_join.json
+++ b/exec/java-exec/src/test/resources/join/hash_join.json
@@ -27,10 +27,10 @@
"pop" : "project",
"@id" : 3,
"exprs" : [ {
- "ref" : "output.A",
+ "ref" : "A",
"expr" : "A"
},
- { "ref" : "output.CCOL", "expr" : "C" }
+ { "ref" : "CCOL", "expr" : "C" }
],
"child" : 1
@@ -39,10 +39,10 @@
"pop" : "project",
"@id" : 4,
"exprs" : [ {
- "ref" : "output.B",
+ "ref" : "B",
"expr" : "B"
},
- { "ref" : "output.DCOL", "expr" : "D" }
+ { "ref" : "DCOL", "expr" : "D" }
],
"child" : 2
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a88102bf/exec/java-exec/src/test/resources/join/hj_multi_condition_join.json
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/resources/join/hj_multi_condition_join.json b/exec/java-exec/src/test/resources/join/hj_multi_condition_join.json
index fd680d0..a6f0cba 100644
--- a/exec/java-exec/src/test/resources/join/hj_multi_condition_join.json
+++ b/exec/java-exec/src/test/resources/join/hj_multi_condition_join.json
@@ -27,10 +27,10 @@
"pop" : "project",
"@id" : 3,
"exprs" : [ {
- "ref" : "output.A",
+ "ref" : "A",
"expr" : "A"
},
- { "ref" : "output.CCOL", "expr" : "C" }
+ { "ref" : "CCOL", "expr" : "C" }
],
"child" : 1
@@ -39,10 +39,10 @@
"pop" : "project",
"@id" : 4,
"exprs" : [ {
- "ref" : "output.B",
+ "ref" : "B",
"expr" : "B"
},
- { "ref" : "output.DCOL", "expr" : "D" }
+ { "ref" : "DCOL", "expr" : "D" }
],
"child" : 2
[03/17] git commit: DRILL-483: Fix NPE when scanning table region on
node without drillbit
Posted by ja...@apache.org.
DRILL-483: Fix NPE when scanning table region on node without drillbit
Project: http://git-wip-us.apache.org/repos/asf/incubator-drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-drill/commit/f8f12df9
Tree: http://git-wip-us.apache.org/repos/asf/incubator-drill/tree/f8f12df9
Diff: http://git-wip-us.apache.org/repos/asf/incubator-drill/diff/f8f12df9
Branch: refs/heads/master
Commit: f8f12df99399fb0097ac4c1388d5ccce9a36d48d
Parents: ed5d7c9
Author: Steven Phillips <sp...@maprtech.com>
Authored: Thu Apr 3 18:38:27 2014 -0700
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Sat Apr 19 21:07:00 2014 -0700
----------------------------------------------------------------------
.../drill/exec/store/hbase/HBaseGroupScan.java | 19 +++++++++++++------
1 file changed, 13 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/f8f12df9/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseGroupScan.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseGroupScan.java b/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseGroupScan.java
index 8c05124..bb0adcc 100644
--- a/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseGroupScan.java
+++ b/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseGroupScan.java
@@ -119,11 +119,13 @@ public class HBaseGroupScan extends AbstractGroupScan {
for (ServerName sn : regionsMap.values()) {
String host = sn.getHostname();
DrillbitEndpoint ep = endpointMap.get(host);
- EndpointAffinity affinity = affinityMap.get(ep);
- if (affinity == null) {
- affinityMap.put(ep, new EndpointAffinity(ep, 1));
- } else {
- affinity.addAffinity(1);
+ if (ep != null) {
+ EndpointAffinity affinity = affinityMap.get(ep);
+ if (affinity == null) {
+ affinityMap.put(ep, new EndpointAffinity(ep, 1));
+ } else {
+ affinity.addAffinity(1);
+ }
}
}
this.endpointAffinities = Lists.newArrayList(affinityMap.values());
@@ -151,12 +153,17 @@ public class HBaseGroupScan extends AbstractGroupScan {
Iterator<Integer> ints = Iterators.cycle(incomingEndpointMap.get(s));
mapIterator.put(s, ints);
}
+ Iterator<Integer> nullIterator = Iterators.cycle(incomingEndpointMap.values());
for (HRegionInfo regionInfo : regionsMap.keySet()) {
logger.debug("creating read entry. start key: {} end key: {}", Bytes.toStringBinary(regionInfo.getStartKey()), Bytes.toStringBinary(regionInfo.getEndKey()));
HBaseSubScan.HBaseSubScanReadEntry p = new HBaseSubScan.HBaseSubScanReadEntry(
tableName, Bytes.toStringBinary(regionInfo.getStartKey()), Bytes.toStringBinary(regionInfo.getEndKey()));
String host = regionsMap.get(regionInfo).getHostname();
- mappings.put(mapIterator.get(host).next(), p);
+ Iterator<Integer> indexIterator = mapIterator.get(host);
+ if (indexIterator == null) {
+ indexIterator = nullIterator;
+ }
+ mappings.put(indexIterator.next(), p);
}
}
[10/17] git commit: DRILL-496: drill stuck in loop when query parquet
file and eventually ran out of memory
Posted by ja...@apache.org.
DRILL-496: drill stuck in loop when query parquet file and eventually ran out of memory
Added condition in NullableColumnReader to break out of loop when there are no more pages to read.
Project: http://git-wip-us.apache.org/repos/asf/incubator-drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-drill/commit/7ec86696
Tree: http://git-wip-us.apache.org/repos/asf/incubator-drill/tree/7ec86696
Diff: http://git-wip-us.apache.org/repos/asf/incubator-drill/diff/7ec86696
Branch: refs/heads/master
Commit: 7ec866968b8691f346523e583252462515279844
Parents: 171217d
Author: Steven Phillips <sp...@maprtech.com>
Authored: Wed Apr 9 15:14:49 2014 -0700
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Sat Apr 19 21:07:28 2014 -0700
----------------------------------------------------------------------
.../apache/drill/exec/store/parquet/NullableColumnReader.java | 4 +++-
.../java/org/apache/drill/exec/store/parquet/PageReadStatus.java | 2 ++
2 files changed, 5 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/7ec86696/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/NullableColumnReader.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/NullableColumnReader.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/NullableColumnReader.java
index 9be8266..b6ae715 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/NullableColumnReader.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/NullableColumnReader.java
@@ -116,7 +116,9 @@ abstract class NullableColumnReader extends ColumnReader{
totalValuesRead += recordsReadInThisIteration;
pageReadStatus.valuesRead += recordsReadInThisIteration;
if (readStartInBytes + readLength >= pageReadStatus.byteLength && bitsUsed == 0) {
- pageReadStatus.next();
+ if (!pageReadStatus.next()) {
+ break;
+ }
} else {
pageReadStatus.readPosInBytes = readStartInBytes + readLength;
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/7ec86696/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/PageReadStatus.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/PageReadStatus.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/PageReadStatus.java
index 1aef7b4..67262f6 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/PageReadStatus.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/PageReadStatus.java
@@ -76,6 +76,8 @@ final class PageReadStatus {
*/
public boolean next() throws IOException {
+ currentPage = null;
+
if(!dataReader.hasRemainder()) return false;
// next, we need to decompress the bytes