[FLINK-8858] [sql-client] Add support for INSERT INTO in SQL Client #6332

twalthr · 2018-07-13T13:36:00Z

What is the purpose of the change

This PR adds support for the SQL INSERT INTO statement in SQL Client. This PR depends on #6323 for finalizing the unified table sinks. The PR adds support for submitting INSERT INTO statements in the CLI shell as well as using the -u command line option. The command line option is the basis for end-to-end testing of the SQL Client (FLINK-8970).

Brief change log

Refactor the XXXResult classes
Add INSERT INTO support to CLI and local executor
Add -u command line parameter

Verifying this change

DependencyTest has been adapted
New test o.a.f.table.client.gateway.local.LocalExecutorITCase#testStreamQueryExecutionSink
New test o.a.f.table.client.gateway.local.ExecutionContextTest#testSourceSinks

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): no
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? yes
If yes, how is the feature documented? not documented

pnowojski

Almost LGTM. One complain that results refactor could be extracted to separate commit, but this PR was more or less straightforward so no big harm :)

pnowojski · 2018-07-13T16:01:05Z

flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java

@@ -85,6 +86,9 @@ public CliClient(SessionContext context, Executor executor) {
 			terminal = TerminalBuilder.builder()
 				.name(CliStrings.CLI_NAME)
 				.build();
+			// make space from previous output and test the writer
+			terminal.writer().println();


why is that needed? Shouldn't we print new line at then of the output instead?

It makes the output on the terminal nicer. We don't know what has been printed before. This starts a terminal session. The output looks now like:

No default environment specified. Searching for '/Users/twalthr/flink/flink/flink-dist/target/flink-1.6-SNAPSHOT-bin/flink-1.6-SNAPSHOT/conf/sql-client-defaults.yaml'...found. Reading default environment from: file:/Users/twalthr/flink/flink/flink-dist/target/flink-1.6-SNAPSHOT-bin/flink-1.6-SNAPSHOT/conf/sql-client-defaults.yaml No session environment specified. [INFO] Executing the following statement: INSERT INTO MyTableName SELECT * FROM MyTableName [INFO] Submitting SQL update statement to the cluster... [INFO] Table update statement has been successfully submitted to the cluster: Cluster ID: StandaloneClusterId / Job ID: fab21f0632da36f9236c343c2850c71d For the current job status visit: http://localhost:8081 Shutting down executor...done.

pnowojski · 2018-07-13T16:34:40Z

flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/SqlClient.java

+		}
+		// execute single update statement
+		else {
+			final boolean success = cli.submitUpdate(options.getUpdateStatement());


we do not wait for the query to complete?

No, this would block the process for unbounded queries and require a (fault-tolerant) monitoring in the SQL Client which is not intended. We just block until the statement has been submitted to the cluster.

pnowojski · 2018-07-13T16:36:56Z

flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java

-				terminal.writer().println(CliStrings.messageError(CliStrings.MESSAGE_UNKNOWN_SQL));
-				continue;
-			}
+		final Optional<SqlCommandCall> parsedStatement = SqlCommandParser.parse(statement);


deduplicate parsing call and if check - there is already a bug here, either missing flush or unnecessary flush

pnowojski · 2018-07-13T16:39:29Z

flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java

+				break;
+			case SOURCE:
+				callSource(cmdCall);
+				break;


missing default?

pnowojski · 2018-07-13T17:23:32Z

flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java

+	 * @param statement SQL update statement
+	 * @return flag to indicate if the submission was successful or not
+	 */
+	public boolean submitUpdate(String statement) {


do we have tests for that? Some ITCase?

The entire CliClient is only tested manually so far.

Hmmm, that's bad :( Is it hard to start local flink mini cluster and execute cli process from JUnit test? (Maybe as a follow up after feature freeze)

I added some unit tests.

pnowojski · 2018-07-13T17:25:52Z

flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java

+
+		try {
+			final ProgramTargetDescriptor programTarget = executor.executeUpdate(context, cmdCall.operands[0]);
+			terminal.writer().println(CliStrings.messageInfo(CliStrings.MESSAGE_STATEMENT_SUBMITTED).toAnsi());


programTarget.writeTo(terminal.writer())

? It would be easier to add more fields in the future and more difficult to forget about printing them.

Initially I had it similar like your proposal, but this would mix a data model class and visualization. ProgramTargetDescriptor should not be responsible how it is represented in the CLI.

hmmm, what about

terminal.writer().println(CliStrings.messageInfo(CliStrings.MESSAGE_STATEMENT_SUBMITTED).toAnsi()); terminal.writer().println(programTarget.toString()) ```?

pnowojski · 2018-07-13T17:28:50Z

...link-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java

+	private <C> ProgramTargetDescriptor executeUpdateInternal(ExecutionContext<C> context, String statement) {
+		final ExecutionContext.EnvironmentInstance envInst = context.createEnvironmentInstance();
+
+		// apply update statement


some of those comments are a little bit unnecessary?

On one side yes but on the other side it allows to read the comments from top to bottom and know what the method is doing without having to look at the actual code.

But there is no additional value of

// apply update statement

over

applyUpdate(envInst.getTableEnvironment(), envInst.getQueryConfig(), statement);

which is completely self explanatory. This is just form of some small code duplication. Same applies to most of those comments.

pnowojski · 2018-07-13T17:31:12Z

...link-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java

+	 * @param tableEnv table environment
+	 * @param query SQL SELECT query
+	 * @return result table object
+	 */
 	private Table createTable(TableEnvironment tableEnv, String query) {


rename query -> selectQuery and drop @param section from java doc?

pnowojski · 2018-07-13T17:31:32Z

...link-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java

+	 * @param queryConfig query configuration
+	 * @param statement SQL update statement (e.g. INSERT INTO)
+	 */
+	private void applyUpdate(TableEnvironment tableEnv, QueryConfig queryConfig, String statement) {


rename statement -> updateStatement and drop @param section from java doc?

pnowojski · 2018-07-13T17:54:16Z

...nk-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ProgramDeployer.java

@@ -73,7 +86,7 @@ public JobExecutionResult fetchExecutionResult() {
 	 * Deploys a job. Depending on the deployment creates a new job cluster. It saves the cluster id in
 	 * the result and blocks until job completion.
 	 */
-	private <T> JobExecutionResult deployJob(ExecutionContext<T> context, JobGraph jobGraph, DynamicResult<T> result) {
+	private <T> void deployJob(ExecutionContext<T> context, JobGraph jobGraph, Result<T> result) {


split this method? Extract if (context.getClusterId() == null) and else branches to separate methods?

pnowojski · 2018-07-13T17:59:52Z

One more thing. Do we need this -u flag? Shouldn't it be enough to support something like:
flink-cli < query01.sql or echo "INSERT INTO bar SELECT * FROM foo" | flink-cli

twalthr · 2018-07-14T07:11:18Z

Thanks for the review @pnowojski. I agree that we should support flink-cli < query01.sql or echo "INSERT INTO bar SELECT * FROM foo" | flink-cli. However, I would move this to a separate issue because I'm not sure how well we support multilines and EOF right now. With the -u flag the user also gets the correct error code after the submission, with flink-cli < query01.sql the CLI would either stay in interactive mode or always return success. I would keep the -u flag for now for testing purposes. We could remove it from the CLI help and leave it as an internal parameter for now. What do you think?

pnowojski

-u seems too specific for me. If it supported any query (-i --input) it would have more sense. However hiding it from CLI help and postponing this issue is fine for me.

Besides that LGTM. Just two minor comments.

pnowojski · 2018-07-14T17:22:14Z

...link-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java

+	private <C> ProgramTargetDescriptor executeUpdateInternal(ExecutionContext<C> context, String statement) {
+		final ExecutionContext.EnvironmentInstance envInst = context.createEnvironmentInstance();
+
+		// apply update statement


But there is no additional value of

// apply update statement

over

applyUpdate(envInst.getTableEnvironment(), envInst.getQueryConfig(), statement);

which is completely self explanatory. This is just form of some small code duplication. Same applies to most of those comments.

This closes apache#6332.

pnowojski reviewed Jul 13, 2018

View reviewed changes

pnowojski approved these changes Jul 14, 2018

View reviewed changes

[FLINK-8858] [sql-client] Add support for INSERT INTO in SQL Client

00e3519

This closes apache#6332.

twalthr force-pushed the FLINK-8858 branch from c51a1ba to 00e3519 Compare July 15, 2018 09:34

twalthr added a commit to twalthr/flink that referenced this pull request Jul 15, 2018

[FLINK-8858] [sql-client] Add support for INSERT INTO in SQL Client

9712c1c

This closes apache#6332.

twalthr added a commit to twalthr/flink that referenced this pull request Jul 15, 2018

[FLINK-8858] [sql-client] Add support for INSERT INTO in SQL Client

3cc4624

This closes apache#6332.

twalthr added a commit to twalthr/flink that referenced this pull request Jul 15, 2018

[FLINK-8858] [sql-client] Add support for INSERT INTO in SQL Client

a657775

This closes apache#6332.

asfgit closed this in 695bc56 Jul 15, 2018

sampathBhat pushed a commit to sampathBhat/flink that referenced this pull request Jul 26, 2018

[FLINK-8858] [sql-client] Add support for INSERT INTO in SQL Client

731660e

This closes apache#6332.

rmetzger added component=API/TableSQL component=SQL/Client labels Mar 18, 2019

rmetzger added component=TableSQL/Client and removed component=API/TableSQL component=SQL/Client labels Apr 30, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-8858] [sql-client] Add support for INSERT INTO in SQL Client #6332

[FLINK-8858] [sql-client] Add support for INSERT INTO in SQL Client #6332

twalthr commented Jul 13, 2018

pnowojski left a comment

pnowojski Jul 13, 2018

twalthr Jul 14, 2018

pnowojski Jul 13, 2018

twalthr Jul 14, 2018

pnowojski Jul 13, 2018

pnowojski Jul 13, 2018

pnowojski Jul 13, 2018

twalthr Jul 14, 2018

pnowojski Jul 14, 2018

twalthr Jul 15, 2018

pnowojski Jul 13, 2018

twalthr Jul 14, 2018

pnowojski Jul 14, 2018

pnowojski Jul 13, 2018

twalthr Jul 14, 2018

pnowojski Jul 14, 2018

pnowojski Jul 13, 2018

pnowojski Jul 13, 2018

pnowojski Jul 13, 2018

pnowojski commented Jul 13, 2018 •

edited

Loading

twalthr commented Jul 14, 2018

pnowojski left a comment

pnowojski Jul 14, 2018

[FLINK-8858] [sql-client] Add support for INSERT INTO in SQL Client #6332

[FLINK-8858] [sql-client] Add support for INSERT INTO in SQL Client #6332

Conversation

twalthr commented Jul 13, 2018

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

pnowojski left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pnowojski commented Jul 13, 2018 • edited Loading

twalthr commented Jul 14, 2018

pnowojski left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pnowojski commented Jul 13, 2018 •

edited

Loading