From 8e47a8693ba5f193892b8e3135bee254d3012e11 Mon Sep 17 00:00:00 2001 From: Ramu Malur Date: Wed, 9 Dec 2015 16:21:52 +0530 Subject: [PATCH 1/5] Initial Version of Drill Interpreter --- drill/pom.xml | 133 +++++++++++ .../zeppelin/drill/DrillInterpreter.java | 212 ++++++++++++++++++ pom.xml | 1 + 3 files changed, 346 insertions(+) create mode 100644 drill/pom.xml create mode 100644 drill/src/main/java/org/apache/zeppelin/drill/DrillInterpreter.java diff --git a/drill/pom.xml b/drill/pom.xml new file mode 100644 index 00000000000..3eb0a8e8b9a --- /dev/null +++ b/drill/pom.xml @@ -0,0 +1,133 @@ + + + + 4.0.0 + + + zeppelin + org.apache.zeppelin + 0.6.0-incubating-SNAPSHOT + + + org.apache.zeppelin + zeppelin-drill + jar + 0.6.0-incubating-SNAPSHOT + Zeppelin: Drill Interpreter + http://www.apache.org + + + 1.2.0 + + + + + org.apache.zeppelin + zeppelin-interpreter + ${project.version} + provided + + + + org.slf4j + slf4j-api + + + + org.slf4j + slf4j-log4j12 + + + + org.apache.drill.exec + drill-jdbc + ${drill.version} + + + + junit + junit + test + + + + + + + + org.apache.maven.plugins + maven-deploy-plugin + 2.7 + + true + + + + + maven-enforcer-plugin + 1.3.1 + + + enforce + none + + + + + + maven-dependency-plugin + 2.8 + + + copy-dependencies + package + + copy-dependencies + + + ${project.build.directory}/../../interpreter/drill + false + false + true + runtime + + + + copy-artifact + package + + copy + + + ${project.build.directory}/../../interpreter/drill + false + false + true + runtime + + + ${project.groupId} + ${project.artifactId} + ${project.version} + ${project.packaging} + + + + + + + + + + diff --git a/drill/src/main/java/org/apache/zeppelin/drill/DrillInterpreter.java b/drill/src/main/java/org/apache/zeppelin/drill/DrillInterpreter.java new file mode 100644 index 00000000000..341410b4f79 --- /dev/null +++ b/drill/src/main/java/org/apache/zeppelin/drill/DrillInterpreter.java @@ -0,0 +1,212 @@ +/* + * 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.zeppelin.drill; + +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.ResultSetMetaData; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.List; +import java.util.Properties; + +import org.apache.zeppelin.interpreter.Interpreter; +import org.apache.zeppelin.interpreter.InterpreterContext; +import org.apache.zeppelin.interpreter.InterpreterPropertyBuilder; +import org.apache.zeppelin.interpreter.InterpreterResult; +import org.apache.zeppelin.interpreter.InterpreterResult.Code; +import org.apache.zeppelin.scheduler.Scheduler; +import org.apache.zeppelin.scheduler.SchedulerFactory; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Interpreter for Apache. Look at https://drill.apache.org for more details. + */ +public class DrillInterpreter + extends Interpreter { + + private static final String RS_ROW_DELIM = "\n"; + + private static final String RS_COL_DELIM = "\t"; + + private static final Logger LOGGER = LoggerFactory.getLogger(DrillInterpreter.class); + + private static final String DRILL_JDBC_DRIVER = "org.apache.drill.jdbc.Driver"; + + private static final String STORAGE_PLUGIN = "drill.storage.plugin"; + + private static final String ZK_CONNECT = "drill.zk.connect"; + + private static final String DIRECTORY = "drill.zk.directory"; + + private static final String CLUSTER_ID = "drill.cluster.id"; + + private static final String JDBC_URI_FORMAT = "jdbc:drill:schema=%s;zk=%s%s/%s"; + + private Connection connection; + + private Statement statement; + + private String initError; + + private boolean initialised; + + private String jdbcURI; + + // Register Drill Interpreter + static { + Interpreter.register("drill", "drill", DrillInterpreter.class.getName(), + new InterpreterPropertyBuilder().add(STORAGE_PLUGIN, "hive", "Storage Plugin") + .add(ZK_CONNECT, "localhost:2181", "Zookeeper Connect") + .add(DIRECTORY, "/Drill", "Drill Directory in Zookeeper") + .add(CLUSTER_ID, "drillbits1", "Drill Cluster ID").build()); + } + + /** + * @param properties + * Details of all Configured Properties + */ + public DrillInterpreter( + Properties properties) { + + super(properties); + this.jdbcURI = String.format(JDBC_URI_FORMAT, properties.getProperty(STORAGE_PLUGIN), + properties.getProperty(ZK_CONNECT), properties.getProperty(DIRECTORY), + properties.getProperty(CLUSTER_ID)); + LOGGER.info("Created DrillInterpreter for JDBC URI {}", this.jdbcURI); + } + + @Override + public void open() { + + LOGGER.info("Initializing Drill JDBC Connection"); + try { + Class.forName(DRILL_JDBC_DRIVER); + this.connection = DriverManager.getConnection(this.jdbcURI); + this.initialised = true; + LOGGER.info("Successfully Initialized Drill JDBC Connection"); + } catch (SQLException e) { + this.initError = e.getMessage(); + LOGGER.error("Cannot Initialize Drill Connection. Reason: {}", this.initError, e); + } catch (ClassNotFoundException e) { + LOGGER.error("Drill Driver Class {} not found in classpath", DRILL_JDBC_DRIVER, e); + } + } + + @Override + public InterpreterResult interpret( + String sql, + InterpreterContext arg1) { + + ResultSet results = null; + InterpreterResult interpreterResult = null; + try { + + if (!this.initialised) { + return new InterpreterResult(Code.ERROR, initError); + } + + this.statement = this.connection.createStatement(); + StringBuilder interpreterResultStr = new StringBuilder("%table "); + + results = this.statement.executeQuery(sql); + ResultSetMetaData rsMetaData = results.getMetaData(); + for (int i = 1; i < rsMetaData.getColumnCount() + 1; i++) { + if (i == 1) { + interpreterResultStr.append(rsMetaData.getColumnName(i)); + } else { + interpreterResultStr.append(RS_COL_DELIM + rsMetaData.getColumnName(i)); + } + } + interpreterResultStr.append(RS_ROW_DELIM); + + while (results.next()) { + for (int i = 1; i < rsMetaData.getColumnCount() + 1; i++) { + interpreterResultStr.append(results.getString(i) + RS_COL_DELIM); + } + interpreterResultStr.append(RS_ROW_DELIM); + } + + interpreterResult = new InterpreterResult(Code.SUCCESS, interpreterResultStr.toString()); + } catch (SQLException ex) { + LOGGER.error("Error running SQL: " + sql, ex); + return new InterpreterResult(Code.ERROR, ex.getMessage()); + } finally { + try { + if (results != null) { + results.close(); + } + if (statement != null) { + statement.close(); + } + } catch (Exception e) { + LOGGER.error("Error Closing Result Set. Ignoring."); + } + } + return interpreterResult; + } + + @Override + public void close() { + + try { + if (this.initialised && this.connection != null) { + this.connection.close(); + this.connection = null; + } + } catch (Exception e) { + LOGGER.error("Error while Closing Drill JDBC Connection", e); + } + } + + @Override + public FormType getFormType() { + + return FormType.SIMPLE; + } + + @Override + public int getProgress( + InterpreterContext context) { + + // TODO Return progress + return 0; + } + + @Override + public Scheduler getScheduler() { + + return SchedulerFactory.singleton() + .createOrGetFIFOScheduler(DrillInterpreter.class.getName() + this.hashCode()); + } + + @Override + public List completion( + String buf, + int cursor) { + + return null; + } + + @Override + public void cancel( + InterpreterContext arg0) { + + // TODO Support Cancel + } +} diff --git a/pom.xml b/pom.xml index 1984cca7913..f1fc47feb12 100755 --- a/pom.xml +++ b/pom.xml @@ -100,6 +100,7 @@ kylin lens cassandra + drill zeppelin-web zeppelin-server zeppelin-distribution From dbf464324cf1bed84d2e335017c16fff1c1666b9 Mon Sep 17 00:00:00 2001 From: Ramu Malur Date: Wed, 9 Dec 2015 16:28:00 +0530 Subject: [PATCH 2/5] Fixed Check style Error --- .../main/java/org/apache/zeppelin/drill/DrillInterpreter.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/drill/src/main/java/org/apache/zeppelin/drill/DrillInterpreter.java b/drill/src/main/java/org/apache/zeppelin/drill/DrillInterpreter.java index 341410b4f79..e4629b4fdd1 100644 --- a/drill/src/main/java/org/apache/zeppelin/drill/DrillInterpreter.java +++ b/drill/src/main/java/org/apache/zeppelin/drill/DrillInterpreter.java @@ -184,7 +184,7 @@ public FormType getFormType() { public int getProgress( InterpreterContext context) { - // TODO Return progress + // TODO(malur): Return progress return 0; } @@ -207,6 +207,6 @@ public List completion( public void cancel( InterpreterContext arg0) { - // TODO Support Cancel + // TODO(malur): Support Cancel } } From 485515954b5ecef96e13686d3fe6633405d7d375 Mon Sep 17 00:00:00 2001 From: Ramu Malur Date: Thu, 17 Dec 2015 15:11:10 +0530 Subject: [PATCH 3/5] Added Unit Tests --- drill/pom.xml | 39 +-- .../zeppelin/drill/DrillInterpreter.java | 126 +++++++--- .../zeppelin/drill/DrillInterpreterTest.java | 228 ++++++++++++++++++ 3 files changed, 347 insertions(+), 46 deletions(-) create mode 100644 drill/src/test/java/org/apache/zeppelin/drill/DrillInterpreterTest.java diff --git a/drill/pom.xml b/drill/pom.xml index 3eb0a8e8b9a..dd245416254 100644 --- a/drill/pom.xml +++ b/drill/pom.xml @@ -1,14 +1,14 @@ - @@ -29,6 +29,8 @@ 1.2.0 + 1.9.5 + 1.0.8 @@ -38,29 +40,36 @@ ${project.version} provided - org.slf4j slf4j-api - org.slf4j slf4j-log4j12 - org.apache.drill.exec drill-jdbc ${drill.version} - junit junit test - + + org.mockito + mockito-all + ${mockito.version} + test + + + com.mockrunner + mockrunner-jdbc + ${mockrunner-jdbc.version} + test + diff --git a/drill/src/main/java/org/apache/zeppelin/drill/DrillInterpreter.java b/drill/src/main/java/org/apache/zeppelin/drill/DrillInterpreter.java index e4629b4fdd1..be130709ffe 100644 --- a/drill/src/main/java/org/apache/zeppelin/drill/DrillInterpreter.java +++ b/drill/src/main/java/org/apache/zeppelin/drill/DrillInterpreter.java @@ -35,29 +35,43 @@ import org.slf4j.LoggerFactory; /** - * Interpreter for Apache. Look at https://drill.apache.org for more details. + * Interpreter for Apache Drill. Look at https://drill.apache.org for more details. + * + * @author malur */ public class DrillInterpreter extends Interpreter { - private static final String RS_ROW_DELIM = "\n"; - - private static final String RS_COL_DELIM = "\t"; - private static final Logger LOGGER = LoggerFactory.getLogger(DrillInterpreter.class); private static final String DRILL_JDBC_DRIVER = "org.apache.drill.jdbc.Driver"; - private static final String STORAGE_PLUGIN = "drill.storage.plugin"; + private static final String DEFAULT_ZK_CONNECT = "localhost:2181"; - private static final String ZK_CONNECT = "drill.zk.connect"; + private static final String DEFAULT_DRILL_DIRECTORY = "/Drill"; - private static final String DIRECTORY = "drill.zk.directory"; + private static final String DEFAULT_DRILL_CLUSTER_ID = "drillbits1"; - private static final String CLUSTER_ID = "drill.cluster.id"; + private static final String DEFAULT_STORAGE_PLUGIN = "hive"; private static final String JDBC_URI_FORMAT = "jdbc:drill:schema=%s;zk=%s%s/%s"; + private static final char RS_ROW_DELIM = '\n'; + + private static final char RS_COL_DELIM = '\t'; + + private static final char WHITESPACE = ' '; + + static final String STORAGE_PLUGIN = "drill.storage.plugin"; + + static final String ZK_CONNECT = "drill.zk.connect"; + + static final String DRILL_DIRECTORY = "drill.zk.directory"; + + static final String DRILL_CLUSTER_ID = "drill.cluster.id"; + + static final String EMPTY_COLUMN_VALUE = ""; + private Connection connection; private Statement statement; @@ -71,10 +85,11 @@ public class DrillInterpreter // Register Drill Interpreter static { Interpreter.register("drill", "drill", DrillInterpreter.class.getName(), - new InterpreterPropertyBuilder().add(STORAGE_PLUGIN, "hive", "Storage Plugin") - .add(ZK_CONNECT, "localhost:2181", "Zookeeper Connect") - .add(DIRECTORY, "/Drill", "Drill Directory in Zookeeper") - .add(CLUSTER_ID, "drillbits1", "Drill Cluster ID").build()); + new InterpreterPropertyBuilder() + .add(STORAGE_PLUGIN, DEFAULT_STORAGE_PLUGIN, "Storage Plugin") + .add(ZK_CONNECT, DEFAULT_ZK_CONNECT, "Zookeeper Connect") + .add(DRILL_DIRECTORY, DEFAULT_DRILL_DIRECTORY, "Drill Directory in Zookeeper") + .add(DRILL_CLUSTER_ID, DEFAULT_DRILL_CLUSTER_ID, "Drill Cluster ID").build()); } /** @@ -85,9 +100,11 @@ public DrillInterpreter( Properties properties) { super(properties); - this.jdbcURI = String.format(JDBC_URI_FORMAT, properties.getProperty(STORAGE_PLUGIN), - properties.getProperty(ZK_CONNECT), properties.getProperty(DIRECTORY), - properties.getProperty(CLUSTER_ID)); + this.jdbcURI = String.format(JDBC_URI_FORMAT, + properties.getProperty(STORAGE_PLUGIN, DEFAULT_STORAGE_PLUGIN), + properties.getProperty(ZK_CONNECT, DEFAULT_ZK_CONNECT), + properties.getProperty(DRILL_DIRECTORY, DEFAULT_DRILL_DIRECTORY), + properties.getProperty(DRILL_CLUSTER_ID, DEFAULT_DRILL_CLUSTER_ID)); LOGGER.info("Created DrillInterpreter for JDBC URI {}", this.jdbcURI); } @@ -95,6 +112,10 @@ public DrillInterpreter( public void open() { LOGGER.info("Initializing Drill JDBC Connection"); + + // Close if open already + this.close(); + try { Class.forName(DRILL_JDBC_DRIVER); this.connection = DriverManager.getConnection(this.jdbcURI); @@ -111,7 +132,7 @@ public void open() { @Override public InterpreterResult interpret( String sql, - InterpreterContext arg1) { + InterpreterContext interpreterContext) { ResultSet results = null; InterpreterResult interpreterResult = null; @@ -121,23 +142,27 @@ public InterpreterResult interpret( return new InterpreterResult(Code.ERROR, initError); } - this.statement = this.connection.createStatement(); + this.statement = getConnection().createStatement(); StringBuilder interpreterResultStr = new StringBuilder("%table "); results = this.statement.executeQuery(sql); - ResultSetMetaData rsMetaData = results.getMetaData(); - for (int i = 1; i < rsMetaData.getColumnCount() + 1; i++) { - if (i == 1) { - interpreterResultStr.append(rsMetaData.getColumnName(i)); - } else { - interpreterResultStr.append(RS_COL_DELIM + rsMetaData.getColumnName(i)); + + ResultSetMetaData md = results.getMetaData(); + + for (int i = 1; i < md.getColumnCount() + 1; i++) { + if (i > 1) { + interpreterResultStr.append(RS_COL_DELIM); } + interpreterResultStr.append(replaceReservedChars(true, md.getColumnName(i))); } interpreterResultStr.append(RS_ROW_DELIM); while (results.next()) { - for (int i = 1; i < rsMetaData.getColumnCount() + 1; i++) { - interpreterResultStr.append(results.getString(i) + RS_COL_DELIM); + for (int i = 1; i < md.getColumnCount() + 1; i++) { + interpreterResultStr.append(replaceReservedChars(true, results.getString(i))); + if (i != md.getColumnCount()) { + interpreterResultStr.append(RS_COL_DELIM); + } } interpreterResultStr.append(RS_ROW_DELIM); } @@ -165,12 +190,13 @@ public InterpreterResult interpret( public void close() { try { - if (this.initialised && this.connection != null) { - this.connection.close(); - this.connection = null; + if (getConnection() != null) { + getConnection().close(); } } catch (Exception e) { LOGGER.error("Error while Closing Drill JDBC Connection", e); + } finally { + this.initError = null; } } @@ -184,7 +210,6 @@ public FormType getFormType() { public int getProgress( InterpreterContext context) { - // TODO(malur): Return progress return 0; } @@ -207,6 +232,45 @@ public List completion( public void cancel( InterpreterContext arg0) { - // TODO(malur): Support Cancel + if (statement != null) { + try { + statement.cancel(); + } catch (SQLException e) { + LOGGER.error("Error while Cancelling the SQL execution", e); + } + } } + + private String replaceReservedChars( + boolean isTableResponseType, + String str) { + + if (str == null) { + return EMPTY_COLUMN_VALUE; + } + return (!isTableResponseType) ? str + : str.replace(RS_COL_DELIM, WHITESPACE).replace(RS_ROW_DELIM, WHITESPACE); + } + + boolean isInitialised() { + + return initialised; + } + + // Test only methods + protected Connection getConnection() { + + return connection; + } + + String getJdbcURI() { + + return this.jdbcURI; + } + + String getInitError() { + + return this.initError; + } + } diff --git a/drill/src/test/java/org/apache/zeppelin/drill/DrillInterpreterTest.java b/drill/src/test/java/org/apache/zeppelin/drill/DrillInterpreterTest.java new file mode 100644 index 00000000000..24e84e69603 --- /dev/null +++ b/drill/src/test/java/org/apache/zeppelin/drill/DrillInterpreterTest.java @@ -0,0 +1,228 @@ + +/* + * 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.zeppelin.drill; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.sql.SQLException; +import java.util.Properties; + +import org.apache.zeppelin.interpreter.InterpreterResult; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import com.mockrunner.jdbc.BasicJDBCTestCaseAdapter; +import com.mockrunner.jdbc.StatementResultSetHandler; +import com.mockrunner.mock.jdbc.MockConnection; +import com.mockrunner.mock.jdbc.MockResultSet; + +/** + * Test cases for {@link DrillInterpreter} + * + * @author malur + */ +public class DrillInterpreterTest + extends BasicJDBCTestCaseAdapter { + + private MockResultSet result = null; + + private Properties properties = null; + + private MockConnection connection = null; + + @Before + public void setup() { + + this.connection = getJDBCMockObjectFactory().getMockConnection(); + + StatementResultSetHandler statementHandler = connection.getStatementResultSetHandler(); + this.result = statementHandler.createResultSet(); + statementHandler.prepareGlobalResultSet(result); + + this.properties = new Properties(); + this.properties.put(DrillInterpreter.ZK_CONNECT, "zkhost:2181"); + this.properties.put(DrillInterpreter.STORAGE_PLUGIN, "hive"); + this.properties.put(DrillInterpreter.DRILL_DIRECTORY, "/DrillDir"); + this.properties.put(DrillInterpreter.DRILL_CLUSTER_ID, "drillbits"); + } + + @Test + public void testConstructor() { + + DrillInterpreter drillInterpreter = new DrillInterpreter(this.properties); + + Assert.assertFalse("DrillInterpreter should not have been initialized", + drillInterpreter.isInitialised()); + Assert.assertNull(drillInterpreter.getInitError()); + Assert.assertNull(drillInterpreter.getConnection()); + Assert.assertEquals("Invalid JDBC URI", + "jdbc:drill:schema=hive;zk=zkhost:2181/DrillDir/drillbits", drillInterpreter.getJdbcURI()); + } + + @Test + public void testConstructorForDefaultProperties() { + + DrillInterpreter drillInterpreter = new DrillInterpreter(new Properties()); + + Assert.assertFalse("DrillInterpreter should not have been initialized", + drillInterpreter.isInitialised()); + Assert.assertNull(drillInterpreter.getConnection()); + Assert.assertEquals("Invalid JDBC URI", + "jdbc:drill:schema=hive;zk=localhost:2181/Drill/drillbits1", drillInterpreter.getJdbcURI()); + } + + @Test + public void testOpen() { + + DrillInterpreter drillInterpreter = spy(new DrillInterpreter(this.properties)); + when(drillInterpreter.getConnection()).thenReturn(this.connection); + + drillInterpreter.open(); + + Assert.assertTrue("DrillInterpreter should have been initialized", + drillInterpreter.isInitialised()); + Assert.assertNotNull(drillInterpreter.getConnection()); + } + + @Test + public void testOpenMultipleTimes() { + + DrillInterpreter drillInterpreter = spy(new DrillInterpreter(this.properties)); + when(drillInterpreter.getConnection()).thenReturn(this.connection); + + drillInterpreter.open(); + drillInterpreter.open(); + drillInterpreter.open(); + + verify(drillInterpreter, times(3)).open(); + verify(drillInterpreter, times(3)).close(); + } + + @Test + public void testConnectionClose() + throws SQLException { + + DrillInterpreter drillInterpreter = spy(new DrillInterpreter(this.properties)); + when(drillInterpreter.getConnection()).thenReturn(this.connection); + + drillInterpreter.close(); + + verifyAllResultSetsClosed(); + verifyAllStatementsClosed(); + verifyConnectionClosed(); + } + + @Test + public void testStatementCancel() + throws SQLException { + + DrillInterpreter drillInterpreter = spy(new DrillInterpreter(this.properties)); + when(drillInterpreter.getConnection()).thenReturn(this.connection); + + drillInterpreter.cancel(null); + + verifyAllResultSetsClosed(); + verifyAllStatementsClosed(); + assertFalse("Cancel operation should not close the connection", + drillInterpreter.getConnection().isClosed()); + } + + @Test + public void testSelectQuery() + throws SQLException { + + DrillInterpreter drillInterpreter = spy(new DrillInterpreter(this.properties)); + when(drillInterpreter.getConnection()).thenReturn(this.connection); + drillInterpreter.open(); + + String sqlQuery = "select * from t"; + + result.addColumn("col1", new String[] { "val11", "val12" }); + result.addColumn("col2", new String[] { "val21", "val22" }); + + InterpreterResult interpreterResult = drillInterpreter.interpret(sqlQuery, null); + + assertEquals(InterpreterResult.Code.SUCCESS, interpreterResult.code()); + assertEquals(InterpreterResult.Type.TABLE, interpreterResult.type()); + assertEquals("col1\tcol2\nval11\tval21\nval12\tval22\n", interpreterResult.message()); + + drillInterpreter.close(); + + verifySQLStatementExecuted(sqlQuery); + verifyAllResultSetsClosed(); + verifyAllStatementsClosed(); + } + + @Test + public void testNullColumnResult() + throws SQLException { + + DrillInterpreter drillInterpreter = spy(new DrillInterpreter(this.properties)); + when(drillInterpreter.getConnection()).thenReturn(this.connection); + drillInterpreter.open(); + + String sqlQuery = "select * from t"; + + result.addColumn("col1", new String[] { "val11", null }); + result.addColumn("col2", new String[] { null, "val22" }); + + InterpreterResult interpreterResult = drillInterpreter.interpret(sqlQuery, null); + + assertEquals(InterpreterResult.Code.SUCCESS, interpreterResult.code()); + assertEquals(InterpreterResult.Type.TABLE, interpreterResult.type()); + assertEquals("col1\tcol2\nval11\t\n\tval22\n", interpreterResult.message()); + + drillInterpreter.close(); + + verifySQLStatementExecuted(sqlQuery); + verifyAllResultSetsClosed(); + verifyAllStatementsClosed(); + } + + @Test + public void testSelectQueryWithSpecialCharacters() + throws SQLException { + + DrillInterpreter drillInterpreter = spy(new DrillInterpreter(this.properties)); + when(drillInterpreter.getConnection()).thenReturn(this.connection); + drillInterpreter.open(); + + String sqlQuery = "select * from t"; + + result.addColumn("co\tl1", new String[] { "val11", "va\tl1\n2" }); + result.addColumn("co\nl2", new String[] { "v\nal21", "val\t22" }); + + InterpreterResult interpreterResult = drillInterpreter.interpret(sqlQuery, null); + + assertEquals(InterpreterResult.Code.SUCCESS, interpreterResult.code()); + assertEquals(InterpreterResult.Type.TABLE, interpreterResult.type()); + assertEquals("co l1\tco l2\nval11\tv al21\nva l1 2\tval 22\n", interpreterResult.message()); + + drillInterpreter.close(); + + verifySQLStatementExecuted(sqlQuery); + verifyAllResultSetsClosed(); + verifyAllStatementsClosed(); + } + +} From 5f81b2e048143d496179d7dfe88b641087f872b3 Mon Sep 17 00:00:00 2001 From: Ramu Malur Date: Mon, 28 Dec 2015 14:26:27 +0530 Subject: [PATCH 4/5] Added support for maxResult --- conf/zeppelin-site.xml.template | 2 +- .../apache/zeppelin/drill/DrillInterpreter.java | 16 +++++++++++++++- .../zeppelin/drill/DrillInterpreterTest.java | 5 +++++ 3 files changed, 21 insertions(+), 2 deletions(-) diff --git a/conf/zeppelin-site.xml.template b/conf/zeppelin-site.xml.template index b6aca75d626..959a1e9f283 100755 --- a/conf/zeppelin-site.xml.template +++ b/conf/zeppelin-site.xml.template @@ -105,7 +105,7 @@ zeppelin.interpreters - org.apache.zeppelin.spark.SparkInterpreter,org.apache.zeppelin.spark.PySparkInterpreter,org.apache.zeppelin.spark.SparkSqlInterpreter,org.apache.zeppelin.spark.DepInterpreter,org.apache.zeppelin.markdown.Markdown,org.apache.zeppelin.angular.AngularInterpreter,org.apache.zeppelin.shell.ShellInterpreter,org.apache.zeppelin.hive.HiveInterpreter,org.apache.zeppelin.tajo.TajoInterpreter,org.apache.zeppelin.flink.FlinkInterpreter,org.apache.zeppelin.lens.LensInterpreter,org.apache.zeppelin.ignite.IgniteInterpreter,org.apache.zeppelin.ignite.IgniteSqlInterpreter,org.apache.zeppelin.cassandra.CassandraInterpreter,org.apache.zeppelin.geode.GeodeOqlInterpreter,org.apache.zeppelin.postgresql.PostgreSqlInterpreter,org.apache.zeppelin.phoenix.PhoenixInterpreter,org.apache.zeppelin.kylin.KylinInterpreter,org.apache.zeppelin.elasticsearch.ElasticsearchInterpreter + org.apache.zeppelin.spark.SparkInterpreter,org.apache.zeppelin.spark.PySparkInterpreter,org.apache.zeppelin.spark.SparkSqlInterpreter,org.apache.zeppelin.spark.DepInterpreter,org.apache.zeppelin.markdown.Markdown,org.apache.zeppelin.angular.AngularInterpreter,org.apache.zeppelin.shell.ShellInterpreter,org.apache.zeppelin.hive.HiveInterpreter,org.apache.zeppelin.tajo.TajoInterpreter,org.apache.zeppelin.flink.FlinkInterpreter,org.apache.zeppelin.lens.LensInterpreter,org.apache.zeppelin.ignite.IgniteInterpreter,org.apache.zeppelin.ignite.IgniteSqlInterpreter,org.apache.zeppelin.cassandra.CassandraInterpreter,org.apache.zeppelin.geode.GeodeOqlInterpreter,org.apache.zeppelin.postgresql.PostgreSqlInterpreter,org.apache.zeppelin.phoenix.PhoenixInterpreter,org.apache.zeppelin.kylin.KylinInterpreter,org.apache.zeppelin.elasticsearch.ElasticsearchInterpreter,org.apache.zeppelin.drill.DrillInterpreter Comma separated interpreter configurations. First interpreter become a default diff --git a/drill/src/main/java/org/apache/zeppelin/drill/DrillInterpreter.java b/drill/src/main/java/org/apache/zeppelin/drill/DrillInterpreter.java index be130709ffe..44a3b8c812a 100644 --- a/drill/src/main/java/org/apache/zeppelin/drill/DrillInterpreter.java +++ b/drill/src/main/java/org/apache/zeppelin/drill/DrillInterpreter.java @@ -70,6 +70,10 @@ public class DrillInterpreter static final String DRILL_CLUSTER_ID = "drill.cluster.id"; + static final String DRILL_MAX_RESULT = "drill.max.result"; + + static final String DEFAULT_MAX_RESULT = "1000"; + static final String EMPTY_COLUMN_VALUE = ""; private Connection connection; @@ -82,6 +86,8 @@ public class DrillInterpreter private String jdbcURI; + private int maxResult; + // Register Drill Interpreter static { Interpreter.register("drill", "drill", DrillInterpreter.class.getName(), @@ -89,7 +95,9 @@ public class DrillInterpreter .add(STORAGE_PLUGIN, DEFAULT_STORAGE_PLUGIN, "Storage Plugin") .add(ZK_CONNECT, DEFAULT_ZK_CONNECT, "Zookeeper Connect") .add(DRILL_DIRECTORY, DEFAULT_DRILL_DIRECTORY, "Drill Directory in Zookeeper") - .add(DRILL_CLUSTER_ID, DEFAULT_DRILL_CLUSTER_ID, "Drill Cluster ID").build()); + .add(DRILL_CLUSTER_ID, DEFAULT_DRILL_CLUSTER_ID, "Drill Cluster ID") + .add(DRILL_MAX_RESULT, DEFAULT_MAX_RESULT, "Max number of SQL result to display.") + .build()); } /** @@ -105,6 +113,7 @@ public DrillInterpreter( properties.getProperty(ZK_CONNECT, DEFAULT_ZK_CONNECT), properties.getProperty(DRILL_DIRECTORY, DEFAULT_DRILL_DIRECTORY), properties.getProperty(DRILL_CLUSTER_ID, DEFAULT_DRILL_CLUSTER_ID)); + this.maxResult = Integer.valueOf(properties.getProperty(DRILL_MAX_RESULT, DEFAULT_MAX_RESULT)); LOGGER.info("Created DrillInterpreter for JDBC URI {}", this.jdbcURI); } @@ -143,6 +152,7 @@ public InterpreterResult interpret( } this.statement = getConnection().createStatement(); + this.statement.setMaxRows(this.maxResult); StringBuilder interpreterResultStr = new StringBuilder("%table "); results = this.statement.executeQuery(sql); @@ -273,4 +283,8 @@ String getInitError() { return this.initError; } + int getMaxResult() { + + return this.maxResult; + } } diff --git a/drill/src/test/java/org/apache/zeppelin/drill/DrillInterpreterTest.java b/drill/src/test/java/org/apache/zeppelin/drill/DrillInterpreterTest.java index 24e84e69603..82f022a25c6 100644 --- a/drill/src/test/java/org/apache/zeppelin/drill/DrillInterpreterTest.java +++ b/drill/src/test/java/org/apache/zeppelin/drill/DrillInterpreterTest.java @@ -64,6 +64,7 @@ public void setup() { this.properties.put(DrillInterpreter.STORAGE_PLUGIN, "hive"); this.properties.put(DrillInterpreter.DRILL_DIRECTORY, "/DrillDir"); this.properties.put(DrillInterpreter.DRILL_CLUSTER_ID, "drillbits"); + this.properties.put(DrillInterpreter.DRILL_MAX_RESULT, "100"); } @Test @@ -77,6 +78,7 @@ public void testConstructor() { Assert.assertNull(drillInterpreter.getConnection()); Assert.assertEquals("Invalid JDBC URI", "jdbc:drill:schema=hive;zk=zkhost:2181/DrillDir/drillbits", drillInterpreter.getJdbcURI()); + Assert.assertEquals("Wrong value for max result", 100, drillInterpreter.getMaxResult()); } @Test @@ -89,6 +91,9 @@ public void testConstructorForDefaultProperties() { Assert.assertNull(drillInterpreter.getConnection()); Assert.assertEquals("Invalid JDBC URI", "jdbc:drill:schema=hive;zk=localhost:2181/Drill/drillbits1", drillInterpreter.getJdbcURI()); + Assert.assertEquals("Wrong default value for max result", + Integer.valueOf(DrillInterpreter.DEFAULT_MAX_RESULT).intValue(), + drillInterpreter.getMaxResult()); } @Test From c0bcf1cccf5c8376c4b95cfef73c7a54fb39b2e6 Mon Sep 17 00:00:00 2001 From: Ramu Malur Date: Mon, 4 Jan 2016 14:16:26 +0530 Subject: [PATCH 5/5] Formatted zeppelin-site.xml.template properly --- conf/zeppelin-site.xml.template | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/conf/zeppelin-site.xml.template b/conf/zeppelin-site.xml.template index 59ea69538fb..10fce9df305 100755 --- a/conf/zeppelin-site.xml.template +++ b/conf/zeppelin-site.xml.template @@ -105,7 +105,7 @@ zeppelin.interpreters -org.apache.zeppelin.spark.SparkInterpreter,org.apache.zeppelin.spark.PySparkInterpreter,org.apache.zeppelin.spark.SparkSqlInterpreter,org.apache.zeppelin.spark.DepInterpreter,org.apache.zeppelin.markdown.Markdown,org.apache.zeppelin.angular.AngularInterpreter,org.apache.zeppelin.shell.ShellInterpreter,org.apache.zeppelin.hive.HiveInterpreter,org.apache.zeppelin.tajo.TajoInterpreter,org.apache.zeppelin.flink.FlinkInterpreter,org.apache.zeppelin.lens.LensInterpreter,org.apache.zeppelin.ignite.IgniteInterpreter,org.apache.zeppelin.ignite.IgniteSqlInterpreter,org.apache.zeppelin.cassandra.CassandraInterpreter,org.apache.zeppelin.geode.GeodeOqlInterpreter,org.apache.zeppelin.postgresql.PostgreSqlInterpreter,org.apache.zeppelin.phoenix.PhoenixInterpreter,org.apache.zeppelin.kylin.KylinInterpreter,org.apache.zeppelin.elasticsearch.ElasticsearchInterpreter,org.apache.zeppelin.scalding.ScaldingInterpreter,org.apache.zeppelin.drill.DrillInterpreter + org.apache.zeppelin.spark.SparkInterpreter,org.apache.zeppelin.spark.PySparkInterpreter,org.apache.zeppelin.spark.SparkSqlInterpreter,org.apache.zeppelin.spark.DepInterpreter,org.apache.zeppelin.markdown.Markdown,org.apache.zeppelin.angular.AngularInterpreter,org.apache.zeppelin.shell.ShellInterpreter,org.apache.zeppelin.hive.HiveInterpreter,org.apache.zeppelin.tajo.TajoInterpreter,org.apache.zeppelin.flink.FlinkInterpreter,org.apache.zeppelin.lens.LensInterpreter,org.apache.zeppelin.ignite.IgniteInterpreter,org.apache.zeppelin.ignite.IgniteSqlInterpreter,org.apache.zeppelin.cassandra.CassandraInterpreter,org.apache.zeppelin.geode.GeodeOqlInterpreter,org.apache.zeppelin.postgresql.PostgreSqlInterpreter,org.apache.zeppelin.phoenix.PhoenixInterpreter,org.apache.zeppelin.kylin.KylinInterpreter,org.apache.zeppelin.elasticsearch.ElasticsearchInterpreter,org.apache.zeppelin.scalding.ScaldingInterpreter,org.apache.zeppelin.drill.DrillInterpreter Comma separated interpreter configurations. First interpreter become a default