You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by ka...@apache.org on 2017/01/30 13:58:44 UTC
[1/3] storm git commit: STORM-2330: Fix storm sql code generation for
UDAF with non standard sql types
Repository: storm
Updated Branches:
refs/heads/master 69568ecf0 -> 4c5e34ee6
STORM-2330: Fix storm sql code generation for UDAF with non standard sql types
Storm-sql uses cacite's code generator. For UDAF with non standard return types, the generated code does not compile.
E.g. For an UDAF that returns a List<T> cacite converts the return type to SQL type "OTHER" and finally inserts a cast to Object[] in the returned code which fails to compile with "java.lang.ClassCastException: java.util.ArrayList cannot be cast to [Ljava.lang.Object;"
Use a custom type factory to return "ANY" as the type (instead of OTHER) if the java type cannot be mapped to a sql type.
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/1d09d786
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/1d09d786
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/1d09d786
Branch: refs/heads/master
Commit: 1d09d786fdbf61f634441e76e2c2a53c35178540
Parents: 1811273
Author: Arun Mahadevan <ar...@apache.org>
Authored: Fri Jan 27 12:19:18 2017 +0530
Committer: Arun Mahadevan <ar...@apache.org>
Committed: Fri Jan 27 12:19:18 2017 +0530
----------------------------------------------------------------------
.../jvm/org/apache/storm/sql/StormSqlImpl.java | 24 ++++-----
.../RexNodeToBlockStatementCompiler.java | 2 +-
.../sql/compiler/StormSqlTypeFactoryImpl.java | 51 ++++++++++++++++++++
.../test/org/apache/storm/sql/TestStormSql.java | 18 +++++++
4 files changed, 82 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/1d09d786/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 af0f1c1..f7bd719 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
@@ -17,36 +17,36 @@
*/
package org.apache.storm.sql;
+import org.apache.calcite.adapter.java.JavaTypeFactory;
import org.apache.calcite.jdbc.CalciteSchema;
import org.apache.calcite.prepare.CalciteCatalogReader;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeSystem;
import org.apache.calcite.schema.Function;
+import org.apache.calcite.schema.SchemaPlus;
+import org.apache.calcite.schema.Table;
import org.apache.calcite.schema.impl.AggregateFunctionImpl;
import org.apache.calcite.schema.impl.ScalarFunctionImpl;
import org.apache.calcite.sql.SqlExplainLevel;
+import org.apache.calcite.sql.SqlNode;
import org.apache.calcite.sql.SqlOperatorTable;
import org.apache.calcite.sql.fun.SqlStdOperatorTable;
import org.apache.calcite.sql.util.ChainedSqlOperatorTable;
-import org.apache.storm.StormSubmitter;
-import org.apache.storm.generated.SubmitOptions;
-import org.apache.calcite.adapter.java.JavaTypeFactory;
-import org.apache.calcite.jdbc.JavaTypeFactoryImpl;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.type.RelDataType;
-import org.apache.calcite.rel.type.RelDataTypeSystem;
-import org.apache.calcite.schema.SchemaPlus;
-import org.apache.calcite.schema.Table;
-import org.apache.calcite.sql.SqlNode;
import org.apache.calcite.tools.FrameworkConfig;
import org.apache.calcite.tools.Frameworks;
import org.apache.calcite.tools.Planner;
+import org.apache.storm.StormSubmitter;
+import org.apache.storm.generated.SubmitOptions;
+import org.apache.storm.sql.compiler.StormSqlTypeFactoryImpl;
import org.apache.storm.sql.compiler.backends.standalone.PlanCompiler;
import org.apache.storm.sql.parser.ColumnConstraint;
import org.apache.storm.sql.parser.ColumnDefinition;
import org.apache.storm.sql.parser.SqlCreateFunction;
import org.apache.storm.sql.parser.SqlCreateTable;
import org.apache.storm.sql.parser.StormParser;
-import org.apache.storm.sql.planner.trident.QueryPlanner;
import org.apache.storm.sql.planner.StormRelUtils;
+import org.apache.storm.sql.planner.trident.QueryPlanner;
import org.apache.storm.sql.runtime.AbstractValuesProcessor;
import org.apache.storm.sql.runtime.ChannelHandler;
import org.apache.storm.sql.runtime.DataSource;
@@ -74,7 +74,7 @@ import java.util.jar.Manifest;
import static org.apache.storm.sql.compiler.CompilerUtil.TableBuilderInfo;
class StormSqlImpl extends StormSql {
- private final JavaTypeFactory typeFactory = new JavaTypeFactoryImpl(
+ private final JavaTypeFactory typeFactory = new StormSqlTypeFactoryImpl(
RelDataTypeSystem.DEFAULT);
private final SchemaPlus schema = Frameworks.createRootSchema(true);
private boolean hasUdf = false;
http://git-wip-us.apache.org/repos/asf/storm/blob/1d09d786/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/compiler/RexNodeToBlockStatementCompiler.java
----------------------------------------------------------------------
diff --git a/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/compiler/RexNodeToBlockStatementCompiler.java b/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/compiler/RexNodeToBlockStatementCompiler.java
index 8400a9b..06ed227 100644
--- a/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/compiler/RexNodeToBlockStatementCompiler.java
+++ b/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/compiler/RexNodeToBlockStatementCompiler.java
@@ -72,7 +72,7 @@ public class RexNodeToBlockStatementCompiler {
final ParameterExpression outputValues_ =
Expressions.parameter(Object[].class, "outputValues");
final JavaTypeFactoryImpl javaTypeFactory =
- new JavaTypeFactoryImpl(rexBuilder.getTypeFactory().getTypeSystem());
+ new StormSqlTypeFactoryImpl(rexBuilder.getTypeFactory().getTypeSystem());
final RexToLixTranslator.InputGetter inputGetter =
new RexToLixTranslator.InputGetterImpl(
http://git-wip-us.apache.org/repos/asf/storm/blob/1d09d786/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/compiler/StormSqlTypeFactoryImpl.java
----------------------------------------------------------------------
diff --git a/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/compiler/StormSqlTypeFactoryImpl.java b/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/compiler/StormSqlTypeFactoryImpl.java
new file mode 100644
index 0000000..21ca063
--- /dev/null
+++ b/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/compiler/StormSqlTypeFactoryImpl.java
@@ -0,0 +1,51 @@
+/**
+ * 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.compiler;
+
+import com.google.common.base.Function;
+import com.google.common.collect.Lists;
+import org.apache.calcite.jdbc.JavaTypeFactoryImpl;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rel.type.RelDataTypeSystem;
+import org.apache.calcite.rel.type.RelRecordType;
+import org.apache.calcite.sql.type.JavaToSqlTypeConversionRules;
+import org.apache.calcite.sql.type.SqlTypeName;
+
+public class StormSqlTypeFactoryImpl extends JavaTypeFactoryImpl {
+
+ public StormSqlTypeFactoryImpl() {
+ }
+
+ public StormSqlTypeFactoryImpl(RelDataTypeSystem typeSystem) {
+ super(typeSystem);
+ }
+
+ @Override
+ public RelDataType toSql(RelDataType type) {
+ if (type instanceof JavaType) {
+ JavaType javaType = (JavaType) type;
+ SqlTypeName sqlTypeName = JavaToSqlTypeConversionRules.instance().lookup(javaType.getJavaClass());
+ if (sqlTypeName == null) {
+ sqlTypeName = SqlTypeName.ANY;
+ }
+ return createTypeWithNullability(createSqlType(sqlTypeName), type.isNullable());
+ }
+ return super.toSql(type);
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/1d09d786/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 1470344..82dc184 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
@@ -35,6 +35,7 @@ import org.junit.Test;
import java.net.URI;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Properties;
@@ -359,6 +360,23 @@ public class TestStormSql {
}
@Test
+ public void testAggFnNonSqlReturnType() throws Exception {
+ List<String> stmt = new ArrayList<>();
+ stmt.add("CREATE EXTERNAL TABLE FOO (ID INT PRIMARY KEY, SALARY INT, PCT DOUBLE, NAME VARCHAR) LOCATION 'mockgroup:///foo'");
+ stmt.add("CREATE FUNCTION TOPN AS 'org.apache.storm.sql.TestUtils$TopN'");
+ stmt.add("SELECT STREAM ID, SUM(SALARY), TOPN(1, SALARY) FROM FOO WHERE ID >= 0 GROUP BY (ID) HAVING MAX(SALARY) > 0");
+ StormSql sql = StormSql.construct();
+ List<Values> values = new ArrayList<>();
+ ChannelHandler h = new TestUtils.CollectDataChannelHandler(values);
+ sql.execute(stmt, h);
+ Assert.assertEquals(4, values.size());
+ Assert.assertEquals(Collections.singletonList(2), values.get(0).get(2));
+ Assert.assertEquals(Collections.singletonList(5), values.get(1).get(2));
+ Assert.assertEquals(Collections.singletonList(8), values.get(2).get(2));
+ Assert.assertEquals(Collections.singletonList(9), values.get(3).get(2));
+ }
+
+ @Test
public void testGroupbySameAggregateOnDifferentColumns() throws Exception {
List<String> stmt = new ArrayList<>();
stmt.add("CREATE EXTERNAL TABLE FOO (ID INT PRIMARY KEY, SALARY INT, PCT DOUBLE, NAME VARCHAR) LOCATION 'mockgroup:///foo'");
[3/3] storm git commit: STORM-2330: CHANGELOG
Posted by ka...@apache.org.
STORM-2330: CHANGELOG
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/4c5e34ee
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/4c5e34ee
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/4c5e34ee
Branch: refs/heads/master
Commit: 4c5e34ee6399d8b19718bd7b75dc7c241a37a515
Parents: fcecf42
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Mon Jan 30 22:58:26 2017 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Mon Jan 30 22:58:26 2017 +0900
----------------------------------------------------------------------
CHANGELOG.md | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/4c5e34ee/CHANGELOG.md
----------------------------------------------------------------------
diff --git a/CHANGELOG.md b/CHANGELOG.md
index e55d793..3d50214 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -194,6 +194,7 @@
* STORM-1769: Added a test to check local nimbus with notifier plugin
## 1.1.0
+ * STORM-2330: Fix storm sql code generation for UDAF with non standard sql types
* STORM-2298: Don't kill Nimbus when ClusterMetricsConsumer is failed to initialize
* STORM-2301: [storm-cassandra] upgrade cassandra driver to 3.1.2
* STORM-1446: Compile the Calcite logical plan to Storm Trident logical plan
[2/3] storm git commit: Merge branch 'STORM-2330' of
https://github.com/arunmahadevan/storm into STORM-2330-merge
Posted by ka...@apache.org.
Merge branch 'STORM-2330' of https://github.com/arunmahadevan/storm into STORM-2330-merge
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/fcecf42e
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/fcecf42e
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/fcecf42e
Branch: refs/heads/master
Commit: fcecf42e6db6f332ded1d76854a95867f98528d1
Parents: 69568ec 1d09d78
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Mon Jan 30 22:51:26 2017 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Mon Jan 30 22:51:26 2017 +0900
----------------------------------------------------------------------
.../jvm/org/apache/storm/sql/StormSqlImpl.java | 24 ++++-----
.../RexNodeToBlockStatementCompiler.java | 2 +-
.../sql/compiler/StormSqlTypeFactoryImpl.java | 51 ++++++++++++++++++++
.../test/org/apache/storm/sql/TestStormSql.java | 18 +++++++
4 files changed, 82 insertions(+), 13 deletions(-)
----------------------------------------------------------------------