You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by jn...@apache.org on 2016/12/20 04:27:30 UTC
[1/8] drill git commit: DRILL-5123: Write query profile after sending
final response to client to improve latency
Repository: drill
Updated Branches:
refs/heads/master cf2b7c70e -> bbcf4b765
DRILL-5123: Write query profile after sending final response to client to improve latency
In testing a particular query, I used a test setup that does not write
to the "persistent store", causing query profiles to not be saved. I
then changed the config to save them (to local disk). This produced
about a 200ms difference in query run time as perceived by the client.
I then moved writing the query profile after sending the client the
final message. This resulted in an approximately 100ms savings, as
perceived by the client, in query run time on short (~3 sec.) queries.
close apache/drill#692
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/417ae930
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/417ae930
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/417ae930
Branch: refs/heads/master
Commit: 417ae930a3839bb0ef8758346fe855611051d037
Parents: cf2b7c7
Author: Paul Rogers <pr...@maprtech.com>
Authored: Mon Dec 12 09:18:38 2016 -0800
Committer: Jinfeng Ni <jn...@apache.org>
Committed: Mon Dec 19 14:45:26 2016 -0800
----------------------------------------------------------------------
.../apache/drill/exec/work/foreman/Foreman.java | 22 +++++++++++++-------
1 file changed, 14 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/417ae930/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java
index 808ba07..c6a3104 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java
@@ -555,7 +555,6 @@ public class Foreman implements Runnable {
final String queueName;
try {
- @SuppressWarnings("resource")
final ClusterCoordinator clusterCoordinator = drillbitContext.getClusterCoordinator();
final DistributedSemaphore distributedSemaphore;
@@ -828,9 +827,6 @@ public class Foreman implements Runnable {
uex = null;
}
- // we store the final result here so we can capture any error/errorId in the profile for later debugging.
- queryManager.writeFinalProfile(uex);
-
/*
* If sending the result fails, we don't really have any way to modify the result we tried to send;
* it is possible it got sent but the result came from a later part of the code path. It is also
@@ -845,6 +841,19 @@ public class Foreman implements Runnable {
logger.warn("Exception sending result to client", resultException);
}
+ // Store the final result here so we can capture any error/errorId in the
+ // profile for later debugging.
+ // Write the query profile AFTER sending results to the user. The observed
+ // user behavior is a possible time-lag between query return and appearance
+ // of the query profile in persistent storage. Also, the query might
+ // succeed, but the profile never appear if the profile write fails. This
+ // behavior is acceptable for an eventually-consistent distributed system.
+ // The key benefit is that the client does not wait for the persistent
+ // storage write; query completion occurs in parallel with profile
+ // persistence.
+
+ queryManager.writeFinalProfile(uex);
+
// Remove the Foreman from the running query list.
bee.retireForeman(Foreman.this);
@@ -1031,10 +1040,8 @@ public class Foreman implements Runnable {
*/
private void setupRootFragment(final PlanFragment rootFragment, final FragmentRoot rootOperator)
throws ExecutionSetupException {
- @SuppressWarnings("resource")
final FragmentContext rootContext = new FragmentContext(drillbitContext, rootFragment, queryContext,
initiatingClient, drillbitContext.getFunctionImplementationRegistry());
- @SuppressWarnings("resource")
final IncomingBuffers buffers = new IncomingBuffers(rootFragment, rootContext);
rootContext.setBuffers(buffers);
@@ -1161,7 +1168,6 @@ public class Foreman implements Runnable {
*/
private void sendRemoteFragments(final DrillbitEndpoint assignment, final Collection<PlanFragment> fragments,
final CountDownLatch latch, final FragmentSubmitFailures fragmentSubmitFailures) {
- @SuppressWarnings("resource")
final Controller controller = drillbitContext.getController();
final InitializeFragments.Builder fb = InitializeFragments.newBuilder();
for(final PlanFragment planFragment : fragments) {
@@ -1187,7 +1193,7 @@ public class Foreman implements Runnable {
final DrillbitEndpoint drillbitEndpoint;
final RpcException rpcException;
- SubmissionException(@SuppressWarnings("unused") final DrillbitEndpoint drillbitEndpoint,
+ SubmissionException(final DrillbitEndpoint drillbitEndpoint,
final RpcException rpcException) {
this.drillbitEndpoint = drillbitEndpoint;
this.rpcException = rpcException;
[7/8] drill git commit: DRILL-5052: Option to debug generated Java
code using an IDE
Posted by jn...@apache.org.
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/test/java/org/apache/drill/exec/compile/TestClassTransformation.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/compile/TestClassTransformation.java b/exec/java-exec/src/test/java/org/apache/drill/exec/compile/TestClassTransformation.java
index bf30601..cc08aa0 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/compile/TestClassTransformation.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/compile/TestClassTransformation.java
@@ -53,7 +53,8 @@ public class TestClassTransformation extends BaseTestQuery {
@Test
public void testJaninoClassCompiler() throws Exception {
logger.debug("Testing JaninoClassCompiler");
- sessionOptions.setOption(OptionValue.createString(OptionType.SESSION, QueryClassLoader.JAVA_COMPILER_OPTION, QueryClassLoader.CompilerPolicy.JANINO.name()));
+ sessionOptions.setOption(OptionValue.createString(OptionType.SESSION, ClassCompilerSelector.JAVA_COMPILER_OPTION, ClassCompilerSelector.CompilerPolicy.JANINO.name()));
+ @SuppressWarnings("resource")
QueryClassLoader loader = new QueryClassLoader(config, sessionOptions);
for (int i = 0; i < ITERATION_COUNT; i++) {
compilationInnerClass(loader);
@@ -64,7 +65,8 @@ public class TestClassTransformation extends BaseTestQuery {
@Test
public void testJDKClassCompiler() throws Exception {
logger.debug("Testing JDKClassCompiler");
- sessionOptions.setOption(OptionValue.createString(OptionType.SESSION, QueryClassLoader.JAVA_COMPILER_OPTION, QueryClassLoader.CompilerPolicy.JDK.name()));
+ sessionOptions.setOption(OptionValue.createString(OptionType.SESSION, ClassCompilerSelector.JAVA_COMPILER_OPTION, ClassCompilerSelector.CompilerPolicy.JDK.name()));
+ @SuppressWarnings("resource")
QueryClassLoader loader = new QueryClassLoader(config, sessionOptions);
for (int i = 0; i < ITERATION_COUNT; i++) {
compilationInnerClass(loader);
@@ -77,9 +79,10 @@ public class TestClassTransformation extends BaseTestQuery {
CodeGenerator<ExampleInner> cg = newCodeGenerator(ExampleInner.class, ExampleTemplateWithInner.class);
ClassSet classSet = new ClassSet(null, cg.getDefinition().getTemplateClassName(), cg.getMaterializedClassName());
String sourceCode = cg.generateAndGet();
- sessionOptions.setOption(OptionValue.createString(OptionType.SESSION, QueryClassLoader.JAVA_COMPILER_OPTION, QueryClassLoader.CompilerPolicy.JDK.name()));
+ sessionOptions.setOption(OptionValue.createString(OptionType.SESSION, ClassCompilerSelector.JAVA_COMPILER_OPTION, ClassCompilerSelector.CompilerPolicy.JDK.name()));
- sessionOptions.setOption(OptionValue.createBoolean(OptionType.SESSION, QueryClassLoader.JAVA_COMPILER_DEBUG_OPTION, false));
+ sessionOptions.setOption(OptionValue.createBoolean(OptionType.SESSION, ClassCompilerSelector.JAVA_COMPILER_DEBUG_OPTION, false));
+ @SuppressWarnings("resource")
QueryClassLoader loader = new QueryClassLoader(config, sessionOptions);
final byte[][] codeWithoutDebug = loader.getClassByteCode(classSet.generated, sourceCode);
loader.close();
@@ -88,7 +91,7 @@ public class TestClassTransformation extends BaseTestQuery {
sizeWithoutDebug += bs.length;
}
- sessionOptions.setOption(OptionValue.createBoolean(OptionType.SESSION, QueryClassLoader.JAVA_COMPILER_DEBUG_OPTION, true));
+ sessionOptions.setOption(OptionValue.createBoolean(OptionType.SESSION, ClassCompilerSelector.JAVA_COMPILER_DEBUG_OPTION, true));
loader = new QueryClassLoader(config, sessionOptions);
final byte[][] codeWithDebug = loader.getClassByteCode(classSet.generated, sourceCode);
loader.close();
@@ -108,7 +111,8 @@ public class TestClassTransformation extends BaseTestQuery {
private void compilationInnerClass(QueryClassLoader loader) throws Exception{
CodeGenerator<ExampleInner> cg = newCodeGenerator(ExampleInner.class, ExampleTemplateWithInner.class);
- ClassTransformer ct = new ClassTransformer(sessionOptions);
+ ClassTransformer ct = new ClassTransformer(config, sessionOptions);
+ @SuppressWarnings("unchecked")
Class<? extends ExampleInner> c = (Class<? extends ExampleInner>) ct.getImplementationClass(loader, cg.getDefinition(), cg.generateAndGet(), cg.getMaterializedClassName());
ExampleInner t = (ExampleInner) c.newInstance();
t.doOutside();
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/test/java/org/apache/drill/exec/compile/TestLargeFileCompilation.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/compile/TestLargeFileCompilation.java b/exec/java-exec/src/test/java/org/apache/drill/exec/compile/TestLargeFileCompilation.java
index f892471..35bd4c9 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/compile/TestLargeFileCompilation.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/compile/TestLargeFileCompilation.java
@@ -107,7 +107,7 @@ public class TestLargeFileCompilation extends BaseTestQuery {
@Test
public void testTEXT_WRITER() throws Exception {
- testNoResult("alter session set `%s`='JDK'", QueryClassLoader.JAVA_COMPILER_OPTION);
+ testNoResult("alter session set `%s`='JDK'", ClassCompilerSelector.JAVA_COMPILER_OPTION);
testNoResult("use dfs_test.tmp");
testNoResult("alter session set `%s`='csv'", ExecConstants.OUTPUT_FORMAT_OPTION);
testNoResult(LARGE_QUERY_WRITER, "wide_table_csv");
@@ -115,7 +115,7 @@ public class TestLargeFileCompilation extends BaseTestQuery {
@Test
public void testPARQUET_WRITER() throws Exception {
- testNoResult("alter session set `%s`='JDK'", QueryClassLoader.JAVA_COMPILER_OPTION);
+ testNoResult("alter session set `%s`='JDK'", ClassCompilerSelector.JAVA_COMPILER_OPTION);
testNoResult("use dfs_test.tmp");
testNoResult("alter session set `%s`='parquet'", ExecConstants.OUTPUT_FORMAT_OPTION);
testNoResult(ITERATION_COUNT, LARGE_QUERY_WRITER, "wide_table_parquet");
@@ -123,31 +123,31 @@ public class TestLargeFileCompilation extends BaseTestQuery {
@Test
public void testGROUP_BY() throws Exception {
- testNoResult("alter session set `%s`='JDK'", QueryClassLoader.JAVA_COMPILER_OPTION);
+ testNoResult("alter session set `%s`='JDK'", ClassCompilerSelector.JAVA_COMPILER_OPTION);
testNoResult(ITERATION_COUNT, LARGE_QUERY_GROUP_BY);
}
@Test
public void testEXTERNAL_SORT() throws Exception {
- testNoResult("alter session set `%s`='JDK'", QueryClassLoader.JAVA_COMPILER_OPTION);
+ testNoResult("alter session set `%s`='JDK'", ClassCompilerSelector.JAVA_COMPILER_OPTION);
testNoResult(ITERATION_COUNT, LARGE_QUERY_ORDER_BY);
}
@Test
public void testTOP_N_SORT() throws Exception {
- testNoResult("alter session set `%s`='JDK'", QueryClassLoader.JAVA_COMPILER_OPTION);
+ testNoResult("alter session set `%s`='JDK'", ClassCompilerSelector.JAVA_COMPILER_OPTION);
testNoResult(ITERATION_COUNT, LARGE_QUERY_ORDER_BY_WITH_LIMIT);
}
@Test
public void testFILTER() throws Exception {
- testNoResult("alter session set `%s`='JDK'", QueryClassLoader.JAVA_COMPILER_OPTION);
+ testNoResult("alter session set `%s`='JDK'", ClassCompilerSelector.JAVA_COMPILER_OPTION);
testNoResult(ITERATION_COUNT, LARGE_QUERY_FILTER);
}
@Test
public void testProject() throws Exception {
- testNoResult("alter session set `%s`='JDK'", QueryClassLoader.JAVA_COMPILER_OPTION);
+ testNoResult("alter session set `%s`='JDK'", ClassCompilerSelector.JAVA_COMPILER_OPTION);
testNoResult(ITERATION_COUNT, LARGE_QUERY_SELECT_LIST);
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/test/java/org/apache/drill/exec/physical/unit/PhysicalOpUnitTestBase.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/unit/PhysicalOpUnitTestBase.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/unit/PhysicalOpUnitTestBase.java
index 2ef21ca..3ca020d 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/unit/PhysicalOpUnitTestBase.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/unit/PhysicalOpUnitTestBase.java
@@ -299,14 +299,14 @@ public class PhysicalOpUnitTestBase extends ExecTest {
result = new Delegate()
{
Object getImplementationClass(CodeGenerator gen) throws IOException, ClassTransformationException {
- return compiler.getImplementationClass(gen);
+ return compiler.createInstance(gen);
}
};
fragContext.getImplementationClass(withAny(CodeGenerator.get(templateClassDefinition, funcReg).getRoot()));
result = new Delegate()
{
Object getImplementationClass(ClassGenerator gen) throws IOException, ClassTransformationException {
- return compiler.getImplementationClass(gen.getCodeGenerator());
+ return compiler.createInstance(gen.getCodeGenerator());
}
};
} catch (ClassTransformationException e) {
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/vector/src/main/java/org/apache/drill/exec/vector/BaseValueVector.java
----------------------------------------------------------------------
diff --git a/exec/vector/src/main/java/org/apache/drill/exec/vector/BaseValueVector.java b/exec/vector/src/main/java/org/apache/drill/exec/vector/BaseValueVector.java
index f520ea4..0321fa8 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/vector/BaseValueVector.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/vector/BaseValueVector.java
@@ -32,7 +32,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public abstract class BaseValueVector implements ValueVector {
- private static final Logger logger = LoggerFactory.getLogger(BaseValueVector.class);
+// private static final Logger logger = LoggerFactory.getLogger(BaseValueVector.class);
public static final int MAX_ALLOCATION_SIZE = Integer.MAX_VALUE;
public static final int INITIAL_VALUE_ALLOCATION = 4096;
@@ -101,6 +101,7 @@ public abstract class BaseValueVector implements ValueVector {
public void generateTestData(int values) {}
//TODO: consider making mutator stateless(if possible) on another issue.
+ @Override
public void reset() {}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/vector/src/main/java/org/apache/drill/exec/vector/complex/AbstractContainerVector.java
----------------------------------------------------------------------
diff --git a/exec/vector/src/main/java/org/apache/drill/exec/vector/complex/AbstractContainerVector.java b/exec/vector/src/main/java/org/apache/drill/exec/vector/complex/AbstractContainerVector.java
index 5dd794e..2c5baa3 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/vector/complex/AbstractContainerVector.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/vector/complex/AbstractContainerVector.java
@@ -60,6 +60,7 @@ public abstract class AbstractContainerVector implements ValueVector {
}
}
+ @Override
public BufferAllocator getAllocator() {
return allocator;
}
@@ -102,6 +103,7 @@ public abstract class AbstractContainerVector implements ValueVector {
}
}
+ @SuppressWarnings("unchecked")
protected <T extends ValueVector> T typeify(ValueVector v, Class<T> clazz) {
if (clazz.isAssignableFrom(v.getClass())) {
return (T) v;
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/vector/src/main/java/org/apache/drill/exec/vector/complex/RepeatedListVector.java
----------------------------------------------------------------------
diff --git a/exec/vector/src/main/java/org/apache/drill/exec/vector/complex/RepeatedListVector.java b/exec/vector/src/main/java/org/apache/drill/exec/vector/complex/RepeatedListVector.java
index 50f357f..0cc3628 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/vector/complex/RepeatedListVector.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/vector/complex/RepeatedListVector.java
@@ -227,6 +227,7 @@ public class RepeatedListVector extends AbstractContainerVector
this.delegate = delegate;
}
+ @Override
public void transfer() {
delegate.transfer();
}
[4/8] drill git commit: DRILL-5051: Fix incorrect computation of
'fetch' in LimitRecordBatch when 'offset' is specified
Posted by jn...@apache.org.
DRILL-5051: Fix incorrect computation of 'fetch' in LimitRecordBatch when 'offset' is specified
close apache/drill#662
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/d8cc7105
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/d8cc7105
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/d8cc7105
Branch: refs/heads/master
Commit: d8cc7105054953dc94afda0785f4d031a4ebbde1
Parents: 1c7309f
Author: hongze.zhz <ho...@alibaba-inc.com>
Authored: Fri Nov 18 20:11:38 2016 +0800
Committer: Jinfeng Ni <jn...@apache.org>
Committed: Mon Dec 19 15:57:32 2016 -0800
----------------------------------------------------------------------
.../physical/impl/limit/LimitRecordBatch.java | 35 ++++----------------
.../java/org/apache/drill/TestBugFixes.java | 10 ++++++
2 files changed, 17 insertions(+), 28 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/d8cc7105/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/limit/LimitRecordBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/limit/LimitRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/limit/LimitRecordBatch.java
index 08ffc0b..254a297 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/limit/LimitRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/limit/LimitRecordBatch.java
@@ -139,18 +139,13 @@ public class LimitRecordBatch extends AbstractSingleRecordBatch<Limit> {
skipBatch = true;
} else {
outgoingSv.allocateNew(recordCount);
- if(incomingSv != null) {
- limitWithSV(recordCount);
- } else {
- limitWithNoSV(recordCount);
- }
+ limit(recordCount);
}
return IterOutcome.OK;
}
- // These two functions are identical except for the computation of the index; merge
- private void limitWithNoSV(int recordCount) {
+ private void limit(int recordCount) {
final int offset = Math.max(0, Math.min(recordCount - 1, recordsToSkip));
recordsToSkip -= offset;
int fetch;
@@ -164,27 +159,11 @@ public class LimitRecordBatch extends AbstractSingleRecordBatch<Limit> {
int svIndex = 0;
for(int i = offset; i < fetch; svIndex++, i++) {
- outgoingSv.setIndex(svIndex, (char) i);
- }
- outgoingSv.setRecordCount(svIndex);
- }
-
- private void limitWithSV(int recordCount) {
- final int offset = Math.max(0, Math.min(recordCount - 1, recordsToSkip));
- recordsToSkip -= offset;
- int fetch;
-
- if(noEndLimit) {
- fetch = recordCount;
- } else {
- fetch = Math.min(recordCount, recordsLeft);
- recordsLeft -= Math.max(0, fetch - offset);
- }
-
- int svIndex = 0;
- for(int i = offset; i < fetch; svIndex++, i++) {
- final char index = incomingSv.getIndex(i);
- outgoingSv.setIndex(svIndex, index);
+ if (incomingSv != null) {
+ outgoingSv.setIndex(svIndex, incomingSv.getIndex(i));
+ } else {
+ outgoingSv.setIndex(svIndex, (char) i);
+ }
}
outgoingSv.setRecordCount(svIndex);
}
http://git-wip-us.apache.org/repos/asf/drill/blob/d8cc7105/exec/java-exec/src/test/java/org/apache/drill/TestBugFixes.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/TestBugFixes.java b/exec/java-exec/src/test/java/org/apache/drill/TestBugFixes.java
index 03b1b61..a9fc5d0 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/TestBugFixes.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/TestBugFixes.java
@@ -221,4 +221,14 @@ public class TestBugFixes extends BaseTestQuery {
.baselineRecords(baseline)
.go();
}
+
+ @Test
+ public void testDRILL5051() throws Exception {
+ testBuilder()
+ .sqlQuery("select count(1) as cnt from (select l_orderkey from (select l_orderkey from cp.`tpch/lineitem.parquet` limit 2) limit 1 offset 1)")
+ .unOrdered()
+ .baselineColumns("cnt")
+ .baselineValues(1L)
+ .go();
+ }
}
[3/8] drill git commit: DRILL-5098: Improving fault tolerance for
connection between client and foreman node.
Posted by jn...@apache.org.
DRILL-5098: Improving fault tolerance for connection between client and foreman node.
Adding tries config option in connection string. Improving fault tolerance in Drill client when trying to make first connection with foreman. The client will try to connect to min(tries, num_drillbits) unique drillbits unless a successfull connection is established.
HYGIENE: Refactoring BasicClient::close to call RemoteConnection::close
close apache/drill#679
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/1c7309fc
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/1c7309fc
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/1c7309fc
Branch: refs/heads/master
Commit: 1c7309fc99ebf3f18cd4658ed3606e97765f286c
Parents: 810198b
Author: Sorabh Hamirwasia <sh...@maprtech.com>
Authored: Thu Dec 1 14:58:00 2016 -0800
Committer: Jinfeng Ni <jn...@apache.org>
Committed: Mon Dec 19 14:49:36 2016 -0800
----------------------------------------------------------------------
.../apache/drill/exec/client/DrillClient.java | 54 ++++-
.../apache/drill/exec/rpc/user/UserClient.java | 3 +-
.../ConnectTriesPropertyTestClusterBits.java | 243 +++++++++++++++++++
.../test/JdbcConnectTriesTestEmbeddedBits.java | 163 +++++++++++++
.../org/apache/drill/exec/rpc/BasicClient.java | 15 +-
.../apache/drill/exec/rpc/RemoteConnection.java | 20 +-
6 files changed, 477 insertions(+), 21 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/1c7309fc/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java b/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java
index 823487a..dee0772 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java
@@ -314,6 +314,7 @@ public class DrillClient implements Closeable, ConnectionThrottle {
}
final List<DrillbitEndpoint> endpoints = new ArrayList<>();
+
if (isDirectConnection) {
// Populate the endpoints list with all the drillbit information provided in the connection string
endpoints.addAll(parseAndVerifyEndpoints(props.getProperty("drillbit"),
@@ -334,7 +335,6 @@ public class DrillClient implements Closeable, ConnectionThrottle {
// shuffle the collection then get the first endpoint
Collections.shuffle(endpoints);
- final DrillbitEndpoint endpoint = endpoints.get(0);
if (props != null) {
final UserProperties.Builder upBuilder = UserProperties.newBuilder();
@@ -357,10 +357,54 @@ public class DrillClient implements Closeable, ConnectionThrottle {
super.afterExecute(r, t);
}
};
- client = new UserClient(clientName, config, supportComplexTypes, allocator, eventLoopGroup, executor);
- logger.debug("Connecting to server {}:{}", endpoint.getAddress(), endpoint.getUserPort());
- connect(endpoint);
- connected = true;
+
+ // "tries" is max number of unique drillbit to try connecting until successfully connected to one of them
+ final String connectTriesConf = (props != null) ? props.getProperty("tries", "5") : "5";
+
+ int connectTriesVal;
+ try {
+ connectTriesVal = Math.min(endpoints.size(), Integer.parseInt(connectTriesConf));
+ } catch (NumberFormatException e) {
+ throw new InvalidConnectionInfoException("Invalid tries value: " + connectTriesConf + " specified in " +
+ "connection string");
+ }
+
+ // If the value provided in the connection string is <=0 then override with 1 since we want to try connecting
+ // at least once
+ connectTriesVal = Math.max(1, connectTriesVal);
+
+ int triedEndpointIndex = 0;
+ DrillbitEndpoint endpoint;
+
+ while (triedEndpointIndex < connectTriesVal) {
+ client = new UserClient(clientName, config, supportComplexTypes, allocator, eventLoopGroup, executor);
+ endpoint = endpoints.get(triedEndpointIndex);
+ logger.debug("Connecting to server {}:{}", endpoint.getAddress(), endpoint.getUserPort());
+
+ try {
+ connect(endpoint);
+ connected = true;
+ logger.info("Successfully connected to server {}:{}", endpoint.getAddress(), endpoint.getUserPort());
+ break;
+ } catch (InvalidConnectionInfoException ex) {
+ logger.error("Connection to {}:{} failed with error {}. Not retrying anymore", endpoint.getAddress(),
+ endpoint.getUserPort(), ex.getMessage());
+ throw ex;
+ } catch (RpcException ex) {
+ ++triedEndpointIndex;
+ logger.error("Attempt {}: Failed to connect to server {}:{}", triedEndpointIndex, endpoint.getAddress(),
+ endpoint.getUserPort());
+
+ // Throw exception when we have exhausted all the tries without having a successful connection
+ if (triedEndpointIndex == connectTriesVal) {
+ throw ex;
+ }
+
+ // Close the connection here to avoid calling close twice in case when all tries are exhausted.
+ // Since DrillClient.close is also calling client.close
+ client.close();
+ }
+ }
}
protected static EventLoopGroup createEventLoop(int size, String prefix) {
http://git-wip-us.apache.org/repos/asf/drill/blob/1c7309fc/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserClient.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserClient.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserClient.java
index e6e62fc..c37e133 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserClient.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserClient.java
@@ -50,6 +50,7 @@ import org.apache.drill.exec.rpc.ProtobufLengthDecoder;
import org.apache.drill.exec.rpc.Response;
import org.apache.drill.exec.rpc.RpcConnectionHandler;
import org.apache.drill.exec.rpc.RpcException;
+import org.apache.drill.exec.rpc.InvalidConnectionInfoException;
import com.google.protobuf.MessageLite;
@@ -158,7 +159,7 @@ public class UserClient extends BasicClientWithConnection<RpcType, UserToBitHand
final String errMsg = String.format("Status: %s, Error Id: %s, Error message: %s",
inbound.getStatus(), inbound.getErrorId(), inbound.getErrorMessage());
logger.error(errMsg);
- throw new RpcException(errMsg);
+ throw new InvalidConnectionInfoException(errMsg);
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/1c7309fc/exec/java-exec/src/test/java/org/apache/drill/exec/client/ConnectTriesPropertyTestClusterBits.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/client/ConnectTriesPropertyTestClusterBits.java b/exec/java-exec/src/test/java/org/apache/drill/exec/client/ConnectTriesPropertyTestClusterBits.java
new file mode 100644
index 0000000..610d455
--- /dev/null
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/client/ConnectTriesPropertyTestClusterBits.java
@@ -0,0 +1,243 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.client;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+
+import org.apache.drill.common.AutoCloseables;
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.exec.ZookeeperHelper;
+import org.apache.drill.exec.coord.ClusterCoordinator;
+import org.apache.drill.exec.exception.DrillbitStartupException;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.rpc.InvalidConnectionInfoException;
+import org.apache.drill.exec.rpc.RpcException;
+import org.apache.drill.exec.server.Drillbit;
+
+import org.apache.drill.exec.server.RemoteServiceSet;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import static junit.framework.TestCase.assertTrue;
+import static junit.framework.TestCase.fail;
+
+public class ConnectTriesPropertyTestClusterBits {
+
+ public static StringBuilder bitInfo;
+ public static final String fakeBitsInfo = "127.0.0.1:5000,127.0.0.1:5001";
+ public static List<Drillbit> drillbits;
+ public static final int drillBitCount = 1;
+ public static ZookeeperHelper zkHelper;
+ public static RemoteServiceSet remoteServiceSet;
+ public static DrillConfig drillConfig;
+
+ @BeforeClass
+ public static void testSetUp() throws Exception {
+ remoteServiceSet = RemoteServiceSet.getLocalServiceSet();
+ zkHelper = new ZookeeperHelper();
+ zkHelper.startZookeeper(1);
+
+ // Creating Drillbits
+ drillConfig = zkHelper.getConfig();
+ try {
+ int drillBitStarted = 0;
+ drillbits = new ArrayList<>();
+ while(drillBitStarted < drillBitCount){
+ drillbits.add(Drillbit.start(drillConfig, remoteServiceSet));
+ ++drillBitStarted;
+ }
+ } catch (DrillbitStartupException e) {
+ throw new RuntimeException("Failed to start drillbits.", e);
+ }
+ bitInfo = new StringBuilder();
+
+ for (int i = 0; i < drillBitCount; ++i) {
+ final DrillbitEndpoint currentEndPoint = drillbits.get(i).getContext().getEndpoint();
+ final String currentBitIp = currentEndPoint.getAddress();
+ final int currentBitPort = currentEndPoint.getUserPort();
+ bitInfo.append(",");
+ bitInfo.append(currentBitIp);
+ bitInfo.append(":");
+ bitInfo.append(currentBitPort);
+ }
+ }
+
+ @AfterClass
+ public static void testCleanUp() throws Exception {
+ AutoCloseables.close(drillbits);
+ }
+
+ @Test
+ public void testSuccessUsingDirectConnectionAndFakeDrillbitPresent() throws Exception {
+ final StringBuilder endpoints = new StringBuilder(fakeBitsInfo);
+ endpoints.append(bitInfo);
+
+ Properties props = new Properties();
+ props.setProperty("drillbit", endpoints.toString());
+ props.setProperty("connect_limit", "3");
+
+ // Test with direct connection
+ DrillClient client = new DrillClient(true);
+ client.connect(props);
+ client.close();
+ }
+
+ @Test
+ public void testSuccessDirectConnectionDefaultConnectTriesAndFakeDrillbits() throws Exception {
+ final StringBuilder endpoints = new StringBuilder(fakeBitsInfo);
+ endpoints.append(bitInfo);
+
+ Properties props = new Properties();
+ props.setProperty("drillbit", endpoints.toString());
+
+ // Test with direct connection
+ DrillClient client = new DrillClient(true);
+ client.connect(props);
+ client.close();
+ }
+
+ @Test
+ public void testFailureUsingDirectConnectionAllFakeBits() throws Exception {
+ final StringBuilder endpoints = new StringBuilder(fakeBitsInfo);
+
+ Properties props = new Properties();
+ props.setProperty("drillbit", endpoints.toString());
+ props.setProperty("tries", "2");
+
+ // Test with direct connection
+ DrillClient client = new DrillClient(true);
+
+ try{
+ client.connect(props);
+ fail();
+ }catch(RpcException ex){
+ assertTrue(ex.getCause() instanceof ExecutionException);
+ client.close();
+ }
+ }
+
+ @Test
+ public void testSuccessUsingZKWithNoFakeBits() throws Exception {
+ Properties props = new Properties();
+ props.setProperty("tries", "2");
+
+ // Test with Cluster Coordinator connection
+ DrillClient client = new DrillClient(drillConfig, remoteServiceSet.getCoordinator());
+ client.connect(props);
+ client.close();
+ }
+
+ @Test
+ public void testSuccessUsingZKWithFakeBits() throws Exception {
+ Properties props = new Properties();
+ props.setProperty("tries", "3");
+
+ // Test with Cluster Coordinator connection
+ DrillClient client = new DrillClient(drillConfig, remoteServiceSet.getCoordinator());
+ // Create couple of fake drillbit endpoints and register with cluster coordinator
+ DrillbitEndpoint fakeEndPoint1 = DrillbitEndpoint.newBuilder().setAddress("127.0.0.1").setUserPort(5000).build();
+ DrillbitEndpoint fakeEndPoint2 = DrillbitEndpoint.newBuilder().setAddress("127.0.0.1").setUserPort(5001).build();
+
+ ClusterCoordinator.RegistrationHandle fakeEndPoint1Handle = remoteServiceSet.getCoordinator()
+ .register(fakeEndPoint1);
+ ClusterCoordinator.RegistrationHandle fakeEndPoint2Handle = remoteServiceSet.getCoordinator()
+ .register(fakeEndPoint2);
+
+ client.connect(props);
+ client.close();
+
+ // Remove the fake drillbits so that other tests are not affected
+ remoteServiceSet.getCoordinator().unregister(fakeEndPoint1Handle);
+ remoteServiceSet.getCoordinator().unregister(fakeEndPoint2Handle);
+ }
+
+ @Test
+ public void testSuccessUsingZKWithDefaultConnectTriesFakeBits() throws Exception {
+ // Test with Cluster Coordinator connection
+ DrillClient client = new DrillClient(drillConfig, remoteServiceSet.getCoordinator());
+
+ // Create couple of fake drillbit endpoints and register with cluster coordinator
+ DrillbitEndpoint fakeEndPoint1 = DrillbitEndpoint.newBuilder().setAddress("127.0.0.1").setUserPort(5000).build();
+ DrillbitEndpoint fakeEndPoint2 = DrillbitEndpoint.newBuilder().setAddress("127.0.0.1").setUserPort(5001).build();
+
+ ClusterCoordinator.RegistrationHandle fakeEndPoint1Handle = remoteServiceSet.getCoordinator()
+ .register(fakeEndPoint1);
+ ClusterCoordinator.RegistrationHandle fakeEndPoint2Handle = remoteServiceSet.getCoordinator()
+ .register(fakeEndPoint2);
+
+ client.connect(null);
+ client.close();
+
+ // Remove the fake drillbits so that other tests are not affected
+ remoteServiceSet.getCoordinator().unregister(fakeEndPoint1Handle);
+ remoteServiceSet.getCoordinator().unregister(fakeEndPoint2Handle);
+ }
+
+ @Test
+ public void testInvalidConnectTriesValue() throws Exception {
+ Properties props = new Properties();
+ props.setProperty("tries", "abc");
+
+ // Test with Cluster Cordinator connection
+ DrillClient client = new DrillClient(drillConfig, remoteServiceSet.getCoordinator());
+
+ try {
+ client.connect(props);
+ fail();
+ } catch (RpcException ex) {
+ assertTrue(ex instanceof InvalidConnectionInfoException);
+ client.close();
+ }
+ }
+
+ @Test
+ public void testConnectFailureUsingZKWithOnlyFakeBits() throws Exception {
+ Properties props = new Properties();
+ props.setProperty("tries", "3");
+
+ // Test with Cluster Coordinator connection
+ RemoteServiceSet localServiceSet = RemoteServiceSet.getLocalServiceSet();
+ DrillClient client = new DrillClient(drillConfig, localServiceSet.getCoordinator());
+
+ // Create couple of fake drillbit endpoints and register with cluster coordinator
+ DrillbitEndpoint fakeEndPoint1 = DrillbitEndpoint.newBuilder().setAddress("127.0.0.1").setUserPort(5000).build();
+ DrillbitEndpoint fakeEndPoint2 = DrillbitEndpoint.newBuilder().setAddress("127.0.0.1").setUserPort(5001).build();
+
+ ClusterCoordinator.RegistrationHandle fakeEndPoint1Handle = localServiceSet.getCoordinator()
+ .register(fakeEndPoint1);
+ ClusterCoordinator.RegistrationHandle fakeEndPoint2Handle = localServiceSet.getCoordinator()
+ .register(fakeEndPoint2);
+
+ try {
+ client.connect(props);
+ fail();
+ } catch (RpcException ex) {
+ assertTrue(ex.getCause() instanceof ExecutionException);
+ client.close();
+ } finally {
+ // Remove the fake drillbits from local cluster cordinator
+ localServiceSet.getCoordinator().unregister(fakeEndPoint1Handle);
+ localServiceSet.getCoordinator().unregister(fakeEndPoint2Handle);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/1c7309fc/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/JdbcConnectTriesTestEmbeddedBits.java
----------------------------------------------------------------------
diff --git a/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/JdbcConnectTriesTestEmbeddedBits.java b/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/JdbcConnectTriesTestEmbeddedBits.java
new file mode 100644
index 0000000..865829c
--- /dev/null
+++ b/exec/jdbc/src/test/java/org/apache/drill/jdbc/test/JdbcConnectTriesTestEmbeddedBits.java
@@ -0,0 +1,163 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.jdbc.test;
+
+import org.apache.drill.exec.rpc.InvalidConnectionInfoException;
+import org.apache.drill.exec.rpc.RpcException;
+import org.apache.drill.jdbc.Driver;
+import org.apache.drill.jdbc.JdbcTestBase;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.sql.SQLException;
+import java.sql.Connection;
+
+import java.util.concurrent.ExecutionException;
+
+import static junit.framework.Assert.assertNotNull;
+import static junit.framework.TestCase.fail;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+public class JdbcConnectTriesTestEmbeddedBits extends JdbcTestBase {
+
+ public static Driver testDrillDriver;
+
+ @BeforeClass
+ public static void testSetUp() throws Exception {
+ testDrillDriver = new Driver();
+ }
+
+ @Test
+ public void testDirectConnectionConnectTriesEqualsDrillbitCount() throws SQLException {
+ Connection connection = null;
+ try {
+ connection = testDrillDriver.connect("jdbc:drill:drillbit=127.0.0.1:5000,127.0.0.1:5001;" + "tries=2",
+ JdbcAssert.getDefaultProperties());
+ fail();
+ } catch (SQLException ex) {
+ assertNull(connection);
+ assertTrue(ex.getCause() instanceof RpcException);
+ assertTrue(ex.getCause().getCause() instanceof ExecutionException);
+ }
+ }
+
+ @Test
+ public void testDirectConnectionConnectTriesGreaterThanDrillbitCount() throws SQLException {
+ Connection connection = null;
+ try {
+ connection = testDrillDriver.connect("jdbc:drill:drillbit=127.0.0.1:5000,127.0.0.1:5001;tries=5",
+ JdbcAssert.getDefaultProperties());
+ fail();
+ } catch (SQLException ex) {
+ assertNull(connection);
+ assertTrue(ex.getCause() instanceof RpcException);
+ assertTrue(ex.getCause().getCause() instanceof ExecutionException);
+ }
+ }
+
+ @Test
+ public void testDirectConnectionConnectTriesLessThanDrillbitCount() throws SQLException {
+ Connection connection = null;
+ try {
+ connection = testDrillDriver.connect("jdbc:drill:drillbit=127.0.0.1:5000,127.0.0.1:5001;tries=1",
+ JdbcAssert.getDefaultProperties());
+ fail();
+ } catch (SQLException ex) {
+ assertNull(connection);
+ assertTrue(ex.getCause() instanceof RpcException);
+ assertTrue(ex.getCause().getCause() instanceof ExecutionException);
+ }
+ }
+
+ @Test
+ public void testDirectConnectionInvalidConnectTries() throws SQLException {
+ Connection connection = null;
+ try {
+ connection = testDrillDriver.connect("jdbc:drill:drillbit=127.0.0.1:5000,127.0.0.1:5001;tries=abc",
+ JdbcAssert.getDefaultProperties());
+ fail();
+ } catch (SQLException ex) {
+ assertNull(connection);
+ assertTrue(ex.getCause() instanceof InvalidConnectionInfoException);
+ }
+ }
+
+ @Test
+ public void testDirectConnectionZeroConnectTries() throws SQLException {
+ Connection connection = null;
+ try {
+ connection = testDrillDriver.connect("jdbc:drill:drillbit=127.0.0.1:5000,127.0.0.1:5001;tries=0",
+ JdbcAssert.getDefaultProperties());
+ fail();
+ } catch (SQLException ex) {
+ assertNull(connection);
+ assertTrue(ex.getCause() instanceof RpcException);
+ assertTrue(ex.getCause().getCause() instanceof ExecutionException);
+ }
+ }
+
+ @Test
+ public void testDirectConnectionNegativeConnectTries() throws SQLException {
+ Connection connection = null;
+ try {
+ connection = testDrillDriver.connect("jdbc:drill:drillbit=127.0.0.1:5000,127.0.0.1:5001;tries=-5",
+ JdbcAssert.getDefaultProperties());
+ fail();
+ } catch (SQLException ex) {
+ assertNull(connection);
+ assertTrue(ex.getCause() instanceof RpcException);
+ assertTrue(ex.getCause().getCause() instanceof ExecutionException);
+ }
+ }
+
+ @Test
+ public void testZKSuccessfulConnectionZeroConnectTries() throws SQLException {
+ Connection connection = testDrillDriver.connect("jdbc:drill:zk=local;tries=0", JdbcAssert.getDefaultProperties());
+ assertNotNull(connection);
+ connection.close();
+ }
+
+ @Test
+ public void testZKSuccessfulConnectionNegativeConnectTries() throws SQLException {
+ Connection connection = testDrillDriver.connect("jdbc:drill:zk=local;tries=-1", JdbcAssert.getDefaultProperties());
+ assertNotNull(connection);
+ connection.close();
+ }
+
+ @Test
+ public void testZKSuccessfulConnectionGreaterThanConnectTries() throws SQLException {
+ Connection connection = testDrillDriver.connect("jdbc:drill:zk=local;tries=7", JdbcAssert.getDefaultProperties());
+ assertNotNull(connection);
+ connection.close();
+ }
+
+ @Test
+ public void testZKConnectionInvalidConnectTries() throws SQLException {
+ Connection connection = null;
+ try {
+ connection = testDrillDriver.connect("jdbc:drill:zk=local;tries=abc", JdbcAssert.getDefaultProperties());
+ fail();
+ } catch (SQLException ex) {
+ assertNull(connection);
+ assertTrue(ex.getCause() instanceof InvalidConnectionInfoException);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/1c7309fc/exec/rpc/src/main/java/org/apache/drill/exec/rpc/BasicClient.java
----------------------------------------------------------------------
diff --git a/exec/rpc/src/main/java/org/apache/drill/exec/rpc/BasicClient.java b/exec/rpc/src/main/java/org/apache/drill/exec/rpc/BasicClient.java
index 0a501fd..a3eb4cb 100644
--- a/exec/rpc/src/main/java/org/apache/drill/exec/rpc/BasicClient.java
+++ b/exec/rpc/src/main/java/org/apache/drill/exec/rpc/BasicClient.java
@@ -150,9 +150,7 @@ public abstract class BasicClient<T extends EnumLite, R extends RemoteConnection
public abstract ProtobufLengthDecoder getDecoder(BufferAllocator allocator);
public boolean isActive() {
- return connection != null
- && connection.getChannel() != null
- && connection.getChannel().isActive();
+ return (connection != null) && connection.isActive();
}
protected abstract void validateHandshake(HANDSHAKE_RESPONSE validateHandshake) throws RpcException;
@@ -302,14 +300,9 @@ public abstract class BasicClient<T extends EnumLite, R extends RemoteConnection
public void close() {
logger.debug("Closing client");
- try {
- connection.getChannel().close().get();
- } catch (final InterruptedException | ExecutionException e) {
- logger.warn("Failure while shutting {}", this.getClass().getName(), e);
-
- // Preserve evidence that the interruption occurred so that code higher up on the call stack can learn of the
- // interruption and respond to it if it wants to.
- Thread.currentThread().interrupt();
+
+ if (connection != null) {
+ connection.close();
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/1c7309fc/exec/rpc/src/main/java/org/apache/drill/exec/rpc/RemoteConnection.java
----------------------------------------------------------------------
diff --git a/exec/rpc/src/main/java/org/apache/drill/exec/rpc/RemoteConnection.java b/exec/rpc/src/main/java/org/apache/drill/exec/rpc/RemoteConnection.java
index ad68140..fbacd23 100644
--- a/exec/rpc/src/main/java/org/apache/drill/exec/rpc/RemoteConnection.java
+++ b/exec/rpc/src/main/java/org/apache/drill/exec/rpc/RemoteConnection.java
@@ -81,7 +81,7 @@ public abstract class RemoteConnection implements ConnectionThrottle, AutoClosea
}
public boolean isActive() {
- return channel.isActive();
+ return (channel != null) && channel.isActive();
}
/**
@@ -181,13 +181,25 @@ public abstract class RemoteConnection implements ConnectionThrottle, AutoClosea
* closed before returning. As part of this call, the channel close handler
* will be triggered which will call channelClosed() above. The latter will
* happen in a separate thread while this method is blocking.
+ *
+ * <p>
+ * The check for isActive is not required here since channel can be in OPEN state without being active. We want
+ * to close in both the scenarios. A channel is in OPEN state when a socket is created for it before binding to an
+ * address.
+ * <li>
+ * For connection oriented transport protocol channel moves to ACTIVE state when a connection is established
+ * using this channel. We need to have channel in ACTIVE state NOT OPEN before we can send any message to
+ * remote endpoint.
+ * </li>
+ * <li>
+ * For connectionless transport protocol a sender can send data as soon as channel moves to OPEN state.
+ * </li>
+ * </p>
*/
@Override
public void close() {
try {
- if (channel.isActive()) {
- channel.close().get();
- }
+ channel.close().get();
} catch (final InterruptedException | ExecutionException e) {
logger.warn("Caught exception while closing channel.", e);
[8/8] drill git commit: DRILL-5052: Option to debug generated Java
code using an IDE
Posted by jn...@apache.org.
DRILL-5052: Option to debug generated Java code using an IDE
Provides a second compilation path for generated code: \u201cplan old Java\u201d
in which generated code inherit from their templates. Such code can be
compiled directly, allowing easy debugging of generated code.
Also show to generate two classes in the External Sort Batch as \u201cplain
old Java\u201d to enable IDE debugging of that generated code. Required
minor clean-up of the templates.
Fixes some broken toString( ) methods in code generation classes
Fixes a variety of small compilation warnings
Adds Java doc to a few classes
Includes clean-up from code review comments.
close apache/drill#660
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/bbcf4b76
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/bbcf4b76
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/bbcf4b76
Branch: refs/heads/master
Commit: bbcf4b765e6946a8b6c7110372c4e1cadbfefa44
Parents: 03928af
Author: Paul Rogers <pr...@maprtech.com>
Authored: Sat Nov 19 18:29:24 2016 -0800
Committer: Jinfeng Ni <jn...@apache.org>
Committed: Mon Dec 19 15:57:36 2016 -0800
----------------------------------------------------------------------
.../org/apache/drill/exec/ExecConstants.java | 1 -
.../exec/compile/AbstractClassCompiler.java | 7 +-
.../drill/exec/compile/CachedClassLoader.java | 69 +++++++
.../apache/drill/exec/compile/ClassBuilder.java | 189 +++++++++++++++++++
.../exec/compile/ClassCompilerSelector.java | 146 ++++++++++++++
.../drill/exec/compile/ClassTransformer.java | 27 ++-
.../apache/drill/exec/compile/CodeCompiler.java | 80 ++++++--
.../drill/exec/compile/DrillJavaFileObject.java | 21 +++
.../drill/exec/compile/JDKClassCompiler.java | 18 +-
.../drill/exec/compile/JaninoClassCompiler.java | 34 +++-
.../apache/drill/exec/compile/MergeAdapter.java | 12 +-
.../drill/exec/compile/QueryClassLoader.java | 91 +--------
.../exec/compile/TemplateClassDefinition.java | 29 ++-
.../exec/compile/sig/CodeGeneratorMethod.java | 2 +-
.../exec/compile/sig/GeneratorMapping.java | 19 ++
.../drill/exec/compile/sig/SignatureHolder.java | 27 ++-
.../apache/drill/exec/expr/ClassGenerator.java | 68 +++++--
.../apache/drill/exec/expr/CodeGenerator.java | 104 ++++++++--
.../drill/exec/expr/DebugStringBuilder.java | 62 ++++++
.../drill/exec/expr/DirectExpression.java | 14 +-
.../apache/drill/exec/ops/FragmentContext.java | 4 +-
.../apache/drill/exec/ops/OperatorContext.java | 10 +-
.../drill/exec/ops/OperatorContextImpl.java | 10 +
.../physical/impl/xsort/ExternalSortBatch.java | 12 +-
.../physical/impl/xsort/SingleBatchSorter.java | 2 +-
.../impl/xsort/SingleBatchSorterTemplate.java | 11 +-
.../server/options/SystemOptionManager.java | 8 +-
.../src/main/resources/drill-module.conf | 6 +-
.../exec/compile/TestClassTransformation.java | 16 +-
.../exec/compile/TestLargeFileCompilation.java | 14 +-
.../physical/unit/PhysicalOpUnitTestBase.java | 4 +-
.../drill/exec/vector/BaseValueVector.java | 3 +-
.../vector/complex/AbstractContainerVector.java | 2 +
.../exec/vector/complex/RepeatedListVector.java | 1 +
34 files changed, 927 insertions(+), 196 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java b/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java
index 0eb9c58..740eb4b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java
@@ -310,7 +310,6 @@ public interface ExecConstants {
OptionValidator ENABLE_NEW_TEXT_READER = new BooleanValidator(ENABLE_NEW_TEXT_READER_KEY, true);
String BOOTSTRAP_STORAGE_PLUGINS_FILE = "bootstrap-storage-plugins.json";
- String MAX_LOADING_CACHE_SIZE_CONFIG = "drill.exec.compile.cache_max_size";
String DRILL_SYS_FILE_SUFFIX = ".sys.drill";
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/java/org/apache/drill/exec/compile/AbstractClassCompiler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/AbstractClassCompiler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/AbstractClassCompiler.java
index a5c96e3..4caf8e1 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/AbstractClassCompiler.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/AbstractClassCompiler.java
@@ -20,12 +20,14 @@ package org.apache.drill.exec.compile;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
+import java.util.Map;
import org.apache.drill.common.util.DrillStringUtils;
import org.apache.drill.exec.compile.ClassTransformer.ClassNames;
import org.apache.drill.exec.exception.ClassTransformationException;
import org.codehaus.commons.compiler.CompileException;
+@SuppressWarnings("unused")
public abstract class AbstractClassCompiler {
protected boolean debug = false;
@@ -74,9 +76,10 @@ public abstract class AbstractClassCompiler {
return out.toString();
}
- protected abstract byte[][] getByteCode(ClassNames className, String sourcecode)
+ protected abstract byte[][] getByteCode(final ClassNames className, final String sourcecode)
throws CompileException, IOException, ClassNotFoundException, ClassTransformationException;
-
+ public abstract Map<String,byte[]> compile(final ClassNames className, final String sourceCode)
+ throws CompileException, IOException, ClassNotFoundException;
protected abstract org.slf4j.Logger getLogger();
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/java/org/apache/drill/exec/compile/CachedClassLoader.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/CachedClassLoader.java b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/CachedClassLoader.java
new file mode 100644
index 0000000..5270aa8
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/CachedClassLoader.java
@@ -0,0 +1,69 @@
+/**
+ * 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.compile;
+
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.util.Map;
+import java.util.concurrent.ConcurrentMap;
+
+import com.google.common.collect.Maps;
+
+/**
+ * Class loader for "plain-old Java" generated classes.
+ * Very basic implementation: allows defining a class from
+ * byte codes and finding the loaded classes. Delegates
+ * all other class requests to the thread context class
+ * loader. This structure ensures that a generated class can
+ * find both its own inner classes as well as all the standard
+ * Drill implementation classes.
+ */
+
+public class CachedClassLoader extends URLClassLoader {
+
+ /**
+ * Cache of generated classes. Semantics: a single thread defines
+ * the classes, many threads may access the classes.
+ */
+
+ private ConcurrentMap<String, Class<?>> cache = Maps.newConcurrentMap();
+
+ public CachedClassLoader() {
+ super(new URL[0], Thread.currentThread().getContextClassLoader());
+ }
+
+ public void addClass(String fqcn, byte[] byteCodes) {
+ Class<?> newClass = defineClass(fqcn, byteCodes, 0, byteCodes.length);
+ cache.put(fqcn, newClass);
+ }
+
+ @Override
+ public Class<?> findClass(String className) throws ClassNotFoundException {
+ Class<?> theClass = cache.get(className);
+ if (theClass != null) {
+ return theClass;
+ }
+ return super.findClass(className);
+ }
+
+ public void addClasses(Map<String, byte[]> results) {
+ for (Map.Entry<String, byte[]> result : results.entrySet()) {
+ addClass(result.getKey(), result.getValue());
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/java/org/apache/drill/exec/compile/ClassBuilder.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/ClassBuilder.java b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/ClassBuilder.java
new file mode 100644
index 0000000..f5024fe
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/ClassBuilder.java
@@ -0,0 +1,189 @@
+/**
+ * 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.compile;
+
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.Map;
+
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.exec.compile.ClassTransformer.ClassNames;
+import org.apache.drill.exec.exception.ClassTransformationException;
+import org.apache.drill.exec.expr.CodeGenerator;
+import org.apache.drill.exec.server.options.OptionManager;
+import org.codehaus.commons.compiler.CompileException;
+
+/**
+ * Implements the "plain-old Java" method of code generation and
+ * compilation. Given a {@link CodeGenerator}, obtains the generated
+ * source code, compiles it with the selected compiler, loads the
+ * byte-codes into a class loader and provides the resulting
+ * class. Compared with the {@link ClassTransformer} mechanism,
+ * this one requires the code generator to have generated a complete
+ * Java class that is capable of direct compilation and loading.
+ * This means the generated class must be a subclass of the template
+ * so that the JVM can use normal Java inheritance to associate the
+ * template and generated methods.
+ * <p>
+ * Here is how to use the plain-old Java technique to debug
+ * generated code:
+ * <ul>
+ * <li>Set the config option <var>drill.exec.compile.save_source</var>
+ * to <var>true</var>.</li>
+ * <li>Set the config option <var>drill.exec.compile.code_dir</var>
+ * to the location where you want to save the generated source
+ * code.</li>
+ * <li>Where you generate code (using a {@link CodeGenerator}),
+ * set the "plain-old Java" options:<pre>
+ * CodeGenerator<Foo> cg = ...
+ * cg.plainOldJavaCapable(true); // Class supports plain-old Java
+ * cg.preferPlainOldJava(true); // Actually generate plain-old Java
+ * ...</pre></li>
+ * <li>In your favorite IDE, add to the code lookup path the
+ * code directory saved earlier. In Eclipse, for example, you do
+ * this in the debug configuration you will use to debug Drill.</li>
+ * <li>Set a breakpoint in template used for the generated code.</li>
+ * <li>Run Drill. The IDE will stop at your breakpoint.</li>
+ * <li>Step into the generated code. Examine class field and
+ * local variables. Have fun!</li>
+ * </ul>
+ * <p>
+ * Note: not all generated code is ready to be compiled as plain-old
+ * Java. Some classes omit from the template the proper <code>throws</code>
+ * declarations. Other minor problems may also crop up. All are easy
+ * to fix. Once you've done so, add the following to mark that you've
+ * done the clean-up:<pre>
+ * cg.plainOldJavaCapable(true); // Class supports plain-old Java</pre>
+ * <p>
+ * The setting to prefer plain-old Java is ignored for generated
+ * classes not marked as plain-old Java capable.
+ */
+
+public class ClassBuilder {
+
+ public static final String SAVE_CODE_OPTION = CodeCompiler.COMPILE_BASE + ".save_source";
+ public static final String CODE_DIR_OPTION = CodeCompiler.COMPILE_BASE + ".code_dir";
+
+ private final DrillConfig config;
+ private final OptionManager options;
+ private final boolean saveCode;
+ private final File codeDir;
+
+ public ClassBuilder(DrillConfig config, OptionManager optionManager) {
+ this.config = config;
+ options = optionManager;
+
+ // The option to save code is a boot-time option because
+ // it is used selectively during debugging, but can cause
+ // excessive I/O in a running server if used to save all code.
+
+ saveCode = config.getBoolean(SAVE_CODE_OPTION);
+ codeDir = new File(config.getString(CODE_DIR_OPTION));
+ }
+
+ /**
+ * Given a code generator which has already generated plain-old Java
+ * code, compile the code, create a class loader, and return the
+ * resulting Java class.
+ *
+ * @param cg a plain-old Java capable code generator that has generated
+ * plain-old Java code
+ * @return the class that the code generator defines
+ * @throws ClassTransformationException
+ */
+
+ public Class<?> getImplementationClass(CodeGenerator<?> cg) throws ClassTransformationException {
+ try {
+ return compileClass(cg);
+ } catch (CompileException | ClassNotFoundException|IOException e) {
+ throw new ClassTransformationException(e);
+ }
+ }
+
+ /**
+ * Performs the actual work of compiling the code and loading the class.
+ *
+ * @param cg the code generator that has built the class(es) to be generated.
+ * @return the class, after code generation and (if needed) compilation.
+ * @throws IOException if an error occurs when optionally writing code to disk.
+ * @throws CompileException if the generated code has compile issues.
+ * @throws ClassNotFoundException if the generated code references unknown classes.
+ * @throws ClassTransformationException generic "something is wrong" error from
+ * Drill class compilation code.
+ */
+ private Class<?> compileClass(CodeGenerator<?> cg) throws IOException, CompileException, ClassNotFoundException, ClassTransformationException {
+
+ // Get the plain-old Java code.
+
+ String code = cg.getGeneratedCode();
+
+ // Get the class names (dotted, file path, etc.)
+
+ String className = cg.getMaterializedClassName();
+ ClassTransformer.ClassNames name = new ClassTransformer.ClassNames(className);
+
+ // A key advantage of this method is that the code can be
+ // saved and debugged, if needed.
+
+ saveCode(code, name);
+
+ // Compile the code and load it into a class loader.
+
+ CachedClassLoader classLoader = new CachedClassLoader();
+ ClassCompilerSelector compilerSelector = new ClassCompilerSelector(classLoader, config, options);
+ Map<String,byte[]> results = compilerSelector.compile(name, code);
+ classLoader.addClasses(results);
+
+ // Get the class from the class loader.
+
+ try {
+ return classLoader.findClass(className);
+ } catch (ClassNotFoundException e) {
+ // This should never occur.
+ throw new IllegalStateException("Code load failed", e);
+ }
+ }
+
+ /**
+ * Save code to a predefined location for debugging. To use the code
+ * for debugging, make sure the save location is on your IDE's source
+ * code search path. Code is saved in usual Java format with each
+ * package as a directory. The provided code directory becomes a
+ * source directory, as in Maven's "src/main/java".
+ *
+ * @param code the source code
+ * @param name the class name
+ */
+
+ private void saveCode(String code, ClassNames name) {
+
+ // Skip if we don't want to save the code.
+
+ if (! saveCode) { return; }
+
+ String pathName = name.slash + ".java";
+ File codeFile = new File(codeDir, pathName);
+ codeFile.getParentFile().mkdirs();
+ try (final FileWriter writer = new FileWriter(codeFile)) {
+ writer.write(code);
+ } catch (IOException e) {
+ System.err.println("Could not save: " + codeFile.getAbsolutePath());
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/java/org/apache/drill/exec/compile/ClassCompilerSelector.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/ClassCompilerSelector.java b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/ClassCompilerSelector.java
new file mode 100644
index 0000000..c8afbc6
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/ClassCompilerSelector.java
@@ -0,0 +1,146 @@
+/**
+ * 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.compile;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Map;
+
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.exec.compile.ClassTransformer.ClassNames;
+import org.apache.drill.exec.exception.ClassTransformationException;
+import org.apache.drill.exec.server.options.OptionManager;
+import org.apache.drill.exec.server.options.OptionValidator;
+import org.apache.drill.exec.server.options.OptionValue;
+import org.apache.drill.exec.server.options.TypeValidators.BooleanValidator;
+import org.apache.drill.exec.server.options.TypeValidators.LongValidator;
+import org.apache.drill.exec.server.options.TypeValidators.StringValidator;
+import org.codehaus.commons.compiler.CompileException;
+
+/**
+ * Selects between the two supported Java compilers: Janino and
+ * the build-in Java compiler.
+ *
+ * <h4>Session Options</h4>
+ * <dl>
+ * <dt>exec.java_compiler</dt>
+ * <dd>The compiler to use. Valid options are defined in the
+ * {@link ClassCompilerSelector.CompilerPolicy} enum.</dd>
+ * <dt>exec.java_compiler_debug</dt>
+ * <dd>If debug logging is enabled, then {@link AbstractClassCompiler} writes the
+ * generated Java code to the log file prior to compilation. This option
+ * adds line numbers to the logged code.</dd>
+ * <dt>exec.java_compiler_janino_maxsize</dt>
+ * <dd>The maximum size of code that the Janio compiler can handle. Larger code is
+ * handled by the JDK compiler. Defaults to 256K.</dd>
+ * </dl>
+ * <h4>Configuration Options</h4>
+ * Configuration options are used when the above session options are unset.
+ * <dl>
+ * <dt>drill.exec.compile.compiler</dt>
+ * <dd>Default for <var>exec.java_compiler</var></dd>
+ * <dt>drill.exec.compile.debug</dt>
+ * <dd>Default for <var>exec.java_compiler_debug</var></dd>
+ * <dt>drill.exec.compile.janino_maxsize</dt>
+ * <dd>Default for <var>exec.java_compiler_janino_maxsize</var></dd>
+ * </dl>
+ */
+
+public class ClassCompilerSelector {
+ public enum CompilerPolicy {
+ DEFAULT, JDK, JANINO;
+ }
+
+ public static final String JAVA_COMPILER_JANINO_MAXSIZE_CONFIG = CodeCompiler.COMPILE_BASE + ".janino_maxsize";
+ public static final String JAVA_COMPILER_DEBUG_CONFIG = CodeCompiler.COMPILE_BASE + ".debug";
+ public static final String JAVA_COMPILER_CONFIG = CodeCompiler.COMPILE_BASE + ".compiler";
+
+ public static final String JAVA_COMPILER_OPTION = "exec.java_compiler";
+ public static final String JAVA_COMPILER_JANINO_MAXSIZE_OPTION = "exec.java_compiler_janino_maxsize";
+ public static final OptionValidator JAVA_COMPILER_JANINO_MAXSIZE = new LongValidator(JAVA_COMPILER_JANINO_MAXSIZE_OPTION, 256*1024);
+
+ public static final String JAVA_COMPILER_DEBUG_OPTION = "exec.java_compiler_debug";
+ public static final OptionValidator JAVA_COMPILER_DEBUG = new BooleanValidator(JAVA_COMPILER_DEBUG_OPTION, true);
+
+ public static final StringValidator JAVA_COMPILER_VALIDATOR = new StringValidator(JAVA_COMPILER_OPTION, CompilerPolicy.DEFAULT.toString()) {
+ @Override
+ public void validate(final OptionValue v, final OptionManager manager) {
+ super.validate(v, manager);
+ try {
+ CompilerPolicy.valueOf(v.string_val.toUpperCase());
+ } catch (IllegalArgumentException e) {
+ throw UserException.validationError()
+ .message("Invalid value '%s' specified for option '%s'. Valid values are %s.",
+ v.string_val, getOptionName(), Arrays.toString(CompilerPolicy.values()))
+ .build(QueryClassLoader.logger);
+ }
+ }
+ };
+
+ private final CompilerPolicy policy;
+ private final long janinoThreshold;
+
+ private final AbstractClassCompiler jdkClassCompiler;
+ private final AbstractClassCompiler janinoClassCompiler;
+
+ public ClassCompilerSelector(ClassLoader classLoader, DrillConfig config, OptionManager sessionOptions) {
+ OptionValue value = sessionOptions.getOption(JAVA_COMPILER_OPTION);
+ this.policy = CompilerPolicy.valueOf((value != null) ? value.string_val.toUpperCase() : config.getString(JAVA_COMPILER_CONFIG).toUpperCase());
+
+ value = sessionOptions.getOption(JAVA_COMPILER_JANINO_MAXSIZE_OPTION);
+ this.janinoThreshold = (value != null) ? value.num_val : config.getLong(JAVA_COMPILER_JANINO_MAXSIZE_CONFIG);
+
+ value = sessionOptions.getOption(JAVA_COMPILER_DEBUG_OPTION);
+ boolean debug = (value != null) ? value.bool_val : config.getBoolean(JAVA_COMPILER_DEBUG_CONFIG);
+
+ this.janinoClassCompiler = (policy == CompilerPolicy.JANINO || policy == CompilerPolicy.DEFAULT) ? new JaninoClassCompiler(classLoader, debug) : null;
+ this.jdkClassCompiler = (policy == CompilerPolicy.JDK || policy == CompilerPolicy.DEFAULT) ? JDKClassCompiler.newInstance(classLoader, debug) : null;
+ }
+
+ byte[][] getClassByteCode(ClassNames className, String sourceCode)
+ throws CompileException, ClassNotFoundException, ClassTransformationException, IOException {
+
+ byte[][] bc = getCompiler(sourceCode).getClassByteCode(className, sourceCode);
+
+ // Uncomment the following to save the generated byte codes.
+
+// final String baseDir = System.getProperty("java.io.tmpdir") + File.separator + className;
+// File classFile = new File(baseDir + className.clazz);
+// classFile.getParentFile().mkdirs();
+// try (BufferedOutputStream out = new BufferedOutputStream(new FileOutputStream(classFile))) {
+// out.write(bc[0]);
+// }
+
+ return bc;
+ }
+
+ public Map<String,byte[]> compile(ClassNames className, String sourceCode)
+ throws CompileException, ClassNotFoundException, ClassTransformationException, IOException {
+ return getCompiler(sourceCode).compile(className, sourceCode);
+ }
+
+ private AbstractClassCompiler getCompiler(String sourceCode) {
+ if (jdkClassCompiler != null &&
+ (policy == CompilerPolicy.JDK || (policy == CompilerPolicy.DEFAULT && sourceCode.length() > janinoThreshold))) {
+ return jdkClassCompiler;
+ } else {
+ return janinoClassCompiler;
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/java/org/apache/drill/exec/compile/ClassTransformer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/ClassTransformer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/ClassTransformer.java
index 02323a9..3c3c30e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/ClassTransformer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/ClassTransformer.java
@@ -22,12 +22,13 @@ import java.util.LinkedList;
import java.util.Map;
import java.util.Set;
+import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.common.util.DrillStringUtils;
import org.apache.drill.common.util.FileUtils;
import org.apache.drill.exec.compile.MergeAdapter.MergedClassResult;
import org.apache.drill.exec.exception.ClassTransformationException;
+import org.apache.drill.exec.expr.CodeGenerator;
import org.apache.drill.exec.server.options.OptionManager;
-import org.apache.drill.exec.server.options.OptionValue;
import org.apache.drill.exec.server.options.TypeValidators.EnumeratedStringValidator;
import org.codehaus.commons.compiler.CompileException;
import org.objectweb.asm.ClassReader;
@@ -39,12 +40,21 @@ import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
+/**
+ * Compiles generated code, merges the resulting class with the
+ * template class, and performs byte-code cleanup on the resulting
+ * byte codes. The most important transform is scalar replacement
+ * which replaces occurences of non-escaping objects with a
+ * collection of member variables.
+ */
+
public class ClassTransformer {
private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ClassTransformer.class);
private static final int MAX_SCALAR_REPLACE_CODE_SIZE = 2*1024*1024; // 2meg
private final ByteCodeLoader byteCodeLoader = new ByteCodeLoader();
+ private final DrillConfig config;
private final OptionManager optionManager;
public final static String SCALAR_REPLACEMENT_OPTION =
@@ -73,13 +83,14 @@ public class ClassTransformer {
return TRY;
case "on":
return ON;
+ default:
+ throw new IllegalArgumentException("Invalid ScalarReplacementOption \"" + s + "\"");
}
-
- throw new IllegalArgumentException("Invalid ScalarReplacementOption \"" + s + "\"");
}
}
- public ClassTransformer(final OptionManager optionManager) {
+ public ClassTransformer(final DrillConfig config, final OptionManager optionManager) {
+ this.config = config;
this.optionManager = optionManager;
}
@@ -210,6 +221,12 @@ public class ClassTransformer {
}
}
+ public Class<?> getImplementationClass(CodeGenerator<?> cg) throws ClassTransformationException {
+ final QueryClassLoader loader = new QueryClassLoader(config, optionManager);
+ return getImplementationClass(loader, cg.getDefinition(),
+ cg.getGeneratedCode(), cg.getMaterializedClassName());
+ }
+
public Class<?> getImplementationClass(
final QueryClassLoader classLoader,
final TemplateClassDefinition<?> templateDefinition,
@@ -248,7 +265,7 @@ public class ClassTransformer {
final ClassNames nextGenerated = nextSet.generated;
final ClassNode generatedNode = classesToMerge.get(nextGenerated.slash);
- /**
+ /*
* TODO
* We're having a problem with some cases of scalar replacement, but we want to get
* the code in so it doesn't rot anymore.
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/java/org/apache/drill/exec/compile/CodeCompiler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/CodeCompiler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/CodeCompiler.java
index af328b1..fb59a4c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/CodeCompiler.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/CodeCompiler.java
@@ -22,7 +22,6 @@ import java.util.List;
import java.util.concurrent.ExecutionException;
import org.apache.drill.common.config.DrillConfig;
-import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.exception.ClassTransformationException;
import org.apache.drill.exec.expr.CodeGenerator;
import org.apache.drill.exec.server.options.OptionManager;
@@ -33,36 +32,72 @@ import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.collect.Lists;
+/**
+ * Global code compiler mechanism shared by all threads and operators.
+ * Holds a single cache of generated code (keyed by code source) to
+ * prevent compiling identical code multiple times. Supports both
+ * the byte-code merging and plain-old Java methods of code
+ * generation and compilation.
+ */
+
public class CodeCompiler {
-// private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(CodeCompiler.class);
+
+ public static final String COMPILE_BASE = "drill.exec.compile";
+ public static final String MAX_LOADING_CACHE_SIZE_CONFIG = COMPILE_BASE + ".cache_max_size";
private final ClassTransformer transformer;
+ private final ClassBuilder classBuilder;
+
+ /**
+ * Google Guava loading cache that defers creating a cache
+ * entry until first needed. Creation is done in a thread-safe
+ * way: if two threads try to create the same class at the same
+ * time, the first does the work, the second waits for the first
+ * to complete, then grabs the new entry.
+ */
+
private final LoadingCache<CodeGenerator<?>, GeneratedClassEntry> cache;
- private final DrillConfig config;
- private final OptionManager optionManager;
public CodeCompiler(final DrillConfig config, final OptionManager optionManager) {
- transformer = new ClassTransformer(optionManager);
- final int cacheMaxSize = config.getInt(ExecConstants.MAX_LOADING_CACHE_SIZE_CONFIG);
+ transformer = new ClassTransformer(config, optionManager);
+ classBuilder = new ClassBuilder(config, optionManager);
+ final int cacheMaxSize = config.getInt(MAX_LOADING_CACHE_SIZE_CONFIG);
cache = CacheBuilder.newBuilder()
.maximumSize(cacheMaxSize)
.build(new Loader());
- this.optionManager = optionManager;
- this.config = config;
}
+ /**
+ * Create a single instance of the generated class.
+ *
+ * @param cg code generator for the class to be instantiated.
+ * @return an instance of the generated class
+ * @throws ClassTransformationException general "something is wrong" exception
+ * for the Drill compilation chain.
+ */
+
@SuppressWarnings("unchecked")
- public <T> T getImplementationClass(final CodeGenerator<?> cg) throws ClassTransformationException, IOException {
- return (T) getImplementationClass(cg, 1).get(0);
+ public <T> T createInstance(final CodeGenerator<?> cg) throws ClassTransformationException {
+ return (T) createInstances(cg, 1).get(0);
}
+ /**
+ * Create multiple instances of the generated class.
+ *
+ * @param cg code generator for the class to be instantiated.
+ * @param count the number of instances desired.
+ * @return a list of instances of the generated class.
+ * @throws ClassTransformationException general "something is wrong" exception
+ * for the Drill compilation chain.
+ */
+
@SuppressWarnings("unchecked")
- public <T> List<T> getImplementationClass(final CodeGenerator<?> cg, int instanceNumber) throws ClassTransformationException, IOException {
+ public <T> List<T> createInstances(final CodeGenerator<?> cg, int count) throws ClassTransformationException {
cg.generate();
try {
final GeneratedClassEntry ce = cache.get(cg);
List<T> tList = Lists.newArrayList();
- for ( int i = 0; i < instanceNumber; i++) {
+ for ( int i = 0; i < count; i++) {
tList.add((T) ce.clazz.newInstance());
}
return tList;
@@ -71,12 +106,27 @@ public class CodeCompiler {
}
}
+ /**
+ * Loader used to create an entry in the class cache when the entry
+ * does not yet exist. Here, we generate the code, compile it,
+ * and place the resulting class into the cache. The class has an
+ * associated class loader which "dangles" from the class itself;
+ * we don't keep track of the class loader itself.
+ */
+
private class Loader extends CacheLoader<CodeGenerator<?>, GeneratedClassEntry> {
@Override
public GeneratedClassEntry load(final CodeGenerator<?> cg) throws Exception {
- final QueryClassLoader loader = new QueryClassLoader(config, optionManager);
- final Class<?> c = transformer.getImplementationClass(loader, cg.getDefinition(),
- cg.getGeneratedCode(), cg.getMaterializedClassName());
+ final Class<?> c;
+ if ( cg.isPlainOldJava( ) ) {
+ // Generate class as plain old Java
+
+ c = classBuilder.getImplementationClass(cg);
+ } else {
+ // Generate class parts and assemble byte-codes.
+
+ c = transformer.getImplementationClass(cg);
+ }
return new GeneratedClassEntry(c);
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/java/org/apache/drill/exec/compile/DrillJavaFileObject.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/DrillJavaFileObject.java b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/DrillJavaFileObject.java
index acc32b5..7b95374 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/DrillJavaFileObject.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/DrillJavaFileObject.java
@@ -24,6 +24,7 @@ import java.io.Reader;
import java.io.StringReader;
import java.net.URI;
import java.net.URISyntaxException;
+import java.util.HashMap;
import java.util.Map;
import javax.tools.SimpleJavaFileObject;
@@ -38,14 +39,18 @@ final class DrillJavaFileObject extends SimpleJavaFileObject {
private Map<String, DrillJavaFileObject> outputFiles;
+ private final String className;
+
public DrillJavaFileObject(final String className, final String sourceCode) {
super(makeURI(className), Kind.SOURCE);
+ this.className = className;
this.sourceCode = sourceCode;
this.outputStream = null;
}
private DrillJavaFileObject(final String name, final Kind kind) {
super(makeURI(name), kind);
+ this.className = name;
this.outputStream = new ByteArrayOutputStream();
this.sourceCode = null;
}
@@ -67,6 +72,22 @@ final class DrillJavaFileObject extends SimpleJavaFileObject {
}
}
+ /**
+ * Return the byte codes for the main class and any nested
+ * classes.
+ *
+ * @return map of fully-qualified class names to byte codes
+ * for the class
+ */
+
+ public Map<String,byte[]> getClassByteCodes() {
+ Map<String,byte[]> results = new HashMap<>();
+ for(DrillJavaFileObject outputFile : outputFiles.values()) {
+ results.put(outputFile.className, outputFile.outputStream.toByteArray());
+ }
+ return results;
+ }
+
public DrillJavaFileObject addOutputJavaFile(String className) {
if (outputFiles == null) {
outputFiles = Maps.newLinkedHashMap();
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/java/org/apache/drill/exec/compile/JDKClassCompiler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/JDKClassCompiler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/JDKClassCompiler.java
index ecd222d..6007078 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/JDKClassCompiler.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/JDKClassCompiler.java
@@ -20,6 +20,7 @@ package org.apache.drill.exec.compile;
import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
+import java.util.Map;
import javax.tools.DiagnosticListener;
import javax.tools.JavaCompiler;
@@ -44,8 +45,7 @@ class JDKClassCompiler extends AbstractClassCompiler {
public static JDKClassCompiler newInstance(ClassLoader classLoader, boolean debug) {
JavaCompiler compiler = ToolProvider.getSystemJavaCompiler();
if (compiler == null) {
- logger.warn("JDK Java compiler not available - probably you're running Drill with a JRE and not a JDK");
- return null;
+ throw new RuntimeException("JDK Java compiler not available - probably you're running Drill with a JRE and not a JDK");
}
return new JDKClassCompiler(compiler, classLoader, debug);
}
@@ -61,6 +61,17 @@ class JDKClassCompiler extends AbstractClassCompiler {
@Override
protected byte[][] getByteCode(final ClassNames className, final String sourceCode)
throws CompileException, IOException, ClassNotFoundException {
+ return doCompile(className, sourceCode).getByteCode();
+ }
+
+ @Override
+ public Map<String,byte[]> compile(final ClassNames className, final String sourceCode)
+ throws CompileException, IOException, ClassNotFoundException {
+ return doCompile(className, sourceCode).getClassByteCodes();
+ }
+
+ private DrillJavaFileObject doCompile(final ClassNames className, final String sourceCode)
+ throws CompileException, IOException, ClassNotFoundException {
try {
// Create one Java source file in memory, which will be compiled later.
DrillJavaFileObject compilationUnit = new DrillJavaFileObject(className.dot, sourceCode);
@@ -74,7 +85,7 @@ class JDKClassCompiler extends AbstractClassCompiler {
throw new ClassNotFoundException(className + ": Class file not created by compilation.");
}
// all good
- return compilationUnit.getByteCode();
+ return compilationUnit;
} catch (RuntimeException rte) {
// Unwrap the compilation exception and throw it.
Throwable cause = rte.getCause();
@@ -93,5 +104,4 @@ class JDKClassCompiler extends AbstractClassCompiler {
@Override
protected org.slf4j.Logger getLogger() { return logger; }
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/java/org/apache/drill/exec/compile/JaninoClassCompiler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/JaninoClassCompiler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/JaninoClassCompiler.java
index 1bb4465..cab8e22 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/JaninoClassCompiler.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/JaninoClassCompiler.java
@@ -19,6 +19,8 @@ package org.apache.drill.exec.compile;
import java.io.IOException;
import java.io.StringReader;
+import java.util.HashMap;
+import java.util.Map;
import org.apache.drill.exec.compile.ClassTransformer.ClassNames;
import org.apache.drill.exec.exception.ClassTransformationException;
@@ -31,7 +33,7 @@ import org.codehaus.janino.Scanner;
import org.codehaus.janino.UnitCompiler;
import org.codehaus.janino.util.ClassFile;
-public class JaninoClassCompiler extends AbstractClassCompiler {
+class JaninoClassCompiler extends AbstractClassCompiler {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(JaninoClassCompiler.class);
private IClassLoader compilationClassLoader;
@@ -42,13 +44,9 @@ public class JaninoClassCompiler extends AbstractClassCompiler {
}
@Override
- protected byte[][] getByteCode(final ClassNames className, final String sourcecode)
+ protected byte[][] getByteCode(final ClassNames className, final String sourceCode)
throws CompileException, IOException, ClassNotFoundException, ClassTransformationException {
- StringReader reader = new StringReader(sourcecode);
- Scanner scanner = new Scanner((String) null, reader);
- Java.CompilationUnit compilationUnit = new Parser(scanner).parseCompilationUnit();
- ClassFile[] classFiles = new UnitCompiler(compilationUnit, compilationClassLoader)
- .compileUnit(this.debug, this.debug, this.debug);
+ ClassFile[] classFiles = doCompile(sourceCode);
byte[][] byteCodes = new byte[classFiles.length][];
for(int i = 0; i < classFiles.length; i++){
@@ -58,5 +56,27 @@ public class JaninoClassCompiler extends AbstractClassCompiler {
}
@Override
+ public Map<String,byte[]> compile(final ClassNames className, final String sourceCode)
+ throws CompileException, IOException, ClassNotFoundException {
+
+ ClassFile[] classFiles = doCompile(sourceCode);
+ Map<String,byte[]> results = new HashMap<>();
+ for(int i = 0; i < classFiles.length; i++) {
+ ClassFile classFile = classFiles[i];
+ results.put(classFile.getThisClassName(), classFile.toByteArray());
+ }
+ return results;
+ }
+
+ private ClassFile[] doCompile(final String sourceCode)
+ throws CompileException, IOException, ClassNotFoundException {
+ StringReader reader = new StringReader(sourceCode);
+ Scanner scanner = new Scanner((String) null, reader);
+ Java.CompilationUnit compilationUnit = new Parser(scanner).parseCompilationUnit();
+ return new UnitCompiler(compilationUnit, compilationClassLoader)
+ .compileUnit(this.debug, this.debug, this.debug);
+ }
+
+ @Override
protected org.slf4j.Logger getLogger() { return logger; }
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/java/org/apache/drill/exec/compile/MergeAdapter.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/MergeAdapter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/MergeAdapter.java
index 82bd413..05e8ac1 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/MergeAdapter.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/MergeAdapter.java
@@ -17,6 +17,8 @@
*/
package org.apache.drill.exec.compile;
+import java.io.File;
+import java.io.IOException;
import java.lang.reflect.Modifier;
import java.util.Collection;
import java.util.Iterator;
@@ -39,11 +41,13 @@ import org.objectweb.asm.tree.FieldNode;
import org.objectweb.asm.tree.MethodNode;
import com.google.common.collect.Sets;
+import com.google.common.io.Files;
/**
* Serves two purposes. Renames all inner classes references to the outer class to the new name. Also adds all the
* methods and fields of the class to merge to the class that is being visited.
*/
+@SuppressWarnings("unused")
class MergeAdapter extends ClassVisitor {
private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(MergeAdapter.class);
private final ClassNode classToMerge;
@@ -253,7 +257,13 @@ class MergeAdapter extends ClassVisitor {
}
// enable when you want all the generated merged class files to also be written to disk.
-// Files.write(outputClass, new File(String.format("/src/scratch/drill-generated-classes/%s-output.class", set.generated.dot)));
+// try {
+// File destDir = new File( "/tmp/scratch/drill-generated-classes" );
+// destDir.mkdirs();
+// Files.write(outputClass, new File(destDir, String.format("%s-output.class", set.generated.dot)));
+// } catch (IOException e) {
+// // Ignore;
+// }
return new MergedClassResult(outputClass, re.getInnerClasses());
} catch (Error | RuntimeException e) {
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/java/org/apache/drill/exec/compile/QueryClassLoader.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/QueryClassLoader.java b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/QueryClassLoader.java
index da03802..31b464b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/QueryClassLoader.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/QueryClassLoader.java
@@ -20,52 +20,23 @@ package org.apache.drill.exec.compile;
import java.io.IOException;
import java.net.URL;
import java.net.URLClassLoader;
-import java.util.Arrays;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.drill.common.config.DrillConfig;
-import org.apache.drill.common.exceptions.UserException;
import org.apache.drill.exec.compile.ClassTransformer.ClassNames;
import org.apache.drill.exec.exception.ClassTransformationException;
import org.apache.drill.exec.server.options.OptionManager;
-import org.apache.drill.exec.server.options.OptionValidator;
-import org.apache.drill.exec.server.options.OptionValue;
-import org.apache.drill.exec.server.options.TypeValidators.BooleanValidator;
-import org.apache.drill.exec.server.options.TypeValidators.LongValidator;
-import org.apache.drill.exec.server.options.TypeValidators.StringValidator;
import org.codehaus.commons.compiler.CompileException;
import com.google.common.collect.MapMaker;
-public class QueryClassLoader extends URLClassLoader {
- private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(QueryClassLoader.class);
-
- public static final String JAVA_COMPILER_OPTION = "exec.java_compiler";
- public static final StringValidator JAVA_COMPILER_VALIDATOR = new StringValidator(JAVA_COMPILER_OPTION, CompilerPolicy.DEFAULT.toString()) {
- @Override
- public void validate(final OptionValue v, final OptionManager manager) {
- super.validate(v, manager);
- try {
- CompilerPolicy.valueOf(v.string_val.toUpperCase());
- } catch (IllegalArgumentException e) {
- throw UserException.validationError()
- .message("Invalid value '%s' specified for option '%s'. Valid values are %s.",
- v.string_val, getOptionName(), Arrays.toString(CompilerPolicy.values()))
- .build(logger);
- }
- }
- };
-
- public static final String JAVA_COMPILER_DEBUG_OPTION = "exec.java_compiler_debug";
- public static final OptionValidator JAVA_COMPILER_DEBUG = new BooleanValidator(JAVA_COMPILER_DEBUG_OPTION, true);
-
- public static final String JAVA_COMPILER_JANINO_MAXSIZE_OPTION = "exec.java_compiler_janino_maxsize";
- public static final OptionValidator JAVA_COMPILER_JANINO_MAXSIZE = new LongValidator(JAVA_COMPILER_JANINO_MAXSIZE_OPTION, 256*1024);
+/**
+ * Per-compilation unit class loader that holds both caching and compilation
+ * steps. */
- public static final String JAVA_COMPILER_CONFIG = "drill.exec.compile.compiler";
- public static final String JAVA_COMPILER_DEBUG_CONFIG = "drill.exec.compile.debug";
- public static final String JAVA_COMPILER_JANINO_MAXSIZE_CONFIG = "drill.exec.compile.janino_maxsize";
+public class QueryClassLoader extends URLClassLoader {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(QueryClassLoader.class);
private ClassCompilerSelector compilerSelector;
@@ -75,7 +46,7 @@ public class QueryClassLoader extends URLClassLoader {
public QueryClassLoader(DrillConfig config, OptionManager sessionOptions) {
super(new URL[0], Thread.currentThread().getContextClassLoader());
- compilerSelector = new ClassCompilerSelector(config, sessionOptions);
+ compilerSelector = new ClassCompilerSelector(this, config, sessionOptions);
}
public long getNextClassIndex() {
@@ -104,54 +75,4 @@ public class QueryClassLoader extends URLClassLoader {
return compilerSelector.getClassByteCode(className, sourceCode);
}
- public enum CompilerPolicy {
- DEFAULT, JDK, JANINO;
- }
-
- private class ClassCompilerSelector {
- private final CompilerPolicy policy;
- private final long janinoThreshold;
-
- private final AbstractClassCompiler jdkClassCompiler;
- private final AbstractClassCompiler janinoClassCompiler;
-
-
- ClassCompilerSelector(DrillConfig config, OptionManager sessionOptions) {
- OptionValue value = sessionOptions.getOption(JAVA_COMPILER_OPTION);
- this.policy = CompilerPolicy.valueOf((value != null) ? value.string_val.toUpperCase() : config.getString(JAVA_COMPILER_CONFIG).toUpperCase());
-
- value = sessionOptions.getOption(JAVA_COMPILER_JANINO_MAXSIZE_OPTION);
- this.janinoThreshold = (value != null) ? value.num_val : config.getLong(JAVA_COMPILER_JANINO_MAXSIZE_CONFIG);
-
- value = sessionOptions.getOption(JAVA_COMPILER_DEBUG_OPTION);
- boolean debug = (value != null) ? value.bool_val : config.getBoolean(JAVA_COMPILER_DEBUG_CONFIG);
-
- this.janinoClassCompiler = (policy == CompilerPolicy.JANINO || policy == CompilerPolicy.DEFAULT) ? new JaninoClassCompiler(QueryClassLoader.this, debug) : null;
- this.jdkClassCompiler = (policy == CompilerPolicy.JDK || policy == CompilerPolicy.DEFAULT) ? JDKClassCompiler.newInstance(QueryClassLoader.this, debug) : null;
- }
-
- private byte[][] getClassByteCode(ClassNames className, String sourceCode)
- throws CompileException, ClassNotFoundException, ClassTransformationException, IOException {
- AbstractClassCompiler classCompiler;
- if (jdkClassCompiler != null &&
- (policy == CompilerPolicy.JDK || (policy == CompilerPolicy.DEFAULT && sourceCode.length() > janinoThreshold))) {
- classCompiler = jdkClassCompiler;
- } else {
- classCompiler = janinoClassCompiler;
- }
-
- byte[][] bc = classCompiler.getClassByteCode(className, sourceCode);
- /*
- * final String baseDir = System.getProperty("java.io.tmpdir") + File.separator + classCompiler.getClass().getSimpleName();
- * File classFile = new File(baseDir + className.clazz);
- * classFile.getParentFile().mkdirs();
- * BufferedOutputStream out = new BufferedOutputStream(new FileOutputStream(classFile));
- * out.write(bc[0]);
- * out.close();
- */
- return bc;
- }
-
- }
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/java/org/apache/drill/exec/compile/TemplateClassDefinition.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/TemplateClassDefinition.java b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/TemplateClassDefinition.java
index ec5bfcd..1979db1 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/TemplateClassDefinition.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/TemplateClassDefinition.java
@@ -21,12 +21,25 @@ import java.util.concurrent.atomic.AtomicLong;
import org.apache.drill.exec.compile.sig.SignatureHolder;
+/**
+ * Defines a code generation "template" which consist of:
+ * <ul>
+ * <li>An interface that defines the generated class.</li>
+ * <li>A template class which implements the interface to provide
+ * "generic" methods that need not be generated.</li>
+ * <li>A signature that lists the methods and vector holders used
+ * by the template.</li>
+ * </ul>
+ *
+ * @param <T> The template interface
+ */
+
public class TemplateClassDefinition<T>{
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TemplateClassDefinition.class);
private final Class<T> iface;
- private final Class<?> template;
+ private final Class<? extends T> template;
private final SignatureHolder signature;
private static final AtomicLong classNumber = new AtomicLong(0);
@@ -41,7 +54,6 @@ public class TemplateClassDefinition<T>{
logger.error("Failure while trying to build signature holder for signature. {}", template.getName(), ex);
}
this.signature = holder;
-
}
public long getNextClassNumber(){
@@ -52,6 +64,9 @@ public class TemplateClassDefinition<T>{
return iface;
}
+ public Class<? extends T> getTemplateClass() {
+ return template;
+ }
public String getTemplateClassName() {
return template.getName();
@@ -63,6 +78,14 @@ public class TemplateClassDefinition<T>{
@Override
public String toString() {
- return "TemplateClassDefinition [template=" + template + ", signature=" + signature + "]";
+ StringBuilder buf = new StringBuilder();
+ buf.append("TemplateClassDefinition [interface=");
+ buf.append((iface == null) ? "null" : iface.getName());
+ buf.append(", template=");
+ buf.append((template == null) ? "null" : template.getName());
+ buf.append(", signature=\n");
+ buf.append(signature);
+ buf.append("]");
+ return buf.toString();
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/java/org/apache/drill/exec/compile/sig/CodeGeneratorMethod.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/sig/CodeGeneratorMethod.java b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/sig/CodeGeneratorMethod.java
index 9df346c..c83498a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/sig/CodeGeneratorMethod.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/sig/CodeGeneratorMethod.java
@@ -82,7 +82,7 @@ public class CodeGeneratorMethod implements Iterable<CodeGeneratorArgument> {
@Override
public String toString() {
- return "CodeGeneratorMethod [" + underlyingMethod.toGenericString() + "]";
+ return "CodeGeneratorMethod [" + ((underlyingMethod == null) ? "null" : underlyingMethod.toGenericString()) + "]";
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/java/org/apache/drill/exec/compile/sig/GeneratorMapping.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/sig/GeneratorMapping.java b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/sig/GeneratorMapping.java
index 9c12116..b9b62a8 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/sig/GeneratorMapping.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/sig/GeneratorMapping.java
@@ -21,6 +21,25 @@ import org.apache.drill.exec.expr.ClassGenerator.BlockType;
import com.google.common.base.Preconditions;
+/**
+ * The code generator works with four conceptual methods which can
+ * have any actual names. This class identify which conceptual methods
+ * are in use and their actual names. Callers obtain the method
+ * names generically using the {@link BlockType} enum. There is,
+ * however, no way to check which methods are in use; the user of
+ * this method must already know this information from another
+ * source.
+ * <table>
+ * <tr><th>Conceptual Method</th>
+ * <th>BlockType</th>
+ * <th>Typical Drill Name</th></tr>
+ * <tr><td>setup</td><td>SETUP</td><td>doSetup</td></tr>
+ * <tr><td>eval</td><td>EVAL</td><td>doEval</td></tr>
+ * <tr><td>reset</td><td>RESET</td><td>?</td></tr>
+ * <tr><td>cleanup</td><td>CLEANUP</td><td>?</td></tr>
+ * </table>
+ */
+
public class GeneratorMapping {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(GeneratorMapping.class);
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/java/org/apache/drill/exec/compile/sig/SignatureHolder.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/sig/SignatureHolder.java b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/sig/SignatureHolder.java
index 7fe8e3b..541a85f 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/sig/SignatureHolder.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/sig/SignatureHolder.java
@@ -20,6 +20,8 @@ package org.apache.drill.exec.compile.sig;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -51,7 +53,6 @@ public class SignatureHolder implements Iterable<CodeGeneratorMethod> {
return new SignatureHolder(signature, innerClasses.toArray(new SignatureHolder[innerClasses.size()]));
}
-
private SignatureHolder(Class<?> signature, SignatureHolder[] childHolders) {
this.childHolders = childHolders;
this.signature = signature;
@@ -67,6 +68,16 @@ public class SignatureHolder implements Iterable<CodeGeneratorMethod> {
methodHolders.add(new CodeGeneratorMethod(m));
}
+ // Alphabetize methods to ensure generated code is comparable.
+ // Also eases debugging as the generated code contain different method
+ // order from run to run.
+
+ Collections.sort( methodHolders, new Comparator<CodeGeneratorMethod>( ) {
+ @Override
+ public int compare(CodeGeneratorMethod o1, CodeGeneratorMethod o2) {
+ return o1.getMethodName().compareTo( o2.getMethodName() );
+ } } );
+
methods = new CodeGeneratorMethod[methodHolders.size()+1];
for (int i =0; i < methodHolders.size(); i++) {
methods[i] = methodHolders.get(i);
@@ -99,7 +110,6 @@ public class SignatureHolder implements Iterable<CodeGeneratorMethod> {
return methods.length;
}
-
public SignatureHolder[] getChildHolders() {
return childHolders;
}
@@ -114,9 +124,16 @@ public class SignatureHolder implements Iterable<CodeGeneratorMethod> {
@Override
public String toString() {
+ StringBuilder buf = new StringBuilder( );
+ buf.append( "SignatureHolder [methods=" );
final int maxLen = 10;
- return "SignatureHolder [methods="
- + (methods != null ? Arrays.asList(methods).subList(0, Math.min(methods.length, maxLen)) : null) + "]";
+ for ( int i = 0; i < maxLen && i < methods.length; i++ ) {
+ if ( i > 0 ) {
+ buf.append( ", \n" );
+ }
+ buf.append( methods[i] );
+ }
+ buf.append( "]" );
+ return buf.toString();
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/java/org/apache/drill/exec/expr/ClassGenerator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/ClassGenerator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/ClassGenerator.java
index 1f82682..96f14fb 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/ClassGenerator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/ClassGenerator.java
@@ -87,7 +87,9 @@ public class ClassGenerator<T>{
}
@SuppressWarnings("unchecked")
- ClassGenerator(CodeGenerator<T> codeGenerator, MappingSet mappingSet, SignatureHolder signature, EvaluationVisitor eval, JDefinedClass clazz, JCodeModel model, OptionManager optionManager) throws JClassAlreadyExistsException {
+ ClassGenerator(CodeGenerator<T> codeGenerator, MappingSet mappingSet, SignatureHolder signature,
+ EvaluationVisitor eval, JDefinedClass clazz, JCodeModel model,
+ OptionManager optionManager) throws JClassAlreadyExistsException {
this.codeGenerator = codeGenerator;
this.clazz = clazz;
this.mappings = mappingSet;
@@ -171,7 +173,7 @@ public class ClassGenerator<T>{
}
public JVar declareVectorValueSetupAndMember(String batchName, TypedFieldId fieldId) {
- return declareVectorValueSetupAndMember( DirectExpression.direct(batchName), fieldId);
+ return declareVectorValueSetupAndMember(DirectExpression.direct(batchName), fieldId);
}
public JVar declareVectorValueSetupAndMember(DirectExpression batchName, TypedFieldId fieldId) {
@@ -202,26 +204,30 @@ public class ClassGenerator<T>{
JInvocation invoke = batchName
.invoke("getValueAccessorById") //
- .arg( vvClass.dotclass())
+ .arg(vvClass.dotclass())
.arg(fieldArr);
- JVar obj = b.decl( //
- objClass, //
- getNextVar("tmp"), //
+ JVar obj = b.decl(
+ objClass,
+ getNextVar("tmp"),
invoke.invoke(vectorAccess));
b._if(obj.eq(JExpr._null()))._then()._throw(JExpr._new(t).arg(JExpr.lit(String.format("Failure while loading vector %s with id: %s.", vv.name(), fieldId.toString()))));
- //b.assign(vv, JExpr.cast(retClass, ((JExpression) JExpr.cast(wrapperClass, obj) ).invoke(vectorAccess)));
- b.assign(vv, JExpr.cast(retClass, obj ));
+ //b.assign(vv, JExpr.cast(retClass, ((JExpression) JExpr.cast(wrapperClass, obj)).invoke(vectorAccess)));
+ b.assign(vv, JExpr.cast(retClass, obj));
vvDeclaration.put(setup, vv);
return vv;
}
public enum BlkCreateMode {
- TRUE, // Create new block
- FALSE, // Do not create block; put into existing block.
- TRUE_IF_BOUND // Create new block only if # of expressions added hit upper-bound (ExecConstants.CODE_GEN_EXP_IN_METHOD_SIZE)
+ /** Create new block */
+ TRUE,
+ /** Do not create block; put into existing block. */
+ FALSE,
+ /** Create new block only if # of expressions added hit upper-bound
+ * ({@link ExecConstants#CODE_GEN_EXP_IN_METHOD_SIZE}). */
+ TRUE_IF_BOUND
}
public HoldingContainer addExpr(LogicalExpression ex) {
@@ -246,6 +252,13 @@ public class ClassGenerator<T>{
rotateBlock(BlkCreateMode.TRUE);
}
+ /**
+ * Create a new code block, closing the current block.
+ *
+ * @param mode the {@link BlkCreateMode block create mode}
+ * for the new block.
+ */
+
private void rotateBlock(BlkCreateMode mode) {
boolean blockRotated = false;
for (LinkedList<SizedJBlock> b : blocks) {
@@ -361,7 +374,7 @@ public class ClassGenerator<T>{
return this.workspaceVectors;
}
- private static class ValueVectorSetup{
+ private static class ValueVectorSetup {
final DirectExpression batch;
final TypedFieldId fieldId;
@@ -411,7 +424,11 @@ public class ClassGenerator<T>{
}
- public static class HoldingContainer{
+ /**
+ * Represents a (Nullable)?(Type)Holder instance.
+ */
+
+ public static class HoldingContainer {
private final JVar holder;
private final JFieldRef value;
private final JFieldRef isSet;
@@ -483,10 +500,33 @@ public class ClassGenerator<T>{
public TypeProtos.MinorType getMinorType() {
return type.getMinorType();
}
+
+ /**
+ * Convert holder to a string for debugging use.
+ */
+
+ @Override
+ public String toString() {
+ DebugStringBuilder buf = new DebugStringBuilder(this);
+ if (isConstant()) {
+ buf.append("const ");
+ }
+ buf.append(holder.type().fullName())
+ .append(" ")
+ .append(holder.name())
+ .append(", ")
+ .append(type.getMode().name())
+ .append(" ")
+ .append(type.getMinorType().name())
+ .append(", ");
+ holder.generate(buf.formatter());
+ buf.append(", ");
+ value.generate(buf.formatter());
+ return buf.toString();
+ }
}
public JType getHolderType(MajorType t) {
return TypeHelper.getHolderType(model, t.getMinorType(), t.getMode());
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/java/org/apache/drill/exec/expr/CodeGenerator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/CodeGenerator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/CodeGenerator.java
index 12178ee..f50cfde 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/CodeGenerator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/CodeGenerator.java
@@ -30,16 +30,30 @@ import com.sun.codemodel.JDefinedClass;
import org.apache.drill.exec.server.options.OptionManager;
/**
- * A code generator is responsible for generating the Java source code required to complete the implementation of an
- * abstract template. It is used with a class transformer to merge precompiled template code with runtime generated and
+ * A code generator is responsible for generating the Java source code required
+ * to complete the implementation of an abstract template.
+ * A code generator can contain one or more ClassGenerators that implement
+ * outer and inner classes associated with a particular runtime generated instance.
+ * <p>
+ * Drill supports two ways to generate and compile the code from a code
+ * generator: via byte-code manipulations or as "plain-old Java."
+ * <p>
+ * When using byte-code transformations, the code generator is used with a
+ * class transformer to merge precompiled template code with runtime generated and
* compiled query specific code to create a runtime instance.
- *
- * A code generator can contain one or more ClassGenerators that implement outer and inner classes associated with a
- * particular runtime generated instance.
+ * <p>
+ * The code generator can optionally be marked as "plain-old Java" capable.
+ * This means that the generated code can be compiled directly as a Java
+ * class without the normal byte-code manipulations. Plain-old Java allows
+ * the option to persist, and debug, the generated code when building new
+ * generated classes or otherwise working with generated code. To turn
+ * on debugging, see the explanation in {@link ClassBuilder}.
*
* @param <T>
- * The interface that results from compiling and merging the runtime code that is generated.
+ * The interface that results from compiling and merging the runtime
+ * code that is generated.
*/
+
public class CodeGenerator<T> {
private static final String PACKAGE_NAME = "org.apache.drill.exec.test.generated";
@@ -50,6 +64,21 @@ public class CodeGenerator<T> {
private final JCodeModel model;
private final ClassGenerator<T> rootGenerator;
+
+ /**
+ * True if the code generated for this class is suitable for compilation
+ * as a plain-old Java class.
+ */
+
+ private boolean plainOldJavaCapable;
+
+ /**
+ * True if the code generated for this class should actually be compiled
+ * via the plain-old Java mechanism. Considered only if the class is
+ * capable of this technique.
+ */
+
+ private boolean usePlainOldJava;
private String generatedCode;
private String generifiedCode;
@@ -58,7 +87,7 @@ public class CodeGenerator<T> {
}
CodeGenerator(MappingSet mappingSet, TemplateClassDefinition<T> definition,
- FunctionImplementationRegistry funcRegistry, OptionManager optionManager) {
+ FunctionImplementationRegistry funcRegistry, OptionManager optionManager) {
Preconditions.checkNotNull(definition.getSignature(),
"The signature for defintion %s was incorrectly initialized.", definition);
this.definition = definition;
@@ -67,6 +96,9 @@ public class CodeGenerator<T> {
try {
this.model = new JCodeModel();
JDefinedClass clazz = model._package(PACKAGE_NAME)._class(className);
+ if ( isPlainOldJava( ) ) {
+ clazz._extends(definition.getTemplateClass( ) );
+ }
rootGenerator = new ClassGenerator<>(this, mappingSet, definition.getSignature(), new EvaluationVisitor(
funcRegistry), clazz, model, optionManager);
} catch (JClassAlreadyExistsException e) {
@@ -74,19 +106,67 @@ public class CodeGenerator<T> {
}
}
+ /**
+ * Indicates that the code for this class can be generated using the
+ * "Plain Old Java" mechanism based on inheritance. The byte-code
+ * method is more lenient, so some code is missing some features such
+ * as proper exception labeling, etc. Set this option to true once
+ * the generation mechanism for a class has been cleaned up to work
+ * via the plain-old Java mechanism.
+ *
+ * @param flag true if the code generated from this instance is
+ * ready to be compiled as a plain-old Java class
+ */
+
+ public void plainOldJavaCapable(boolean flag) {
+ plainOldJavaCapable = flag;
+ }
+
+ /**
+ * Identifies that this generated class should be generated via the
+ * plain-old Java mechanism. This flag only has meaning if the
+ * generated class is capable of plain-old Java generation.
+ *
+ * @param flag true if the class should be generated and compiled
+ * as a plain-old Java class (rather than via byte-code manipulations)
+ */
+
+ public void preferPlainOldJava(boolean flag) {
+ usePlainOldJava = flag;
+ }
+
+ public boolean isPlainOldJava() {
+ return plainOldJavaCapable && usePlainOldJava;
+ }
+
public ClassGenerator<T> getRoot() {
return rootGenerator;
}
- public void generate() throws IOException {
+ public void generate() {
+
+ // If this generated class uses the "straight Java" technique
+ // (no byte code manipulation), then the class must extend the
+ // template so it plays by normal Java rules for finding the
+ // template methods via inheritance rather than via code injection.
+
+ if (isPlainOldJava()) {
+ rootGenerator.clazz._extends(definition.getTemplateClass( ));
+ }
+
rootGenerator.flushCode();
SingleClassStringWriter w = new SingleClassStringWriter();
- model.build(w);
+ try {
+ model.build(w);
+ } catch (IOException e) {
+ // No I/O errors should occur during model building
+ // unless something is terribly wrong.
+ throw new IllegalStateException(e);
+ }
this.generatedCode = w.getCode().toString();
this.generifiedCode = generatedCode.replaceAll(this.className, "GenericGenerated");
-
}
public String generateAndGet() throws IOException {
@@ -156,7 +236,7 @@ public class CodeGenerator<T> {
if (other.definition != null){
return false;
}
- } else if (!definition.equals(other.definition)){
+ } else if (!definition.equals(other.definition)) {
return false;
}
if (generifiedCode == null) {
@@ -164,7 +244,7 @@ public class CodeGenerator<T> {
return false;
}
- } else if (!generifiedCode.equals(other.generifiedCode)){
+ } else if (!generifiedCode.equals(other.generifiedCode)) {
return false;
}
return true;
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/java/org/apache/drill/exec/expr/DebugStringBuilder.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/DebugStringBuilder.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/DebugStringBuilder.java
new file mode 100644
index 0000000..057c621
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/DebugStringBuilder.java
@@ -0,0 +1,62 @@
+/**
+ * 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.expr;
+
+import java.io.PrintWriter;
+import java.io.StringWriter;
+
+import com.sun.codemodel.JFormatter;
+
+/**
+ * Utility class to build a debug string for an object
+ * in a standard format. That format is:
+ * <pre>[<i>className</i>:
+ * <i>variable=<value>... ]</pre>
+ */
+
+public class DebugStringBuilder {
+
+ private final StringWriter strWriter;
+ private final PrintWriter writer;
+ private final JFormatter fmt;
+
+ public DebugStringBuilder( Object obj ) {
+ strWriter = new StringWriter( );
+ writer = new PrintWriter( strWriter );
+ writer.print( "[" );
+ writer.print( obj.getClass().getSimpleName() );
+ writer.print( ": " );
+ fmt = new JFormatter( writer );
+ }
+
+ public DebugStringBuilder append( String s ) {
+ writer.print( s );
+ return this;
+ }
+
+ @Override
+ public String toString( ) {
+ writer.print( "]" );
+ writer.flush();
+ return strWriter.toString();
+ }
+
+ public JFormatter formatter() { return fmt; }
+ public PrintWriter writer() { return writer; }
+
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/java/org/apache/drill/exec/expr/DirectExpression.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/DirectExpression.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/DirectExpression.java
index c4c3e7a..b99cd13 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/DirectExpression.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/DirectExpression.java
@@ -20,17 +20,24 @@ package org.apache.drill.exec.expr;
import com.sun.codemodel.JExpressionImpl;
import com.sun.codemodel.JFormatter;
-public class DirectExpression extends JExpressionImpl{
+/**
+ * Encapsulates a Java expression, defined as anything that is
+ * valid in the following code:<br>
+ * <code>(<i>expr</i>)</code>
+ */
+
+public class DirectExpression extends JExpressionImpl {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DirectExpression.class);
- final String source;
+ private final String source;
private DirectExpression(final String source) {
super();
this.source = source;
}
- public void generate( JFormatter f ) {
+ @Override
+ public void generate(JFormatter f) {
f.p('(').p(source).p(')');
}
@@ -67,5 +74,4 @@ public class DirectExpression extends JExpressionImpl{
}
return true;
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/java/org/apache/drill/exec/ops/FragmentContext.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/ops/FragmentContext.java b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/FragmentContext.java
index 8229b5b..e288095 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/ops/FragmentContext.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/FragmentContext.java
@@ -313,7 +313,7 @@ public class FragmentContext implements AutoCloseable, UdfUtilities {
public <T> T getImplementationClass(final CodeGenerator<T> cg)
throws ClassTransformationException, IOException {
- return context.getCompiler().getImplementationClass(cg);
+ return context.getCompiler().createInstance(cg);
}
public <T> List<T> getImplementationClass(final ClassGenerator<T> cg, final int instanceCount) throws ClassTransformationException, IOException {
@@ -321,7 +321,7 @@ public class FragmentContext implements AutoCloseable, UdfUtilities {
}
public <T> List<T> getImplementationClass(final CodeGenerator<T> cg, final int instanceCount) throws ClassTransformationException, IOException {
- return context.getCompiler().getImplementationClass(cg, instanceCount);
+ return context.getCompiler().createInstances(cg, instanceCount);
}
public AccountingUserConnection getUserDataTunnel() {
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/java/org/apache/drill/exec/ops/OperatorContext.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/ops/OperatorContext.java b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/OperatorContext.java
index 92a7269..d6045fc 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/ops/OperatorContext.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/OperatorContext.java
@@ -17,22 +17,22 @@
*/
package org.apache.drill.exec.ops;
-import com.google.common.util.concurrent.ListenableFuture;
-import io.netty.buffer.DrillBuf;
-
import java.io.IOException;
import java.util.Iterator;
import java.util.concurrent.Callable;
-import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.physical.base.PhysicalOperator;
-import org.apache.drill.exec.testing.ExecutionControls;
import org.apache.drill.exec.store.dfs.DrillFileSystem;
+import org.apache.drill.exec.testing.ExecutionControls;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.UserGroupInformation;
+import com.google.common.util.concurrent.ListenableFuture;
+
+import io.netty.buffer.DrillBuf;
+
public abstract class OperatorContext {
public abstract DrillBuf replace(DrillBuf old, int newSize);
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/java/org/apache/drill/exec/ops/OperatorContextImpl.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/ops/OperatorContextImpl.java b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/OperatorContextImpl.java
index 390b71c..c19cc1f 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/ops/OperatorContextImpl.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/OperatorContextImpl.java
@@ -89,33 +89,41 @@ class OperatorContextImpl extends OperatorContext implements AutoCloseable {
scanDecodeExecutor = context.getDrillbitContext().getScanDecodeExecutor();
}
+ @Override
public DrillBuf replace(DrillBuf old, int newSize) {
return manager.replace(old, newSize);
}
+ @Override
public DrillBuf getManagedBuffer() {
return manager.getManagedBuffer();
}
+ @Override
public DrillBuf getManagedBuffer(int size) {
return manager.getManagedBuffer(size);
}
// Allow an operator to use the thread pool
+ @Override
public ExecutorService getExecutor() {
return executor;
}
+ @Override
public ExecutorService getScanExecutor() {
return scanExecutor;
}
+ @Override
public ExecutorService getScanDecodeExecutor() {
return scanDecodeExecutor;
}
+ @Override
public ExecutionControls getExecutionControls() {
return executionControls;
}
+ @Override
public BufferAllocator getAllocator() {
if (allocator == null) {
throw new UnsupportedOperationException("Operator context does not have an allocator");
@@ -151,10 +159,12 @@ class OperatorContextImpl extends OperatorContext implements AutoCloseable {
closed = true;
}
+ @Override
public OperatorStats getStats() {
return stats;
}
+ @Override
public <RESULT> ListenableFuture<RESULT> runCallableAs(final UserGroupInformation proxyUgi,
final Callable<RESULT> callable) {
synchronized (this) {
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/ExternalSortBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/ExternalSortBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/ExternalSortBatch.java
index 54aa72d..95d64bd 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/ExternalSortBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/ExternalSortBatch.java
@@ -711,18 +711,20 @@ public class ExternalSortBatch extends AbstractRecordBatch<ExternalSort> {
g.rotateBlock();
g.getEvalBlock()._return(JExpr.lit(0));
+ cg.plainOldJavaCapable(true); // This class can generate plain-old Java.
+ // Uncomment out this line to debug the generated code.
+// cg.preferPlainOldJava(true);
return context.getImplementationClass(cg);
-
-
}
public SingleBatchSorter createNewSorter(FragmentContext context, VectorAccessible batch)
throws ClassTransformationException, IOException, SchemaChangeException{
CodeGenerator<SingleBatchSorter> cg = CodeGenerator.get(SingleBatchSorter.TEMPLATE_DEFINITION, context.getFunctionRegistry(), context.getOptions());
- ClassGenerator<SingleBatchSorter> g = cg.getRoot();
-
- generateComparisons(g, batch);
+ cg.plainOldJavaCapable(true); // This class can generate plain-old Java.
+ // Uncomment out this line to debug the generated code.
+// cg.preferPlainOldJava(true);
+ generateComparisons(cg.getRoot(), batch);
return context.getImplementationClass(cg);
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/SingleBatchSorter.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/SingleBatchSorter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/SingleBatchSorter.java
index b4986ba..e59d1b1 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/SingleBatchSorter.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/SingleBatchSorter.java
@@ -26,7 +26,7 @@ import org.apache.drill.exec.record.selection.SelectionVector2;
public interface SingleBatchSorter {
public void setup(FragmentContext context, SelectionVector2 vector2, VectorAccessible incoming) throws SchemaChangeException;
- public void sort(SelectionVector2 vector2);
+ public void sort(SelectionVector2 vector2) throws SchemaChangeException;
public static TemplateClassDefinition<SingleBatchSorter> TEMPLATE_DEFINITION = new TemplateClassDefinition<SingleBatchSorter>(SingleBatchSorter.class, SingleBatchSorterTemplate.class);
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/SingleBatchSorterTemplate.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/SingleBatchSorterTemplate.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/SingleBatchSorterTemplate.java
index 9a6bc8c..5a8b0c3 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/SingleBatchSorterTemplate.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/SingleBatchSorterTemplate.java
@@ -37,6 +37,7 @@ public abstract class SingleBatchSorterTemplate implements SingleBatchSorter, In
private SelectionVector2 vector2;
+ @Override
public void setup(FragmentContext context, SelectionVector2 vector2, VectorAccessible incoming) throws SchemaChangeException{
Preconditions.checkNotNull(vector2);
this.vector2 = vector2;
@@ -68,10 +69,14 @@ public abstract class SingleBatchSorterTemplate implements SingleBatchSorter, In
public int compare(int leftIndex, int rightIndex) {
char sv1 = vector2.getIndex(leftIndex);
char sv2 = vector2.getIndex(rightIndex);
- return doEval(sv1, sv2);
+ try {
+ return doEval(sv1, sv2);
+ } catch (SchemaChangeException e) {
+ throw new RuntimeException( e );
+ }
}
- public abstract void doSetup(@Named("context") FragmentContext context, @Named("incoming") VectorAccessible incoming, @Named("outgoing") RecordBatch outgoing);
- public abstract int doEval(@Named("leftIndex") char leftIndex, @Named("rightIndex") char rightIndex);
+ public abstract void doSetup(@Named("context") FragmentContext context, @Named("incoming") VectorAccessible incoming, @Named("outgoing") RecordBatch outgoing) throws SchemaChangeException;
+ public abstract int doEval(@Named("leftIndex") char leftIndex, @Named("rightIndex") char rightIndex) throws SchemaChangeException;
}
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/SystemOptionManager.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/SystemOptionManager.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/SystemOptionManager.java
index 2c322c7..57c72d5 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/SystemOptionManager.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/SystemOptionManager.java
@@ -32,8 +32,8 @@ import org.apache.drill.common.config.LogicalPlanPersistence;
import org.apache.drill.common.map.CaseInsensitiveMap;
import org.apache.drill.common.exceptions.UserException;
import org.apache.drill.exec.ExecConstants;
+import org.apache.drill.exec.compile.ClassCompilerSelector;
import org.apache.drill.exec.compile.ClassTransformer;
-import org.apache.drill.exec.compile.QueryClassLoader;
import org.apache.drill.exec.planner.physical.PlannerSettings;
import org.apache.drill.exec.server.options.OptionValue.OptionType;
import org.apache.drill.exec.server.options.TypeValidators.BooleanValidator;
@@ -145,9 +145,9 @@ public class SystemOptionManager extends BaseOptionManager implements AutoClosea
ExecConstants.ADMIN_USERS_VALIDATOR,
ExecConstants.ADMIN_USER_GROUPS_VALIDATOR,
ExecConstants.IMPERSONATION_POLICY_VALIDATOR,
- QueryClassLoader.JAVA_COMPILER_VALIDATOR,
- QueryClassLoader.JAVA_COMPILER_JANINO_MAXSIZE,
- QueryClassLoader.JAVA_COMPILER_DEBUG,
+ ClassCompilerSelector.JAVA_COMPILER_VALIDATOR,
+ ClassCompilerSelector.JAVA_COMPILER_JANINO_MAXSIZE,
+ ClassCompilerSelector.JAVA_COMPILER_DEBUG,
ExecConstants.ENABLE_VERBOSE_ERRORS,
ExecConstants.ENABLE_WINDOW_FUNCTIONS_VALIDATOR,
ClassTransformer.SCALAR_REPLACEMENT_VALIDATOR,
http://git-wip-us.apache.org/repos/asf/drill/blob/bbcf4b76/exec/java-exec/src/main/resources/drill-module.conf
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/resources/drill-module.conf b/exec/java-exec/src/main/resources/drill-module.conf
index ab73cec..7e6d7c6 100644
--- a/exec/java-exec/src/main/resources/drill-module.conf
+++ b/exec/java-exec/src/main/resources/drill-module.conf
@@ -166,7 +166,11 @@ drill.exec: {
compiler: "DEFAULT",
debug: true,
janino_maxsize: 262144,
- cache_max_size: 1000
+ cache_max_size: 1000,
+ // Enable to write generated source to disk. See ClassBuilder
+ save_source: false,
+ // Where to save the generated source. See ClassBuilder
+ code_dir: "/tmp/drill/codegen"
},
sort: {
purge.threshold : 1000,
[6/8] drill git commit: DRILL-5032: Drill query on hive parquet table
failed with OutOfMemoryError: Java heap space
Posted by jn...@apache.org.
DRILL-5032: Drill query on hive parquet table failed with OutOfMemoryError: Java heap space
close apache/drill#654
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/03928af0
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/03928af0
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/03928af0
Branch: refs/heads/master
Commit: 03928af0b5cafd52e5b153aa852e5642b505f2c6
Parents: d8cc710
Author: Serhii-Harnyk <se...@gmail.com>
Authored: Thu Oct 27 19:20:27 2016 +0000
Committer: Jinfeng Ni <jn...@apache.org>
Committed: Mon Dec 19 15:57:36 2016 -0800
----------------------------------------------------------------------
.../codegen/templates/HiveRecordReaders.java | 4 +-
.../planner/sql/HivePartitionDescriptor.java | 12 +-
...onvertHiveParquetScanToDrillParquetScan.java | 14 +-
.../drill/exec/store/hive/ColumnListsCache.java | 95 ++++
.../store/hive/DrillHiveMetaStoreClient.java | 40 +-
.../exec/store/hive/HiveAbstractReader.java | 15 +-
.../store/hive/HiveDrillNativeParquetScan.java | 4 +-
.../hive/HiveDrillNativeScanBatchCreator.java | 6 +-
.../exec/store/hive/HiveMetadataProvider.java | 11 +-
.../drill/exec/store/hive/HivePartition.java | 61 +++
.../drill/exec/store/hive/HiveReadEntry.java | 31 +-
.../apache/drill/exec/store/hive/HiveScan.java | 17 +-
.../exec/store/hive/HiveScanBatchCreator.java | 8 +-
.../drill/exec/store/hive/HiveSubScan.java | 15 +-
.../apache/drill/exec/store/hive/HiveTable.java | 382 ---------------
.../store/hive/HiveTableWithColumnCache.java | 76 +++
.../drill/exec/store/hive/HiveTableWrapper.java | 466 +++++++++++++++++++
.../drill/exec/store/hive/HiveUtilities.java | 41 +-
.../exec/store/hive/schema/DrillHiveTable.java | 7 +-
.../drill/exec/TestHivePartitionPruning.java | 29 +-
.../exec/hive/TestInfoSchemaOnHiveStorage.java | 2 +
.../exec/store/hive/HiveTestDataGenerator.java | 9 +
.../store/hive/schema/TestColumnListCache.java | 111 +++++
23 files changed, 980 insertions(+), 476 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/03928af0/contrib/storage-hive/core/src/main/codegen/templates/HiveRecordReaders.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/codegen/templates/HiveRecordReaders.java b/contrib/storage-hive/core/src/main/codegen/templates/HiveRecordReaders.java
index 0dc8c08..7d6733e 100644
--- a/contrib/storage-hive/core/src/main/codegen/templates/HiveRecordReaders.java
+++ b/contrib/storage-hive/core/src/main/codegen/templates/HiveRecordReaders.java
@@ -43,8 +43,6 @@ import org.apache.drill.common.expression.SchemaPath;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.vector.AllocationHelper;
import org.apache.drill.exec.vector.ValueVector;
-import org.apache.hadoop.hive.metastore.api.Partition;
-import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.security.UserGroupInformation;
@@ -75,7 +73,7 @@ public class Hive${entry.hiveReader}Reader extends HiveAbstractReader {
Object value;
</#if>
- public Hive${entry.hiveReader}Reader(Table table, Partition partition, InputSplit inputSplit, List<SchemaPath> projectedColumns,
+ public Hive${entry.hiveReader}Reader(HiveTableWithColumnCache table, HivePartition partition, InputSplit inputSplit, List<SchemaPath> projectedColumns,
FragmentContext context, final HiveConf hiveConf,
UserGroupInformation proxyUgi) throws ExecutionSetupException {
super(table, partition, inputSplit, projectedColumns, context, hiveConf, proxyUgi);
http://git-wip-us.apache.org/repos/asf/drill/blob/03928af0/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/HivePartitionDescriptor.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/HivePartitionDescriptor.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/HivePartitionDescriptor.java
index d42aea7..b22c14d 100644
--- a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/HivePartitionDescriptor.java
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/HivePartitionDescriptor.java
@@ -31,10 +31,10 @@ import org.apache.drill.exec.planner.PartitionLocation;
import org.apache.drill.exec.planner.logical.DrillRel;
import org.apache.drill.exec.planner.logical.DrillScanRel;
import org.apache.drill.exec.planner.physical.PlannerSettings;
+import org.apache.drill.exec.store.hive.HiveTableWrapper;
import org.apache.drill.exec.store.hive.HiveUtilities;
import org.apache.drill.exec.store.hive.HiveReadEntry;
import org.apache.drill.exec.store.hive.HiveScan;
-import org.apache.drill.exec.store.hive.HiveTable;
import org.apache.drill.exec.vector.ValueVector;
import org.apache.hadoop.hive.metastore.api.Partition;
@@ -64,7 +64,7 @@ public class HivePartitionDescriptor extends AbstractPartitionDescriptor {
this.scanRel = scanRel;
this.managedBuffer = managedBuffer.reallocIfNeeded(256);
this.defaultPartitionValue = defaultPartitionValue;
- for (HiveTable.FieldSchemaWrapper wrapper : ((HiveScan) scanRel.getGroupScan()).hiveReadEntry.table.partitionKeys) {
+ for (HiveTableWrapper.FieldSchemaWrapper wrapper : ((HiveScan) scanRel.getGroupScan()).hiveReadEntry.table.partitionKeys) {
partitionMap.put(wrapper.name, i);
i++;
}
@@ -115,7 +115,7 @@ public class HivePartitionDescriptor extends AbstractPartitionDescriptor {
}
for(ValueVector v : vectors) {
- if(v == null){
+ if (v == null) {
continue;
}
v.getMutator().setValueCount(partitions.size());
@@ -166,10 +166,10 @@ public class HivePartitionDescriptor extends AbstractPartitionDescriptor {
private GroupScan createNewGroupScan(List<PartitionLocation> newPartitionLocations) throws ExecutionSetupException {
HiveScan hiveScan = (HiveScan) scanRel.getGroupScan();
HiveReadEntry origReadEntry = hiveScan.hiveReadEntry;
- List<HiveTable.HivePartition> oldPartitions = origReadEntry.partitions;
- List<HiveTable.HivePartition> newPartitions = new LinkedList<>();
+ List<HiveTableWrapper.HivePartitionWrapper> oldPartitions = origReadEntry.partitions;
+ List<HiveTableWrapper.HivePartitionWrapper> newPartitions = Lists.newLinkedList();
- for (HiveTable.HivePartition part: oldPartitions) {
+ for (HiveTableWrapper.HivePartitionWrapper part: oldPartitions) {
String partitionLocation = part.getPartition().getSd().getLocation();
for (PartitionLocation newPartitionLocation: newPartitionLocations) {
if (partitionLocation.equals(newPartitionLocation.getEntirePartitionLocation())) {
http://git-wip-us.apache.org/repos/asf/drill/blob/03928af0/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/logical/ConvertHiveParquetScanToDrillParquetScan.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/logical/ConvertHiveParquetScanToDrillParquetScan.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/logical/ConvertHiveParquetScanToDrillParquetScan.java
index 228308f..df85ca0 100644
--- a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/logical/ConvertHiveParquetScanToDrillParquetScan.java
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/logical/ConvertHiveParquetScanToDrillParquetScan.java
@@ -39,10 +39,10 @@ import org.apache.drill.exec.store.StoragePluginOptimizerRule;
import org.apache.drill.exec.store.hive.HiveDrillNativeParquetScan;
import org.apache.drill.exec.store.hive.HiveReadEntry;
import org.apache.drill.exec.store.hive.HiveScan;
-import org.apache.drill.exec.store.hive.HiveTable.HivePartition;
+import org.apache.drill.exec.store.hive.HiveTableWithColumnCache;
+import org.apache.drill.exec.store.hive.HiveTableWrapper.HivePartitionWrapper;
import org.apache.drill.exec.store.hive.HiveUtilities;
import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.metastore.MetaStoreUtils;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
import org.apache.hadoop.hive.metastore.api.Table;
@@ -97,23 +97,23 @@ public class ConvertHiveParquetScanToDrillParquetScan extends StoragePluginOptim
final HiveScan hiveScan = (HiveScan) scanRel.getGroupScan();
final HiveConf hiveConf = hiveScan.getHiveConf();
- final Table hiveTable = hiveScan.hiveReadEntry.getTable();
+ final HiveTableWithColumnCache hiveTable = hiveScan.hiveReadEntry.getTable();
final Class<? extends InputFormat<?,?>> tableInputFormat =
- getInputFormatFromSD(MetaStoreUtils.getTableMetadata(hiveTable), hiveScan.hiveReadEntry, hiveTable.getSd(),
+ getInputFormatFromSD(HiveUtilities.getTableMetadata(hiveTable), hiveScan.hiveReadEntry, hiveTable.getSd(),
hiveConf);
if (tableInputFormat == null || !tableInputFormat.equals(MapredParquetInputFormat.class)) {
return false;
}
- final List<HivePartition> partitions = hiveScan.hiveReadEntry.getHivePartitionWrappers();
+ final List<HivePartitionWrapper> partitions = hiveScan.hiveReadEntry.getHivePartitionWrappers();
if (partitions == null) {
return true;
}
final List<FieldSchema> tableSchema = hiveTable.getSd().getCols();
// Make sure all partitions have the same input format as the table input format
- for (HivePartition partition : partitions) {
+ for (HivePartitionWrapper partition : partitions) {
final StorageDescriptor partitionSD = partition.getPartition().getSd();
Class<? extends InputFormat<?, ?>> inputFormat = getInputFormatFromSD(
HiveUtilities.getPartitionMetadata(partition.getPartition(), hiveTable), hiveScan.hiveReadEntry, partitionSD,
@@ -179,7 +179,7 @@ public class ConvertHiveParquetScanToDrillParquetScan extends StoragePluginOptim
getPartitionColMapping(hiveTable, partitionColumnLabel);
final DrillScanRel nativeScanRel = createNativeScanRel(partitionColMapping, hiveScanRel);
- if(hiveScanRel.getRowType().getFieldCount() == 0) {
+ if (hiveScanRel.getRowType().getFieldCount() == 0) {
call.transformTo(nativeScanRel);
} else {
final DrillProjectRel projectRel = createProjectRel(hiveScanRel, partitionColMapping, nativeScanRel);
http://git-wip-us.apache.org/repos/asf/drill/blob/03928af0/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/ColumnListsCache.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/ColumnListsCache.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/ColumnListsCache.java
new file mode 100644
index 0000000..ae4baa1
--- /dev/null
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/ColumnListsCache.java
@@ -0,0 +1,95 @@
+/*
+* 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.store.hive;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.Table;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * The class represents "cache" for partition and table columns.
+ * Used to reduce physical plan for Hive tables.
+ * Only unique partition lists of columns stored in the column lists cache.
+ * Table columns should be stored at index 0.
+ */
+public class ColumnListsCache {
+ // contains immutable column lists
+ private final List<List<FieldSchema>> fields;
+
+ // keys of the map are column lists and values are them positions in list fields
+ private final Map<List<FieldSchema>, Integer> keys;
+
+ public ColumnListsCache(Table table) {
+ this();
+ // table columns stored at index 0.
+ addOrGet(table.getSd().getCols());
+ }
+
+ public ColumnListsCache() {
+ this.fields = Lists.newArrayList();
+ this.keys = Maps.newHashMap();
+ }
+
+ /**
+ * Checks if column list has been added before and returns position of column list.
+ * If list is unique, adds list to the fields list and returns it position.
+ * Returns -1, if {@param columns} equals null.
+ *
+ * @param columns list of columns
+ * @return index of {@param columns} or -1, if {@param columns} equals null
+ */
+ public int addOrGet(List<FieldSchema> columns) {
+ if (columns == null) {
+ return -1;
+ }
+ Integer index = keys.get(columns);
+ if (index != null) {
+ return index;
+ } else {
+ index = fields.size();
+ final List<FieldSchema> immutableList = ImmutableList.copyOf(columns);
+ fields.add(immutableList);
+ keys.put(immutableList, index);
+ return index;
+ }
+ }
+
+ /**
+ * Returns list of columns at the specified position in fields list,
+ * or null if index is negative or greater than fields list size.
+ *
+ * @param index index of column list to return
+ * @return list of columns at the specified position in fields list
+ * or null if index is negative or greater than fields list size
+ */
+ public List<FieldSchema> getColumns(int index) {
+ if (index >= 0 && index < fields.size()) {
+ return fields.get(index);
+ } else {
+ return null;
+ }
+ }
+
+ public List<List<FieldSchema>> getFields() {
+ return Lists.newArrayList(fields);
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/03928af0/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/DrillHiveMetaStoreClient.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/DrillHiveMetaStoreClient.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/DrillHiveMetaStoreClient.java
index d7ba659..92efdc7 100644
--- a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/DrillHiveMetaStoreClient.java
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/DrillHiveMetaStoreClient.java
@@ -23,8 +23,6 @@ import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
-import org.apache.calcite.schema.Schema;
-import org.apache.commons.lang3.tuple.Pair;
import org.apache.drill.common.exceptions.DrillRuntimeException;
import org.apache.drill.common.exceptions.UserException;
import org.apache.drill.exec.util.ImpersonationUtil;
@@ -270,9 +268,9 @@ public abstract class DrillHiveMetaStoreClient extends HiveMetaStoreClient {
/** Helper method which gets table metadata. Retries once if the first call to fetch the metadata fails */
protected static HiveReadEntry getHiveReadEntryHelper(final IMetaStoreClient mClient, final String dbName,
final String tableName) throws TException {
- Table t = null;
+ Table table = null;
try {
- t = mClient.getTable(dbName, tableName);
+ table = mClient.getTable(dbName, tableName);
} catch (MetaException | NoSuchObjectException e) {
throw e;
} catch (TException e) {
@@ -283,10 +281,10 @@ public abstract class DrillHiveMetaStoreClient extends HiveMetaStoreClient {
logger.warn("Failure while attempting to close existing hive metastore connection. May leak connection.", ex);
}
mClient.reconnect();
- t = mClient.getTable(dbName, tableName);
+ table = mClient.getTable(dbName, tableName);
}
- if (t == null) {
+ if (table == null) {
throw new UnknownTableException(String.format("Unable to find table '%s'.", tableName));
}
@@ -306,16 +304,34 @@ public abstract class DrillHiveMetaStoreClient extends HiveMetaStoreClient {
partitions = mClient.listPartitions(dbName, tableName, (short) -1);
}
- List<HiveTable.HivePartition> hivePartitions = Lists.newArrayList();
- for (Partition part : partitions) {
- hivePartitions.add(new HiveTable.HivePartition(part));
+ List<HiveTableWrapper.HivePartitionWrapper> hivePartitionWrappers = Lists.newArrayList();
+ HiveTableWithColumnCache hiveTable = new HiveTableWithColumnCache(table, new ColumnListsCache(table));
+ for (Partition partition : partitions) {
+ hivePartitionWrappers.add(createPartitionWithSpecColumns(hiveTable, partition));
}
- if (hivePartitions.size() == 0) {
- hivePartitions = null;
+ if (hivePartitionWrappers.isEmpty()) {
+ hivePartitionWrappers = null;
}
- return new HiveReadEntry(new HiveTable(t), hivePartitions);
+ return new HiveReadEntry(new HiveTableWrapper(hiveTable), hivePartitionWrappers);
+ }
+
+ /**
+ * Helper method which stores partition columns in table columnListCache. If table columnListCache has exactly the
+ * same columns as partition, in partition stores columns index that corresponds to identical column list.
+ * If table columnListCache hasn't such column list, the column list adds to table columnListCache and in partition
+ * stores columns index that corresponds to column list.
+ *
+ * @param table hive table instance
+ * @param partition partition instance
+ * @return hive partition wrapper
+ */
+ public static HiveTableWrapper.HivePartitionWrapper createPartitionWithSpecColumns(HiveTableWithColumnCache table, Partition partition) {
+ int listIndex = table.getColumnListsCache().addOrGet(partition.getSd().getCols());
+ HivePartition hivePartition = new HivePartition(partition, listIndex);
+ HiveTableWrapper.HivePartitionWrapper hivePartitionWrapper = new HiveTableWrapper.HivePartitionWrapper(hivePartition);
+ return hivePartitionWrapper;
}
/**
http://git-wip-us.apache.org/repos/asf/drill/blob/03928af0/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveAbstractReader.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveAbstractReader.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveAbstractReader.java
index 107fc66..8c6df84 100644
--- a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveAbstractReader.java
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveAbstractReader.java
@@ -39,10 +39,7 @@ import org.apache.drill.exec.vector.AllocationHelper;
import org.apache.drill.exec.vector.ValueVector;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
-import org.apache.hadoop.hive.metastore.MetaStoreUtils;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
-import org.apache.hadoop.hive.metastore.api.Partition;
-import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.hive.serde2.ColumnProjectionUtils;
import org.apache.hadoop.hive.serde2.SerDe;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
@@ -67,8 +64,8 @@ public abstract class HiveAbstractReader extends AbstractRecordReader {
protected final DrillBuf managedBuffer;
- protected Table table;
- protected Partition partition;
+ protected HiveTableWithColumnCache table;
+ protected HivePartition partition;
protected InputSplit inputSplit;
protected List<String> selectedColumnNames;
protected List<StructField> selectedStructFieldRefs = Lists.newArrayList();
@@ -106,9 +103,9 @@ public abstract class HiveAbstractReader extends AbstractRecordReader {
protected static final int TARGET_RECORD_COUNT = 4000;
- public HiveAbstractReader(Table table, Partition partition, InputSplit inputSplit, List<SchemaPath> projectedColumns,
- FragmentContext context, final HiveConf hiveConf,
- UserGroupInformation proxyUgi) throws ExecutionSetupException {
+ public HiveAbstractReader(HiveTableWithColumnCache table, HivePartition partition, InputSplit inputSplit, List<SchemaPath> projectedColumns,
+ FragmentContext context, final HiveConf hiveConf,
+ UserGroupInformation proxyUgi) throws ExecutionSetupException {
this.table = table;
this.partition = partition;
this.inputSplit = inputSplit;
@@ -130,7 +127,7 @@ public abstract class HiveAbstractReader extends AbstractRecordReader {
Properties tableProperties;
try {
- tableProperties = MetaStoreUtils.getTableMetadata(table);
+ tableProperties = HiveUtilities.getTableMetadata(table);
final Properties partitionProperties =
(partition == null) ? tableProperties :
HiveUtilities.getPartitionMetadata(partition, table);
http://git-wip-us.apache.org/repos/asf/drill/blob/03928af0/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveDrillNativeParquetScan.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveDrillNativeParquetScan.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveDrillNativeParquetScan.java
index 17cae22..ccec61a 100644
--- a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveDrillNativeParquetScan.java
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveDrillNativeParquetScan.java
@@ -28,7 +28,7 @@ import org.apache.drill.exec.physical.base.PhysicalOperator;
import org.apache.drill.exec.physical.base.ScanStats;
import org.apache.drill.exec.physical.base.SubScan;
import org.apache.drill.exec.store.StoragePluginRegistry;
-import org.apache.drill.exec.store.hive.HiveTable.HivePartition;
+import org.apache.drill.exec.store.hive.HiveTableWrapper.HivePartitionWrapper;
import java.io.IOException;
import java.util.List;
@@ -103,7 +103,7 @@ public class HiveDrillNativeParquetScan extends HiveScan {
@Override
public String toString() {
- final List<HivePartition> partitions = hiveReadEntry.getHivePartitionWrappers();
+ final List<HivePartitionWrapper> partitions = hiveReadEntry.getHivePartitionWrappers();
int numPartitions = partitions == null ? 0 : partitions.size();
return "HiveDrillNativeParquetScan [table=" + hiveReadEntry.getHiveTableWrapper()
+ ", columns=" + columns
http://git-wip-us.apache.org/repos/asf/drill/blob/03928af0/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveDrillNativeScanBatchCreator.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveDrillNativeScanBatchCreator.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveDrillNativeScanBatchCreator.java
index 42db5d0..66f41e2 100644
--- a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveDrillNativeScanBatchCreator.java
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveDrillNativeScanBatchCreator.java
@@ -42,8 +42,6 @@ import org.apache.drill.exec.util.ImpersonationUtil;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.metastore.api.Partition;
-import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.hive.ql.io.parquet.ProjectionPusher;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapred.InputSplit;
@@ -62,9 +60,9 @@ public class HiveDrillNativeScanBatchCreator implements BatchCreator<HiveDrillNa
@Override
public ScanBatch getBatch(FragmentContext context, HiveDrillNativeParquetSubScan config, List<RecordBatch> children)
throws ExecutionSetupException {
- final Table table = config.getTable();
+ final HiveTableWithColumnCache table = config.getTable();
final List<InputSplit> splits = config.getInputSplits();
- final List<Partition> partitions = config.getPartitions();
+ final List<HivePartition> partitions = config.getPartitions();
final List<SchemaPath> columns = config.getColumns();
final String partitionDesignator = context.getOptions()
.getOption(ExecConstants.FILESYSTEM_PARTITION_COLUMN_LABEL).string_val;
http://git-wip-us.apache.org/repos/asf/drill/blob/03928af0/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveMetadataProvider.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveMetadataProvider.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveMetadataProvider.java
index 49f7689..e80b37b 100644
--- a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveMetadataProvider.java
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveMetadataProvider.java
@@ -30,7 +30,6 @@ import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.MetaStoreUtils;
import org.apache.hadoop.hive.metastore.api.Partition;
import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
-import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.InputFormat;
import org.apache.hadoop.mapred.InputSplit;
@@ -83,7 +82,7 @@ public class HiveMetadataProvider {
public HiveStats getStats(final HiveReadEntry hiveReadEntry) throws IOException {
final Stopwatch timeGetStats = Stopwatch.createStarted();
- final Table table = hiveReadEntry.getTable();
+ final HiveTableWithColumnCache table = hiveReadEntry.getTable();
try {
if (!isPartitionedTable) {
final Properties properties = MetaStoreUtils.getTableMetadata(table);
@@ -96,7 +95,7 @@ public class HiveMetadataProvider {
return getStatsEstimateFromInputSplits(getTableInputSplits());
} else {
final HiveStats aggStats = new HiveStats(0, 0);
- for(Partition partition : hiveReadEntry.getPartitions()) {
+ for(HivePartition partition : hiveReadEntry.getPartitions()) {
final Properties properties = HiveUtilities.getPartitionMetadata(partition, table);
HiveStats stats = getStatsFromProps(properties);
@@ -124,7 +123,7 @@ public class HiveMetadataProvider {
return tableInputSplits;
}
- final Properties properties = MetaStoreUtils.getTableMetadata(hiveReadEntry.getTable());
+ final Properties properties = HiveUtilities.getTableMetadata(hiveReadEntry.getTable());
tableInputSplits = splitInputWithUGI(properties, hiveReadEntry.getTable().getSd(), null);
return tableInputSplits;
@@ -133,7 +132,7 @@ public class HiveMetadataProvider {
/** Helper method which returns the InputSplits for given partition. InputSplits are cached to speed up subsequent
* metadata cache requests for the same partition(s).
*/
- private List<InputSplitWrapper> getPartitionInputSplits(final Partition partition) throws Exception {
+ private List<InputSplitWrapper> getPartitionInputSplits(final HivePartition partition) throws Exception {
if (partitionInputSplitMap.containsKey(partition)) {
return partitionInputSplitMap.get(partition);
}
@@ -161,7 +160,7 @@ public class HiveMetadataProvider {
}
final List<InputSplitWrapper> splits = Lists.newArrayList();
- for (Partition p : hiveReadEntry.getPartitions()) {
+ for (HivePartition p : hiveReadEntry.getPartitions()) {
splits.addAll(getPartitionInputSplits(p));
}
return splits;
http://git-wip-us.apache.org/repos/asf/drill/blob/03928af0/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HivePartition.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HivePartition.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HivePartition.java
new file mode 100644
index 0000000..ad539b1
--- /dev/null
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HivePartition.java
@@ -0,0 +1,61 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to you under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+package org.apache.drill.exec.store.hive;
+
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * This class is wrapper of {@link Partition} class and used for
+ * storage of such additional information as index of list in column lists cache.
+ */
+public class HivePartition extends Partition {
+ // index of partition column list in the table's column list cache
+ private int columnListIndex;
+
+ public HivePartition(
+ List<String> values,
+ String dbName,
+ String tableName,
+ int createTime,
+ int lastAccessTime,
+ StorageDescriptor sd,
+ Map<String,String> parameters,
+ int columnListIndex)
+ {
+ super(values, dbName, tableName, createTime, lastAccessTime, sd, parameters);
+ this.columnListIndex = columnListIndex;
+ }
+
+ public HivePartition(Partition other, int columnListIndex) {
+ super(other);
+ this.columnListIndex = columnListIndex;
+ }
+
+ /**
+ * To reduce physical plan for Hive tables, in partitions does not stored list of columns
+ * but stored index of that list in the table's column list cache.
+ *
+ * @return index of partition column list in the table's column list cache
+ */
+ public int getColumnListIndex() {
+ return columnListIndex;
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/03928af0/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveReadEntry.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveReadEntry.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveReadEntry.java
index 4df33ec..0cf7433 100644
--- a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveReadEntry.java
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveReadEntry.java
@@ -21,9 +21,7 @@ import java.util.List;
import org.apache.calcite.schema.Schema.TableType;
-import org.apache.drill.exec.store.hive.HiveTable.HivePartition;
-import org.apache.hadoop.hive.metastore.api.Partition;
-import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.drill.exec.store.hive.HiveTableWrapper.HivePartitionWrapper;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonIgnore;
@@ -33,42 +31,47 @@ import com.google.common.collect.Lists;
public class HiveReadEntry {
@JsonProperty("table")
- public HiveTable table;
+ public HiveTableWrapper table;
@JsonProperty("partitions")
- public List<HiveTable.HivePartition> partitions;
+ public List<HivePartitionWrapper> partitions;
@JsonIgnore
- private List<Partition> partitionsUnwrapped = Lists.newArrayList();
+ private List<HivePartition> partitionsUnwrapped = Lists.newArrayList();
@JsonCreator
- public HiveReadEntry(@JsonProperty("table") HiveTable table,
- @JsonProperty("partitions") List<HiveTable.HivePartition> partitions) {
+ public HiveReadEntry(@JsonProperty("table") HiveTableWrapper table,
+ @JsonProperty("partitions") List<HivePartitionWrapper> partitions) {
this.table = table;
this.partitions = partitions;
if (partitions != null) {
- for(HiveTable.HivePartition part : partitions) {
+ for(HivePartitionWrapper part : partitions) {
partitionsUnwrapped.add(part.getPartition());
}
}
}
@JsonIgnore
- public Table getTable() {
+ public HiveTableWithColumnCache getTable() {
return table.getTable();
}
@JsonIgnore
- public List<Partition> getPartitions() {
+ public HiveTableWrapper getTableWrapper() {
+ return table;
+ }
+
+ @JsonIgnore
+ public List<HivePartition> getPartitions() {
return partitionsUnwrapped;
}
@JsonIgnore
- public HiveTable getHiveTableWrapper() {
+ public HiveTableWrapper getHiveTableWrapper() {
return table;
}
@JsonIgnore
- public List<HivePartition> getHivePartitionWrappers() {
+ public List<HivePartitionWrapper> getHivePartitionWrappers() {
return partitions;
}
@@ -81,7 +84,7 @@ public class HiveReadEntry {
return TableType.TABLE;
}
- public String getPartitionLocation(HiveTable.HivePartition partition) {
+ public String getPartitionLocation(HivePartitionWrapper partition) {
String partitionPath = table.getTable().getSd().getLocation();
for (String value: partition.values) {
http://git-wip-us.apache.org/repos/asf/drill/blob/03928af0/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveScan.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveScan.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveScan.java
index 1a58cbd..c6cc8a2 100644
--- a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveScan.java
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveScan.java
@@ -40,9 +40,10 @@ import org.apache.drill.exec.store.AbstractRecordReader;
import org.apache.drill.exec.store.StoragePluginRegistry;
import org.apache.drill.exec.store.hive.HiveMetadataProvider.HiveStats;
import org.apache.drill.exec.store.hive.HiveMetadataProvider.InputSplitWrapper;
-import org.apache.drill.exec.store.hive.HiveTable.HivePartition;
+import org.apache.drill.exec.store.hive.HiveTableWrapper.HivePartitionWrapper;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.Partition;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.mapred.InputSplit;
@@ -56,6 +57,8 @@ import com.google.common.collect.Lists;
import com.google.common.io.ByteArrayDataOutput;
import com.google.common.io.ByteStreams;
+import static org.apache.drill.exec.store.hive.DrillHiveMetaStoreClient.createPartitionWithSpecColumns;
+
@JsonTypeName("hive-scan")
public class HiveScan extends AbstractGroupScan {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(HiveScan.class);
@@ -151,12 +154,14 @@ public class HiveScan extends AbstractGroupScan {
public SubScan getSpecificScan(final int minorFragmentId) throws ExecutionSetupException {
try {
final List<InputSplitWrapper> splits = mappings.get(minorFragmentId);
- List<HivePartition> parts = Lists.newArrayList();
+ List<HivePartitionWrapper> parts = Lists.newArrayList();
final List<String> encodedInputSplits = Lists.newArrayList();
final List<String> splitTypes = Lists.newArrayList();
for (final InputSplitWrapper split : splits) {
- if (split.getPartition() != null) {
- parts.add(new HivePartition(split.getPartition()));
+ final Partition splitPartition = split.getPartition();
+ if (splitPartition != null) {
+ HiveTableWithColumnCache table = hiveReadEntry.getTable();
+ parts.add(createPartitionWithSpecColumns(new HiveTableWithColumnCache(table, new ColumnListsCache(table)), splitPartition));
}
encodedInputSplits.add(serializeInputSplit(split.getSplit()));
@@ -166,7 +171,7 @@ public class HiveScan extends AbstractGroupScan {
parts = null;
}
- final HiveReadEntry subEntry = new HiveReadEntry(hiveReadEntry.table, parts);
+ final HiveReadEntry subEntry = new HiveReadEntry(hiveReadEntry.getTableWrapper(), parts);
return new HiveSubScan(getUserName(), encodedInputSplits, subEntry, splitTypes, columns, storagePlugin);
} catch (IOException | ReflectiveOperationException e) {
throw new ExecutionSetupException(e);
@@ -259,7 +264,7 @@ public class HiveScan extends AbstractGroupScan {
@Override
public String toString() {
- List<HivePartition> partitions = hiveReadEntry.getHivePartitionWrappers();
+ List<HivePartitionWrapper> partitions = hiveReadEntry.getHivePartitionWrappers();
int numPartitions = partitions == null ? 0 : partitions.size();
return "HiveScan [table=" + hiveReadEntry.getHiveTableWrapper()
+ ", columns=" + columns
http://git-wip-us.apache.org/repos/asf/drill/blob/03928af0/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveScanBatchCreator.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveScanBatchCreator.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveScanBatchCreator.java
index 7aece71..47ea323 100644
--- a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveScanBatchCreator.java
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveScanBatchCreator.java
@@ -30,8 +30,6 @@ import org.apache.drill.exec.record.RecordBatch;
import org.apache.drill.exec.store.RecordReader;
import org.apache.drill.exec.util.ImpersonationUtil;
import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.metastore.api.Partition;
-import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.hive.ql.io.RCFileInputFormat;
import org.apache.hadoop.hive.ql.io.avro.AvroContainerInputFormat;
import org.apache.hadoop.hive.ql.io.orc.OrcInputFormat;
@@ -63,9 +61,9 @@ public class HiveScanBatchCreator implements BatchCreator<HiveSubScan> {
public ScanBatch getBatch(FragmentContext context, HiveSubScan config, List<RecordBatch> children)
throws ExecutionSetupException {
List<RecordReader> readers = Lists.newArrayList();
- Table table = config.getTable();
+ HiveTableWithColumnCache table = config.getTable();
List<InputSplit> splits = config.getInputSplits();
- List<Partition> partitions = config.getPartitions();
+ List<HivePartition> partitions = config.getPartitions();
boolean hasPartitions = (partitions != null && partitions.size() > 0);
int i = 0;
final UserGroupInformation proxyUgi = ImpersonationUtil.createProxyUgi(config.getUserName(),
@@ -80,7 +78,7 @@ public class HiveScanBatchCreator implements BatchCreator<HiveSubScan> {
}
Constructor<? extends HiveAbstractReader> readerConstructor = null;
try {
- readerConstructor = readerClass.getConstructor(Table.class, Partition.class,
+ readerConstructor = readerClass.getConstructor(HiveTableWithColumnCache.class, HivePartition.class,
InputSplit.class, List.class, FragmentContext.class, HiveConf.class,
UserGroupInformation.class);
for (InputSplit split : splits) {
http://git-wip-us.apache.org/repos/asf/drill/blob/03928af0/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveSubScan.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveSubScan.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveSubScan.java
index 74b68a6..107188c 100644
--- a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveSubScan.java
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveSubScan.java
@@ -19,7 +19,6 @@ package org.apache.drill.exec.store.hive;
import java.io.IOException;
import java.lang.reflect.Constructor;
-import java.util.Collections;
import java.util.Iterator;
import java.util.List;
@@ -27,19 +26,13 @@ import com.fasterxml.jackson.annotation.JacksonInject;
import org.apache.commons.codec.binary.Base64;
import org.apache.drill.common.exceptions.ExecutionSetupException;
import org.apache.drill.common.expression.SchemaPath;
-import org.apache.drill.exec.ops.FragmentContext;
-import org.apache.drill.exec.ops.OperatorContext;
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.SubScan;
-import org.apache.drill.exec.physical.impl.ScanBatch;
import org.apache.drill.exec.proto.UserBitShared.CoreOperatorType;
-import org.apache.drill.exec.store.RecordReader;
import org.apache.drill.exec.store.StoragePluginRegistry;
import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.metastore.api.Partition;
-import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.mapred.InputSplit;
import com.fasterxml.jackson.annotation.JsonCreator;
@@ -58,9 +51,9 @@ public class HiveSubScan extends AbstractBase implements SubScan {
@JsonIgnore
protected List<InputSplit> inputSplits = Lists.newArrayList();
@JsonIgnore
- protected Table table;
+ protected HiveTableWithColumnCache table;
@JsonIgnore
- protected List<Partition> partitions;
+ protected List<HivePartition> partitions;
@JsonIgnore
protected HiveStoragePlugin storagePlugin;
@@ -112,11 +105,11 @@ public class HiveSubScan extends AbstractBase implements SubScan {
return splits;
}
- public Table getTable() {
+ public HiveTableWithColumnCache getTable() {
return table;
}
- public List<Partition> getPartitions() {
+ public List<HivePartition> getPartitions() {
return partitions;
}
http://git-wip-us.apache.org/repos/asf/drill/blob/03928af0/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveTable.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveTable.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveTable.java
deleted file mode 100644
index b6dd079..0000000
--- a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveTable.java
+++ /dev/null
@@ -1,382 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.drill.exec.store.hive;
-
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import org.apache.hadoop.hive.metastore.api.FieldSchema;
-import org.apache.hadoop.hive.metastore.api.Order;
-import org.apache.hadoop.hive.metastore.api.Partition;
-import org.apache.hadoop.hive.metastore.api.SerDeInfo;
-import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
-import org.apache.hadoop.hive.metastore.api.Table;
-
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonIgnore;
-import com.fasterxml.jackson.annotation.JsonProperty;
-import com.fasterxml.jackson.annotation.JsonTypeName;
-import com.google.common.collect.Lists;
-
-@JsonTypeName("table")
-public class HiveTable {
-
- @JsonIgnore
- private Table table;
-
- @JsonProperty
- public String tableName;
- @JsonProperty
- public String dbName;
- @JsonProperty
- public String owner;
- @JsonProperty
- public int createTime;
- @JsonProperty
- public int lastAccessTime;
- @JsonProperty
- public int retention;
- @JsonProperty
- public StorageDescriptorWrapper sd;
- @JsonProperty
- public List<FieldSchemaWrapper> partitionKeys;
- @JsonProperty
- public Map<String,String> parameters;
- @JsonProperty
- public String viewOriginalText;
- @JsonProperty
- public String viewExpandedText;
- @JsonProperty
- public String tableType;
-
- @JsonIgnore
- public final Map<String, String> partitionNameTypeMap = new HashMap<>();
-
- @JsonCreator
- public HiveTable(@JsonProperty("tableName") String tableName, @JsonProperty("dbName") String dbName, @JsonProperty("owner") String owner, @JsonProperty("createTime") int createTime,
- @JsonProperty("lastAccessTime") int lastAccessTime, @JsonProperty("retention") int retention, @JsonProperty("sd") StorageDescriptorWrapper sd,
- @JsonProperty("partitionKeys") List<FieldSchemaWrapper> partitionKeys, @JsonProperty("parameters") Map<String, String> parameters,
- @JsonProperty("viewOriginalText") String viewOriginalText, @JsonProperty("viewExpandedText") String viewExpandedText, @JsonProperty("tableType") String tableType
- ) {
- this.tableName = tableName;
- this.dbName = dbName;
- this.owner = owner;
- this.createTime = createTime;
- this.lastAccessTime = lastAccessTime;
- this.retention = retention;
- this.sd = sd;
- this.partitionKeys = partitionKeys;
- this.parameters = parameters;
- this.viewOriginalText = viewOriginalText;
- this.viewExpandedText = viewExpandedText;
- this.tableType = tableType;
-
- List<FieldSchema> partitionKeysUnwrapped = Lists.newArrayList();
- for (FieldSchemaWrapper w : partitionKeys) {
- partitionKeysUnwrapped.add(w.getFieldSchema());
- partitionNameTypeMap.put(w.name, w.type);
- }
- StorageDescriptor sdUnwrapped = sd.getSd();
- this.table = new Table(tableName, dbName, owner, createTime, lastAccessTime, retention, sdUnwrapped, partitionKeysUnwrapped,
- parameters, viewOriginalText, viewExpandedText, tableType);
- }
-
- public HiveTable(Table table) {
- if (table == null) {
- return;
- }
- this.table = table;
- this.tableName = table.getTableName();
- this.dbName = table.getDbName();
- this.owner = table.getOwner();
- this.createTime = table.getCreateTime();
- this.lastAccessTime = table.getLastAccessTime();
- this.retention = table.getRetention();
- this.sd = new StorageDescriptorWrapper(table.getSd());
- this.partitionKeys = Lists.newArrayList();
- for (FieldSchema f : table.getPartitionKeys()) {
- this.partitionKeys.add(new FieldSchemaWrapper(f));
- partitionNameTypeMap.put(f.getName(), f.getType());
- }
- this.parameters = table.getParameters();
- this.viewOriginalText = table.getViewOriginalText();
- this.viewExpandedText = table.getViewExpandedText();
- this.tableType = table.getTableType();
- }
-
- @JsonIgnore
- public Table getTable() {
- return table;
- }
-
- @Override
- public String toString() {
- StringBuilder sb = new StringBuilder("Table(");
-
- sb.append("dbName:");
- sb.append(this.dbName);
- sb.append(", ");
-
- sb.append("tableName:");
- sb.append(this.tableName);
- sb.append(")");
-
- return sb.toString();
- }
-
- public static class HivePartition {
-
- @JsonIgnore
- private Partition partition;
-
- @JsonProperty
- public List<String> values;
- @JsonProperty
- public String tableName;
- @JsonProperty
- public String dbName;
- @JsonProperty
- public int createTime;
- @JsonProperty
- public int lastAccessTime;
- @JsonProperty
- public StorageDescriptorWrapper sd;
- @JsonProperty
- public Map<String,String> parameters;
-
- @JsonCreator
- public HivePartition(@JsonProperty("values") List<String> values, @JsonProperty("tableName") String tableName, @JsonProperty("dbName") String dbName, @JsonProperty("createTime") int createTime,
- @JsonProperty("lastAccessTime") int lastAccessTime, @JsonProperty("sd") StorageDescriptorWrapper sd,
- @JsonProperty("parameters") Map<String, String> parameters
- ) {
- this.values = values;
- this.tableName = tableName;
- this.dbName = dbName;
- this.createTime = createTime;
- this.lastAccessTime = lastAccessTime;
- this.sd = sd;
- this.parameters = parameters;
-
- StorageDescriptor sdUnwrapped = sd.getSd();
- this.partition = new org.apache.hadoop.hive.metastore.api.Partition(values, tableName, dbName, createTime, lastAccessTime, sdUnwrapped, parameters);
- }
-
- public HivePartition(Partition partition) {
- if (partition == null) {
- return;
- }
- this.partition = partition;
- this.values = partition.getValues();
- this.tableName = partition.getTableName();
- this.dbName = partition.getDbName();
- this.createTime = partition.getCreateTime();
- this.lastAccessTime = partition.getLastAccessTime();
- this.sd = new StorageDescriptorWrapper(partition.getSd());
- this.parameters = partition.getParameters();
- }
-
- @JsonIgnore
- public Partition getPartition() {
- return partition;
- }
-
- @Override
- public String toString() {
- StringBuilder sb = new StringBuilder("Partition(");
- sb.append("values:");
- sb.append(this.values);
- sb.append(")");
- return sb.toString();
- }
- }
-
- public static class StorageDescriptorWrapper {
- @JsonIgnore
- private StorageDescriptor sd;
- @JsonProperty
- public List<FieldSchemaWrapper> cols;
- @JsonProperty
- public String location;
- @JsonProperty
- public String inputFormat;
- @JsonProperty
- public String outputFormat;
- @JsonProperty
- public boolean compressed;
- @JsonProperty
- public int numBuckets;
- @JsonProperty
- public SerDeInfoWrapper serDeInfo;
- // @JsonProperty
-// public List<String> bucketCols;
- @JsonProperty
- public List<OrderWrapper> sortCols;
- @JsonProperty
- public Map<String,String> parameters;
-
- @JsonCreator
- public StorageDescriptorWrapper(@JsonProperty("cols") List<FieldSchemaWrapper> cols, @JsonProperty("location") String location, @JsonProperty("inputFormat") String inputFormat,
- @JsonProperty("outputFormat") String outputFormat, @JsonProperty("compressed") boolean compressed, @JsonProperty("numBuckets") int numBuckets,
- @JsonProperty("serDeInfo") SerDeInfoWrapper serDeInfo, @JsonProperty("sortCols") List<OrderWrapper> sortCols,
- @JsonProperty("parameters") Map<String,String> parameters) {
- this.cols = cols;
- this.location = location;
- this.inputFormat = inputFormat;
- this.outputFormat = outputFormat;
- this.compressed = compressed;
- this.numBuckets = numBuckets;
- this.serDeInfo = serDeInfo;
-// this.bucketCols = bucketCols;
- this.sortCols = sortCols;
- this.parameters = parameters;
- List<FieldSchema> colsUnwrapped = Lists.newArrayList();
- for (FieldSchemaWrapper w: cols) {
- colsUnwrapped.add(w.getFieldSchema());
- }
- SerDeInfo serDeInfoUnwrapped = serDeInfo.getSerDeInfo();
- List<Order> sortColsUnwrapped = Lists.newArrayList();
- for (OrderWrapper w : sortCols) {
- sortColsUnwrapped.add(w.getOrder());
- }
-// this.sd = new StorageDescriptor(colsUnwrapped, location, inputFormat, outputFormat, compressed, numBuckets, serDeInfoUnwrapped,
-// bucketCols, sortColsUnwrapped, parameters);
- this.sd = new StorageDescriptor(colsUnwrapped, location, inputFormat, outputFormat, compressed, numBuckets, serDeInfoUnwrapped,
- null, sortColsUnwrapped, parameters);
- }
-
- public StorageDescriptorWrapper(StorageDescriptor sd) {
- this.sd = sd;
- this.cols = Lists.newArrayList();
- for (FieldSchema f : sd.getCols()) {
- this.cols.add(new FieldSchemaWrapper(f));
- }
- this.location = sd.getLocation();
- this.inputFormat = sd.getInputFormat();
- this.outputFormat = sd.getOutputFormat();
- this.compressed = sd.isCompressed();
- this.numBuckets = sd.getNumBuckets();
- this.serDeInfo = new SerDeInfoWrapper(sd.getSerdeInfo());
-// this.bucketCols = sd.getBucketCols();
- this.sortCols = Lists.newArrayList();
- for (Order o : sd.getSortCols()) {
- this.sortCols.add(new OrderWrapper(o));
- }
- this.parameters = sd.getParameters();
- }
-
- @JsonIgnore
- public StorageDescriptor getSd() {
- return sd;
- }
-
- }
-
- public static class SerDeInfoWrapper {
- @JsonIgnore
- private SerDeInfo serDeInfo;
- @JsonProperty
- public String name;
- @JsonProperty
- public String serializationLib;
- @JsonProperty
- public Map<String,String> parameters;
-
- @JsonCreator
- public SerDeInfoWrapper(@JsonProperty("name") String name, @JsonProperty("serializationLib") String serializationLib, @JsonProperty("parameters") Map<String, String> parameters) {
- this.name = name;
- this.serializationLib = serializationLib;
- this.parameters = parameters;
- this.serDeInfo = new SerDeInfo(name, serializationLib, parameters);
- }
-
- public SerDeInfoWrapper(SerDeInfo serDeInfo) {
- this.serDeInfo = serDeInfo;
- this.name = serDeInfo.getName();
- this.serializationLib = serDeInfo.getSerializationLib();
- this.parameters = serDeInfo.getParameters();
- }
-
- @JsonIgnore
- public SerDeInfo getSerDeInfo() {
- return serDeInfo;
- }
- }
-
- public static class FieldSchemaWrapper {
- @JsonIgnore
- private FieldSchema fieldSchema;
- @JsonProperty
- public String name;
- @JsonProperty
- public String type;
- @JsonProperty
- public String comment;
-
- @JsonCreator
- public FieldSchemaWrapper(@JsonProperty("name") String name, @JsonProperty("type") String type, @JsonProperty("comment") String comment) {
- this.name = name;
- this.type = type;
- this.comment = comment;
- this.fieldSchema = new FieldSchema(name, type, comment);
- }
-
- public FieldSchemaWrapper(FieldSchema fieldSchema) {
- this.fieldSchema = fieldSchema;
- this.name = fieldSchema.getName();
- this.type = fieldSchema.getType();
- this.comment = fieldSchema.getComment();
- }
-
- @JsonIgnore
- public FieldSchema getFieldSchema() {
- return fieldSchema;
- }
- }
-
- public static class OrderWrapper {
- @JsonIgnore
- private Order ord;
- @JsonProperty
- public String col;
- @JsonProperty
- public int order;
-
- @JsonCreator
- public OrderWrapper(@JsonProperty("col") String col, @JsonProperty("order") int order) {
- this.col = col;
- this.order = order;
- }
-
- public OrderWrapper(Order ord) {
- this.ord = ord;
- this.col = ord.getCol();
- this.order = ord.getOrder();
- }
-
- @JsonIgnore
- public Order getOrder() {
- return ord;
- }
- }
-
- public Map<String, String> getPartitionNameTypeMap() {
- return partitionNameTypeMap;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/drill/blob/03928af0/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveTableWithColumnCache.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveTableWithColumnCache.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveTableWithColumnCache.java
new file mode 100644
index 0000000..91888ef
--- /dev/null
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveTableWithColumnCache.java
@@ -0,0 +1,76 @@
+/*
+* 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.store.hive;
+
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * This class is wrapper of {@link Table} class and used for
+ * storage of such additional information as column lists cache.
+ */
+public class HiveTableWithColumnCache extends Table {
+
+ private ColumnListsCache columnListsCache;
+
+ public HiveTableWithColumnCache() {
+ super();
+ }
+
+ public HiveTableWithColumnCache(
+ String tableName,
+ String dbName,
+ String owner,
+ int createTime,
+ int lastAccessTime,
+ int retention,
+ StorageDescriptor sd,
+ List<FieldSchema> partitionKeys,
+ Map<String,String> parameters,
+ String viewOriginalText,
+ String viewExpandedText,
+ String tableType,
+ ColumnListsCache columnListsCache) {
+ super(tableName, dbName, owner, createTime, lastAccessTime, retention, sd,
+ partitionKeys, parameters, viewOriginalText, viewExpandedText, tableType);
+ this.columnListsCache = columnListsCache;
+ }
+
+ public HiveTableWithColumnCache(HiveTableWithColumnCache other) {
+ super(other);
+ columnListsCache = other.getColumnListsCache();
+ }
+
+ public HiveTableWithColumnCache(Table other, ColumnListsCache columnListsCache) {
+ super(other);
+ this.columnListsCache = columnListsCache;
+ }
+
+ /**
+ * To reduce physical plan for Hive tables, unique partition lists of columns stored in the
+ * table's column lists cache.
+ *
+ * @return table's column lists cache
+ */
+ public ColumnListsCache getColumnListsCache() {
+ return columnListsCache;
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/03928af0/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveTableWrapper.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveTableWrapper.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveTableWrapper.java
new file mode 100644
index 0000000..7f2afa6
--- /dev/null
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveTableWrapper.java
@@ -0,0 +1,466 @@
+/**
+ * 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.store.hive;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.Order;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.SerDeInfo;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import com.google.common.collect.Lists;
+
+@JsonTypeName("table")
+public class HiveTableWrapper {
+
+ @JsonIgnore
+ private HiveTableWithColumnCache table;
+
+ @JsonProperty
+ public String tableName;
+ @JsonProperty
+ public String dbName;
+ @JsonProperty
+ public String owner;
+ @JsonProperty
+ public int createTime;
+ @JsonProperty
+ public int lastAccessTime;
+ @JsonProperty
+ public int retention;
+ @JsonProperty
+ public StorageDescriptorWrapper sd;
+ @JsonProperty
+ public List<FieldSchemaWrapper> partitionKeys;
+ @JsonProperty
+ public Map<String,String> parameters;
+ @JsonProperty
+ public String viewOriginalText;
+ @JsonProperty
+ public String viewExpandedText;
+ @JsonProperty
+ public String tableType;
+ @JsonProperty
+ public ColumnsCacheWrapper columnsCache;
+
+ @JsonIgnore
+ public final Map<String, String> partitionNameTypeMap = new HashMap<>();
+
+ @JsonCreator
+ public HiveTableWrapper(@JsonProperty("tableName") String tableName, @JsonProperty("dbName") String dbName, @JsonProperty("owner") String owner,
+ @JsonProperty("createTime") int createTime, @JsonProperty("lastAccessTime") int lastAccessTime,
+ @JsonProperty("retention") int retention, @JsonProperty("sd") StorageDescriptorWrapper sd,
+ @JsonProperty("partitionKeys") List<FieldSchemaWrapper> partitionKeys, @JsonProperty("parameters") Map<String, String> parameters,
+ @JsonProperty("viewOriginalText") String viewOriginalText, @JsonProperty("viewExpandedText") String viewExpandedText,
+ @JsonProperty("tableType") String tableType, @JsonProperty("columnsCache") ColumnsCacheWrapper columnsCache
+ ) {
+ this.tableName = tableName;
+ this.dbName = dbName;
+ this.owner = owner;
+ this.createTime = createTime;
+ this.lastAccessTime = lastAccessTime;
+ this.retention = retention;
+ this.sd = sd;
+ this.partitionKeys = partitionKeys;
+ this.parameters = parameters;
+ this.viewOriginalText = viewOriginalText;
+ this.viewExpandedText = viewExpandedText;
+ this.tableType = tableType;
+ this.columnsCache = columnsCache;
+
+ List<FieldSchema> partitionKeysUnwrapped = Lists.newArrayList();
+ for (FieldSchemaWrapper w : partitionKeys) {
+ partitionKeysUnwrapped.add(w.getFieldSchema());
+ partitionNameTypeMap.put(w.name, w.type);
+ }
+ StorageDescriptor sdUnwrapped = sd.getSd();
+ this.table = new HiveTableWithColumnCache(tableName, dbName, owner, createTime, lastAccessTime, retention, sdUnwrapped, partitionKeysUnwrapped,
+ parameters, viewOriginalText, viewExpandedText, tableType, columnsCache.getColumnListsCache());
+ }
+
+ public HiveTableWrapper(HiveTableWithColumnCache table) {
+ if (table == null) {
+ return;
+ }
+ this.table = table;
+ this.tableName = table.getTableName();
+ this.dbName = table.getDbName();
+ this.owner = table.getOwner();
+ this.createTime = table.getCreateTime();
+ this.lastAccessTime = table.getLastAccessTime();
+ this.retention = table.getRetention();
+ this.sd = new StorageDescriptorWrapper(table.getSd());
+ this.partitionKeys = Lists.newArrayList();
+ for (FieldSchema f : table.getPartitionKeys()) {
+ this.partitionKeys.add(new FieldSchemaWrapper(f));
+ partitionNameTypeMap.put(f.getName(), f.getType());
+ }
+ this.parameters = table.getParameters();
+ this.viewOriginalText = table.getViewOriginalText();
+ this.viewExpandedText = table.getViewExpandedText();
+ this.tableType = table.getTableType();
+ this.columnsCache = new ColumnsCacheWrapper(table.getColumnListsCache());
+ }
+
+ @JsonIgnore
+ public HiveTableWithColumnCache getTable() {
+ return table;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder("Table(");
+
+ sb.append("dbName:");
+ sb.append(this.dbName);
+ sb.append(", ");
+
+ sb.append("tableName:");
+ sb.append(this.tableName);
+ sb.append(")");
+
+ return sb.toString();
+ }
+
+ /**
+ * Wrapper for {@link Partition} class. Used for serialization and deserialization of {@link HivePartition}.
+ */
+ public static class HivePartitionWrapper {
+
+ @JsonIgnore
+ private HivePartition partition;
+
+ @JsonProperty
+ public List<String> values;
+
+ @JsonProperty
+ public String tableName;
+
+ @JsonProperty
+ public String dbName;
+
+ @JsonProperty
+ public int createTime;
+
+ @JsonProperty
+ public int lastAccessTime;
+
+ @JsonProperty
+ public StorageDescriptorWrapper sd;
+
+ @JsonProperty
+ public Map<String, String> parameters;
+
+ @JsonProperty
+ private int columnListIndex;
+
+ @JsonCreator
+ public HivePartitionWrapper(@JsonProperty("values") List<String> values, @JsonProperty("tableName") String tableName,
+ @JsonProperty("dbName") String dbName, @JsonProperty("createTime") int createTime,
+ @JsonProperty("lastAccessTime") int lastAccessTime, @JsonProperty("sd") StorageDescriptorWrapper sd,
+ @JsonProperty("parameters") Map<String, String> parameters, @JsonProperty("columnListIndex") int columnListIndex) {
+ this.values = values;
+ this.tableName = tableName;
+ this.dbName = dbName;
+ this.createTime = createTime;
+ this.lastAccessTime = lastAccessTime;
+ this.sd = sd;
+ this.parameters = parameters;
+ this.columnListIndex = columnListIndex;
+
+ StorageDescriptor sdUnwrapped = sd.getSd();
+ this.partition = new HivePartition(values, tableName, dbName, createTime, lastAccessTime, sdUnwrapped, parameters, columnListIndex);
+ }
+
+ public HivePartitionWrapper(HivePartition partition) {
+ if (partition == null) {
+ return;
+ }
+ this.partition = partition;
+ this.values = partition.getValues();
+ this.tableName = partition.getTableName();
+ this.dbName = partition.getDbName();
+ this.createTime = partition.getCreateTime();
+ this.lastAccessTime = partition.getLastAccessTime();
+ this.sd = new StorageDescriptorWrapper(partition.getSd());
+ this.parameters = partition.getParameters();
+ this.columnListIndex = partition.getColumnListIndex();
+ }
+
+ @JsonIgnore
+ public HivePartition getPartition() {
+ return partition;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder("Partition(");
+ sb.append("values:");
+ sb.append(this.values);
+ sb.append(")");
+ return sb.toString();
+ }
+ }
+
+ /**
+ * Wrapper for {@link StorageDescriptor} class.
+ * Used in {@link HivePartitionWrapper} and {@link HiveTableWrapper}
+ * for serialization and deserialization of {@link StorageDescriptor}.
+ */
+ public static class StorageDescriptorWrapper {
+
+ @JsonIgnore
+ private StorageDescriptor sd;
+
+ // column lists stored in ColumnListsCache
+ @JsonIgnore
+ public List<FieldSchemaWrapper> columns;
+
+ @JsonProperty
+ public String location;
+
+ @JsonProperty
+ public String inputFormat;
+
+ @JsonProperty
+ public String outputFormat;
+
+ @JsonProperty
+ public boolean compressed;
+
+ @JsonProperty
+ public int numBuckets;
+
+ @JsonProperty
+ public SerDeInfoWrapper serDeInfo;
+
+ @JsonProperty
+ public List<OrderWrapper> sortCols;
+
+ @JsonProperty
+ public Map<String, String> parameters;
+
+ @JsonCreator
+ public StorageDescriptorWrapper(@JsonProperty("columns") List<FieldSchemaWrapper> columns, @JsonProperty("location") String location, @JsonProperty("inputFormat") String inputFormat,
+ @JsonProperty("outputFormat") String outputFormat, @JsonProperty("compressed") boolean compressed, @JsonProperty("numBuckets") int numBuckets,
+ @JsonProperty("serDeInfo") SerDeInfoWrapper serDeInfo, @JsonProperty("sortCols") List<OrderWrapper> sortCols,
+ @JsonProperty("parameters") Map<String,String> parameters) {
+ this.columns = columns;
+ this.location = location;
+ this.inputFormat = inputFormat;
+ this.outputFormat = outputFormat;
+ this.compressed = compressed;
+ this.numBuckets = numBuckets;
+ this.serDeInfo = serDeInfo;
+ this.sortCols = sortCols;
+ this.parameters = parameters;
+ List<FieldSchema> colsUnwrapped;
+ if (columns != null) {
+ colsUnwrapped = Lists.newArrayList();
+ for (FieldSchemaWrapper fieldSchema : columns) {
+ colsUnwrapped.add(fieldSchema.getFieldSchema());
+ }
+ } else {
+ colsUnwrapped = null;
+ }
+ SerDeInfo serDeInfoUnwrapped = serDeInfo.getSerDeInfo();
+ List<Order> sortColsUnwrapped;
+ if (sortCols != null) {
+ sortColsUnwrapped = Lists.newArrayList();
+ for (OrderWrapper order : sortCols) {
+ sortColsUnwrapped.add(order.getOrder());
+ }
+ } else {
+ sortColsUnwrapped = null;
+ }
+ sd = new StorageDescriptor(colsUnwrapped, location, inputFormat, outputFormat,
+ compressed, numBuckets, serDeInfoUnwrapped, null, sortColsUnwrapped, parameters);
+ }
+
+ public StorageDescriptorWrapper(StorageDescriptor storageDescriptor) {
+ sd = storageDescriptor;
+ location = storageDescriptor.getLocation();
+ inputFormat = storageDescriptor.getInputFormat();
+ outputFormat = storageDescriptor.getOutputFormat();
+ compressed = storageDescriptor.isCompressed();
+ numBuckets = storageDescriptor.getNumBuckets();
+ serDeInfo = new SerDeInfoWrapper(storageDescriptor.getSerdeInfo());
+ if (sd.getSortCols() != null) {
+ sortCols = Lists.newArrayList();
+ for (Order order : sd.getSortCols()) {
+ sortCols.add(new OrderWrapper(order));
+ }
+ }
+ parameters = storageDescriptor.getParameters();
+ if (sd.getCols() != null) {
+ this.columns = Lists.newArrayList();
+ for (FieldSchema fieldSchema : sd.getCols()) {
+ this.columns.add(new FieldSchemaWrapper(fieldSchema));
+ }
+ }
+ }
+
+ @JsonIgnore
+ public StorageDescriptor getSd() {
+ return sd;
+ }
+ }
+
+ public static class SerDeInfoWrapper {
+ @JsonIgnore
+ private SerDeInfo serDeInfo;
+ @JsonProperty
+ public String name;
+ @JsonProperty
+ public String serializationLib;
+ @JsonProperty
+ public Map<String,String> parameters;
+
+ @JsonCreator
+ public SerDeInfoWrapper(@JsonProperty("name") String name, @JsonProperty("serializationLib") String serializationLib, @JsonProperty("parameters") Map<String, String> parameters) {
+ this.name = name;
+ this.serializationLib = serializationLib;
+ this.parameters = parameters;
+ this.serDeInfo = new SerDeInfo(name, serializationLib, parameters);
+ }
+
+ public SerDeInfoWrapper(SerDeInfo serDeInfo) {
+ this.serDeInfo = serDeInfo;
+ this.name = serDeInfo.getName();
+ this.serializationLib = serDeInfo.getSerializationLib();
+ this.parameters = serDeInfo.getParameters();
+ }
+
+ @JsonIgnore
+ public SerDeInfo getSerDeInfo() {
+ return serDeInfo;
+ }
+ }
+
+ public static class FieldSchemaWrapper {
+ @JsonIgnore
+ private FieldSchema fieldSchema;
+ @JsonProperty
+ public String name;
+ @JsonProperty
+ public String type;
+ @JsonProperty
+ public String comment;
+
+ @JsonCreator
+ public FieldSchemaWrapper(@JsonProperty("name") String name, @JsonProperty("type") String type, @JsonProperty("comment") String comment) {
+ this.name = name;
+ this.type = type;
+ this.comment = comment;
+ this.fieldSchema = new FieldSchema(name, type, comment);
+ }
+
+ public FieldSchemaWrapper(FieldSchema fieldSchema) {
+ this.fieldSchema = fieldSchema;
+ this.name = fieldSchema.getName();
+ this.type = fieldSchema.getType();
+ this.comment = fieldSchema.getComment();
+ }
+
+ @JsonIgnore
+ public FieldSchema getFieldSchema() {
+ return fieldSchema;
+ }
+ }
+
+ public static class OrderWrapper {
+ @JsonIgnore
+ private Order ord;
+ @JsonProperty
+ public String col;
+ @JsonProperty
+ public int order;
+
+ @JsonCreator
+ public OrderWrapper(@JsonProperty("col") String col, @JsonProperty("order") int order) {
+ this.col = col;
+ this.order = order;
+ }
+
+ public OrderWrapper(Order ord) {
+ this.ord = ord;
+ this.col = ord.getCol();
+ this.order = ord.getOrder();
+ }
+
+ @JsonIgnore
+ public Order getOrder() {
+ return ord;
+ }
+ }
+
+ public Map<String, String> getPartitionNameTypeMap() {
+ return partitionNameTypeMap;
+ }
+
+ /**
+ * Wrapper for {@link ColumnListsCache} class.
+ * Used in {@link HiveTableWrapper} for serialization and deserialization of {@link ColumnListsCache}.
+ */
+ public static class ColumnsCacheWrapper {
+ @JsonIgnore
+ private final ColumnListsCache columnListsCache;
+
+ @JsonProperty
+ private final List<List<FieldSchemaWrapper>> keys;
+
+ @JsonCreator
+ public ColumnsCacheWrapper(@JsonProperty("keys") List<List<FieldSchemaWrapper>> keys) {
+ this.keys = keys;
+ this.columnListsCache = new ColumnListsCache();
+ for (List<FieldSchemaWrapper> columns : keys) {
+ final List<FieldSchema> columnsUnwrapped = Lists.newArrayList();
+ for (FieldSchemaWrapper field : columns) {
+ columnsUnwrapped.add(field.getFieldSchema());
+ }
+ columnListsCache.addOrGet(columnsUnwrapped);
+ }
+ }
+
+ public ColumnsCacheWrapper(ColumnListsCache columnListsCache) {
+ this.columnListsCache = columnListsCache;
+ final List<List<FieldSchemaWrapper>> keysWrapped = Lists.newArrayList();
+ for (List<FieldSchema> columns : columnListsCache.getFields()) {
+ final List<FieldSchemaWrapper> columnsWrapped = Lists.newArrayList();
+ for (FieldSchema field : columns) {
+ columnsWrapped.add(new FieldSchemaWrapper(field));
+ }
+ keysWrapped.add(columnsWrapped);
+ }
+ this.keys = keysWrapped;
+ }
+
+ @JsonIgnore
+ public ColumnListsCache getColumnListsCache() {
+ return columnListsCache;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/03928af0/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveUtilities.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveUtilities.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveUtilities.java
index 2e23aff..1d5e6bf 100644
--- a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveUtilities.java
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveUtilities.java
@@ -71,6 +71,7 @@ import org.joda.time.DateTimeZone;
import java.math.BigDecimal;
import java.sql.Date;
import java.sql.Timestamp;
+import java.util.List;
import java.util.Map;
import java.util.Properties;
@@ -398,12 +399,14 @@ public class HiveUtilities {
* Wrapper around {@link MetaStoreUtils#getPartitionMetadata(Partition, Table)} which also adds parameters from table
* to properties returned by {@link MetaStoreUtils#getPartitionMetadata(Partition, Table)}.
*
- * @param partition
- * @param table
- * @return
+ * @param partition the source of partition level parameters
+ * @param table the source of table level parameters
+ * @return properties
*/
- public static Properties getPartitionMetadata(final Partition partition, final Table table) {
- final Properties properties = MetaStoreUtils.getPartitionMetadata(partition, table);
+ public static Properties getPartitionMetadata(final HivePartition partition, final HiveTableWithColumnCache table) {
+ final Properties properties;
+ restoreColumns(table, partition);
+ properties = MetaStoreUtils.getPartitionMetadata(partition, table);
// SerDe expects properties from Table, but above call doesn't add Table properties.
// Include Table properties in final list in order to not to break SerDes that depend on
@@ -417,6 +420,34 @@ public class HiveUtilities {
return properties;
}
+ /**
+ * Sets columns from table cache to table and partition.
+ *
+ * @param partition partition which will set column list
+ * @param table the source of column lists cache
+ */
+ public static void restoreColumns(HiveTableWithColumnCache table, HivePartition partition) {
+ // exactly the same column lists for partitions or table
+ // stored only one time to reduce physical plan serialization
+ if (partition != null && partition.getSd().getCols() == null) {
+ partition.getSd().setCols(table.getColumnListsCache().getColumns(partition.getColumnListIndex()));
+ }
+ if (table.getSd().getCols() == null) {
+ table.getSd().setCols(table.getColumnListsCache().getColumns(0));
+ }
+ }
+
+ /**
+ * Wrapper around {@link MetaStoreUtils#getSchema(StorageDescriptor, StorageDescriptor, Map, String, String, List)}
+ * which also sets columns from table cache to table and returns properties returned by
+ * {@link MetaStoreUtils#getSchema(StorageDescriptor, StorageDescriptor, Map, String, String, List)}.
+ */
+ public static Properties getTableMetadata(HiveTableWithColumnCache table) {
+ restoreColumns(table, null);
+ return MetaStoreUtils.getSchema(table.getSd(), table.getSd(), table.getParameters(),
+ table.getDbName(), table.getTableName(), table.getPartitionKeys());
+ }
+
public static void throwUnsupportedHiveDataTypeError(String unsupportedType) {
StringBuilder errMsg = new StringBuilder();
errMsg.append(String.format("Unsupported Hive data type %s. ", unsupportedType));
http://git-wip-us.apache.org/repos/asf/drill/blob/03928af0/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/schema/DrillHiveTable.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/schema/DrillHiveTable.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/schema/DrillHiveTable.java
index 29f7757..af02c0a 100644
--- a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/schema/DrillHiveTable.java
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/schema/DrillHiveTable.java
@@ -24,6 +24,7 @@ import java.util.List;
import org.apache.drill.exec.planner.logical.DrillTable;
import org.apache.drill.exec.store.hive.HiveReadEntry;
import org.apache.drill.exec.store.hive.HiveStoragePlugin;
+import org.apache.drill.exec.store.hive.HiveTableWithColumnCache;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.ql.metadata.Table;
import org.apache.hadoop.hive.serde2.typeinfo.DecimalTypeInfo;
@@ -43,11 +44,11 @@ import com.google.common.collect.Lists;
public class DrillHiveTable extends DrillTable{
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DrillHiveTable.class);
- protected final Table hiveTable;
+ protected final HiveTableWithColumnCache hiveTable;
public DrillHiveTable(String storageEngineName, HiveStoragePlugin plugin, String userName, HiveReadEntry readEntry) {
super(storageEngineName, plugin, userName, readEntry);
- this.hiveTable = new Table(readEntry.getTable());
+ this.hiveTable = new HiveTableWithColumnCache(readEntry.getTable());
}
@Override
@@ -55,7 +56,7 @@ public class DrillHiveTable extends DrillTable{
List<RelDataType> typeList = Lists.newArrayList();
List<String> fieldNameList = Lists.newArrayList();
- List<FieldSchema> hiveFields = hiveTable.getCols();
+ List<FieldSchema> hiveFields = hiveTable.getColumnListsCache().getColumns(0);
for(FieldSchema hiveField : hiveFields) {
fieldNameList.add(hiveField.getName());
typeList.add(getNullableRelDataTypeFromHiveType(
http://git-wip-us.apache.org/repos/asf/drill/blob/03928af0/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/TestHivePartitionPruning.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/TestHivePartitionPruning.java b/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/TestHivePartitionPruning.java
index 7ac1896..a32f538 100644
--- a/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/TestHivePartitionPruning.java
+++ b/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/TestHivePartitionPruning.java
@@ -17,16 +17,19 @@
*/
package org.apache.drill.exec;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import org.apache.drill.exec.hive.HiveTestBase;
import org.apache.drill.exec.planner.physical.PlannerSettings;
+import org.apache.drill.exec.rpc.user.QueryDataBatch;
import org.junit.AfterClass;
import org.junit.BeforeClass;
-import org.junit.Ignore;
import org.junit.Test;
+import java.util.List;
+
public class TestHivePartitionPruning extends HiveTestBase {
// enable decimal data type
@BeforeClass
@@ -149,6 +152,30 @@ public class TestHivePartitionPruning extends HiveTestBase {
.go();
}
+ @Test // DRILL-5032
+ public void testPartitionColumnsCaching() throws Exception {
+ final String query = "EXPLAIN PLAN FOR SELECT * FROM hive.partition_with_few_schemas";
+
+ List<QueryDataBatch> queryDataBatches = testSqlWithResults(query);
+ String resultString = getResultString(queryDataBatches, "|");
+
+ // different for both partitions column strings from physical plan
+ String columnString = "\"name\" : \"a\"";
+ String secondColumnString = "\"name\" : \"a1\"";
+
+ int columnIndex = resultString.indexOf(columnString);
+ assertTrue(columnIndex >= 0);
+ columnIndex = resultString.indexOf(columnString, columnIndex + 1);
+ // checks that column added to physical plan only one time
+ assertEquals(-1, columnIndex);
+
+ int secondColumnIndex = resultString.indexOf(secondColumnString);
+ assertTrue(secondColumnIndex >= 0);
+ secondColumnIndex = resultString.indexOf(secondColumnString, secondColumnIndex + 1);
+ // checks that column added to physical plan only one time
+ assertEquals(-1, secondColumnIndex);
+ }
+
@AfterClass
public static void disableDecimalDataType() throws Exception {
test(String.format("alter session set `%s` = false", PlannerSettings.ENABLE_DECIMAL_DATA_TYPE_KEY));
http://git-wip-us.apache.org/repos/asf/drill/blob/03928af0/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/hive/TestInfoSchemaOnHiveStorage.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/hive/TestInfoSchemaOnHiveStorage.java b/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/hive/TestInfoSchemaOnHiveStorage.java
index 0a94867..fb4bb17 100644
--- a/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/hive/TestInfoSchemaOnHiveStorage.java
+++ b/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/hive/TestInfoSchemaOnHiveStorage.java
@@ -43,6 +43,7 @@ public class TestInfoSchemaOnHiveStorage extends HiveTestBase {
.baselineValues("hive.default", "kv_sh")
.baselineValues("hive.default", "countstar_parquet")
.baselineValues("hive.default", "simple_json")
+ .baselineValues("hive.default", "partition_with_few_schemas")
.go();
testBuilder()
@@ -243,6 +244,7 @@ public class TestInfoSchemaOnHiveStorage extends HiveTestBase {
.baselineValues("DRILL", "hive.default", "readtest_parquet", "TABLE")
.baselineValues("DRILL", "hive.default", "hiveview", "VIEW")
.baselineValues("DRILL", "hive.default", "partition_pruning_test", "TABLE")
+ .baselineValues("DRILL", "hive.default", "partition_with_few_schemas", "TABLE")
.baselineValues("DRILL", "hive.default", "kv_parquet", "TABLE")
.baselineValues("DRILL", "hive.default", "countstar_parquet", "TABLE")
.baselineValues("DRILL", "hive.default", "kv_sh", "TABLE")
http://git-wip-us.apache.org/repos/asf/drill/blob/03928af0/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/store/hive/HiveTestDataGenerator.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/store/hive/HiveTestDataGenerator.java b/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/store/hive/HiveTestDataGenerator.java
index 7a5b72d..435c66b 100644
--- a/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/store/hive/HiveTestDataGenerator.java
+++ b/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/store/hive/HiveTestDataGenerator.java
@@ -438,6 +438,15 @@ public class HiveTestDataGenerator {
executeQuery(hiveDriver, "INSERT OVERWRITE TABLE partition_pruning_test PARTITION(c, d, e) " +
"SELECT a, b, c, d, e FROM partition_pruning_test_loadtable");
+ executeQuery(hiveDriver,
+ "CREATE TABLE IF NOT EXISTS partition_with_few_schemas(a DATE, b TIMESTAMP) "+
+ "partitioned by (c INT, d INT, e INT) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' STORED AS TEXTFILE");
+ executeQuery(hiveDriver, "INSERT OVERWRITE TABLE partition_with_few_schemas PARTITION(c, d, e) " +
+ "SELECT a, b, c, d, e FROM partition_pruning_test_loadtable");
+ executeQuery(hiveDriver,"alter table partition_with_few_schemas partition(c=1, d=1, e=1) change a a1 INT");
+ executeQuery(hiveDriver,"alter table partition_with_few_schemas partition(c=1, d=1, e=2) change a a1 INT");
+ executeQuery(hiveDriver,"alter table partition_with_few_schemas partition(c=2, d=2, e=2) change a a1 INT");
+
// Add a partition with custom location
executeQuery(hiveDriver,
String.format("ALTER TABLE partition_pruning_test ADD PARTITION (c=99, d=98, e=97) LOCATION '%s'",
[5/8] drill git commit: DRILL-5032: Drill query on hive parquet table
failed with OutOfMemoryError: Java heap space
Posted by jn...@apache.org.
http://git-wip-us.apache.org/repos/asf/drill/blob/03928af0/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/store/hive/schema/TestColumnListCache.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/store/hive/schema/TestColumnListCache.java b/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/store/hive/schema/TestColumnListCache.java
new file mode 100644
index 0000000..2fdab54
--- /dev/null
+++ b/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/store/hive/schema/TestColumnListCache.java
@@ -0,0 +1,111 @@
+/*
+* 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.store.hive.schema;
+
+import com.google.common.collect.Lists;
+import org.apache.drill.exec.store.hive.ColumnListsCache;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.junit.Test;
+
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+public class TestColumnListCache {
+
+ @Test
+ public void testTableColumnsIndex() {
+ ColumnListsCache cache = new ColumnListsCache();
+ List<FieldSchema> columns = Lists.newArrayList();
+ columns.add(new FieldSchema("f1", "int", null));
+ columns.add(new FieldSchema("f2", "int", null));
+ assertEquals(0, cache.addOrGet(columns));
+ }
+
+ @Test
+ public void testPartitionColumnsIndex() {
+ ColumnListsCache cache = new ColumnListsCache();
+ List<FieldSchema> columns = Lists.newArrayList();
+ columns.add(new FieldSchema("f1", "int", null));
+ columns.add(new FieldSchema("f2", "int", null));
+ cache.addOrGet(columns);
+ columns.add(new FieldSchema("f3", "int", null));
+ assertEquals(1, cache.addOrGet(columns));
+ }
+
+ @Test
+ public void testColumnListUnique() {
+ ColumnListsCache cache = new ColumnListsCache();
+ List<FieldSchema> columns = Lists.newArrayList();
+ columns.add(new FieldSchema("f1", "int", null));
+ columns.add(new FieldSchema("f2", "int", null));
+ cache.addOrGet(columns);
+ cache.addOrGet(Lists.newArrayList(columns));
+ assertEquals(0, cache.addOrGet(Lists.newArrayList(columns)));
+ }
+
+ @Test
+ public void testPartitionColumnListAccess() {
+ ColumnListsCache cache = new ColumnListsCache();
+ List<FieldSchema> columns = Lists.newArrayList();
+ columns.add(new FieldSchema("f1", "int", null));
+ columns.add(new FieldSchema("f2", "int", null));
+ cache.addOrGet(columns);
+ cache.addOrGet(columns);
+ columns.add(new FieldSchema("f3", "int", null));
+ cache.addOrGet(columns);
+ cache.addOrGet(columns);
+ columns.add(new FieldSchema("f4", "int", null));
+ cache.addOrGet(columns);
+ cache.addOrGet(columns);
+ assertEquals(columns, cache.getColumns(2));
+ }
+
+ @Test
+ public void testPartitionColumnCaching() {
+ ColumnListsCache cache = new ColumnListsCache();
+ List<FieldSchema> columns = Lists.newArrayList();
+ columns.add(new FieldSchema("f1", "int", null));
+ columns.add(new FieldSchema("f2", "int", null));
+ // sum of all indexes from cache
+ int indexSum = cache.addOrGet(columns);
+ indexSum += cache.addOrGet(columns);
+ List<FieldSchema> sameColumns = Lists.newArrayList(columns);
+ indexSum += cache.addOrGet(sameColumns);
+ List<FieldSchema> otherColumns = Lists.newArrayList();
+ otherColumns.add(new FieldSchema("f3", "int", null));
+ otherColumns.add(new FieldSchema("f4", "int", null));
+ // sum of all indexes from cache
+ int secondIndexSum = cache.addOrGet(otherColumns);
+ secondIndexSum += cache.addOrGet(otherColumns);
+ List<FieldSchema> sameOtherColumns = Lists.newArrayList();
+ sameOtherColumns.add(new FieldSchema("f3", "int", null));
+ sameOtherColumns.add(new FieldSchema("f4", "int", null));
+ secondIndexSum += cache.addOrGet(sameOtherColumns);
+ secondIndexSum += cache.addOrGet(Lists.newArrayList(sameOtherColumns));
+ secondIndexSum += cache.addOrGet(otherColumns);
+ secondIndexSum += cache.addOrGet(otherColumns);
+ indexSum += cache.addOrGet(sameColumns);
+ indexSum += cache.addOrGet(columns);
+ // added only two kinds of column lists
+ assertNull(cache.getColumns(3));
+ // sum of the indices of the first column list
+ assertEquals(0, indexSum);
+ assertEquals(6, secondIndexSum);
+ }
+}
[2/8] drill git commit: DRILL-5117: Compile error when query a json
file with 1000+columns
Posted by jn...@apache.org.
DRILL-5117: Compile error when query a json file with 1000+columns
close apache/drill#686
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/810198b1
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/810198b1
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/810198b1
Branch: refs/heads/master
Commit: 810198b18bfbe38712256c58b102bca079d934c1
Parents: 417ae93
Author: Serhii-Harnyk <se...@gmail.com>
Authored: Thu Dec 8 20:08:34 2016 +0000
Committer: Jinfeng Ni <jn...@apache.org>
Committed: Mon Dec 19 14:49:15 2016 -0800
----------------------------------------------------------------------
.../src/main/java/org/apache/drill/exec/expr/SizedJBlock.java | 5 ++++-
.../org/apache/drill/exec/compile/TestLargeFileCompilation.java | 4 ----
2 files changed, 4 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/810198b1/exec/java-exec/src/main/java/org/apache/drill/exec/expr/SizedJBlock.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/SizedJBlock.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/SizedJBlock.java
index cf110c6..5d806a3 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/SizedJBlock.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/SizedJBlock.java
@@ -32,7 +32,10 @@ public class SizedJBlock {
public SizedJBlock(JBlock block) {
this.block = block;
- this.count = 0;
+ // Project, Filter and Aggregator receives JBlock, using ClassGenerator.addExpr() method,
+ // but the Copier is doing kind of short-cut handling, by accessing the eval() and setup() directly.
+ // To take into account JBlocks, that were filled in Copier, sets count to 1.
+ this.count = 1;
}
public JBlock getBlock() {
http://git-wip-us.apache.org/repos/asf/drill/blob/810198b1/exec/java-exec/src/test/java/org/apache/drill/exec/compile/TestLargeFileCompilation.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/compile/TestLargeFileCompilation.java b/exec/java-exec/src/test/java/org/apache/drill/exec/compile/TestLargeFileCompilation.java
index 6c7fd9a..f892471 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/compile/TestLargeFileCompilation.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/compile/TestLargeFileCompilation.java
@@ -20,7 +20,6 @@ package org.apache.drill.exec.compile;
import org.apache.drill.BaseTestQuery;
import org.apache.drill.common.util.TestTools;
import org.apache.drill.exec.ExecConstants;
-import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestRule;
@@ -129,14 +128,12 @@ public class TestLargeFileCompilation extends BaseTestQuery {
}
@Test
- @Ignore("DRILL-1808")
public void testEXTERNAL_SORT() throws Exception {
testNoResult("alter session set `%s`='JDK'", QueryClassLoader.JAVA_COMPILER_OPTION);
testNoResult(ITERATION_COUNT, LARGE_QUERY_ORDER_BY);
}
@Test
- @Ignore("DRILL-1808")
public void testTOP_N_SORT() throws Exception {
testNoResult("alter session set `%s`='JDK'", QueryClassLoader.JAVA_COMPILER_OPTION);
testNoResult(ITERATION_COUNT, LARGE_QUERY_ORDER_BY_WITH_LIMIT);
@@ -153,5 +150,4 @@ public class TestLargeFileCompilation extends BaseTestQuery {
testNoResult("alter session set `%s`='JDK'", QueryClassLoader.JAVA_COMPILER_OPTION);
testNoResult(ITERATION_COUNT, LARGE_QUERY_SELECT_LIST);
}
-
}