[STORM-1585] Address review comments
Project: http://git-wip-us.apache.org/repos/asf/storm/repo Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/8842b731 Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/8842b731 Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/8842b731 Branch: refs/heads/master Commit: 8842b73184e6b4b0bfa28778db44949693d65258 Parents: 0047279 Author: Arun Mahadevan <[email protected]> Authored: Mon Mar 21 14:30:19 2016 +0530 Committer: Arun Mahadevan <[email protected]> Committed: Mon Mar 21 14:34:27 2016 +0530 ---------------------------------------------------------------------- .../storm-sql-core/src/codegen/data/Parser.tdd | 1 + .../src/codegen/includes/parserImpls.ftl | 12 ++++-- .../jvm/org/apache/storm/sql/StormSqlImpl.java | 3 ++ .../storm/sql/parser/SqlCreateFunction.java | 30 ++++++++++++- .../test/org/apache/storm/sql/TestStormSql.java | 44 +++++++++++++++++++- 5 files changed, 83 insertions(+), 7 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/storm/blob/8842b731/external/sql/storm-sql-core/src/codegen/data/Parser.tdd ---------------------------------------------------------------------- diff --git a/external/sql/storm-sql-core/src/codegen/data/Parser.tdd b/external/sql/storm-sql-core/src/codegen/data/Parser.tdd index 2ddf111..1e921c7 100644 --- a/external/sql/storm-sql-core/src/codegen/data/Parser.tdd +++ b/external/sql/storm-sql-core/src/codegen/data/Parser.tdd @@ -34,6 +34,7 @@ "OUTPUTFORMAT", "STORED", "TBLPROPERTIES", + "JAR" ] # List of methods for parsing custom SQL statements. http://git-wip-us.apache.org/repos/asf/storm/blob/8842b731/external/sql/storm-sql-core/src/codegen/includes/parserImpls.ftl ---------------------------------------------------------------------- diff --git a/external/sql/storm-sql-core/src/codegen/includes/parserImpls.ftl b/external/sql/storm-sql-core/src/codegen/includes/parserImpls.ftl index c26d3a7..0013231 100644 --- a/external/sql/storm-sql-core/src/codegen/includes/parserImpls.ftl +++ b/external/sql/storm-sql-core/src/codegen/includes/parserImpls.ftl @@ -93,13 +93,19 @@ SqlNode SqlCreateFunction() : SqlParserPos pos; SqlIdentifier functionName; SqlNode className; + SqlNode jarName = null; } { <CREATE> { pos = getPos(); } <FUNCTION> - functionName = CompoundIdentifier() + functionName = CompoundIdentifier() <AS> - className = StringLiteral() { - return new SqlCreateFunction(pos, functionName, className); + className = StringLiteral() + [ + <USING> <JAR> + jarName = StringLiteral() + ] + { + return new SqlCreateFunction(pos, functionName, className, jarName); } } \ No newline at end of file http://git-wip-us.apache.org/repos/asf/storm/blob/8842b731/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/StormSqlImpl.java ---------------------------------------------------------------------- diff --git a/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/StormSqlImpl.java b/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/StormSqlImpl.java index b4bba8e..bf5a921 100644 --- a/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/StormSqlImpl.java +++ b/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/StormSqlImpl.java @@ -165,6 +165,9 @@ class StormSqlImpl extends StormSql { } private void handleCreateFunction(SqlCreateFunction sqlCreateFunction) throws ClassNotFoundException { + if(sqlCreateFunction.jarName() != null) { + throw new UnsupportedOperationException("UDF 'USING JAR' not implemented"); + } schema.add(sqlCreateFunction.functionName().toUpperCase(), ScalarFunctionImpl.create(Class.forName(sqlCreateFunction.className()), "evaluate")); hasUdf = true; http://git-wip-us.apache.org/repos/asf/storm/blob/8842b731/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/parser/SqlCreateFunction.java ---------------------------------------------------------------------- diff --git a/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/parser/SqlCreateFunction.java b/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/parser/SqlCreateFunction.java index 5dcd7d1..a53802c 100644 --- a/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/parser/SqlCreateFunction.java +++ b/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/parser/SqlCreateFunction.java @@ -1,3 +1,20 @@ +/** + * 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.storm.sql.parser; import org.apache.calcite.sql.SqlCall; @@ -21,7 +38,7 @@ public class SqlCreateFunction extends SqlCall { public SqlCall createCall( SqlLiteral functionQualifier, SqlParserPos pos, SqlNode... o) { assert functionQualifier == null; - return new SqlCreateFunction(pos, (SqlIdentifier) o[0], o[1]); + return new SqlCreateFunction(pos, (SqlIdentifier) o[0], o[1], o[2]); } @Override @@ -30,16 +47,21 @@ public class SqlCreateFunction extends SqlCall { SqlCreateFunction t = (SqlCreateFunction) call; UnparseUtil u = new UnparseUtil(writer, leftPrec, rightPrec); u.keyword("CREATE", "FUNCTION").node(t.functionName).keyword("AS").node(t.className); + if (t.jarName != null) { + u.keyword("USING", "JAR").node(t.jarName); + } } }; private final SqlIdentifier functionName; private final SqlNode className; + private final SqlNode jarName; - public SqlCreateFunction(SqlParserPos pos, SqlIdentifier functionName, SqlNode className) { + public SqlCreateFunction(SqlParserPos pos, SqlIdentifier functionName, SqlNode className, SqlNode jarName) { super(pos); this.functionName = functionName; this.className = className; + this.jarName = jarName; } @Override @@ -65,4 +87,8 @@ public class SqlCreateFunction extends SqlCall { public String className() { return ((NlsString)SqlLiteral.value(className)).getValue(); } + + public String jarName() { + return jarName == null ? null : ((NlsString)SqlLiteral.value(jarName)).getValue(); + } } http://git-wip-us.apache.org/repos/asf/storm/blob/8842b731/external/sql/storm-sql-core/src/test/org/apache/storm/sql/TestStormSql.java ---------------------------------------------------------------------- diff --git a/external/sql/storm-sql-core/src/test/org/apache/storm/sql/TestStormSql.java b/external/sql/storm-sql-core/src/test/org/apache/storm/sql/TestStormSql.java index ce1e27f..872b83d 100644 --- a/external/sql/storm-sql-core/src/test/org/apache/storm/sql/TestStormSql.java +++ b/external/sql/storm-sql-core/src/test/org/apache/storm/sql/TestStormSql.java @@ -6,9 +6,9 @@ * 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. @@ -18,6 +18,8 @@ package org.apache.storm.sql; import com.google.common.collect.ImmutableMap; +import org.apache.calcite.sql.validate.SqlValidatorException; +import org.apache.calcite.tools.ValidationException; import org.apache.storm.sql.runtime.ChannelHandler; import org.apache.storm.sql.runtime.DataSource; import org.apache.storm.sql.runtime.DataSourcesProvider; @@ -135,6 +137,33 @@ public class TestStormSql { Assert.assertEquals(0, values.size()); } + @Test(expected = ValidationException.class) + public void testExternalUdfType() throws Exception { + List<String> stmt = new ArrayList<>(); + stmt.add("CREATE EXTERNAL TABLE FOO (ID INT, NAME VARCHAR) LOCATION 'mock:///foo'"); + stmt.add("CREATE FUNCTION MYPLUS AS 'org.apache.storm.sql.TestUtils$MyPlus'"); + stmt.add("SELECT STREAM MYPLUS(NAME, 1) FROM FOO WHERE ID = 0"); + StormSql sql = StormSql.construct(); + List<Values> values = new ArrayList<>(); + ChannelHandler h = new TestUtils.CollectDataChannelHandler(values); + sql.execute(stmt, h); + System.out.println(values); + + } + + @Test + public void testExternalUdfType2() throws Exception { + List<String> stmt = new ArrayList<>(); + stmt.add("CREATE EXTERNAL TABLE FOO (ID INT, NAME VARCHAR) LOCATION 'mock:///foo'"); + stmt.add("CREATE FUNCTION MYPLUS AS 'org.apache.storm.sql.TestUtils$MyPlus'"); + stmt.add("SELECT STREAM ID FROM FOO WHERE MYPLUS(ID, 1) = 'x'"); + StormSql sql = StormSql.construct(); + List<Values> values = new ArrayList<>(); + ChannelHandler h = new TestUtils.CollectDataChannelHandler(values); + sql.execute(stmt, h); + Assert.assertEquals(0, values.size()); + } + @Test public void testExternalUdf() throws Exception { List<String> stmt = new ArrayList<>(); @@ -150,4 +179,15 @@ public class TestStormSql { Assert.assertEquals(5, values.get(1).get(0)); } + @Test(expected = UnsupportedOperationException.class) + public void testExternalUdfUsingJar() throws Exception { + List<String> stmt = new ArrayList<>(); + stmt.add("CREATE EXTERNAL TABLE FOO (ID INT) LOCATION 'mock:///foo'"); + stmt.add("CREATE FUNCTION MYPLUS AS 'org.apache.storm.sql.TestUtils$MyPlus' USING JAR 'foo'"); + stmt.add("SELECT STREAM MYPLUS(ID, 1) FROM FOO WHERE ID > 2"); + StormSql sql = StormSql.construct(); + List<Values> values = new ArrayList<>(); + ChannelHandler h = new TestUtils.CollectDataChannelHandler(values); + sql.execute(stmt, h); + } }
