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 2016/10/11 21:22:51 UTC
[1/3] storm git commit: STORM-1444 Support EXPLAIN statement in
StormSQL
Repository: storm
Updated Branches:
refs/heads/1.x-branch 7d88b68df -> c069734c8
STORM-1444 Support EXPLAIN statement in StormSQL
* when '--explain' is specified to topology name, activate explain mode
* when explain mode is activated, it analyzes each query statement, and shows query plan
* it doesn't submit topologies
* Showing plan is provided by Calcite: it just uses that feature
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/0fc50bfc
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/0fc50bfc
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/0fc50bfc
Branch: refs/heads/1.x-branch
Commit: 0fc50bfc33ce04dc964ac0254f10a31cadb0849c
Parents: 7d88b68
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Mon Sep 12 17:14:17 2016 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Wed Oct 12 06:20:47 2016 +0900
----------------------------------------------------------------------
bin/storm.py | 10 +++-
docs/storm-sql.md | 45 ++++++++++++++++
external/sql/README.md | 46 ++++++++++++++++
external/sql/storm-sql-core/pom.xml | 4 ++
.../src/jvm/org/apache/storm/sql/StormSql.java | 5 ++
.../jvm/org/apache/storm/sql/StormSqlImpl.java | 37 +++++++++++++
.../org/apache/storm/sql/StormSqlRunner.java | 55 +++++++++++++++++---
7 files changed, 193 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/0fc50bfc/bin/storm.py
----------------------------------------------------------------------
diff --git a/bin/storm.py b/bin/storm.py
index 61aae98..424d6c5 100755
--- a/bin/storm.py
+++ b/bin/storm.py
@@ -322,9 +322,10 @@ def jar(jarfile, klass, *args):
["-Dstorm.dependency.artifacts=" + json.dumps(artifact_to_file_jars)])
def sql(sql_file, topology_name):
- """Syntax: [storm sql sql-file topology-name]
+ """Syntax: [storm sql sql-file topology-name], or [storm sql sql-file --explain] when activating explain mode
Compiles the SQL statements into a Trident topology and submits it to Storm.
+ If user activates explain mode, SQL Runner analyzes each query statement and shows query plan instead of submitting topology.
--jars and --artifacts options available for jar are also applied to sql command.
Please refer "help jar" to see how to use --jars and --artifacts options.
@@ -348,11 +349,16 @@ def sql(sql_file, topology_name):
# include this for running StormSqlRunner, but not for generated topology
extrajars.extend(sql_core_jars)
+ if topology_name == "--explain":
+ args = ["--file", sql_file, "--explain"]
+ else:
+ args = ["--file", sql_file, "--topology", topology_name]
+
exec_storm_class(
"org.apache.storm.sql.StormSqlRunner",
jvmtype="-client",
extrajars=extrajars,
- args=[sql_file, topology_name],
+ args=args,
daemon=False,
jvmopts=["-Dstorm.dependency.jars=" + ",".join(local_jars)] +
["-Dstorm.dependency.artifacts=" + json.dumps(artifact_to_file_jars)])
http://git-wip-us.apache.org/repos/asf/storm/blob/0fc50bfc/docs/storm-sql.md
----------------------------------------------------------------------
diff --git a/docs/storm-sql.md b/docs/storm-sql.md
index 3eaf678..bf8119a 100644
--- a/docs/storm-sql.md
+++ b/docs/storm-sql.md
@@ -18,6 +18,8 @@ $ bin/storm sql <sql-file> <topo-name>
In which `sql-file` contains a list of SQL statements to be executed, and `topo-name` is the name of the topology.
+StormSQL activates `explain mode` and shows query plan instead of submitting topology when user specifies `topo-name` as `--explain`.
+Detailed explanation is available from `Showing Query Plan (explain mode)` section.
## Supported Features
@@ -127,6 +129,49 @@ Above command submits the SQL statements to StormSQL. Users need to modify each
By now you should be able to see the `order_filtering` topology in the Storm UI.
+## Showing Query Plan (explain mode)
+
+Like `explain` on SQL statement, StormSQL provides `explain mode` when running Storm SQL Runner. In explain mode, StormSQL analyzes each query statement (only DML) and show plan instead of submitting topology.
+
+In order to run `explain mode`, you need to provide topology name as `--explain` and run `storm sql` as same as submitting.
+
+For example, when you run the example seen above with explain mode:
+
+```
+$ bin/storm sql order_filtering.sql --explain --artifacts "org.apache.storm:storm-sql-kafka:2.0.0-SNAPSHOT,org.apache.storm:storm-kafka:2.0.0-SNAPSHOT,org.apache.kafka:kafka_2.10:0.8.2.2\!org.slf4j:slf4j-log4j12,org.apache.kafka:kafka-clients:0.8.2.2"
+```
+
+StormSQL prints out like below:
+
+```
+
+===========================================================
+query>
+CREATE EXTERNAL TABLE ORDERS (ID INT PRIMARY KEY, UNIT_PRICE INT, QUANTITY INT) LOCATION 'kafka://localhost:2181/brokers?topic=orders' TBLPROPERTIES '{"producer":{"bootstrap.servers":"localhost:9092","acks":"1","key.serializer":"org.apache.storm.kafka.IntSerializer","value.serializer":"org.apache.storm.kafka.ByteBufferSerializer"}}'
+-----------------------------------------------------------
+16:53:43.951 [main] INFO o.a.s.s.r.DataSourcesRegistry - Registering scheme kafka with org.apache.storm.sql.kafka.KafkaDataSourcesProvider@4d1bf319
+No plan presented on DDL
+===========================================================
+===========================================================
+query>
+CREATE EXTERNAL TABLE LARGE_ORDERS (ID INT PRIMARY KEY, TOTAL INT) LOCATION 'kafka://localhost:2181/brokers?topic=large_orders' TBLPROPERTIES '{"producer":{"bootstrap.servers":"localhost:9092","acks":"1","key.serializer":"org.apache.storm.kafka.IntSerializer","value.serializer":"org.apache.storm.kafka.ByteBufferSerializer"}}'
+-----------------------------------------------------------
+No plan presented on DDL
+===========================================================
+===========================================================
+query>
+INSERT INTO LARGE_ORDERS SELECT ID, UNIT_PRICE * QUANTITY AS TOTAL FROM ORDERS WHERE UNIT_PRICE * QUANTITY > 50
+-----------------------------------------------------------
+plan>
+LogicalTableModify(table=[[LARGE_ORDERS]], operation=[INSERT], updateColumnList=[[]], flattened=[true]), id = 8
+ LogicalProject(ID=[$0], TOTAL=[*($1, $2)]), id = 7
+ LogicalFilter(condition=[>(*($1, $2), 50)]), id = 6
+ EnumerableTableScan(table=[[ORDERS]]), id = 5
+
+===========================================================
+
+```
+
## Current Limitations
- Windowing is yet to be implemented.
http://git-wip-us.apache.org/repos/asf/storm/blob/0fc50bfc/external/sql/README.md
----------------------------------------------------------------------
diff --git a/external/sql/README.md b/external/sql/README.md
index d0c0d06..a17f1ff 100644
--- a/external/sql/README.md
+++ b/external/sql/README.md
@@ -12,6 +12,9 @@ $ bin/storm sql <sql-file> <topo-name>
In which `sql-file` contains a list of SQL statements to be executed, and `topo-name` is the name of the topology.
+StormSQL activates `explain mode` and shows query plan instead of submitting topology when user specifies `topo-name` as `--explain`.
+Detailed explanation is available from `Showing Query Plan (explain mode)` section.
+
## Supported Features
The following features are supported in the current repository:
@@ -123,6 +126,49 @@ Above command submits the SQL statements to StormSQL. Users need to modify each
By now you should be able to see the `order_filtering` topology in the Storm UI.
+## Showing Query Plan (explain mode)
+
+Like `explain` on SQL statement, StormSQL provides `explain mode` when running Storm SQL Runner. In explain mode, StormSQL analyzes each query statement (only DML) and show plan instead of submitting topology.
+
+In order to run `explain mode`, you need to provide topology name as `--explain` and run `storm sql` as same as submitting.
+
+For example, when you run the example seen above with explain mode:
+
+```
+$ bin/storm sql order_filtering.sql --explain --artifacts "org.apache.storm:storm-sql-kafka:2.0.0-SNAPSHOT,org.apache.storm:storm-kafka:2.0.0-SNAPSHOT,org.apache.kafka:kafka_2.10:0.8.2.2\!org.slf4j:slf4j-log4j12,org.apache.kafka:kafka-clients:0.8.2.2"
+```
+
+StormSQL prints out like below:
+
+```
+
+===========================================================
+query>
+CREATE EXTERNAL TABLE ORDERS (ID INT PRIMARY KEY, UNIT_PRICE INT, QUANTITY INT) LOCATION 'kafka://localhost:2181/brokers?topic=orders' TBLPROPERTIES '{"producer":{"bootstrap.servers":"localhost:9092","acks":"1","key.serializer":"org.apache.storm.kafka.IntSerializer","value.serializer":"org.apache.storm.kafka.ByteBufferSerializer"}}'
+-----------------------------------------------------------
+16:53:43.951 [main] INFO o.a.s.s.r.DataSourcesRegistry - Registering scheme kafka with org.apache.storm.sql.kafka.KafkaDataSourcesProvider@4d1bf319
+No plan presented on DDL
+===========================================================
+===========================================================
+query>
+CREATE EXTERNAL TABLE LARGE_ORDERS (ID INT PRIMARY KEY, TOTAL INT) LOCATION 'kafka://localhost:2181/brokers?topic=large_orders' TBLPROPERTIES '{"producer":{"bootstrap.servers":"localhost:9092","acks":"1","key.serializer":"org.apache.storm.kafka.IntSerializer","value.serializer":"org.apache.storm.kafka.ByteBufferSerializer"}}'
+-----------------------------------------------------------
+No plan presented on DDL
+===========================================================
+===========================================================
+query>
+INSERT INTO LARGE_ORDERS SELECT ID, UNIT_PRICE * QUANTITY AS TOTAL FROM ORDERS WHERE UNIT_PRICE * QUANTITY > 50
+-----------------------------------------------------------
+plan>
+LogicalTableModify(table=[[LARGE_ORDERS]], operation=[INSERT], updateColumnList=[[]], flattened=[true]), id = 8
+ LogicalProject(ID=[$0], TOTAL=[*($1, $2)]), id = 7
+ LogicalFilter(condition=[>(*($1, $2), 50)]), id = 6
+ EnumerableTableScan(table=[[ORDERS]]), id = 5
+
+===========================================================
+
+```
+
## Current Limitations
- Windowing is yet to be implemented.
http://git-wip-us.apache.org/repos/asf/storm/blob/0fc50bfc/external/sql/storm-sql-core/pom.xml
----------------------------------------------------------------------
diff --git a/external/sql/storm-sql-core/pom.xml b/external/sql/storm-sql-core/pom.xml
index c4fd024..f2e29ca 100644
--- a/external/sql/storm-sql-core/pom.xml
+++ b/external/sql/storm-sql-core/pom.xml
@@ -97,6 +97,10 @@
<version>${jackson.version}</version>
</dependency>
<dependency>
+ <groupId>commons-cli</groupId>
+ <artifactId>commons-cli</artifactId>
+ </dependency>
+ <dependency>
<groupId>commons-lang</groupId>
<artifactId>commons-lang</artifactId>
<scope>runtime</scope>
http://git-wip-us.apache.org/repos/asf/storm/blob/0fc50bfc/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/StormSql.java
----------------------------------------------------------------------
diff --git a/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/StormSql.java b/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/StormSql.java
index 57b64f6..5dec4af 100644
--- a/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/StormSql.java
+++ b/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/StormSql.java
@@ -47,6 +47,11 @@ public abstract class StormSql {
StormSubmitter.ProgressListener progressListener, String asUser)
throws Exception;
+ /**
+ * Print out query plan for each query.
+ */
+ public abstract void explain(Iterable<String> statements) throws Exception;
+
public static StormSql construct() {
return new StormSqlImpl();
}
http://git-wip-us.apache.org/repos/asf/storm/blob/0fc50bfc/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 e404999..61f3b6f 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
@@ -18,10 +18,12 @@
package org.apache.storm.sql;
import org.apache.calcite.jdbc.CalciteSchema;
+import org.apache.calcite.plan.RelOptUtil;
import org.apache.calcite.prepare.CalciteCatalogReader;
import org.apache.calcite.schema.Function;
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.SqlOperatorTable;
import org.apache.calcite.sql.fun.SqlStdOperatorTable;
import org.apache.calcite.sql.util.ChainedSqlOperatorTable;
@@ -140,6 +142,41 @@ class StormSqlImpl extends StormSql {
}
}
+ @Override
+ public void explain(Iterable<String> statements) throws Exception {
+ Map<String, ISqlTridentDataSource> dataSources = new HashMap<>();
+ for (String sql : statements) {
+ StormParser parser = new StormParser(sql);
+ SqlNode node = parser.impl().parseSqlStmtEof();
+
+ System.out.println("===========================================================");
+ System.out.println("query>");
+ System.out.println(sql);
+ System.out.println("-----------------------------------------------------------");
+
+ if (node instanceof SqlCreateTable) {
+ handleCreateTableForTrident((SqlCreateTable) node, dataSources);
+ System.out.println("No plan presented on DDL");
+ } else if (node instanceof SqlCreateFunction) {
+ handleCreateFunction((SqlCreateFunction) node);
+ System.out.println("No plan presented on DDL");
+ } else {
+ FrameworkConfig config = buildFrameWorkConfig();
+ Planner planner = Frameworks.getPlanner(config);
+ SqlNode parse = planner.parse(sql);
+ SqlNode validate = planner.validate(parse);
+ RelNode tree = planner.convert(validate);
+
+ // TODO: change to all attributes when we change to cost-based planner
+ String plan = RelOptUtil.toString(tree, SqlExplainLevel.NON_COST_ATTRIBUTES);
+ System.out.println("plan>");
+ System.out.println(plan);
+ }
+
+ System.out.println("===========================================================");
+ }
+ }
+
private void packageEmptyTopology(Path jar) throws IOException {
Manifest manifest = new Manifest();
Attributes attr = manifest.getMainAttributes();
http://git-wip-us.apache.org/repos/asf/storm/blob/0fc50bfc/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/StormSqlRunner.java
----------------------------------------------------------------------
diff --git a/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/StormSqlRunner.java b/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/StormSqlRunner.java
index 22981d5..5618647 100644
--- a/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/StormSqlRunner.java
+++ b/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/StormSqlRunner.java
@@ -17,6 +17,11 @@
*/
package org.apache.storm.sql;
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.CommandLineParser;
+import org.apache.commons.cli.DefaultParser;
+import org.apache.commons.cli.HelpFormatter;
+import org.apache.commons.cli.Options;
import org.apache.storm.generated.SubmitOptions;
import org.apache.storm.generated.TopologyInitialStatus;
import org.apache.storm.utils.Utils;
@@ -28,17 +33,53 @@ import java.util.List;
import java.util.Map;
public class StormSqlRunner {
+ private static final String OPTION_SQL_FILE_SHORT = "f";
+ private static final String OPTION_SQL_FILE_LONG = "file";
+ private static final String OPTION_SQL_TOPOLOGY_NAME_SHORT = "t";
+ private static final String OPTION_SQL_TOPOLOGY_NAME_LONG = "topology";
+ private static final String OPTION_SQL_EXPLAIN_SHORT = "e";
+ private static final String OPTION_SQL_EXPLAIN_LONG = "explain";
+
public static void main(String[] args) throws Exception {
- if (args.length != 2) {
- System.err.println("storm-sql <sql-file> <topo-name>");
- return;
+ Options options = buildOptions();
+ CommandLineParser parser = new DefaultParser();
+ CommandLine commandLine = parser.parse(options, args);
+
+ if (!commandLine.hasOption(OPTION_SQL_FILE_LONG)) {
+ printUsageAndExit(options, OPTION_SQL_FILE_LONG + " is required");
}
- List<String> stmts = Files.readAllLines(Paths.get(args[0]), StandardCharsets.UTF_8);
- String topoName = args[1];
+
+ String filePath = commandLine.getOptionValue(OPTION_SQL_FILE_LONG);
+ List<String> stmts = Files.readAllLines(Paths.get(filePath), StandardCharsets.UTF_8);
StormSql sql = StormSql.construct();
@SuppressWarnings("unchecked")
Map<String, ?> conf = Utils.readStormConfig();
- SubmitOptions options = new SubmitOptions(TopologyInitialStatus.ACTIVE);
- sql.submit(topoName, stmts, conf, options, null, null);
+
+ if (commandLine.hasOption(OPTION_SQL_EXPLAIN_LONG)) {
+ sql.explain(stmts);
+ } else if (commandLine.hasOption(OPTION_SQL_TOPOLOGY_NAME_LONG)) {
+ String topoName = commandLine.getOptionValue(OPTION_SQL_TOPOLOGY_NAME_LONG);
+ SubmitOptions submitOptions = new SubmitOptions(TopologyInitialStatus.ACTIVE);
+ sql.submit(topoName, stmts, conf, submitOptions, null, null);
+ } else {
+ printUsageAndExit(options, "Either " + OPTION_SQL_TOPOLOGY_NAME_LONG + " or " + OPTION_SQL_EXPLAIN_LONG +
+ " must be presented");
+ }
}
+
+ private static void printUsageAndExit(Options options, String message) {
+ System.out.println(message);
+ HelpFormatter formatter = new HelpFormatter();
+ formatter.printHelp("storm-sql-runner ", options);
+ System.exit(1);
+ }
+
+ private static Options buildOptions() {
+ Options options = new Options();
+ options.addOption(OPTION_SQL_FILE_SHORT, OPTION_SQL_FILE_LONG, true, "REQUIRED SQL file which has sql statements");
+ options.addOption(OPTION_SQL_TOPOLOGY_NAME_SHORT, OPTION_SQL_TOPOLOGY_NAME_LONG, true, "Topology name to submit");
+ options.addOption(OPTION_SQL_EXPLAIN_SHORT, OPTION_SQL_EXPLAIN_LONG, false, "Activate explain mode (topology name will be ignored)");
+ return options;
+ }
+
}
[2/3] storm git commit: Merge branch 'STORM-1444-1.x' into 1.x-branch
Posted by ka...@apache.org.
Merge branch 'STORM-1444-1.x' into 1.x-branch
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/da881f93
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/da881f93
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/da881f93
Branch: refs/heads/1.x-branch
Commit: da881f937252777ffa985dd172307f3d7a5800d3
Parents: 7d88b68 0fc50bf
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Wed Oct 12 06:22:07 2016 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Wed Oct 12 06:22:07 2016 +0900
----------------------------------------------------------------------
bin/storm.py | 10 +++-
docs/storm-sql.md | 45 ++++++++++++++++
external/sql/README.md | 46 ++++++++++++++++
external/sql/storm-sql-core/pom.xml | 4 ++
.../src/jvm/org/apache/storm/sql/StormSql.java | 5 ++
.../jvm/org/apache/storm/sql/StormSqlImpl.java | 37 +++++++++++++
.../org/apache/storm/sql/StormSqlRunner.java | 55 +++++++++++++++++---
7 files changed, 193 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
[3/3] storm git commit: add STORM-1444 to CHANGELOG
Posted by ka...@apache.org.
add STORM-1444 to CHANGELOG
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/c069734c
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/c069734c
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/c069734c
Branch: refs/heads/1.x-branch
Commit: c069734c89594b808a71ec0d59accb5a2e9a5d1e
Parents: da881f9
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Wed Oct 12 06:22:37 2016 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Wed Oct 12 06:22:37 2016 +0900
----------------------------------------------------------------------
CHANGELOG.md | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/c069734c/CHANGELOG.md
----------------------------------------------------------------------
diff --git a/CHANGELOG.md b/CHANGELOG.md
index c7421e4..c907b89 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,4 +1,5 @@
## 1.1.0
+ * STORM-1444: Support EXPLAIN statement in StormSQL
* STORM-2099: Introduce new sql external module: storm-sql-redis
* STORM-2097: Improve logging in trident core and examples
* STORM-2144: Fix Storm-sql group-by behavior in standalone mode