diff --git a/conf/zeppelin-site.xml.template b/conf/zeppelin-site.xml.template index 74fa2e76053..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.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 diff --git a/drill/pom.xml b/drill/pom.xml new file mode 100644 index 00000000000..dd245416254 --- /dev/null +++ b/drill/pom.xml @@ -0,0 +1,142 @@ + + + + 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 + 1.9.5 + 1.0.8 + + + + + 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.mockito + mockito-all + ${mockito.version} + test + + + com.mockrunner + mockrunner-jdbc + ${mockrunner-jdbc.version} + 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..44a3b8c812a --- /dev/null +++ b/drill/src/main/java/org/apache/zeppelin/drill/DrillInterpreter.java @@ -0,0 +1,290 @@ +/* + * 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 Drill. Look at https://drill.apache.org for more details. + * + * @author malur + */ +public class DrillInterpreter + extends Interpreter { + + 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 DEFAULT_ZK_CONNECT = "localhost:2181"; + + private static final String DEFAULT_DRILL_DIRECTORY = "/Drill"; + + private static final String DEFAULT_DRILL_CLUSTER_ID = "drillbits1"; + + 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 DRILL_MAX_RESULT = "drill.max.result"; + + static final String DEFAULT_MAX_RESULT = "1000"; + + static final String EMPTY_COLUMN_VALUE = ""; + + private Connection connection; + + private Statement statement; + + private String initError; + + private boolean initialised; + + private String jdbcURI; + + private int maxResult; + + // Register Drill Interpreter + static { + Interpreter.register("drill", "drill", DrillInterpreter.class.getName(), + 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") + .add(DRILL_MAX_RESULT, DEFAULT_MAX_RESULT, "Max number of SQL result to display.") + .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, 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)); + this.maxResult = Integer.valueOf(properties.getProperty(DRILL_MAX_RESULT, DEFAULT_MAX_RESULT)); + LOGGER.info("Created DrillInterpreter for JDBC URI {}", this.jdbcURI); + } + + @Override + 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); + 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 interpreterContext) { + + ResultSet results = null; + InterpreterResult interpreterResult = null; + try { + + if (!this.initialised) { + return new InterpreterResult(Code.ERROR, initError); + } + + this.statement = getConnection().createStatement(); + this.statement.setMaxRows(this.maxResult); + StringBuilder interpreterResultStr = new StringBuilder("%table "); + + results = this.statement.executeQuery(sql); + + 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 < 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); + } + + 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 (getConnection() != null) { + getConnection().close(); + } + } catch (Exception e) { + LOGGER.error("Error while Closing Drill JDBC Connection", e); + } finally { + this.initError = null; + } + } + + @Override + public FormType getFormType() { + + return FormType.SIMPLE; + } + + @Override + public int getProgress( + InterpreterContext context) { + + 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) { + + 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; + } + + 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 new file mode 100644 index 00000000000..82f022a25c6 --- /dev/null +++ b/drill/src/test/java/org/apache/zeppelin/drill/DrillInterpreterTest.java @@ -0,0 +1,233 @@ + +/* + * 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"); + this.properties.put(DrillInterpreter.DRILL_MAX_RESULT, "100"); + } + + @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()); + Assert.assertEquals("Wrong value for max result", 100, drillInterpreter.getMaxResult()); + } + + @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()); + Assert.assertEquals("Wrong default value for max result", + Integer.valueOf(DrillInterpreter.DEFAULT_MAX_RESULT).intValue(), + drillInterpreter.getMaxResult()); + } + + @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(); + } + +} diff --git a/pom.xml b/pom.xml index 88d38aa85f6..9092bb799ff 100755 --- a/pom.xml +++ b/pom.xml @@ -100,6 +100,7 @@ kylin lens cassandra + drill elasticsearch zeppelin-web zeppelin-server