You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2023/01/18 11:09:30 UTC
[GitHub] [flink] fsk119 opened a new pull request, #21717: [FLINK-29945][sql-client] Supports to submit SQL to a embedded SQL Ga…
fsk119 opened a new pull request, #21717:
URL: https://github.com/apache/flink/pull/21717
…teway in the SQL Client
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
*After rework, the SQL Client uses the SQL Gateway as the base.*
## Brief change log
*(for example:)*
- *The TaskInfo is stored in the blob store on job creation time as a persistent artifact*
- *Deployments RPC transmits only the blob storage reference*
- *TaskManagers retrieve the TaskInfo from the blob cache*
## Verifying this change
Please make sure both new and modified tests in this PR follows the conventions defined in our code quality guide: https://flink.apache.org/contributing/code-style-and-quality-common.html#testing
This change is a trivial rework / code cleanup without any test coverage.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / **no**)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**)
- The serializers: (yes / **no** / don't know)
- The runtime per-record code paths (performance sensitive): (yes / **no** / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / **no** / don't know)
- The S3 file system connector: (yes / **no** / don't know)
## Documentation
- Does this pull request introduce a new feature? (yes / **no**)
- If yes, how is the feature documented? (**not applicable** / docs / JavaDocs / not documented)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fsk119 commented on a diff in pull request #21717: [FLINK-29945][sql-client] Supports to submit SQL to a embedded SQL Ga…
Posted by "fsk119 (via GitHub)" <gi...@apache.org>.
fsk119 commented on code in PR #21717:
URL: https://github.com/apache/flink/pull/21717#discussion_r1089677325
##########
flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/ExecutorImplITCase.java:
##########
@@ -414,50 +462,130 @@ void testStopJob() throws Exception {
final String insert = "INSERT INTO snk SELECT a FROM src;";
try {
- executor.executeOperation(executor.parseStatement(srcDdl));
- executor.executeOperation(executor.parseStatement(snkDdl));
- TableResult result = executor.executeOperation(executor.parseStatement(insert));
- JobClient jobClient = result.getJobClient().get();
- JobID jobId = jobClient.getJobID();
+ executor.configureSession(srcDdl);
+ executor.configureSession(snkDdl);
+ ClientResult result = executor.executeStatement(insert);
+ JobID jobID = result.getJobId();
// wait till the job turns into running status or the test times out
- JobStatus jobStatus;
- do {
- Thread.sleep(2_000L);
- jobStatus = jobClient.getJobStatus().get();
- } while (jobStatus != JobStatus.RUNNING);
-
- Optional<String> savepoint = executor.stopJob(jobId.toString(), true, true);
- assertThat(savepoint).isPresent();
+ TestUtils.waitUntilAllTasksAreRunning(clusterClient, jobID);
+ StringData savepointPath =
+ CollectionUtil.iteratorToList(
+ executor.executeStatement(
+ String.format("STOP JOB '%s' WITH SAVEPOINT", jobID)))
+ .get(0)
+ .getString(0);
+ assertThat(
+ Files.exists(
+ Paths.get(
+ URI.create(
+ Preconditions.checkNotNull(savepointPath)
+ .toString()))))
+ .isTrue();
Review Comment:
Thanks for pointing it out. I think the root cause is related the implementation of the sql gateway and we can discuss in another thread. I have opened a ticket to track this: FLINK-30811
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fsk119 merged pull request #21717: [FLINK-29945][sql-client] Supports to submit SQL to a embedded SQL Ga…
Posted by "fsk119 (via GitHub)" <gi...@apache.org>.
fsk119 merged PR #21717:
URL: https://github.com/apache/flink/pull/21717
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #21717: [FLINK-29945][sql-client] Supports to submit SQL to a embedded SQL Ga…
Posted by GitBox <gi...@apache.org>.
flinkbot commented on PR #21717:
URL: https://github.com/apache/flink/pull/21717#issuecomment-1386884294
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ca159e9f1709684b216e6836ae5f409be92a5b",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "54ca159e9f1709684b216e6836ae5f409be92a5b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 54ca159e9f1709684b216e6836ae5f409be92a5b UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] LadyForest commented on a diff in pull request #21717: [FLINK-29945][sql-client] Supports to submit SQL to a embedded SQL Ga…
Posted by GitBox <gi...@apache.org>.
LadyForest commented on code in PR #21717:
URL: https://github.com/apache/flink/pull/21717#discussion_r1081032613
##########
flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/parser/SqlCommandParserImpl.java:
##########
@@ -18,35 +18,136 @@
package org.apache.flink.table.client.cli.parser;
-import org.apache.flink.table.api.SqlParserException;
-import org.apache.flink.table.client.gateway.Executor;
+import org.apache.flink.sql.parser.impl.FlinkSqlParserImplTokenManager;
+import org.apache.flink.sql.parser.impl.SimpleCharStream;
+import org.apache.flink.sql.parser.impl.Token;
+import org.apache.flink.table.api.SqlParserEOFException;
import org.apache.flink.table.client.gateway.SqlExecutionException;
-import org.apache.flink.table.operations.Operation;
+import java.io.StringReader;
+import java.util.Iterator;
import java.util.Optional;
-/** SqlCommandParserImpl wrappers an {@link Executor} supports parse a statement to an Operation. */
-public class SqlCommandParserImpl implements SqlCommandParser {
- private final Executor executor;
+import static org.apache.flink.sql.parser.impl.FlinkSqlParserImplConstants.EOF;
+import static org.apache.flink.sql.parser.impl.FlinkSqlParserImplConstants.IDENTIFIER;
+import static org.apache.flink.sql.parser.impl.FlinkSqlParserImplConstants.SEMICOLON;
- public SqlCommandParserImpl(Executor executor) {
- this.executor = executor;
- }
+/**
+ * The {@link SqlCommandParserImpl} uses {@link FlinkSqlParserImplTokenManager} to do lexical
+ * analysis. It cannot recognize special hive keywords yet because Hive has a slightly different
+ * vocabulary compared to Flink's, which causes the The {@link SqlCommandParserImpl}
Review Comment:
Nit: remove duplicated "The"
##########
flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/parser/SqlCommandParserImpl.java:
##########
@@ -18,35 +18,136 @@
package org.apache.flink.table.client.cli.parser;
-import org.apache.flink.table.api.SqlParserException;
-import org.apache.flink.table.client.gateway.Executor;
+import org.apache.flink.sql.parser.impl.FlinkSqlParserImplTokenManager;
+import org.apache.flink.sql.parser.impl.SimpleCharStream;
+import org.apache.flink.sql.parser.impl.Token;
+import org.apache.flink.table.api.SqlParserEOFException;
import org.apache.flink.table.client.gateway.SqlExecutionException;
-import org.apache.flink.table.operations.Operation;
+import java.io.StringReader;
+import java.util.Iterator;
import java.util.Optional;
-/** SqlCommandParserImpl wrappers an {@link Executor} supports parse a statement to an Operation. */
-public class SqlCommandParserImpl implements SqlCommandParser {
- private final Executor executor;
+import static org.apache.flink.sql.parser.impl.FlinkSqlParserImplConstants.EOF;
+import static org.apache.flink.sql.parser.impl.FlinkSqlParserImplConstants.IDENTIFIER;
+import static org.apache.flink.sql.parser.impl.FlinkSqlParserImplConstants.SEMICOLON;
- public SqlCommandParserImpl(Executor executor) {
- this.executor = executor;
- }
+/**
+ * The {@link SqlCommandParserImpl} uses {@link FlinkSqlParserImplTokenManager} to do lexical
+ * analysis. It cannot recognize special hive keywords yet because Hive has a slightly different
+ * vocabulary compared to Flink's, which causes the The {@link SqlCommandParserImpl}
+ * misunderstanding some Hive's keywords to IDENTIFIER. But the ClientParser is only responsible to
+ * check whether the statement is completed or not and only cares about a few statements. So it's
+ * acceptable to tolerate the inaccuracy here.
+ */
+public class SqlCommandParserImpl implements SqlCommandParser {
- @Override
- public Optional<Operation> parseCommand(String stmt) throws SqlParserException {
+ public Optional<Command> parseStatement(String statement) throws SqlExecutionException {
// normalize
- stmt = stmt.trim();
+ statement = statement.trim();
// meet empty statement, e.g ";\n"
- if (stmt.isEmpty() || stmt.equals(";")) {
+ if (statement.isEmpty() || statement.equals(";")) {
return Optional.empty();
+ } else {
+ return Optional.of(getCommand(new TokenIterator(statement.trim())));
Review Comment:
Nit: no need to `trim` again?
##########
flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/parser/SqlCommandParserImplTest.java:
##########
@@ -28,105 +28,99 @@
import java.util.Arrays;
import java.util.List;
-import java.util.Optional;
import static org.apache.flink.core.testutils.FlinkAssertions.anyCauseMatches;
-import static org.apache.flink.table.client.cli.parser.StatementType.BEGIN_STATEMENT_SET;
-import static org.apache.flink.table.client.cli.parser.StatementType.CLEAR;
-import static org.apache.flink.table.client.cli.parser.StatementType.END;
-import static org.apache.flink.table.client.cli.parser.StatementType.EXPLAIN;
-import static org.apache.flink.table.client.cli.parser.StatementType.HELP;
-import static org.apache.flink.table.client.cli.parser.StatementType.OTHER;
-import static org.apache.flink.table.client.cli.parser.StatementType.QUIT;
-import static org.apache.flink.table.client.cli.parser.StatementType.SELECT;
-import static org.apache.flink.table.client.cli.parser.StatementType.SHOW_CREATE;
+import static org.apache.flink.table.client.cli.parser.Command.CLEAR;
+import static org.apache.flink.table.client.cli.parser.Command.HELP;
+import static org.apache.flink.table.client.cli.parser.Command.OTHER;
+import static org.apache.flink.table.client.cli.parser.Command.QUIT;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/**
- * Testing whether {@link ClientParser} can parse statement to get {@link StatementType} correctly.
+ * Testing whether {@link SqlCommandParserImpl} can parse statement to get {@link Command}
+ * correctly.
*/
-public class ClientParserTest {
+public class SqlCommandParserImplTest {
private static final String EXECUTE_STATEMENT_SET =
"EXECUTE STATEMENT SET BEGIN\n INSERT INTO StreamingTable SELECT * FROM (VALUES (1, 'Hello World'));";
- private final ClientParser clientParser = new ClientParser();
+ private final SqlCommandParserImpl sqlCommandParserImpl = new SqlCommandParserImpl();
@ParameterizedTest
@MethodSource("positiveCases")
public void testParseStatement(TestSpec testData) {
- Optional<StatementType> type = clientParser.parseStatement(testData.statement);
- assertThat(type.orElse(null)).isEqualTo(testData.type);
+ Command command = sqlCommandParserImpl.parseStatement(testData.statement).orElse(null);
+ assertThat(command).isEqualTo(testData.command);
}
@ParameterizedTest
@MethodSource("negativeCases")
public void testParseIncompleteStatement(String statement) {
- assertThatThrownBy(() -> clientParser.parseStatement(statement))
+ assertThatThrownBy(() -> sqlCommandParserImpl.parseStatement(statement))
.satisfies(anyCauseMatches(SqlExecutionException.class))
.cause()
.satisfies(anyCauseMatches(SqlParserEOFException.class));
}
private static List<TestSpec> positiveCases() {
return Arrays.asList(
- TestSpec.of(";", OTHER),
+ TestSpec.of(";", null),
TestSpec.of("; ;", OTHER),
// comment and multi lines tests
- TestSpec.of("SHOW --ignore;\n CREATE TABLE tbl;", SHOW_CREATE),
- TestSpec.of("SHOW\n create\t TABLE `tbl`;", SHOW_CREATE),
+ TestSpec.of("SHOW --ignore;\n CREATE TABLE tbl;", OTHER),
+ TestSpec.of("SHOW\n create\t TABLE `tbl`;", OTHER),
TestSpec.of("SHOW -- create\n TABLES;", OTHER),
// special characters tests
- TestSpec.of("SELECT * FROM `tbl`;", SELECT),
- TestSpec.of("SHOW /* ignore */ CREATE TABLE \"tbl\";", SHOW_CREATE),
- TestSpec.of("SELECT '\\';", SELECT),
+ TestSpec.of("SELECT * FROM `tbl`;", OTHER),
+ TestSpec.of("SHOW /* ignore */ CREATE TABLE \"tbl\";", OTHER),
+ TestSpec.of("SELECT '\\';", OTHER),
// normal tests
TestSpec.of("quit;", QUIT), // non case sensitive test
TestSpec.of("QUIT;", QUIT),
TestSpec.of("Quit;", QUIT),
TestSpec.of("QuIt;", QUIT),
TestSpec.of("clear;", CLEAR),
TestSpec.of("help;", HELP),
- TestSpec.of("EXPLAIN PLAN FOR 'what_ever';", EXPLAIN),
- TestSpec.of("SHOW CREATE TABLE(what_ever);", SHOW_CREATE),
- TestSpec.of("SHOW CREATE VIEW (what_ever);", SHOW_CREATE),
- TestSpec.of("SHOW CREATE syntax_error;", SHOW_CREATE),
+ TestSpec.of("EXPLAIN PLAN FOR 'what_ever';", OTHER),
+ TestSpec.of("SHOW CREATE TABLE(what_ever);", OTHER),
+ TestSpec.of("SHOW CREATE VIEW (what_ever);", OTHER),
+ TestSpec.of("SHOW CREATE syntax_error;", OTHER),
TestSpec.of("SHOW TABLES;", OTHER),
- TestSpec.of("BEGIN STATEMENT SET;", BEGIN_STATEMENT_SET),
+ TestSpec.of("BEGIN STATEMENT SET;", OTHER),
TestSpec.of("BEGIN statement;", OTHER),
- TestSpec.of("END;", END),
+ TestSpec.of("END;", OTHER),
TestSpec.of("END statement;", OTHER),
// statement set tests
TestSpec.of(EXECUTE_STATEMENT_SET, OTHER),
- TestSpec.of("EXPLAIN " + EXECUTE_STATEMENT_SET, EXPLAIN),
- TestSpec.of("EXPLAIN BEGIN STATEMENT SET;", EXPLAIN),
+ TestSpec.of("EXPLAIN " + EXECUTE_STATEMENT_SET, OTHER),
+ TestSpec.of("EXPLAIN BEGIN STATEMENT SET;", OTHER),
TestSpec.of(EXECUTE_STATEMENT_SET + "\nEND;", OTHER),
- TestSpec.of("EXPLAIN " + EXECUTE_STATEMENT_SET + "\nEND;", EXPLAIN),
- TestSpec.of("EXPLAIN BEGIN STATEMENT SET;\nEND;", EXPLAIN));
+ TestSpec.of("EXPLAIN " + EXECUTE_STATEMENT_SET + "\nEND;", OTHER),
+ TestSpec.of("EXPLAIN BEGIN STATEMENT SET;\nEND;", OTHER));
}
private static List<String> negativeCases() {
- return Arrays.asList(
- "", "\n", " ", "-- comment;", "SHOW TABLES -- comment;", "SHOW TABLES");
+ return Arrays.asList("-- comment;", "SHOW TABLES -- comment;", "SHOW TABLES");
}
/** Used to load generated data. */
private static class TestSpec {
String statement;
- StatementType type;
+ Command command;
- TestSpec(String statement, @Nullable StatementType type) {
+ TestSpec(String statement, @Nullable Command type) {
this.statement = statement;
- this.type = type;
+ this.command = type;
Review Comment:
Nit: align the naming?
##########
flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/parser/SqlCommandParserImplTest.java:
##########
@@ -28,105 +28,99 @@
import java.util.Arrays;
import java.util.List;
-import java.util.Optional;
import static org.apache.flink.core.testutils.FlinkAssertions.anyCauseMatches;
-import static org.apache.flink.table.client.cli.parser.StatementType.BEGIN_STATEMENT_SET;
-import static org.apache.flink.table.client.cli.parser.StatementType.CLEAR;
-import static org.apache.flink.table.client.cli.parser.StatementType.END;
-import static org.apache.flink.table.client.cli.parser.StatementType.EXPLAIN;
-import static org.apache.flink.table.client.cli.parser.StatementType.HELP;
-import static org.apache.flink.table.client.cli.parser.StatementType.OTHER;
-import static org.apache.flink.table.client.cli.parser.StatementType.QUIT;
-import static org.apache.flink.table.client.cli.parser.StatementType.SELECT;
-import static org.apache.flink.table.client.cli.parser.StatementType.SHOW_CREATE;
+import static org.apache.flink.table.client.cli.parser.Command.CLEAR;
+import static org.apache.flink.table.client.cli.parser.Command.HELP;
+import static org.apache.flink.table.client.cli.parser.Command.OTHER;
+import static org.apache.flink.table.client.cli.parser.Command.QUIT;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/**
- * Testing whether {@link ClientParser} can parse statement to get {@link StatementType} correctly.
+ * Testing whether {@link SqlCommandParserImpl} can parse statement to get {@link Command}
+ * correctly.
*/
-public class ClientParserTest {
+public class SqlCommandParserImplTest {
private static final String EXECUTE_STATEMENT_SET =
"EXECUTE STATEMENT SET BEGIN\n INSERT INTO StreamingTable SELECT * FROM (VALUES (1, 'Hello World'));";
- private final ClientParser clientParser = new ClientParser();
+ private final SqlCommandParserImpl sqlCommandParserImpl = new SqlCommandParserImpl();
@ParameterizedTest
@MethodSource("positiveCases")
public void testParseStatement(TestSpec testData) {
- Optional<StatementType> type = clientParser.parseStatement(testData.statement);
- assertThat(type.orElse(null)).isEqualTo(testData.type);
+ Command command = sqlCommandParserImpl.parseStatement(testData.statement).orElse(null);
+ assertThat(command).isEqualTo(testData.command);
}
@ParameterizedTest
@MethodSource("negativeCases")
public void testParseIncompleteStatement(String statement) {
- assertThatThrownBy(() -> clientParser.parseStatement(statement))
+ assertThatThrownBy(() -> sqlCommandParserImpl.parseStatement(statement))
.satisfies(anyCauseMatches(SqlExecutionException.class))
.cause()
.satisfies(anyCauseMatches(SqlParserEOFException.class));
}
private static List<TestSpec> positiveCases() {
return Arrays.asList(
- TestSpec.of(";", OTHER),
+ TestSpec.of(";", null),
TestSpec.of("; ;", OTHER),
// comment and multi lines tests
- TestSpec.of("SHOW --ignore;\n CREATE TABLE tbl;", SHOW_CREATE),
- TestSpec.of("SHOW\n create\t TABLE `tbl`;", SHOW_CREATE),
+ TestSpec.of("SHOW --ignore;\n CREATE TABLE tbl;", OTHER),
+ TestSpec.of("SHOW\n create\t TABLE `tbl`;", OTHER),
TestSpec.of("SHOW -- create\n TABLES;", OTHER),
// special characters tests
- TestSpec.of("SELECT * FROM `tbl`;", SELECT),
- TestSpec.of("SHOW /* ignore */ CREATE TABLE \"tbl\";", SHOW_CREATE),
- TestSpec.of("SELECT '\\';", SELECT),
+ TestSpec.of("SELECT * FROM `tbl`;", OTHER),
+ TestSpec.of("SHOW /* ignore */ CREATE TABLE \"tbl\";", OTHER),
+ TestSpec.of("SELECT '\\';", OTHER),
// normal tests
TestSpec.of("quit;", QUIT), // non case sensitive test
TestSpec.of("QUIT;", QUIT),
TestSpec.of("Quit;", QUIT),
TestSpec.of("QuIt;", QUIT),
TestSpec.of("clear;", CLEAR),
TestSpec.of("help;", HELP),
- TestSpec.of("EXPLAIN PLAN FOR 'what_ever';", EXPLAIN),
- TestSpec.of("SHOW CREATE TABLE(what_ever);", SHOW_CREATE),
- TestSpec.of("SHOW CREATE VIEW (what_ever);", SHOW_CREATE),
- TestSpec.of("SHOW CREATE syntax_error;", SHOW_CREATE),
+ TestSpec.of("EXPLAIN PLAN FOR 'what_ever';", OTHER),
+ TestSpec.of("SHOW CREATE TABLE(what_ever);", OTHER),
+ TestSpec.of("SHOW CREATE VIEW (what_ever);", OTHER),
+ TestSpec.of("SHOW CREATE syntax_error;", OTHER),
TestSpec.of("SHOW TABLES;", OTHER),
- TestSpec.of("BEGIN STATEMENT SET;", BEGIN_STATEMENT_SET),
+ TestSpec.of("BEGIN STATEMENT SET;", OTHER),
TestSpec.of("BEGIN statement;", OTHER),
- TestSpec.of("END;", END),
+ TestSpec.of("END;", OTHER),
TestSpec.of("END statement;", OTHER),
// statement set tests
TestSpec.of(EXECUTE_STATEMENT_SET, OTHER),
- TestSpec.of("EXPLAIN " + EXECUTE_STATEMENT_SET, EXPLAIN),
- TestSpec.of("EXPLAIN BEGIN STATEMENT SET;", EXPLAIN),
+ TestSpec.of("EXPLAIN " + EXECUTE_STATEMENT_SET, OTHER),
+ TestSpec.of("EXPLAIN BEGIN STATEMENT SET;", OTHER),
TestSpec.of(EXECUTE_STATEMENT_SET + "\nEND;", OTHER),
- TestSpec.of("EXPLAIN " + EXECUTE_STATEMENT_SET + "\nEND;", EXPLAIN),
- TestSpec.of("EXPLAIN BEGIN STATEMENT SET;\nEND;", EXPLAIN));
+ TestSpec.of("EXPLAIN " + EXECUTE_STATEMENT_SET + "\nEND;", OTHER),
+ TestSpec.of("EXPLAIN BEGIN STATEMENT SET;\nEND;", OTHER));
}
private static List<String> negativeCases() {
- return Arrays.asList(
- "", "\n", " ", "-- comment;", "SHOW TABLES -- comment;", "SHOW TABLES");
+ return Arrays.asList("-- comment;", "SHOW TABLES -- comment;", "SHOW TABLES");
}
/** Used to load generated data. */
private static class TestSpec {
String statement;
- StatementType type;
+ Command command;
Review Comment:
Nit: add `@Nullable` annotation
##########
flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/util/TestHiveCatalogFactory.java:
##########
@@ -0,0 +1,124 @@
+/*
+ * 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.flink.table.client.util;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.sql.parser.hive.ddl.SqlCreateHiveTable;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
+import org.apache.flink.table.catalog.hive.HiveCatalog;
+import org.apache.flink.table.catalog.hive.HiveTestUtils;
+import org.apache.flink.table.catalog.hive.factories.HiveCatalogFactory;
+import org.apache.flink.table.catalog.hive.factories.HiveCatalogFactoryOptions;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.types.DataType;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/**
+ * A test factory that is the same as {@link HiveCatalogFactory} except returning a {@link
+ * HiveCatalog} always with an embedded Hive metastore to test logic of {@link HiveCatalogFactory}.
+ */
+public class TestHiveCatalogFactory extends HiveCatalogFactory {
+ public static final String ADDITIONAL_TEST_DATABASE = "additional_test_database";
+ public static final String TEST_TABLE = "test_table";
+ static final String TABLE_WITH_PARAMETERIZED_TYPES = "param_types_table";
+
+ @Override
+ public String factoryIdentifier() {
+ return "hive-test";
+ }
+
+ @Override
+ public Catalog createCatalog(Context context) {
+ final Configuration configuration = Configuration.fromMap(context.getOptions());
+
+ // Developers may already have their own production/testing hive-site.xml set in their
+ // environment,
+ // and Flink tests should avoid using those hive-site.xml.
+ // Thus, explicitly create a testing HiveConf for unit tests here
+ Catalog hiveCatalog =
+ HiveTestUtils.createHiveCatalog(
+ context.getName(),
+ configuration.getString(HiveCatalogFactoryOptions.HIVE_VERSION));
+
+ // Creates an additional database to test tableEnv.useDatabase() will switch current
+ // database of the catalog
+ hiveCatalog.open();
+ try {
+ hiveCatalog.createDatabase(
+ ADDITIONAL_TEST_DATABASE,
+ new CatalogDatabaseImpl(new HashMap<>(), null),
Review Comment:
Nit: `Collections.emptyMap()`
##########
flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/parser/SqlMultiLineParser.java:
##########
@@ -44,11 +43,9 @@ public class SqlMultiLineParser extends DefaultParser {
/** Sql command parser. */
private final SqlCommandParser parser;
/** Exception caught in parsing. */
- private Throwable parseException = null;
- /** Operation parsed. */
- private Operation parsedOperation = null;
- /** Command read from terminal. */
- private String command;
+ private SqlExecutionException parseException = null;
+ /** Parsed statement type. */
+ private Command statementType;
Review Comment:
Nit: add `@Nullable` annotation
##########
flink-table/flink-sql-client/src/test/resources/sql/table.q:
##########
@@ -830,15 +895,18 @@ CREATE TABLE IF NOT EXISTS daily_orders (
PRIMARY KEY(dt, `user`) NOT ENFORCED
) PARTITIONED BY (dt) WITH (
'connector' = 'filesystem',
- 'path' = '$VAR_BATCH_PATH',
+ 'path' = '/var/folders/1k/kcnx0zh91c1dkl0_m00ds2mh0000gp/T/junit8174794114152489198/3d95f3a7-ec94-4352-b6b8-2d933b6c89e33458890576674823971',
Review Comment:
revert this value
##########
flink-table/flink-sql-client/src/test/resources/sql/table.q:
##########
@@ -79,7 +79,10 @@ show tables;
# test SHOW CREATE TABLE
show create table orders;
-CREATE TABLE `default_catalog`.`default_database`.`orders` (
++-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
Review Comment:
Is this by design? Personally, I felt the column length be better calculated according to the content length. Besides, the right border of the table is missing?
I thought it should be the following format that aligned with what we have now.
```text
Flink SQL> show create table foo;
+-----------------------------------------------------------+
| result |
+-----------------------------------------------------------+
| CREATE TABLE `default_catalog`.`default_database`.`foo` ( |
| `f0` INT, |
| `f1` VARCHAR(2147483647) |
| ) WITH ( |
| 'connector' = 'datagen' |
| ) |
+-----------------------------------------------------------+
1 row in set
```
##########
flink-table/flink-sql-client/src/test/resources/sql/table.q:
##########
@@ -79,7 +79,10 @@ show tables;
# test SHOW CREATE TABLE
show create table orders;
-CREATE TABLE `default_catalog`.`default_database`.`orders` (
++-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
Review Comment:
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fsk119 commented on pull request #21717: [FLINK-29945][sql-client] Supports to submit SQL to a embedded SQL Ga…
Posted by "fsk119 (via GitHub)" <gi...@apache.org>.
fsk119 commented on PR #21717:
URL: https://github.com/apache/flink/pull/21717#issuecomment-1399118306
All test passes. It would be nice if you can take a look when you are free. @LadyForest
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fsk119 commented on pull request #21717: [FLINK-29945][sql-client] Supports to submit SQL to a embedded SQL Ga…
Posted by "fsk119 (via GitHub)" <gi...@apache.org>.
fsk119 commented on PR #21717:
URL: https://github.com/apache/flink/pull/21717#issuecomment-1407530104
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] LadyForest commented on pull request #21717: [FLINK-29945][sql-client] Supports to submit SQL to a embedded SQL Ga…
Posted by GitBox <gi...@apache.org>.
LadyForest commented on PR #21717:
URL: https://github.com/apache/flink/pull/21717#issuecomment-1397141360
P.S. Wrong cmd will get CLI blocked forever. The stacktrace is attached
```text
Flink SQL> desc extended foo;
Exception in thread "main" org.apache.flink.table.client.SqlClientException: Could not read from command line.
at org.apache.flink.table.client.cli.CliClient.getAndExecuteStatements(CliClient.java:221)
at org.apache.flink.table.client.cli.CliClient.executeInteractive(CliClient.java:186)
at org.apache.flink.table.client.cli.CliClient.executeInInteractiveMode(CliClient.java:132)
at org.apache.flink.table.client.SqlClient.openCli(SqlClient.java:157)
at org.apache.flink.table.client.SqlClient.start(SqlClient.java:102)
at org.apache.flink.table.client.SqlClient.startClient(SqlClient.java:193)
at org.apache.flink.table.client.SqlClient.main(SqlClient.java:167)
Caused by: org.apache.flink.table.client.gateway.SqlExecutionException: Failed to get response.
at org.apache.flink.table.client.gateway.ExecutorImpl.getResponse(ExecutorImpl.java:325)
at org.apache.flink.table.client.gateway.ExecutorImpl.getSessionConfig(ExecutorImpl.java:184)
at org.apache.flink.table.client.cli.CliClient.printExecutionException(CliClient.java:408)
at org.apache.flink.table.client.cli.CliClient.executeStatement(CliClient.java:272)
at org.apache.flink.table.client.cli.CliClient.getAndExecuteStatements(CliClient.java:206)
... 6 more
Caused by: org.apache.flink.runtime.rest.util.RestClientException: [org.apache.flink.runtime.rest.handler.RestHandlerException: Failed to getSessionConfig.
at org.apache.flink.table.gateway.rest.handler.session.GetSessionConfigHandler.handleRequest(GetSessionConfigHandler.java:66)
at org.apache.flink.table.gateway.rest.handler.AbstractSqlGatewayRestHandler.respondToRequest(AbstractSqlGatewayRestHandler.java:84)
at org.apache.flink.table.gateway.rest.handler.AbstractSqlGatewayRestHandler.respondToRequest(AbstractSqlGatewayRestHandler.java:52)
at org.apache.flink.runtime.rest.handler.AbstractHandler.respondAsLeader(AbstractHandler.java:196)
at org.apache.flink.runtime.rest.handler.LeaderRetrievalHandler.lambda$channelRead0$0(LeaderRetrievalHandler.java:83)
at java.util.Optional.ifPresent(Optional.java:159)
at org.apache.flink.util.OptionalConsumer.ifPresent(OptionalConsumer.java:45)
at org.apache.flink.runtime.rest.handler.LeaderRetrievalHandler.channelRead0(LeaderRetrievalHandler.java:80)
at org.apache.flink.runtime.rest.handler.LeaderRetrievalHandler.channelRead0(LeaderRetrievalHandler.java:49)
at org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.runtime.rest.handler.router.RouterHandler.routed(RouterHandler.java:115)
at org.apache.flink.runtime.rest.handler.router.RouterHandler.channelRead0(RouterHandler.java:94)
at org.apache.flink.runtime.rest.handler.router.RouterHandler.channelRead0(RouterHandler.java:55)
at org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.shaded.netty4.io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.runtime.rest.FileUploadHandler.channelRead0(FileUploadHandler.java:208)
at org.apache.flink.runtime.rest.FileUploadHandler.channelRead0(FileUploadHandler.java:69)
at org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.shaded.netty4.io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:436)
at org.apache.flink.shaded.netty4.io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:336)
at org.apache.flink.shaded.netty4.io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:308)
at org.apache.flink.shaded.netty4.io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:251)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.shaded.netty4.io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:788)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:724)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:650)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562)
at org.apache.flink.shaded.netty4.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
at org.apache.flink.shaded.netty4.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.flink.table.gateway.api.utils.SqlGatewayException: Failed to getSessionConfig.
at org.apache.flink.table.gateway.service.SqlGatewayServiceImpl.getSessionConfig(SqlGatewayServiceImpl.java:118)
at org.apache.flink.table.gateway.rest.handler.session.GetSessionConfigHandler.handleRequest(GetSessionConfigHandler.java:61)
... 48 more
Caused by: org.apache.flink.table.gateway.api.utils.SqlGatewayException: Session 'aaf18eff-e23c-46db-93ed-42fd2f6b867d' does not exist.
at org.apache.flink.table.gateway.service.session.SessionManager.getSession(SessionManager.java:127)
at org.apache.flink.table.gateway.service.SqlGatewayServiceImpl.getSession(SqlGatewayServiceImpl.java:373)
at org.apache.flink.table.gateway.service.SqlGatewayServiceImpl.getSessionConfig(SqlGatewayServiceImpl.java:115)
... 49 more
]
at org.apache.flink.runtime.rest.RestClient.parseResponse(RestClient.java:536)
at org.apache.flink.runtime.rest.RestClient.lambda$submitRequest$3(RestClient.java:516)
at java.util.concurrent.CompletableFuture.uniCompose(CompletableFuture.java:952)
at java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:926)
at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fsk119 commented on a diff in pull request #21717: [FLINK-29945][sql-client] Supports to submit SQL to a embedded SQL Ga…
Posted by "fsk119 (via GitHub)" <gi...@apache.org>.
fsk119 commented on code in PR #21717:
URL: https://github.com/apache/flink/pull/21717#discussion_r1083995101
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/session/ConfigureSessionRequestBody.java:
##########
@@ -39,6 +39,10 @@ public class ConfigureSessionRequestBody implements RequestBody {
@Nullable
private final Long timeout;
+ public ConfigureSessionRequestBody(String statement) {
+ this(statement, null);
+ }
+
public ConfigureSessionRequestBody(
Review Comment:
The annotation `JsonCreator` is used when serde the json object. However, the developer uses the creator here to build the object conveniently. Do you mean we should add annotation `JsonCreator` for the constructor `ConfigureSessionRequestBody(@JsonProperty(FIELD_NAME_STATEMENT) String statement, @Nullable @JsonProperty(FIELD_NAME_EXECUTION_TIMEOUT) Long timeout)`?
##########
flink-table/flink-sql-client/src/test/resources/sql/set.q:
##########
@@ -118,47 +118,61 @@ SELECT hive_add_one(1);
Received a total of 1 row
!ok
-REMOVE JAR '$VAR_UDF_JAR_PATH';
-[INFO] The specified jar is removed from session classloader.
-!info
+# TODO: support this in the FLINK-30692
+#REMOVE JAR '$VAR_UDF_JAR_PATH';
+#[INFO] The specified jar is removed from session classloader.
+#!info
-SHOW JARS;
-Empty set
-!ok
+#SHOW JARS;
+#Empty set
+#!ok
+
+reset table.resources.download-dir;
+[INFO] Execute statement succeed.
+!info
# list the configured configuration
set;
-'execution.attached' = 'true'
-'execution.savepoint-restore-mode' = 'NO_CLAIM'
-'execution.savepoint.ignore-unclaimed-state' = 'false'
-'execution.shutdown-on-attached-exit' = 'false'
-'execution.target' = 'remote'
-'jobmanager.rpc.address' = '$VAR_JOBMANAGER_RPC_ADDRESS'
-'k1' = 'v1'
-'pipeline.classpaths' = ''
-'pipeline.jars' = ''
-'rest.port' = '$VAR_REST_PORT'
-'sql-client.execution.result-mode' = 'tableau'
-'table.exec.legacy-cast-behaviour' = 'DISABLED'
-'table.sql-dialect' = 'hive'
++--------------------------------------------+-----------+
+| key | value |
++--------------------------------------------+-----------+
+| execution.attached | true |
+| execution.savepoint-restore-mode | NO_CLAIM |
+| execution.savepoint.ignore-unclaimed-state | false |
+| execution.shutdown-on-attached-exit | false |
+| execution.target | remote |
+| jobmanager.rpc.address | $VAR_JOBMANAGER_RPC_ADDRESS |
+| k1 | v1 |
+| pipeline.classpaths | |
+| pipeline.jars | |
+| rest.port | $VAR_REST_PORT |
+| sql-client.execution.result-mode | tableau |
+| table.exec.legacy-cast-behaviour | DISABLED |
+| table.sql-dialect | hive |
++--------------------------------------------+-----------+
+13 rows in set
Review Comment:
Thanks for pointing it out! I find I use the wrong API in the ExecutorImpl. I should use `ScheduledExecutorService#scheduleAtFixedRate` rather than `ScheduledExecutorService#schedule`. The latter API only triggers the session once.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fsk119 commented on pull request #21717: [FLINK-29945][sql-client] Supports to submit SQL to a embedded SQL Ga…
Posted by "fsk119 (via GitHub)" <gi...@apache.org>.
fsk119 commented on PR #21717:
URL: https://github.com/apache/flink/pull/21717#issuecomment-1407553872
Thanks for your patience and detailed tests! It's nice of you!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fsk119 commented on pull request #21717: [FLINK-29945][sql-client] Supports to submit SQL to a embedded SQL Ga…
Posted by GitBox <gi...@apache.org>.
fsk119 commented on PR #21717:
URL: https://github.com/apache/flink/pull/21717#issuecomment-1398397981
> P.S. Wrong cmd will get CLI blocked forever.
Thanks for reporting the case. With the commit 0e04c7, the Executor will try to close the Operation in the background when users hit ctrl + c.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] LadyForest commented on a diff in pull request #21717: [FLINK-29945][sql-client] Supports to submit SQL to a embedded SQL Ga…
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #21717:
URL: https://github.com/apache/flink/pull/21717#discussion_r1089615342
##########
flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/ExecutorImplITCase.java:
##########
@@ -414,50 +462,130 @@ void testStopJob() throws Exception {
final String insert = "INSERT INTO snk SELECT a FROM src;";
try {
- executor.executeOperation(executor.parseStatement(srcDdl));
- executor.executeOperation(executor.parseStatement(snkDdl));
- TableResult result = executor.executeOperation(executor.parseStatement(insert));
- JobClient jobClient = result.getJobClient().get();
- JobID jobId = jobClient.getJobID();
+ executor.configureSession(srcDdl);
+ executor.configureSession(snkDdl);
+ ClientResult result = executor.executeStatement(insert);
+ JobID jobID = result.getJobId();
// wait till the job turns into running status or the test times out
- JobStatus jobStatus;
- do {
- Thread.sleep(2_000L);
- jobStatus = jobClient.getJobStatus().get();
- } while (jobStatus != JobStatus.RUNNING);
-
- Optional<String> savepoint = executor.stopJob(jobId.toString(), true, true);
- assertThat(savepoint).isPresent();
+ TestUtils.waitUntilAllTasksAreRunning(clusterClient, jobID);
+ StringData savepointPath =
+ CollectionUtil.iteratorToList(
+ executor.executeStatement(
+ String.format("STOP JOB '%s' WITH SAVEPOINT", jobID)))
+ .get(0)
+ .getString(0);
+ assertThat(
+ Files.exists(
+ Paths.get(
+ URI.create(
+ Preconditions.checkNotNull(savepointPath)
+ .toString()))))
+ .isTrue();
Review Comment:
```suggestion
assertThat(savepointPath)
.isNotNull()
.matches(
stringData ->
Files.exists(Paths.get(URI.create(stringData.toString()))));
```
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/session/ConfigureSessionRequestBody.java:
##########
@@ -39,6 +39,10 @@ public class ConfigureSessionRequestBody implements RequestBody {
@Nullable
private final Long timeout;
+ public ConfigureSessionRequestBody(String statement) {
+ this(statement, null);
+ }
+
public ConfigureSessionRequestBody(
Review Comment:
> The annotation `JsonCreator` is used when serde the json object. However, the developer uses the creator here to build the object conveniently. Do you mean we should add annotation `JsonCreator` for the constructor `ConfigureSessionRequestBody(@JsonProperty(FIELD_NAME_STATEMENT) String statement, @Nullable @JsonProperty(FIELD_NAME_EXECUTION_TIMEOUT) Long timeout)`?
Yes. The base interface `RequestBody` says
> Subclass instances are converted to JSON using jackson-databind. Subclasses must have a constructor that accepts all fields of the JSON request, that should be annotated with @JsonCreator.
##########
flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/ExecutorImplITCase.java:
##########
@@ -414,50 +462,130 @@ void testStopJob() throws Exception {
final String insert = "INSERT INTO snk SELECT a FROM src;";
try {
- executor.executeOperation(executor.parseStatement(srcDdl));
- executor.executeOperation(executor.parseStatement(snkDdl));
- TableResult result = executor.executeOperation(executor.parseStatement(insert));
- JobClient jobClient = result.getJobClient().get();
- JobID jobId = jobClient.getJobID();
+ executor.configureSession(srcDdl);
+ executor.configureSession(snkDdl);
+ ClientResult result = executor.executeStatement(insert);
+ JobID jobID = result.getJobId();
// wait till the job turns into running status or the test times out
- JobStatus jobStatus;
- do {
- Thread.sleep(2_000L);
- jobStatus = jobClient.getJobStatus().get();
- } while (jobStatus != JobStatus.RUNNING);
-
- Optional<String> savepoint = executor.stopJob(jobId.toString(), true, true);
- assertThat(savepoint).isPresent();
+ TestUtils.waitUntilAllTasksAreRunning(clusterClient, jobID);
+ StringData savepointPath =
+ CollectionUtil.iteratorToList(
+ executor.executeStatement(
+ String.format("STOP JOB '%s' WITH SAVEPOINT", jobID)))
+ .get(0)
+ .getString(0);
+ assertThat(
+ Files.exists(
+ Paths.get(
+ URI.create(
+ Preconditions.checkNotNull(savepointPath)
+ .toString()))))
+ .isTrue();
Review Comment:
Btw, I found the updated configuration is not read
```sql
Flink SQL> CREATE TABLE src (a STRING) WITH ('connector' = 'datagen');
[INFO] Execute statement succeed.
Flink SQL> CREATE TABLE snk (a STRING) WITH ('connector' = 'blackhole');
[INFO] Execute statement succeed.
Flink SQL> INSERT INTO snk SELECT a FROM src;
[INFO] Submitting SQL update statement to the cluster...
[INFO] SQL update statement has been successfully submitted to the cluster:
Job ID: 9bafe0a3408cf8db4ae8b4fc4110322f
Flink SQL> show jobs;
+----------------------------------+--------------------------------------------------+---------+-------------------------+
| job id | job name | status | start time |
+----------------------------------+--------------------------------------------------+---------+-------------------------+
| 9bafe0a3408cf8db4ae8b4fc4110322f | insert-into_default_catalog.default_database.snk | RUNNING | 2023-01-28T03:47:56.161 |
+----------------------------------+--------------------------------------------------+---------+-------------------------+
1 row in set
Flink SQL> stop job '9bafe0a3408cf8db4ae8b4fc4110322f' with savepoint;
[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.runtime.rest.util.RestClientException: [org.apache.flink.runtime.rest.handler.RestHandlerException: Config key [state.savepoints.dir] is not set. Property [targetDirectory] must be provided.
Flink SQL> set 'state.savepoints.dir' = '/tmp/test';
[INFO] Execute statement succeed.
Flink SQL> stop job '9bafe0a3408cf8db4ae8b4fc4110322f' with savepoint;
[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.runtime.rest.util.RestClientException: [org.apache.flink.runtime.rest.handler.RestHandlerException: Config key [state.savepoints.dir] is not set. Property [targetDirectory] must be provided.
Flink SQL> set;
+--------------------------------------------+----------------------------------------------------------------------------------------------------------------------------------------------------+
| key | value |
+--------------------------------------------+----------------------------------------------------------------------------------------------------------------------------------------------------+
| execution.attached | true |
| execution.savepoint-restore-mode | NO_CLAIM |
| execution.savepoint.ignore-unclaimed-state | false |
| execution.shutdown-on-attached-exit | false |
| execution.target | remote |
| jobmanager.bind-host | localhost |
| jobmanager.execution.failover-strategy | region |
| jobmanager.memory.process.size | 1600m |
| jobmanager.rpc.address | localhost |
| jobmanager.rpc.port | 6123 |
| parallelism.default | 1 |
| pipeline.classpaths | |
| pipeline.jars | file:${baseDir}/flink/flink-dist/target/flink-1.17-SNAPSHOT-bin/flink-1.17-SNAPSHOT/opt/flink-python-1.17-SNAPSHOT.jar |
| rest.address | localhost |
| rest.bind-address | localhost |
| sql-gateway.endpoint.rest.address | localhost |
| sql-gateway.endpoint.rest.port | 58950 |
| state.savepoints.dir | /tmp/test |
| table.resources.download-dir | /var/folders/xd/9dp1y4vd3h56kjkvdk426l500000gp/T/sql-gateway-3623ad21-994f-46e3-bfa5-2fb01a99d3c6 |
| taskmanager.bind-host | localhost |
| taskmanager.host | localhost |
| taskmanager.memory.process.size | 1728m |
| taskmanager.numberOfTaskSlots | 1 |
+--------------------------------------------+----------------------------------------------------------------------------------------------------------------------------------------------------+
23 rows in set
Flink SQL> set sql-client.verbose=true;
[INFO] Execute statement succeed.
Flink SQL> stop job '9bafe0a3408cf8db4ae8b4fc4110322f' with savepoint;
[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.runtime.rest.util.RestClientException: [Internal server error., <Exception on server side:
org.apache.flink.table.gateway.api.utils.SqlGatewayException: org.apache.flink.table.gateway.api.utils.SqlGatewayException: Failed to fetchResults.
at org.apache.flink.table.gateway.rest.handler.statement.FetchResultsHandler.handleRequest(FetchResultsHandler.java:85)
at org.apache.flink.table.gateway.rest.handler.AbstractSqlGatewayRestHandler.respondToRequest(AbstractSqlGatewayRestHandler.java:84)
at org.apache.flink.table.gateway.rest.handler.AbstractSqlGatewayRestHandler.respondToRequest(AbstractSqlGatewayRestHandler.java:52)
at org.apache.flink.runtime.rest.handler.AbstractHandler.respondAsLeader(AbstractHandler.java:196)
at org.apache.flink.runtime.rest.handler.LeaderRetrievalHandler.lambda$channelRead0$0(LeaderRetrievalHandler.java:83)
at java.util.Optional.ifPresent(Optional.java:159)
at org.apache.flink.util.OptionalConsumer.ifPresent(OptionalConsumer.java:45)
at org.apache.flink.runtime.rest.handler.LeaderRetrievalHandler.channelRead0(LeaderRetrievalHandler.java:80)
at org.apache.flink.runtime.rest.handler.LeaderRetrievalHandler.channelRead0(LeaderRetrievalHandler.java:49)
at org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.runtime.rest.handler.router.RouterHandler.routed(RouterHandler.java:115)
at org.apache.flink.runtime.rest.handler.router.RouterHandler.channelRead0(RouterHandler.java:94)
at org.apache.flink.runtime.rest.handler.router.RouterHandler.channelRead0(RouterHandler.java:55)
at org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.shaded.netty4.io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.runtime.rest.FileUploadHandler.channelRead0(FileUploadHandler.java:208)
at org.apache.flink.runtime.rest.FileUploadHandler.channelRead0(FileUploadHandler.java:69)
at org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.shaded.netty4.io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:436)
at org.apache.flink.shaded.netty4.io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:336)
at org.apache.flink.shaded.netty4.io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:308)
at org.apache.flink.shaded.netty4.io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:251)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.shaded.netty4.io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:788)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:724)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:650)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562)
at org.apache.flink.shaded.netty4.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
at org.apache.flink.shaded.netty4.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.flink.table.gateway.api.utils.SqlGatewayException: Failed to fetchResults.
at org.apache.flink.table.gateway.service.SqlGatewayServiceImpl.fetchResults(SqlGatewayServiceImpl.java:229)
at org.apache.flink.table.gateway.rest.handler.statement.FetchResultsHandler.handleRequest(FetchResultsHandler.java:83)
... 48 more
Caused by: org.apache.flink.table.gateway.service.utils.SqlExecutionException: Failed to execute the operation d6d8fe6f-b0d0-4040-8420-80c605cf6b34.
at org.apache.flink.table.gateway.service.operation.OperationManager$Operation.processThrowable(OperationManager.java:398)
at org.apache.flink.table.gateway.service.operation.OperationManager$Operation.lambda$run$0(OperationManager.java:251)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
... 1 more
Caused by: org.apache.flink.table.gateway.service.utils.SqlExecutionException: Could not stop job 9bafe0a3408cf8db4ae8b4fc4110322f for operation d6d8fe6f-b0d0-4040-8420-80c605cf6b34.
at org.apache.flink.table.gateway.service.operation.OperationExecutor.callStopJobOperation(OperationExecutor.java:520)
at org.apache.flink.table.gateway.service.operation.OperationExecutor.executeOperation(OperationExecutor.java:332)
at org.apache.flink.table.gateway.service.operation.OperationExecutor.executeStatement(OperationExecutor.java:181)
at org.apache.flink.table.gateway.service.SqlGatewayServiceImpl.lambda$executeStatement$1(SqlGatewayServiceImpl.java:212)
at org.apache.flink.table.gateway.service.operation.OperationManager.lambda$submitOperation$1(OperationManager.java:110)
at org.apache.flink.table.gateway.service.operation.OperationManager$Operation.lambda$run$0(OperationManager.java:242)
... 7 more
Caused by: org.apache.flink.table.gateway.service.utils.SqlExecutionException: Failed to run cluster action.
at org.apache.flink.table.gateway.service.operation.OperationExecutor.runClusterAction(OperationExecutor.java:600)
at org.apache.flink.table.gateway.service.operation.OperationExecutor.callStopJobOperation(OperationExecutor.java:487)
... 12 more
Caused by: org.apache.flink.util.FlinkException: Could not stop job 9bafe0a3408cf8db4ae8b4fc4110322f in session d6d8fe6f-b0d0-4040-8420-80c605cf6b34.
at org.apache.flink.table.gateway.service.operation.OperationExecutor.lambda$callStopJobOperation$10(OperationExecutor.java:510)
at org.apache.flink.table.gateway.service.operation.OperationExecutor.runClusterAction(OperationExecutor.java:598)
... 13 more
Caused by: java.util.concurrent.ExecutionException: org.apache.flink.runtime.rest.util.RestClientException: [org.apache.flink.runtime.rest.handler.RestHandlerException: Config key [state.savepoints.dir] is not set. Property [targetDirectory] must be provided.
at org.apache.flink.runtime.rest.handler.job.savepoints.SavepointHandlers$StopWithSavepointHandler.triggerOperation(SavepointHandlers.java:200)
at org.apache.flink.runtime.rest.handler.job.savepoints.SavepointHandlers$SavepointHandlerBase.handleRequest(SavepointHandlers.java:149)
at org.apache.flink.runtime.rest.handler.job.savepoints.SavepointHandlers$StopWithSavepointHandler.handleRequest(SavepointHandlers.java:170)
at org.apache.flink.runtime.rest.handler.AbstractRestHandler.respondToRequest(AbstractRestHandler.java:83)
at org.apache.flink.runtime.rest.handler.AbstractHandler.respondAsLeader(AbstractHandler.java:196)
at org.apache.flink.runtime.rest.handler.LeaderRetrievalHandler.lambda$channelRead0$0(LeaderRetrievalHandler.java:83)
at java.util.Optional.ifPresent(Optional.java:159)
at org.apache.flink.util.OptionalConsumer.ifPresent(OptionalConsumer.java:45)
at org.apache.flink.runtime.rest.handler.LeaderRetrievalHandler.channelRead0(LeaderRetrievalHandler.java:80)
at org.apache.flink.runtime.rest.handler.LeaderRetrievalHandler.channelRead0(LeaderRetrievalHandler.java:49)
at org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.runtime.rest.handler.router.RouterHandler.routed(RouterHandler.java:115)
at org.apache.flink.runtime.rest.handler.router.RouterHandler.channelRead0(RouterHandler.java:94)
at org.apache.flink.runtime.rest.handler.router.RouterHandler.channelRead0(RouterHandler.java:55)
at org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.shaded.netty4.io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.runtime.rest.FileUploadHandler.channelRead0(FileUploadHandler.java:208)
at org.apache.flink.runtime.rest.FileUploadHandler.channelRead0(FileUploadHandler.java:69)
at org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.shaded.netty4.io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:436)
at org.apache.flink.shaded.netty4.io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:336)
at org.apache.flink.shaded.netty4.io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:308)
at org.apache.flink.shaded.netty4.io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:251)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.shaded.netty4.io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:788)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:724)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:650)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562)
at org.apache.flink.shaded.netty4.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
at org.apache.flink.shaded.netty4.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at java.lang.Thread.run(Thread.java:748)
]
at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1915)
at org.apache.flink.table.gateway.service.operation.OperationExecutor.lambda$callStopJobOperation$10(OperationExecutor.java:502)
... 14 more
Caused by: org.apache.flink.runtime.rest.util.RestClientException: [org.apache.flink.runtime.rest.handler.RestHandlerException: Config key [state.savepoints.dir] is not set. Property [targetDirectory] must be provided.
at org.apache.flink.runtime.rest.handler.job.savepoints.SavepointHandlers$StopWithSavepointHandler.triggerOperation(SavepointHandlers.java:200)
at org.apache.flink.runtime.rest.handler.job.savepoints.SavepointHandlers$SavepointHandlerBase.handleRequest(SavepointHandlers.java:149)
at org.apache.flink.runtime.rest.handler.job.savepoints.SavepointHandlers$StopWithSavepointHandler.handleRequest(SavepointHandlers.java:170)
at org.apache.flink.runtime.rest.handler.AbstractRestHandler.respondToRequest(AbstractRestHandler.java:83)
at org.apache.flink.runtime.rest.handler.AbstractHandler.respondAsLeader(AbstractHandler.java:196)
at org.apache.flink.runtime.rest.handler.LeaderRetrievalHandler.lambda$channelRead0$0(LeaderRetrievalHandler.java:83)
at java.util.Optional.ifPresent(Optional.java:159)
at org.apache.flink.util.OptionalConsumer.ifPresent(OptionalConsumer.java:45)
at org.apache.flink.runtime.rest.handler.LeaderRetrievalHandler.channelRead0(LeaderRetrievalHandler.java:80)
at org.apache.flink.runtime.rest.handler.LeaderRetrievalHandler.channelRead0(LeaderRetrievalHandler.java:49)
at org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.runtime.rest.handler.router.RouterHandler.routed(RouterHandler.java:115)
at org.apache.flink.runtime.rest.handler.router.RouterHandler.channelRead0(RouterHandler.java:94)
at org.apache.flink.runtime.rest.handler.router.RouterHandler.channelRead0(RouterHandler.java:55)
at org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.shaded.netty4.io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.runtime.rest.FileUploadHandler.channelRead0(FileUploadHandler.java:208)
at org.apache.flink.runtime.rest.FileUploadHandler.channelRead0(FileUploadHandler.java:69)
at org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.shaded.netty4.io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:436)
at org.apache.flink.shaded.netty4.io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:336)
at org.apache.flink.shaded.netty4.io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:308)
at org.apache.flink.shaded.netty4.io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:251)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.shaded.netty4.io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:788)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:724)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:650)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562)
at org.apache.flink.shaded.netty4.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
at org.apache.flink.shaded.netty4.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at java.lang.Thread.run(Thread.java:748)
]
at org.apache.flink.runtime.rest.RestClient.parseResponse(RestClient.java:536)
at org.apache.flink.runtime.rest.RestClient.lambda$submitRequest$3(RestClient.java:516)
at java.util.concurrent.CompletableFuture.uniCompose(CompletableFuture.java:952)
at java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:926)
at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
... 3 more
End of exception on server side>]
at org.apache.flink.runtime.rest.RestClient.parseResponse(RestClient.java:536)
at org.apache.flink.runtime.rest.RestClient.lambda$submitRequest$3(RestClient.java:516)
at java.util.concurrent.CompletableFuture.uniCompose(CompletableFuture.java:952)
at java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:926)
at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
```
##########
flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/ExecutorImplITCase.java:
##########
@@ -414,50 +462,130 @@ void testStopJob() throws Exception {
final String insert = "INSERT INTO snk SELECT a FROM src;";
try {
- executor.executeOperation(executor.parseStatement(srcDdl));
- executor.executeOperation(executor.parseStatement(snkDdl));
- TableResult result = executor.executeOperation(executor.parseStatement(insert));
- JobClient jobClient = result.getJobClient().get();
- JobID jobId = jobClient.getJobID();
+ executor.configureSession(srcDdl);
+ executor.configureSession(snkDdl);
+ ClientResult result = executor.executeStatement(insert);
+ JobID jobID = result.getJobId();
// wait till the job turns into running status or the test times out
- JobStatus jobStatus;
- do {
- Thread.sleep(2_000L);
- jobStatus = jobClient.getJobStatus().get();
- } while (jobStatus != JobStatus.RUNNING);
-
- Optional<String> savepoint = executor.stopJob(jobId.toString(), true, true);
- assertThat(savepoint).isPresent();
+ TestUtils.waitUntilAllTasksAreRunning(clusterClient, jobID);
+ StringData savepointPath =
+ CollectionUtil.iteratorToList(
+ executor.executeStatement(
+ String.format("STOP JOB '%s' WITH SAVEPOINT", jobID)))
+ .get(0)
+ .getString(0);
+ assertThat(
+ Files.exists(
+ Paths.get(
+ URI.create(
+ Preconditions.checkNotNull(savepointPath)
+ .toString()))))
+ .isTrue();
Review Comment:
And `stop job` seems not to work
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fsk119 commented on pull request #21717: [FLINK-29945][sql-client] Supports to submit SQL to a embedded SQL Ga…
Posted by "fsk119 (via GitHub)" <gi...@apache.org>.
fsk119 commented on PR #21717:
URL: https://github.com/apache/flink/pull/21717#issuecomment-1401848129
I also run the TPC-h e2e test in my private pipeline. It all passes. I think with the commit a94aff3, the sql client exits without the risk of the resource leak if the connection is stable.
https://dev.azure.com/1059623455/Flink/_build/results?buildId=501&view=logs&s=ae4f8708-9994-57d3-c2d7-b892156e7812
https://dev.azure.com/1059623455/Flink/_build/results?buildId=500&view=logs&j=0e31ee24-31a6-528c-a4bf-45cde9b2a14e&t=ff03a8fa-e84e-5199-efb2-5433077ce8e2
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] LadyForest commented on a diff in pull request #21717: [FLINK-29945][sql-client] Supports to submit SQL to a embedded SQL Ga…
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #21717:
URL: https://github.com/apache/flink/pull/21717#discussion_r1089690301
##########
flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/ExecutorImpl.java:
##########
@@ -0,0 +1,447 @@
+/*
+ * 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.flink.table.client.gateway;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.runtime.rest.RestClient;
+import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.runtime.rest.messages.MessageParameters;
+import org.apache.flink.runtime.rest.messages.RequestBody;
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+import org.apache.flink.runtime.rest.util.RestClientException;
+import org.apache.flink.table.api.SqlParserEOFException;
+import org.apache.flink.table.client.SqlClientException;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.gateway.SqlGateway;
+import org.apache.flink.table.gateway.api.operation.OperationHandle;
+import org.apache.flink.table.gateway.api.results.ResultSet;
+import org.apache.flink.table.gateway.api.session.SessionHandle;
+import org.apache.flink.table.gateway.rest.header.operation.CancelOperationHeaders;
+import org.apache.flink.table.gateway.rest.header.operation.CloseOperationHeaders;
+import org.apache.flink.table.gateway.rest.header.session.CloseSessionHeaders;
+import org.apache.flink.table.gateway.rest.header.session.ConfigureSessionHeaders;
+import org.apache.flink.table.gateway.rest.header.session.GetSessionConfigHeaders;
+import org.apache.flink.table.gateway.rest.header.session.OpenSessionHeaders;
+import org.apache.flink.table.gateway.rest.header.session.TriggerSessionHeartbeatHeaders;
+import org.apache.flink.table.gateway.rest.header.statement.CompleteStatementHeaders;
+import org.apache.flink.table.gateway.rest.header.statement.ExecuteStatementHeaders;
+import org.apache.flink.table.gateway.rest.header.statement.FetchResultsHeaders;
+import org.apache.flink.table.gateway.rest.message.operation.OperationMessageParameters;
+import org.apache.flink.table.gateway.rest.message.session.CloseSessionResponseBody;
+import org.apache.flink.table.gateway.rest.message.session.ConfigureSessionRequestBody;
+import org.apache.flink.table.gateway.rest.message.session.GetSessionConfigResponseBody;
+import org.apache.flink.table.gateway.rest.message.session.OpenSessionRequestBody;
+import org.apache.flink.table.gateway.rest.message.session.OpenSessionResponseBody;
+import org.apache.flink.table.gateway.rest.message.session.SessionMessageParameters;
+import org.apache.flink.table.gateway.rest.message.statement.CompleteStatementRequestBody;
+import org.apache.flink.table.gateway.rest.message.statement.ExecuteStatementRequestBody;
+import org.apache.flink.table.gateway.rest.message.statement.ExecuteStatementResponseBody;
+import org.apache.flink.table.gateway.rest.message.statement.FetchResultsMessageParameters;
+import org.apache.flink.table.gateway.rest.message.statement.FetchResultsResponseBody;
+import org.apache.flink.table.gateway.rest.serde.ResultInfo;
+import org.apache.flink.table.gateway.rest.util.RowFormat;
+import org.apache.flink.table.gateway.rest.util.SqlGatewayRestEndpointUtils;
+import org.apache.flink.table.gateway.service.context.DefaultContext;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.util.function.SupplierWithException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.Iterator;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+
+import static org.apache.flink.table.gateway.rest.handler.session.CloseSessionHandler.CLOSE_MESSAGE;
+
+/** Executor to connect to {@link SqlGateway} and execute statements. */
+public class ExecutorImpl implements Executor {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ExecutorImpl.class);
+ private static final long HEARTBEAT_INTERVAL_MILLISECONDS = 60_000L;
+
+ private final DefaultContext defaultContext;
+ private final InetSocketAddress gatewayAddress;
+ private final long heartbeatInterval;
+ private final ExecutorService service;
+ private final ScheduledExecutorService heartbeatScheduler;
+
+ private RestClient restClient;
+ private SessionHandle sessionHandle;
+
+ @VisibleForTesting
+ public ExecutorImpl(
+ DefaultContext defaultContext,
+ InetSocketAddress gatewayAddress,
+ long heartbeatInterval) {
+ this.defaultContext = defaultContext;
+ this.gatewayAddress = gatewayAddress;
+ this.heartbeatInterval = heartbeatInterval;
+ this.service = Executors.newCachedThreadPool();
+ this.heartbeatScheduler = Executors.newSingleThreadScheduledExecutor();
+ }
+
+ public ExecutorImpl(DefaultContext defaultContext, InetSocketAddress gatewayAddress) {
+ this(defaultContext, gatewayAddress, HEARTBEAT_INTERVAL_MILLISECONDS);
+ }
+
+ public void openSession(@Nullable String sessionId) {
+ try {
+ restClient = new RestClient(defaultContext.getFlinkConfig(), service);
+ } catch (Exception e) {
+ throw new SqlExecutionException("Can not create the Rest Client.", e);
+ }
+
+ LOG.info("Open session to {}.", gatewayAddress);
+ OpenSessionRequestBody request =
+ new OpenSessionRequestBody(sessionId, defaultContext.getFlinkConfig().toMap());
+ try {
+ // open session to address:port
+ OpenSessionResponseBody response =
+ sendRequest(
+ OpenSessionHeaders.getInstance(),
+ EmptyMessageParameters.getInstance(),
+ request)
+ .get();
+ sessionHandle = new SessionHandle(UUID.fromString(response.getSessionHandle()));
+ // register heartbeat service
+ heartbeatScheduler.scheduleAtFixedRate(
+ () ->
+ getResponse(
+ sendRequest(
+ TriggerSessionHeartbeatHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ EmptyRequestBody.getInstance())),
+ heartbeatInterval,
+ heartbeatInterval,
+ TimeUnit.MILLISECONDS);
+ // register dependencies
+ defaultContext
+ .getDependencies()
+ .forEach(jar -> configureSession(String.format("ADD JAR '%s'", jar)));
+ } catch (Exception e) {
+ throw new SqlClientException(
+ String.format("Failed to open session to %s", gatewayAddress), e);
+ }
+ }
+
+ public void closeSession() throws SqlExecutionException {
+ if (sessionHandle == null) {
+ return;
+ }
+ try {
+ CompletableFuture<CloseSessionResponseBody> response =
+ sendRequest(
+ CloseSessionHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ EmptyRequestBody.getInstance());
+
+ if (!response.get().getStatus().equals(CLOSE_MESSAGE)) {
+ LOG.warn("The status of close session response isn't {}.", CLOSE_MESSAGE);
+ }
+ } catch (Exception e) {
+ LOG.warn(
+ String.format(
+ "Unexpected error occurs when closing session %s.", sessionHandle),
+ e);
+ // ignore any throwable to keep the cleanup running
+ } finally {
+ sessionHandle = null;
+ }
+ }
+
+ public void configureSession(String statement) {
+ try {
+ sendRequest(
+ ConfigureSessionHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ new ConfigureSessionRequestBody(statement))
+ .get();
+ } catch (Exception e) {
+ throw new SqlExecutionException(
+ String.format("Failed to open session to %s", gatewayAddress), e);
+ }
+ }
+
+ public ReadableConfig getSessionConfig() {
+ try {
+ GetSessionConfigResponseBody response =
+ getResponse(
+ sendRequest(
+ GetSessionConfigHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ EmptyRequestBody.getInstance()));
+ return Configuration.fromMap(response.getProperties());
+ } catch (Exception e) {
+ throw new SqlExecutionException("Failed to get the get session config.", e);
+ }
+ }
+
+ public ClientResult executeStatement(String statement) {
+ ExecuteStatementRequestBody request = new ExecuteStatementRequestBody(statement);
+ CompletableFuture<ExecuteStatementResponseBody> executeStatementResponse =
+ sendRequest(
+ ExecuteStatementHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ request);
+ // It's possible that the execution is canceled during the submission.
+ // Close the Operation in background to make sure the execution can continue.
+ getResponse(
+ executeStatementResponse,
+ e -> {
+ service.submit(
+ () -> {
+ try {
+ ExecuteStatementResponseBody executeStatementResponseBody =
+ executeStatementResponse.get();
+ OperationHandle operationHandle =
+ new OperationHandle(
+ UUID.fromString(
+ executeStatementResponseBody
+ .getOperationHandle()));
+ // close operation in background to make sure users can not
Review Comment:
I see this logic repeated several times. What about adding a util method like
```java
private CompletableFuture<OperationStatusResponseBody> closeOperationAsync(
OperationHandle operationHandle) {
return sendRequest(
CloseOperationHeaders.getInstance(),
new OperationMessageParameters(sessionHandle, operationHandle),
EmptyRequestBody.getInstance());
}
```
##########
flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/ExecutorImpl.java:
##########
@@ -0,0 +1,447 @@
+/*
+ * 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.flink.table.client.gateway;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.runtime.rest.RestClient;
+import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.runtime.rest.messages.MessageParameters;
+import org.apache.flink.runtime.rest.messages.RequestBody;
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+import org.apache.flink.runtime.rest.util.RestClientException;
+import org.apache.flink.table.api.SqlParserEOFException;
+import org.apache.flink.table.client.SqlClientException;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.gateway.SqlGateway;
+import org.apache.flink.table.gateway.api.operation.OperationHandle;
+import org.apache.flink.table.gateway.api.results.ResultSet;
+import org.apache.flink.table.gateway.api.session.SessionHandle;
+import org.apache.flink.table.gateway.rest.header.operation.CancelOperationHeaders;
+import org.apache.flink.table.gateway.rest.header.operation.CloseOperationHeaders;
+import org.apache.flink.table.gateway.rest.header.session.CloseSessionHeaders;
+import org.apache.flink.table.gateway.rest.header.session.ConfigureSessionHeaders;
+import org.apache.flink.table.gateway.rest.header.session.GetSessionConfigHeaders;
+import org.apache.flink.table.gateway.rest.header.session.OpenSessionHeaders;
+import org.apache.flink.table.gateway.rest.header.session.TriggerSessionHeartbeatHeaders;
+import org.apache.flink.table.gateway.rest.header.statement.CompleteStatementHeaders;
+import org.apache.flink.table.gateway.rest.header.statement.ExecuteStatementHeaders;
+import org.apache.flink.table.gateway.rest.header.statement.FetchResultsHeaders;
+import org.apache.flink.table.gateway.rest.message.operation.OperationMessageParameters;
+import org.apache.flink.table.gateway.rest.message.session.CloseSessionResponseBody;
+import org.apache.flink.table.gateway.rest.message.session.ConfigureSessionRequestBody;
+import org.apache.flink.table.gateway.rest.message.session.GetSessionConfigResponseBody;
+import org.apache.flink.table.gateway.rest.message.session.OpenSessionRequestBody;
+import org.apache.flink.table.gateway.rest.message.session.OpenSessionResponseBody;
+import org.apache.flink.table.gateway.rest.message.session.SessionMessageParameters;
+import org.apache.flink.table.gateway.rest.message.statement.CompleteStatementRequestBody;
+import org.apache.flink.table.gateway.rest.message.statement.ExecuteStatementRequestBody;
+import org.apache.flink.table.gateway.rest.message.statement.ExecuteStatementResponseBody;
+import org.apache.flink.table.gateway.rest.message.statement.FetchResultsMessageParameters;
+import org.apache.flink.table.gateway.rest.message.statement.FetchResultsResponseBody;
+import org.apache.flink.table.gateway.rest.serde.ResultInfo;
+import org.apache.flink.table.gateway.rest.util.RowFormat;
+import org.apache.flink.table.gateway.rest.util.SqlGatewayRestEndpointUtils;
+import org.apache.flink.table.gateway.service.context.DefaultContext;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.util.function.SupplierWithException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.Iterator;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+
+import static org.apache.flink.table.gateway.rest.handler.session.CloseSessionHandler.CLOSE_MESSAGE;
+
+/** Executor to connect to {@link SqlGateway} and execute statements. */
+public class ExecutorImpl implements Executor {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ExecutorImpl.class);
+ private static final long HEARTBEAT_INTERVAL_MILLISECONDS = 60_000L;
+
+ private final DefaultContext defaultContext;
+ private final InetSocketAddress gatewayAddress;
+ private final long heartbeatInterval;
+ private final ExecutorService service;
+ private final ScheduledExecutorService heartbeatScheduler;
+
+ private RestClient restClient;
+ private SessionHandle sessionHandle;
+
+ @VisibleForTesting
+ public ExecutorImpl(
+ DefaultContext defaultContext,
+ InetSocketAddress gatewayAddress,
+ long heartbeatInterval) {
+ this.defaultContext = defaultContext;
+ this.gatewayAddress = gatewayAddress;
+ this.heartbeatInterval = heartbeatInterval;
+ this.service = Executors.newCachedThreadPool();
+ this.heartbeatScheduler = Executors.newSingleThreadScheduledExecutor();
+ }
+
+ public ExecutorImpl(DefaultContext defaultContext, InetSocketAddress gatewayAddress) {
+ this(defaultContext, gatewayAddress, HEARTBEAT_INTERVAL_MILLISECONDS);
+ }
+
+ public void openSession(@Nullable String sessionId) {
+ try {
+ restClient = new RestClient(defaultContext.getFlinkConfig(), service);
+ } catch (Exception e) {
+ throw new SqlExecutionException("Can not create the Rest Client.", e);
+ }
+
+ LOG.info("Open session to {}.", gatewayAddress);
+ OpenSessionRequestBody request =
+ new OpenSessionRequestBody(sessionId, defaultContext.getFlinkConfig().toMap());
+ try {
+ // open session to address:port
+ OpenSessionResponseBody response =
+ sendRequest(
+ OpenSessionHeaders.getInstance(),
+ EmptyMessageParameters.getInstance(),
+ request)
+ .get();
+ sessionHandle = new SessionHandle(UUID.fromString(response.getSessionHandle()));
+ // register heartbeat service
+ heartbeatScheduler.scheduleAtFixedRate(
+ () ->
+ getResponse(
+ sendRequest(
+ TriggerSessionHeartbeatHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ EmptyRequestBody.getInstance())),
+ heartbeatInterval,
+ heartbeatInterval,
+ TimeUnit.MILLISECONDS);
+ // register dependencies
+ defaultContext
+ .getDependencies()
+ .forEach(jar -> configureSession(String.format("ADD JAR '%s'", jar)));
+ } catch (Exception e) {
+ throw new SqlClientException(
+ String.format("Failed to open session to %s", gatewayAddress), e);
+ }
+ }
+
+ public void closeSession() throws SqlExecutionException {
+ if (sessionHandle == null) {
+ return;
+ }
+ try {
+ CompletableFuture<CloseSessionResponseBody> response =
+ sendRequest(
+ CloseSessionHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ EmptyRequestBody.getInstance());
+
+ if (!response.get().getStatus().equals(CLOSE_MESSAGE)) {
+ LOG.warn("The status of close session response isn't {}.", CLOSE_MESSAGE);
+ }
+ } catch (Exception e) {
+ LOG.warn(
+ String.format(
+ "Unexpected error occurs when closing session %s.", sessionHandle),
+ e);
+ // ignore any throwable to keep the cleanup running
+ } finally {
+ sessionHandle = null;
+ }
+ }
+
+ public void configureSession(String statement) {
+ try {
+ sendRequest(
+ ConfigureSessionHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ new ConfigureSessionRequestBody(statement))
+ .get();
+ } catch (Exception e) {
+ throw new SqlExecutionException(
+ String.format("Failed to open session to %s", gatewayAddress), e);
+ }
+ }
+
+ public ReadableConfig getSessionConfig() {
+ try {
+ GetSessionConfigResponseBody response =
+ getResponse(
+ sendRequest(
+ GetSessionConfigHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ EmptyRequestBody.getInstance()));
+ return Configuration.fromMap(response.getProperties());
+ } catch (Exception e) {
+ throw new SqlExecutionException("Failed to get the get session config.", e);
+ }
+ }
+
+ public ClientResult executeStatement(String statement) {
+ ExecuteStatementRequestBody request = new ExecuteStatementRequestBody(statement);
+ CompletableFuture<ExecuteStatementResponseBody> executeStatementResponse =
+ sendRequest(
+ ExecuteStatementHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ request);
+ // It's possible that the execution is canceled during the submission.
+ // Close the Operation in background to make sure the execution can continue.
+ getResponse(
+ executeStatementResponse,
+ e -> {
+ service.submit(
+ () -> {
+ try {
+ ExecuteStatementResponseBody executeStatementResponseBody =
+ executeStatementResponse.get();
+ OperationHandle operationHandle =
Review Comment:
Nit: can we simplify the creation of `OperationHandle` by introducing a util method like
```java
private OperationHandle getOperationHandle(Supplier<String> handleSupplier) {
return new OperationHandle(UUID.fromString(handleSupplier.get()));
}
```
```java
OperationHandle operationHandle =
getOperationHandle(
executeStatementResponseBody
::getOperationHandle);
```
and
```java
OperationHandle operationHandle =
getOperationHandle(
() -> getResponse(executeStatementResponse).getOperationHandle());
```
##########
flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/ExecutorImpl.java:
##########
@@ -0,0 +1,447 @@
+/*
+ * 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.flink.table.client.gateway;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.runtime.rest.RestClient;
+import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.runtime.rest.messages.MessageParameters;
+import org.apache.flink.runtime.rest.messages.RequestBody;
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+import org.apache.flink.runtime.rest.util.RestClientException;
+import org.apache.flink.table.api.SqlParserEOFException;
+import org.apache.flink.table.client.SqlClientException;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.gateway.SqlGateway;
+import org.apache.flink.table.gateway.api.operation.OperationHandle;
+import org.apache.flink.table.gateway.api.results.ResultSet;
+import org.apache.flink.table.gateway.api.session.SessionHandle;
+import org.apache.flink.table.gateway.rest.header.operation.CancelOperationHeaders;
+import org.apache.flink.table.gateway.rest.header.operation.CloseOperationHeaders;
+import org.apache.flink.table.gateway.rest.header.session.CloseSessionHeaders;
+import org.apache.flink.table.gateway.rest.header.session.ConfigureSessionHeaders;
+import org.apache.flink.table.gateway.rest.header.session.GetSessionConfigHeaders;
+import org.apache.flink.table.gateway.rest.header.session.OpenSessionHeaders;
+import org.apache.flink.table.gateway.rest.header.session.TriggerSessionHeartbeatHeaders;
+import org.apache.flink.table.gateway.rest.header.statement.CompleteStatementHeaders;
+import org.apache.flink.table.gateway.rest.header.statement.ExecuteStatementHeaders;
+import org.apache.flink.table.gateway.rest.header.statement.FetchResultsHeaders;
+import org.apache.flink.table.gateway.rest.message.operation.OperationMessageParameters;
+import org.apache.flink.table.gateway.rest.message.session.CloseSessionResponseBody;
+import org.apache.flink.table.gateway.rest.message.session.ConfigureSessionRequestBody;
+import org.apache.flink.table.gateway.rest.message.session.GetSessionConfigResponseBody;
+import org.apache.flink.table.gateway.rest.message.session.OpenSessionRequestBody;
+import org.apache.flink.table.gateway.rest.message.session.OpenSessionResponseBody;
+import org.apache.flink.table.gateway.rest.message.session.SessionMessageParameters;
+import org.apache.flink.table.gateway.rest.message.statement.CompleteStatementRequestBody;
+import org.apache.flink.table.gateway.rest.message.statement.ExecuteStatementRequestBody;
+import org.apache.flink.table.gateway.rest.message.statement.ExecuteStatementResponseBody;
+import org.apache.flink.table.gateway.rest.message.statement.FetchResultsMessageParameters;
+import org.apache.flink.table.gateway.rest.message.statement.FetchResultsResponseBody;
+import org.apache.flink.table.gateway.rest.serde.ResultInfo;
+import org.apache.flink.table.gateway.rest.util.RowFormat;
+import org.apache.flink.table.gateway.rest.util.SqlGatewayRestEndpointUtils;
+import org.apache.flink.table.gateway.service.context.DefaultContext;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.util.function.SupplierWithException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.Iterator;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+
+import static org.apache.flink.table.gateway.rest.handler.session.CloseSessionHandler.CLOSE_MESSAGE;
+
+/** Executor to connect to {@link SqlGateway} and execute statements. */
+public class ExecutorImpl implements Executor {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ExecutorImpl.class);
+ private static final long HEARTBEAT_INTERVAL_MILLISECONDS = 60_000L;
+
+ private final DefaultContext defaultContext;
+ private final InetSocketAddress gatewayAddress;
+ private final long heartbeatInterval;
+ private final ExecutorService service;
+ private final ScheduledExecutorService heartbeatScheduler;
+
+ private RestClient restClient;
+ private SessionHandle sessionHandle;
+
+ @VisibleForTesting
+ public ExecutorImpl(
+ DefaultContext defaultContext,
+ InetSocketAddress gatewayAddress,
+ long heartbeatInterval) {
+ this.defaultContext = defaultContext;
+ this.gatewayAddress = gatewayAddress;
+ this.heartbeatInterval = heartbeatInterval;
+ this.service = Executors.newCachedThreadPool();
+ this.heartbeatScheduler = Executors.newSingleThreadScheduledExecutor();
+ }
+
+ public ExecutorImpl(DefaultContext defaultContext, InetSocketAddress gatewayAddress) {
+ this(defaultContext, gatewayAddress, HEARTBEAT_INTERVAL_MILLISECONDS);
+ }
+
+ public void openSession(@Nullable String sessionId) {
+ try {
+ restClient = new RestClient(defaultContext.getFlinkConfig(), service);
+ } catch (Exception e) {
+ throw new SqlExecutionException("Can not create the Rest Client.", e);
+ }
+
+ LOG.info("Open session to {}.", gatewayAddress);
+ OpenSessionRequestBody request =
+ new OpenSessionRequestBody(sessionId, defaultContext.getFlinkConfig().toMap());
+ try {
+ // open session to address:port
+ OpenSessionResponseBody response =
+ sendRequest(
+ OpenSessionHeaders.getInstance(),
+ EmptyMessageParameters.getInstance(),
+ request)
+ .get();
+ sessionHandle = new SessionHandle(UUID.fromString(response.getSessionHandle()));
+ // register heartbeat service
+ heartbeatScheduler.scheduleAtFixedRate(
+ () ->
+ getResponse(
+ sendRequest(
+ TriggerSessionHeartbeatHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ EmptyRequestBody.getInstance())),
+ heartbeatInterval,
+ heartbeatInterval,
+ TimeUnit.MILLISECONDS);
+ // register dependencies
+ defaultContext
+ .getDependencies()
+ .forEach(jar -> configureSession(String.format("ADD JAR '%s'", jar)));
+ } catch (Exception e) {
+ throw new SqlClientException(
+ String.format("Failed to open session to %s", gatewayAddress), e);
+ }
+ }
+
+ public void closeSession() throws SqlExecutionException {
+ if (sessionHandle == null) {
+ return;
+ }
+ try {
+ CompletableFuture<CloseSessionResponseBody> response =
+ sendRequest(
+ CloseSessionHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ EmptyRequestBody.getInstance());
+
+ if (!response.get().getStatus().equals(CLOSE_MESSAGE)) {
+ LOG.warn("The status of close session response isn't {}.", CLOSE_MESSAGE);
+ }
+ } catch (Exception e) {
+ LOG.warn(
+ String.format(
+ "Unexpected error occurs when closing session %s.", sessionHandle),
+ e);
+ // ignore any throwable to keep the cleanup running
+ } finally {
+ sessionHandle = null;
+ }
+ }
+
+ public void configureSession(String statement) {
+ try {
+ sendRequest(
+ ConfigureSessionHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ new ConfigureSessionRequestBody(statement))
+ .get();
+ } catch (Exception e) {
+ throw new SqlExecutionException(
+ String.format("Failed to open session to %s", gatewayAddress), e);
+ }
+ }
+
+ public ReadableConfig getSessionConfig() {
+ try {
+ GetSessionConfigResponseBody response =
+ getResponse(
+ sendRequest(
+ GetSessionConfigHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ EmptyRequestBody.getInstance()));
+ return Configuration.fromMap(response.getProperties());
+ } catch (Exception e) {
+ throw new SqlExecutionException("Failed to get the get session config.", e);
+ }
+ }
+
+ public ClientResult executeStatement(String statement) {
+ ExecuteStatementRequestBody request = new ExecuteStatementRequestBody(statement);
+ CompletableFuture<ExecuteStatementResponseBody> executeStatementResponse =
+ sendRequest(
+ ExecuteStatementHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ request);
+ // It's possible that the execution is canceled during the submission.
+ // Close the Operation in background to make sure the execution can continue.
+ getResponse(
+ executeStatementResponse,
+ e -> {
+ service.submit(
+ () -> {
+ try {
+ ExecuteStatementResponseBody executeStatementResponseBody =
+ executeStatementResponse.get();
+ OperationHandle operationHandle =
+ new OperationHandle(
+ UUID.fromString(
+ executeStatementResponseBody
+ .getOperationHandle()));
+ // close operation in background to make sure users can not
+ // interrupt the execution.
+ sendRequest(
+ CloseOperationHeaders.getInstance(),
+ new OperationMessageParameters(
+ sessionHandle, operationHandle),
+ EmptyRequestBody.getInstance());
+ } catch (Exception newException) {
+ // ignore
+ }
+ });
+ throw new SqlExecutionException("Interrupted to get response.", e);
+ });
+
+ OperationHandle operationHandle =
+ new OperationHandle(
+ UUID.fromString(
+ getResponse(executeStatementResponse).getOperationHandle()));
+ FetchResultsResponseBody fetchResultsResponse = fetchUtilResultsReady(operationHandle);
+ ResultInfo firstResult = fetchResultsResponse.getResults();
+
+ return new ClientResult(
+ fetchResultsResponse.getResults().getResultSchema(),
+ new RowDataInfoIterator(
+ operationHandle,
+ firstResult,
+ SqlGatewayRestEndpointUtils.parseToken(
+ fetchResultsResponse.getNextResultUri())),
+ fetchResultsResponse.isQueryResult(),
+ fetchResultsResponse.getResultKind(),
+ fetchResultsResponse.getJobID());
+ }
+
+ public List<String> completeStatement(String statement, int position) {
+ return getResponse(
+ sendRequest(
+ CompleteStatementHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ new CompleteStatementRequestBody(statement, position)))
+ .getCandidates();
+ }
+
+ @Override
+ public void close() {
+ closeSession();
+ service.shutdownNow();
+ heartbeatScheduler.shutdownNow();
+ }
+
+ @VisibleForTesting
+ public SessionHandle getSessionHandle() {
+ return sessionHandle;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ private class RowDataInfoIterator implements CloseableIterator<RowData> {
+
+ private final OperationHandle operationHandle;
+ private Iterator<RowData> current;
+ @Nullable private Long nextToken;
+
+ public RowDataInfoIterator(
+ OperationHandle operationHandle, ResultInfo resultInfo, @Nullable Long nextToken) {
+ this.operationHandle = operationHandle;
+ this.current = resultInfo.getData().iterator();
+ this.nextToken = nextToken;
+ }
+
+ @Override
+ public void close() throws Exception {
+ getResponse(
+ sendRequest(
+ CloseOperationHeaders.getInstance(),
+ new OperationMessageParameters(sessionHandle, operationHandle),
+ EmptyRequestBody.getInstance()));
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (!current.hasNext()) {
+ while (nextToken != null && !current.hasNext()) {
+ FetchResultsResponseBody fetchResultsResponseBody =
+ fetchResults(operationHandle, nextToken);
+ nextToken =
+ SqlGatewayRestEndpointUtils.parseToken(
+ fetchResultsResponseBody.getNextResultUri());
+ current = fetchResultsResponseBody.getResults().getData().iterator();
+ }
+ }
+ return current.hasNext();
+ }
+
+ @Override
+ public RowData next() {
+ return current.next();
+ }
+
+ private FetchResultsResponseBody fetchResults(OperationHandle operationHandle, long token) {
+ return getFetchResultResponse(
+ operationHandle,
+ () -> {
+ try {
+ return sendRequest(
+ FetchResultsHeaders.getDefaultInstance(),
+ new FetchResultsMessageParameters(
+ sessionHandle,
+ operationHandle,
+ token,
+ RowFormat.PLAIN_TEXT),
+ EmptyRequestBody.getInstance())
+ .get();
+ } catch (InterruptedException e) {
+ // cancel operation in background
+ sendRequest(
+ CancelOperationHeaders.getInstance(),
+ new OperationMessageParameters(sessionHandle, operationHandle),
+ EmptyRequestBody.getInstance());
+ throw new SqlExecutionException("Interrupted to fetch results.", e);
+ }
+ });
+ }
+ }
+
+ private <
+ M extends MessageHeaders<R, P, U>,
+ U extends MessageParameters,
+ R extends RequestBody,
+ P extends ResponseBody>
+ CompletableFuture<P> sendRequest(M messageHeaders, U messageParameters, R request) {
+ try {
+ return restClient.sendRequest(
+ gatewayAddress.getHostName(),
+ gatewayAddress.getPort(),
+ messageHeaders,
+ messageParameters,
+ request);
+ } catch (IOException ioException) {
+ throw new SqlExecutionException("Failed to connect to the SQL Gateway.", ioException);
+ }
+ }
+
+ private FetchResultsResponseBody fetchUtilResultsReady(OperationHandle operationHandle) {
+ FetchResultsResponseBody response;
+ do {
+ response =
+ getFetchResultResponse(
+ operationHandle,
+ () -> {
+ try {
+ Thread.sleep(100);
+ return sendRequest(
+ FetchResultsHeaders.getDefaultInstance(),
+ new FetchResultsMessageParameters(
Review Comment:
Can we reuse `parameters` here?
##########
flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/ExecutorImpl.java:
##########
@@ -0,0 +1,447 @@
+/*
+ * 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.flink.table.client.gateway;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.runtime.rest.RestClient;
+import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.runtime.rest.messages.MessageParameters;
+import org.apache.flink.runtime.rest.messages.RequestBody;
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+import org.apache.flink.runtime.rest.util.RestClientException;
+import org.apache.flink.table.api.SqlParserEOFException;
+import org.apache.flink.table.client.SqlClientException;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.gateway.SqlGateway;
+import org.apache.flink.table.gateway.api.operation.OperationHandle;
+import org.apache.flink.table.gateway.api.results.ResultSet;
+import org.apache.flink.table.gateway.api.session.SessionHandle;
+import org.apache.flink.table.gateway.rest.header.operation.CancelOperationHeaders;
+import org.apache.flink.table.gateway.rest.header.operation.CloseOperationHeaders;
+import org.apache.flink.table.gateway.rest.header.session.CloseSessionHeaders;
+import org.apache.flink.table.gateway.rest.header.session.ConfigureSessionHeaders;
+import org.apache.flink.table.gateway.rest.header.session.GetSessionConfigHeaders;
+import org.apache.flink.table.gateway.rest.header.session.OpenSessionHeaders;
+import org.apache.flink.table.gateway.rest.header.session.TriggerSessionHeartbeatHeaders;
+import org.apache.flink.table.gateway.rest.header.statement.CompleteStatementHeaders;
+import org.apache.flink.table.gateway.rest.header.statement.ExecuteStatementHeaders;
+import org.apache.flink.table.gateway.rest.header.statement.FetchResultsHeaders;
+import org.apache.flink.table.gateway.rest.message.operation.OperationMessageParameters;
+import org.apache.flink.table.gateway.rest.message.session.CloseSessionResponseBody;
+import org.apache.flink.table.gateway.rest.message.session.ConfigureSessionRequestBody;
+import org.apache.flink.table.gateway.rest.message.session.GetSessionConfigResponseBody;
+import org.apache.flink.table.gateway.rest.message.session.OpenSessionRequestBody;
+import org.apache.flink.table.gateway.rest.message.session.OpenSessionResponseBody;
+import org.apache.flink.table.gateway.rest.message.session.SessionMessageParameters;
+import org.apache.flink.table.gateway.rest.message.statement.CompleteStatementRequestBody;
+import org.apache.flink.table.gateway.rest.message.statement.ExecuteStatementRequestBody;
+import org.apache.flink.table.gateway.rest.message.statement.ExecuteStatementResponseBody;
+import org.apache.flink.table.gateway.rest.message.statement.FetchResultsMessageParameters;
+import org.apache.flink.table.gateway.rest.message.statement.FetchResultsResponseBody;
+import org.apache.flink.table.gateway.rest.serde.ResultInfo;
+import org.apache.flink.table.gateway.rest.util.RowFormat;
+import org.apache.flink.table.gateway.rest.util.SqlGatewayRestEndpointUtils;
+import org.apache.flink.table.gateway.service.context.DefaultContext;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.util.function.SupplierWithException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.Iterator;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+
+import static org.apache.flink.table.gateway.rest.handler.session.CloseSessionHandler.CLOSE_MESSAGE;
+
+/** Executor to connect to {@link SqlGateway} and execute statements. */
+public class ExecutorImpl implements Executor {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ExecutorImpl.class);
+ private static final long HEARTBEAT_INTERVAL_MILLISECONDS = 60_000L;
+
+ private final DefaultContext defaultContext;
+ private final InetSocketAddress gatewayAddress;
+ private final long heartbeatInterval;
+ private final ExecutorService service;
+ private final ScheduledExecutorService heartbeatScheduler;
+
+ private RestClient restClient;
+ private SessionHandle sessionHandle;
+
+ @VisibleForTesting
+ public ExecutorImpl(
+ DefaultContext defaultContext,
+ InetSocketAddress gatewayAddress,
+ long heartbeatInterval) {
+ this.defaultContext = defaultContext;
+ this.gatewayAddress = gatewayAddress;
+ this.heartbeatInterval = heartbeatInterval;
+ this.service = Executors.newCachedThreadPool();
+ this.heartbeatScheduler = Executors.newSingleThreadScheduledExecutor();
+ }
+
+ public ExecutorImpl(DefaultContext defaultContext, InetSocketAddress gatewayAddress) {
+ this(defaultContext, gatewayAddress, HEARTBEAT_INTERVAL_MILLISECONDS);
+ }
+
+ public void openSession(@Nullable String sessionId) {
+ try {
+ restClient = new RestClient(defaultContext.getFlinkConfig(), service);
+ } catch (Exception e) {
+ throw new SqlExecutionException("Can not create the Rest Client.", e);
+ }
+
+ LOG.info("Open session to {}.", gatewayAddress);
+ OpenSessionRequestBody request =
+ new OpenSessionRequestBody(sessionId, defaultContext.getFlinkConfig().toMap());
+ try {
+ // open session to address:port
+ OpenSessionResponseBody response =
+ sendRequest(
+ OpenSessionHeaders.getInstance(),
+ EmptyMessageParameters.getInstance(),
+ request)
+ .get();
+ sessionHandle = new SessionHandle(UUID.fromString(response.getSessionHandle()));
+ // register heartbeat service
+ heartbeatScheduler.scheduleAtFixedRate(
+ () ->
+ getResponse(
+ sendRequest(
+ TriggerSessionHeartbeatHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ EmptyRequestBody.getInstance())),
+ heartbeatInterval,
+ heartbeatInterval,
+ TimeUnit.MILLISECONDS);
+ // register dependencies
+ defaultContext
+ .getDependencies()
+ .forEach(jar -> configureSession(String.format("ADD JAR '%s'", jar)));
+ } catch (Exception e) {
+ throw new SqlClientException(
+ String.format("Failed to open session to %s", gatewayAddress), e);
+ }
+ }
+
+ public void closeSession() throws SqlExecutionException {
+ if (sessionHandle == null) {
+ return;
+ }
+ try {
+ CompletableFuture<CloseSessionResponseBody> response =
+ sendRequest(
+ CloseSessionHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ EmptyRequestBody.getInstance());
+
+ if (!response.get().getStatus().equals(CLOSE_MESSAGE)) {
+ LOG.warn("The status of close session response isn't {}.", CLOSE_MESSAGE);
+ }
+ } catch (Exception e) {
+ LOG.warn(
+ String.format(
+ "Unexpected error occurs when closing session %s.", sessionHandle),
+ e);
+ // ignore any throwable to keep the cleanup running
+ } finally {
+ sessionHandle = null;
+ }
+ }
+
+ public void configureSession(String statement) {
+ try {
+ sendRequest(
+ ConfigureSessionHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ new ConfigureSessionRequestBody(statement))
+ .get();
+ } catch (Exception e) {
+ throw new SqlExecutionException(
+ String.format("Failed to open session to %s", gatewayAddress), e);
+ }
+ }
+
+ public ReadableConfig getSessionConfig() {
+ try {
+ GetSessionConfigResponseBody response =
+ getResponse(
+ sendRequest(
+ GetSessionConfigHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ EmptyRequestBody.getInstance()));
+ return Configuration.fromMap(response.getProperties());
+ } catch (Exception e) {
+ throw new SqlExecutionException("Failed to get the get session config.", e);
+ }
+ }
+
+ public ClientResult executeStatement(String statement) {
+ ExecuteStatementRequestBody request = new ExecuteStatementRequestBody(statement);
+ CompletableFuture<ExecuteStatementResponseBody> executeStatementResponse =
+ sendRequest(
+ ExecuteStatementHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ request);
+ // It's possible that the execution is canceled during the submission.
+ // Close the Operation in background to make sure the execution can continue.
+ getResponse(
+ executeStatementResponse,
+ e -> {
+ service.submit(
+ () -> {
+ try {
+ ExecuteStatementResponseBody executeStatementResponseBody =
+ executeStatementResponse.get();
+ OperationHandle operationHandle =
+ new OperationHandle(
+ UUID.fromString(
+ executeStatementResponseBody
+ .getOperationHandle()));
+ // close operation in background to make sure users can not
+ // interrupt the execution.
+ sendRequest(
+ CloseOperationHeaders.getInstance(),
+ new OperationMessageParameters(
+ sessionHandle, operationHandle),
+ EmptyRequestBody.getInstance());
+ } catch (Exception newException) {
+ // ignore
+ }
+ });
+ throw new SqlExecutionException("Interrupted to get response.", e);
+ });
+
+ OperationHandle operationHandle =
+ new OperationHandle(
+ UUID.fromString(
+ getResponse(executeStatementResponse).getOperationHandle()));
+ FetchResultsResponseBody fetchResultsResponse = fetchUtilResultsReady(operationHandle);
+ ResultInfo firstResult = fetchResultsResponse.getResults();
+
+ return new ClientResult(
+ fetchResultsResponse.getResults().getResultSchema(),
+ new RowDataInfoIterator(
+ operationHandle,
+ firstResult,
+ SqlGatewayRestEndpointUtils.parseToken(
+ fetchResultsResponse.getNextResultUri())),
+ fetchResultsResponse.isQueryResult(),
+ fetchResultsResponse.getResultKind(),
+ fetchResultsResponse.getJobID());
+ }
+
+ public List<String> completeStatement(String statement, int position) {
+ return getResponse(
+ sendRequest(
+ CompleteStatementHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ new CompleteStatementRequestBody(statement, position)))
+ .getCandidates();
+ }
+
+ @Override
+ public void close() {
+ closeSession();
+ service.shutdownNow();
+ heartbeatScheduler.shutdownNow();
+ }
+
+ @VisibleForTesting
+ public SessionHandle getSessionHandle() {
+ return sessionHandle;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ private class RowDataInfoIterator implements CloseableIterator<RowData> {
+
+ private final OperationHandle operationHandle;
+ private Iterator<RowData> current;
+ @Nullable private Long nextToken;
+
+ public RowDataInfoIterator(
+ OperationHandle operationHandle, ResultInfo resultInfo, @Nullable Long nextToken) {
+ this.operationHandle = operationHandle;
+ this.current = resultInfo.getData().iterator();
+ this.nextToken = nextToken;
+ }
+
+ @Override
+ public void close() throws Exception {
+ getResponse(
+ sendRequest(
+ CloseOperationHeaders.getInstance(),
+ new OperationMessageParameters(sessionHandle, operationHandle),
+ EmptyRequestBody.getInstance()));
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (!current.hasNext()) {
+ while (nextToken != null && !current.hasNext()) {
+ FetchResultsResponseBody fetchResultsResponseBody =
+ fetchResults(operationHandle, nextToken);
+ nextToken =
+ SqlGatewayRestEndpointUtils.parseToken(
+ fetchResultsResponseBody.getNextResultUri());
+ current = fetchResultsResponseBody.getResults().getData().iterator();
+ }
+ }
+ return current.hasNext();
+ }
+
+ @Override
+ public RowData next() {
+ return current.next();
+ }
+
+ private FetchResultsResponseBody fetchResults(OperationHandle operationHandle, long token) {
+ return getFetchResultResponse(
+ operationHandle,
+ () -> {
+ try {
+ return sendRequest(
+ FetchResultsHeaders.getDefaultInstance(),
+ new FetchResultsMessageParameters(
+ sessionHandle,
+ operationHandle,
+ token,
+ RowFormat.PLAIN_TEXT),
+ EmptyRequestBody.getInstance())
+ .get();
+ } catch (InterruptedException e) {
+ // cancel operation in background
+ sendRequest(
+ CancelOperationHeaders.getInstance(),
+ new OperationMessageParameters(sessionHandle, operationHandle),
+ EmptyRequestBody.getInstance());
+ throw new SqlExecutionException("Interrupted to fetch results.", e);
+ }
+ });
+ }
+ }
+
+ private <
+ M extends MessageHeaders<R, P, U>,
+ U extends MessageParameters,
+ R extends RequestBody,
+ P extends ResponseBody>
+ CompletableFuture<P> sendRequest(M messageHeaders, U messageParameters, R request) {
+ try {
+ return restClient.sendRequest(
+ gatewayAddress.getHostName(),
+ gatewayAddress.getPort(),
+ messageHeaders,
+ messageParameters,
+ request);
+ } catch (IOException ioException) {
+ throw new SqlExecutionException("Failed to connect to the SQL Gateway.", ioException);
+ }
+ }
+
+ private FetchResultsResponseBody fetchUtilResultsReady(OperationHandle operationHandle) {
+ FetchResultsResponseBody response;
+ do {
+ response =
+ getFetchResultResponse(
+ operationHandle,
+ () -> {
+ try {
+ Thread.sleep(100);
+ return sendRequest(
+ FetchResultsHeaders.getDefaultInstance(),
+ new FetchResultsMessageParameters(
+ sessionHandle,
+ operationHandle,
+ 0L,
+ RowFormat.PLAIN_TEXT),
+ EmptyRequestBody.getInstance())
+ .get();
+ } catch (InterruptedException e) {
+ // CliClient will not close the results. Try best to close it.
+ sendRequest(
+ CloseOperationHeaders.getInstance(),
+ new OperationMessageParameters(
Review Comment:
ditto?
##########
flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/ExecutorImpl.java:
##########
@@ -0,0 +1,447 @@
+/*
+ * 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.flink.table.client.gateway;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.runtime.rest.RestClient;
+import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.runtime.rest.messages.MessageParameters;
+import org.apache.flink.runtime.rest.messages.RequestBody;
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+import org.apache.flink.runtime.rest.util.RestClientException;
+import org.apache.flink.table.api.SqlParserEOFException;
+import org.apache.flink.table.client.SqlClientException;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.gateway.SqlGateway;
+import org.apache.flink.table.gateway.api.operation.OperationHandle;
+import org.apache.flink.table.gateway.api.results.ResultSet;
+import org.apache.flink.table.gateway.api.session.SessionHandle;
+import org.apache.flink.table.gateway.rest.header.operation.CancelOperationHeaders;
+import org.apache.flink.table.gateway.rest.header.operation.CloseOperationHeaders;
+import org.apache.flink.table.gateway.rest.header.session.CloseSessionHeaders;
+import org.apache.flink.table.gateway.rest.header.session.ConfigureSessionHeaders;
+import org.apache.flink.table.gateway.rest.header.session.GetSessionConfigHeaders;
+import org.apache.flink.table.gateway.rest.header.session.OpenSessionHeaders;
+import org.apache.flink.table.gateway.rest.header.session.TriggerSessionHeartbeatHeaders;
+import org.apache.flink.table.gateway.rest.header.statement.CompleteStatementHeaders;
+import org.apache.flink.table.gateway.rest.header.statement.ExecuteStatementHeaders;
+import org.apache.flink.table.gateway.rest.header.statement.FetchResultsHeaders;
+import org.apache.flink.table.gateway.rest.message.operation.OperationMessageParameters;
+import org.apache.flink.table.gateway.rest.message.session.CloseSessionResponseBody;
+import org.apache.flink.table.gateway.rest.message.session.ConfigureSessionRequestBody;
+import org.apache.flink.table.gateway.rest.message.session.GetSessionConfigResponseBody;
+import org.apache.flink.table.gateway.rest.message.session.OpenSessionRequestBody;
+import org.apache.flink.table.gateway.rest.message.session.OpenSessionResponseBody;
+import org.apache.flink.table.gateway.rest.message.session.SessionMessageParameters;
+import org.apache.flink.table.gateway.rest.message.statement.CompleteStatementRequestBody;
+import org.apache.flink.table.gateway.rest.message.statement.ExecuteStatementRequestBody;
+import org.apache.flink.table.gateway.rest.message.statement.ExecuteStatementResponseBody;
+import org.apache.flink.table.gateway.rest.message.statement.FetchResultsMessageParameters;
+import org.apache.flink.table.gateway.rest.message.statement.FetchResultsResponseBody;
+import org.apache.flink.table.gateway.rest.serde.ResultInfo;
+import org.apache.flink.table.gateway.rest.util.RowFormat;
+import org.apache.flink.table.gateway.rest.util.SqlGatewayRestEndpointUtils;
+import org.apache.flink.table.gateway.service.context.DefaultContext;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.util.function.SupplierWithException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.Iterator;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+
+import static org.apache.flink.table.gateway.rest.handler.session.CloseSessionHandler.CLOSE_MESSAGE;
+
+/** Executor to connect to {@link SqlGateway} and execute statements. */
+public class ExecutorImpl implements Executor {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ExecutorImpl.class);
+ private static final long HEARTBEAT_INTERVAL_MILLISECONDS = 60_000L;
+
+ private final DefaultContext defaultContext;
+ private final InetSocketAddress gatewayAddress;
+ private final long heartbeatInterval;
+ private final ExecutorService service;
+ private final ScheduledExecutorService heartbeatScheduler;
+
+ private RestClient restClient;
+ private SessionHandle sessionHandle;
+
+ @VisibleForTesting
+ public ExecutorImpl(
+ DefaultContext defaultContext,
+ InetSocketAddress gatewayAddress,
+ long heartbeatInterval) {
+ this.defaultContext = defaultContext;
+ this.gatewayAddress = gatewayAddress;
+ this.heartbeatInterval = heartbeatInterval;
+ this.service = Executors.newCachedThreadPool();
+ this.heartbeatScheduler = Executors.newSingleThreadScheduledExecutor();
+ }
+
+ public ExecutorImpl(DefaultContext defaultContext, InetSocketAddress gatewayAddress) {
+ this(defaultContext, gatewayAddress, HEARTBEAT_INTERVAL_MILLISECONDS);
+ }
+
+ public void openSession(@Nullable String sessionId) {
+ try {
+ restClient = new RestClient(defaultContext.getFlinkConfig(), service);
+ } catch (Exception e) {
+ throw new SqlExecutionException("Can not create the Rest Client.", e);
+ }
+
+ LOG.info("Open session to {}.", gatewayAddress);
+ OpenSessionRequestBody request =
+ new OpenSessionRequestBody(sessionId, defaultContext.getFlinkConfig().toMap());
+ try {
+ // open session to address:port
+ OpenSessionResponseBody response =
+ sendRequest(
+ OpenSessionHeaders.getInstance(),
+ EmptyMessageParameters.getInstance(),
+ request)
+ .get();
+ sessionHandle = new SessionHandle(UUID.fromString(response.getSessionHandle()));
+ // register heartbeat service
+ heartbeatScheduler.scheduleAtFixedRate(
+ () ->
+ getResponse(
+ sendRequest(
+ TriggerSessionHeartbeatHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ EmptyRequestBody.getInstance())),
+ heartbeatInterval,
+ heartbeatInterval,
+ TimeUnit.MILLISECONDS);
+ // register dependencies
+ defaultContext
+ .getDependencies()
+ .forEach(jar -> configureSession(String.format("ADD JAR '%s'", jar)));
+ } catch (Exception e) {
+ throw new SqlClientException(
+ String.format("Failed to open session to %s", gatewayAddress), e);
+ }
+ }
+
+ public void closeSession() throws SqlExecutionException {
+ if (sessionHandle == null) {
+ return;
+ }
+ try {
+ CompletableFuture<CloseSessionResponseBody> response =
+ sendRequest(
+ CloseSessionHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ EmptyRequestBody.getInstance());
+
+ if (!response.get().getStatus().equals(CLOSE_MESSAGE)) {
+ LOG.warn("The status of close session response isn't {}.", CLOSE_MESSAGE);
+ }
+ } catch (Exception e) {
+ LOG.warn(
+ String.format(
+ "Unexpected error occurs when closing session %s.", sessionHandle),
+ e);
+ // ignore any throwable to keep the cleanup running
+ } finally {
+ sessionHandle = null;
+ }
+ }
+
+ public void configureSession(String statement) {
+ try {
+ sendRequest(
+ ConfigureSessionHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ new ConfigureSessionRequestBody(statement))
+ .get();
+ } catch (Exception e) {
+ throw new SqlExecutionException(
+ String.format("Failed to open session to %s", gatewayAddress), e);
+ }
+ }
+
+ public ReadableConfig getSessionConfig() {
+ try {
+ GetSessionConfigResponseBody response =
+ getResponse(
+ sendRequest(
+ GetSessionConfigHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ EmptyRequestBody.getInstance()));
+ return Configuration.fromMap(response.getProperties());
+ } catch (Exception e) {
+ throw new SqlExecutionException("Failed to get the get session config.", e);
+ }
+ }
+
+ public ClientResult executeStatement(String statement) {
+ ExecuteStatementRequestBody request = new ExecuteStatementRequestBody(statement);
+ CompletableFuture<ExecuteStatementResponseBody> executeStatementResponse =
+ sendRequest(
+ ExecuteStatementHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ request);
+ // It's possible that the execution is canceled during the submission.
+ // Close the Operation in background to make sure the execution can continue.
+ getResponse(
+ executeStatementResponse,
+ e -> {
+ service.submit(
+ () -> {
+ try {
+ ExecuteStatementResponseBody executeStatementResponseBody =
+ executeStatementResponse.get();
+ OperationHandle operationHandle =
+ new OperationHandle(
+ UUID.fromString(
+ executeStatementResponseBody
+ .getOperationHandle()));
+ // close operation in background to make sure users can not
+ // interrupt the execution.
+ sendRequest(
+ CloseOperationHeaders.getInstance(),
+ new OperationMessageParameters(
+ sessionHandle, operationHandle),
+ EmptyRequestBody.getInstance());
+ } catch (Exception newException) {
+ // ignore
+ }
+ });
+ throw new SqlExecutionException("Interrupted to get response.", e);
+ });
+
+ OperationHandle operationHandle =
+ new OperationHandle(
+ UUID.fromString(
+ getResponse(executeStatementResponse).getOperationHandle()));
+ FetchResultsResponseBody fetchResultsResponse = fetchUtilResultsReady(operationHandle);
+ ResultInfo firstResult = fetchResultsResponse.getResults();
+
+ return new ClientResult(
+ fetchResultsResponse.getResults().getResultSchema(),
+ new RowDataInfoIterator(
+ operationHandle,
+ firstResult,
+ SqlGatewayRestEndpointUtils.parseToken(
+ fetchResultsResponse.getNextResultUri())),
+ fetchResultsResponse.isQueryResult(),
+ fetchResultsResponse.getResultKind(),
+ fetchResultsResponse.getJobID());
+ }
+
+ public List<String> completeStatement(String statement, int position) {
+ return getResponse(
+ sendRequest(
+ CompleteStatementHeaders.getInstance(),
+ new SessionMessageParameters(sessionHandle),
+ new CompleteStatementRequestBody(statement, position)))
+ .getCandidates();
+ }
+
+ @Override
+ public void close() {
+ closeSession();
+ service.shutdownNow();
+ heartbeatScheduler.shutdownNow();
+ }
+
+ @VisibleForTesting
+ public SessionHandle getSessionHandle() {
+ return sessionHandle;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ private class RowDataInfoIterator implements CloseableIterator<RowData> {
+
+ private final OperationHandle operationHandle;
+ private Iterator<RowData> current;
+ @Nullable private Long nextToken;
+
+ public RowDataInfoIterator(
+ OperationHandle operationHandle, ResultInfo resultInfo, @Nullable Long nextToken) {
+ this.operationHandle = operationHandle;
+ this.current = resultInfo.getData().iterator();
+ this.nextToken = nextToken;
+ }
+
+ @Override
+ public void close() throws Exception {
+ getResponse(
+ sendRequest(
+ CloseOperationHeaders.getInstance(),
+ new OperationMessageParameters(sessionHandle, operationHandle),
+ EmptyRequestBody.getInstance()));
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (!current.hasNext()) {
+ while (nextToken != null && !current.hasNext()) {
+ FetchResultsResponseBody fetchResultsResponseBody =
+ fetchResults(operationHandle, nextToken);
+ nextToken =
+ SqlGatewayRestEndpointUtils.parseToken(
+ fetchResultsResponseBody.getNextResultUri());
+ current = fetchResultsResponseBody.getResults().getData().iterator();
+ }
+ }
+ return current.hasNext();
+ }
+
+ @Override
+ public RowData next() {
+ return current.next();
+ }
+
+ private FetchResultsResponseBody fetchResults(OperationHandle operationHandle, long token) {
+ return getFetchResultResponse(
+ operationHandle,
+ () -> {
+ try {
+ return sendRequest(
+ FetchResultsHeaders.getDefaultInstance(),
+ new FetchResultsMessageParameters(
+ sessionHandle,
+ operationHandle,
+ token,
+ RowFormat.PLAIN_TEXT),
+ EmptyRequestBody.getInstance())
+ .get();
+ } catch (InterruptedException e) {
+ // cancel operation in background
+ sendRequest(
+ CancelOperationHeaders.getInstance(),
+ new OperationMessageParameters(sessionHandle, operationHandle),
+ EmptyRequestBody.getInstance());
+ throw new SqlExecutionException("Interrupted to fetch results.", e);
+ }
+ });
+ }
+ }
+
+ private <
+ M extends MessageHeaders<R, P, U>,
+ U extends MessageParameters,
+ R extends RequestBody,
+ P extends ResponseBody>
+ CompletableFuture<P> sendRequest(M messageHeaders, U messageParameters, R request) {
+ try {
+ return restClient.sendRequest(
+ gatewayAddress.getHostName(),
+ gatewayAddress.getPort(),
+ messageHeaders,
+ messageParameters,
+ request);
+ } catch (IOException ioException) {
+ throw new SqlExecutionException("Failed to connect to the SQL Gateway.", ioException);
+ }
+ }
+
+ private FetchResultsResponseBody fetchUtilResultsReady(OperationHandle operationHandle) {
+ FetchResultsResponseBody response;
+ do {
+ response =
+ getFetchResultResponse(
+ operationHandle,
+ () -> {
+ try {
+ Thread.sleep(100);
+ return sendRequest(
+ FetchResultsHeaders.getDefaultInstance(),
+ new FetchResultsMessageParameters(
+ sessionHandle,
+ operationHandle,
+ 0L,
+ RowFormat.PLAIN_TEXT),
+ EmptyRequestBody.getInstance())
+ .get();
+ } catch (InterruptedException e) {
+ // CliClient will not close the results. Try best to close it.
+ sendRequest(
+ CloseOperationHeaders.getInstance(),
+ new OperationMessageParameters(
+ sessionHandle, operationHandle),
+ EmptyRequestBody.getInstance());
+ throw new SqlExecutionException(
+ "Interrupted to fetch results.", e);
+ }
+ });
+ } while (response.getResultType().equals(ResultSet.ResultType.NOT_READY));
+ return response;
+ }
+
+ private <T> T getResponse(CompletableFuture<T> future) {
+ return getResponse(
+ future,
+ e -> {
+ throw new SqlExecutionException("Interrupted to get response.", e);
+ });
+ }
+
+ private <T> T getResponse(
+ CompletableFuture<T> future,
+ Function<InterruptedException, T> interruptedExceptionHandler) {
+ try {
+ return future.get();
+ } catch (ExecutionException executionException) {
+ Throwable cause = executionException.getCause();
+ throw new SqlExecutionException("Failed to get response.", cause);
+ } catch (InterruptedException e) {
+ return interruptedExceptionHandler.apply(e);
+ }
+ }
+
+ private FetchResultsResponseBody getFetchResultResponse(
+ OperationHandle operationHandle,
+ SupplierWithException<FetchResultsResponseBody, ExecutionException> responseSupplier) {
Review Comment:
Can we simplify this like
```java
private FetchResultsResponseBody getFetchResponse(
OperationHandle operationHandle,
long token,
boolean init,
FunctionWithException<
InterruptedException, FetchResultsResponseBody, SqlExecutionException>
responseComposer) {
try {
if (!init) {
Thread.sleep(100);
}
return sendRequest(
FetchResultsHeaders.getDefaultInstance(),
new FetchResultsMessageParameters(
sessionHandle, operationHandle, token, RowFormat.PLAIN_TEXT),
EmptyRequestBody.getInstance())
.get();
} catch (InterruptedException e) {
return responseComposer.apply(e);
} catch (ExecutionException e) {
Throwable cause = e.getCause();
if (cause instanceof RestClientException
&& cause.getMessage().contains("Encountered \"<EOF>\"")) {
throw new SqlParserEOFException(cause.getMessage(), cause);
} else {
throw new SqlExecutionException(
String.format(
"Failed to get response for the operation %s.", operationHandle),
cause);
}
}
}
```
And
```java
private FetchResultsResponseBody fetchResults(OperationHandle operationHandle, long token) {
return getFetchResponse(
operationHandle,
token,
true,
(e) -> {
sendRequest(
CancelOperationHeaders.getInstance(),
new OperationMessageParameters(sessionHandle, operationHandle),
EmptyRequestBody.getInstance());
throw new SqlExecutionException("Interrupted to fetch results.", e);
});
}
```
```java
private FetchResultsResponseBody fetchUtilResultsReady(OperationHandle operationHandle) {
FetchResultsResponseBody response;
do {
response =
getFetchResponse(
operationHandle,
0L,
false,
(e) -> {
// CliClient will not close the results. Try best to close it.
closeOperationAsync(operationHandle);
throw new SqlExecutionException("Interrupted to fetch results.", e);
});
} while (response.getResultType().equals(ResultSet.ResultType.NOT_READY));
return response;
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fsk119 commented on a diff in pull request #21717: [FLINK-29945][sql-client] Supports to submit SQL to a embedded SQL Ga…
Posted by GitBox <gi...@apache.org>.
fsk119 commented on code in PR #21717:
URL: https://github.com/apache/flink/pull/21717#discussion_r1082055861
##########
flink-table/flink-sql-client/src/test/resources/sql/table.q:
##########
@@ -79,7 +79,10 @@ show tables;
# test SHOW CREATE TABLE
show create table orders;
-CREATE TABLE `default_catalog`.`default_database`.`orders` (
++-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
Review Comment:
In the FLIP, we do reach an agreement about the presentation. BTW, I think it's the first version of the presentation when the input line is too long. From my experience, we should support printing the results with multiple lines.
In PG/Presto, the client uses '+' to indicate the current line is part of the last line. Considering the current change is so large, I think we can move the improvements to the future.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] LadyForest commented on a diff in pull request #21717: [FLINK-29945][sql-client] Supports to submit SQL to a embedded SQL Ga…
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #21717:
URL: https://github.com/apache/flink/pull/21717#discussion_r1083646639
##########
flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/utils/SqlGatewayServiceExtension.java:
##########
@@ -80,7 +81,8 @@ public void beforeAll(ExtensionContext context) throws Exception {
map.put(ENV_FLINK_CONF_DIR, confFolder.getAbsolutePath());
CommonTestUtils.setEnv(map);
- sessionManager = new SessionManager(DefaultContext.load(new Configuration()));
+ sessionManager =
+ new SessionManager(DefaultContext.load(new Configuration(), new ArrayList<>()));
Review Comment:
Nit: `Collections.emptyList()`
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/session/ConfigureSessionRequestBody.java:
##########
@@ -39,6 +39,10 @@ public class ConfigureSessionRequestBody implements RequestBody {
@Nullable
private final Long timeout;
+ public ConfigureSessionRequestBody(String statement) {
+ this(statement, null);
+ }
+
public ConfigureSessionRequestBody(
Review Comment:
Nit: add `@JsonCreator` annotation
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/statement/ExecuteStatementRequestBody.java:
##########
@@ -46,6 +46,10 @@ public class ExecuteStatementRequestBody implements RequestBody {
@Nullable
private final Map<String, String> executionConfig;
+ public ExecuteStatementRequestBody(String statement) {
+ this(statement, 0L, null);
+ }
+
public ExecuteStatementRequestBody(
Review Comment:
ditto
##########
flink-table/flink-sql-client/src/test/resources/sql/set.q:
##########
@@ -118,47 +118,61 @@ SELECT hive_add_one(1);
Received a total of 1 row
!ok
-REMOVE JAR '$VAR_UDF_JAR_PATH';
-[INFO] The specified jar is removed from session classloader.
-!info
+# TODO: support this in the FLINK-30692
+#REMOVE JAR '$VAR_UDF_JAR_PATH';
+#[INFO] The specified jar is removed from session classloader.
+#!info
-SHOW JARS;
-Empty set
-!ok
+#SHOW JARS;
+#Empty set
+#!ok
+
+reset table.resources.download-dir;
+[INFO] Execute statement succeed.
+!info
# list the configured configuration
set;
-'execution.attached' = 'true'
-'execution.savepoint-restore-mode' = 'NO_CLAIM'
-'execution.savepoint.ignore-unclaimed-state' = 'false'
-'execution.shutdown-on-attached-exit' = 'false'
-'execution.target' = 'remote'
-'jobmanager.rpc.address' = '$VAR_JOBMANAGER_RPC_ADDRESS'
-'k1' = 'v1'
-'pipeline.classpaths' = ''
-'pipeline.jars' = ''
-'rest.port' = '$VAR_REST_PORT'
-'sql-client.execution.result-mode' = 'tableau'
-'table.exec.legacy-cast-behaviour' = 'DISABLED'
-'table.sql-dialect' = 'hive'
++--------------------------------------------+-----------+
+| key | value |
++--------------------------------------------+-----------+
+| execution.attached | true |
+| execution.savepoint-restore-mode | NO_CLAIM |
+| execution.savepoint.ignore-unclaimed-state | false |
+| execution.shutdown-on-attached-exit | false |
+| execution.target | remote |
+| jobmanager.rpc.address | $VAR_JOBMANAGER_RPC_ADDRESS |
+| k1 | v1 |
+| pipeline.classpaths | |
+| pipeline.jars | |
+| rest.port | $VAR_REST_PORT |
+| sql-client.execution.result-mode | tableau |
+| table.exec.legacy-cast-behaviour | DISABLED |
+| table.sql-dialect | hive |
++--------------------------------------------+-----------+
+13 rows in set
Review Comment:
What I've tested.
```text
Flink SQL> set;
Exception in thread "main" org.apache.flink.table.client.SqlClientException: Could not read from command line.
at org.apache.flink.table.client.cli.CliClient.getAndExecuteStatements(CliClient.java:221)
at org.apache.flink.table.client.cli.CliClient.executeInteractive(CliClient.java:186)
at org.apache.flink.table.client.cli.CliClient.executeInInteractiveMode(CliClient.java:132)
at org.apache.flink.table.client.SqlClient.openCli(SqlClient.java:153)
at org.apache.flink.table.client.SqlClient.start(SqlClient.java:100)
at org.apache.flink.table.client.SqlClient.startClient(SqlClient.java:189)
at org.apache.flink.table.client.SqlClient.main(SqlClient.java:163)
Caused by: org.apache.flink.table.client.gateway.SqlExecutionException: Failed to get the get session config.
at org.apache.flink.table.client.gateway.ExecutorImpl.getSessionConfig(ExecutorImpl.java:190)
at org.apache.flink.table.client.cli.CliClient.printExecutionException(CliClient.java:408)
at org.apache.flink.table.client.cli.CliClient.executeStatement(CliClient.java:272)
at org.apache.flink.table.client.cli.CliClient.getAndExecuteStatements(CliClient.java:206)
... 6 more
Caused by: org.apache.flink.table.client.gateway.SqlExecutionException: Failed to get response.
at org.apache.flink.table.client.gateway.ExecutorImpl.getResponse(ExecutorImpl.java:399)
at org.apache.flink.table.client.gateway.ExecutorImpl.getResponse(ExecutorImpl.java:385)
at org.apache.flink.table.client.gateway.ExecutorImpl.getSessionConfig(ExecutorImpl.java:183)
... 9 more
Caused by: org.apache.flink.runtime.rest.util.RestClientException: [org.apache.flink.runtime.rest.handler.RestHandlerException: Failed to getSessionConfig.
at org.apache.flink.table.gateway.rest.handler.session.GetSessionConfigHandler.handleRequest(GetSessionConfigHandler.java:66)
at org.apache.flink.table.gateway.rest.handler.AbstractSqlGatewayRestHandler.respondToRequest(AbstractSqlGatewayRestHandler.java:84)
at org.apache.flink.table.gateway.rest.handler.AbstractSqlGatewayRestHandler.respondToRequest(AbstractSqlGatewayRestHandler.java:52)
at org.apache.flink.runtime.rest.handler.AbstractHandler.respondAsLeader(AbstractHandler.java:196)
at org.apache.flink.runtime.rest.handler.LeaderRetrievalHandler.lambda$channelRead0$0(LeaderRetrievalHandler.java:83)
at java.util.Optional.ifPresent(Optional.java:159)
at org.apache.flink.util.OptionalConsumer.ifPresent(OptionalConsumer.java:45)
at org.apache.flink.runtime.rest.handler.LeaderRetrievalHandler.channelRead0(LeaderRetrievalHandler.java:80)
at org.apache.flink.runtime.rest.handler.LeaderRetrievalHandler.channelRead0(LeaderRetrievalHandler.java:49)
at org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.runtime.rest.handler.router.RouterHandler.routed(RouterHandler.java:115)
at org.apache.flink.runtime.rest.handler.router.RouterHandler.channelRead0(RouterHandler.java:94)
at org.apache.flink.runtime.rest.handler.router.RouterHandler.channelRead0(RouterHandler.java:55)
at org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.shaded.netty4.io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.runtime.rest.FileUploadHandler.channelRead0(FileUploadHandler.java:208)
at org.apache.flink.runtime.rest.FileUploadHandler.channelRead0(FileUploadHandler.java:69)
at org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.shaded.netty4.io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:436)
at org.apache.flink.shaded.netty4.io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:336)
at org.apache.flink.shaded.netty4.io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:308)
at org.apache.flink.shaded.netty4.io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:251)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.shaded.netty4.io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:788)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:724)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:650)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562)
at org.apache.flink.shaded.netty4.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
at org.apache.flink.shaded.netty4.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.flink.table.gateway.api.utils.SqlGatewayException: Failed to getSessionConfig.
at org.apache.flink.table.gateway.service.SqlGatewayServiceImpl.getSessionConfig(SqlGatewayServiceImpl.java:119)
at org.apache.flink.table.gateway.rest.handler.session.GetSessionConfigHandler.handleRequest(GetSessionConfigHandler.java:61)
... 48 more
Caused by: org.apache.flink.table.gateway.api.utils.SqlGatewayException: Session 'd436a4a9-904d-4455-90c5-ad753eda1798' does not exist.
at org.apache.flink.table.gateway.service.session.SessionManager.getSession(SessionManager.java:127)
at org.apache.flink.table.gateway.service.SqlGatewayServiceImpl.getSession(SqlGatewayServiceImpl.java:374)
at org.apache.flink.table.gateway.service.SqlGatewayServiceImpl.getSessionConfig(SqlGatewayServiceImpl.java:116)
... 49 more
]
at org.apache.flink.runtime.rest.RestClient.parseResponse(RestClient.java:536)
at org.apache.flink.runtime.rest.RestClient.lambda$submitRequest$3(RestClient.java:516)
at java.util.concurrent.CompletableFuture.uniCompose(CompletableFuture.java:952)
at java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:926)
at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Shutting down the session...
done.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] LadyForest commented on a diff in pull request #21717: [FLINK-29945][sql-client] Supports to submit SQL to a embedded SQL Ga…
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #21717:
URL: https://github.com/apache/flink/pull/21717#discussion_r1083641565
##########
flink-table/flink-sql-client/src/test/resources/sql/set.q:
##########
@@ -118,47 +118,61 @@ SELECT hive_add_one(1);
Received a total of 1 row
!ok
-REMOVE JAR '$VAR_UDF_JAR_PATH';
-[INFO] The specified jar is removed from session classloader.
-!info
+# TODO: support this in the FLINK-30692
+#REMOVE JAR '$VAR_UDF_JAR_PATH';
+#[INFO] The specified jar is removed from session classloader.
+#!info
-SHOW JARS;
-Empty set
-!ok
+#SHOW JARS;
+#Empty set
+#!ok
+
+reset table.resources.download-dir;
+[INFO] Execute statement succeed.
+!info
# list the configured configuration
set;
-'execution.attached' = 'true'
-'execution.savepoint-restore-mode' = 'NO_CLAIM'
-'execution.savepoint.ignore-unclaimed-state' = 'false'
-'execution.shutdown-on-attached-exit' = 'false'
-'execution.target' = 'remote'
-'jobmanager.rpc.address' = '$VAR_JOBMANAGER_RPC_ADDRESS'
-'k1' = 'v1'
-'pipeline.classpaths' = ''
-'pipeline.jars' = ''
-'rest.port' = '$VAR_REST_PORT'
-'sql-client.execution.result-mode' = 'tableau'
-'table.exec.legacy-cast-behaviour' = 'DISABLED'
-'table.sql-dialect' = 'hive'
++--------------------------------------------+-----------+
+| key | value |
++--------------------------------------------+-----------+
+| execution.attached | true |
+| execution.savepoint-restore-mode | NO_CLAIM |
+| execution.savepoint.ignore-unclaimed-state | false |
+| execution.shutdown-on-attached-exit | false |
+| execution.target | remote |
+| jobmanager.rpc.address | $VAR_JOBMANAGER_RPC_ADDRESS |
+| k1 | v1 |
+| pipeline.classpaths | |
+| pipeline.jars | |
+| rest.port | $VAR_REST_PORT |
+| sql-client.execution.result-mode | tableau |
+| table.exec.legacy-cast-behaviour | DISABLED |
+| table.sql-dialect | hive |
++--------------------------------------------+-----------+
+13 rows in set
Review Comment:
What I've tested. I'm sure that the execution is successive and within 10 min.
```text
Flink SQL> set;
Exception in thread "main" org.apache.flink.table.client.SqlClientException: Could not read from command line.
at org.apache.flink.table.client.cli.CliClient.getAndExecuteStatements(CliClient.java:221)
at org.apache.flink.table.client.cli.CliClient.executeInteractive(CliClient.java:186)
at org.apache.flink.table.client.cli.CliClient.executeInInteractiveMode(CliClient.java:132)
at org.apache.flink.table.client.SqlClient.openCli(SqlClient.java:153)
at org.apache.flink.table.client.SqlClient.start(SqlClient.java:100)
at org.apache.flink.table.client.SqlClient.startClient(SqlClient.java:189)
at org.apache.flink.table.client.SqlClient.main(SqlClient.java:163)
Caused by: org.apache.flink.table.client.gateway.SqlExecutionException: Failed to get the get session config.
at org.apache.flink.table.client.gateway.ExecutorImpl.getSessionConfig(ExecutorImpl.java:190)
at org.apache.flink.table.client.cli.CliClient.printExecutionException(CliClient.java:408)
at org.apache.flink.table.client.cli.CliClient.executeStatement(CliClient.java:272)
at org.apache.flink.table.client.cli.CliClient.getAndExecuteStatements(CliClient.java:206)
... 6 more
Caused by: org.apache.flink.table.client.gateway.SqlExecutionException: Failed to get response.
at org.apache.flink.table.client.gateway.ExecutorImpl.getResponse(ExecutorImpl.java:399)
at org.apache.flink.table.client.gateway.ExecutorImpl.getResponse(ExecutorImpl.java:385)
at org.apache.flink.table.client.gateway.ExecutorImpl.getSessionConfig(ExecutorImpl.java:183)
... 9 more
Caused by: org.apache.flink.runtime.rest.util.RestClientException: [org.apache.flink.runtime.rest.handler.RestHandlerException: Failed to getSessionConfig.
at org.apache.flink.table.gateway.rest.handler.session.GetSessionConfigHandler.handleRequest(GetSessionConfigHandler.java:66)
at org.apache.flink.table.gateway.rest.handler.AbstractSqlGatewayRestHandler.respondToRequest(AbstractSqlGatewayRestHandler.java:84)
at org.apache.flink.table.gateway.rest.handler.AbstractSqlGatewayRestHandler.respondToRequest(AbstractSqlGatewayRestHandler.java:52)
at org.apache.flink.runtime.rest.handler.AbstractHandler.respondAsLeader(AbstractHandler.java:196)
at org.apache.flink.runtime.rest.handler.LeaderRetrievalHandler.lambda$channelRead0$0(LeaderRetrievalHandler.java:83)
at java.util.Optional.ifPresent(Optional.java:159)
at org.apache.flink.util.OptionalConsumer.ifPresent(OptionalConsumer.java:45)
at org.apache.flink.runtime.rest.handler.LeaderRetrievalHandler.channelRead0(LeaderRetrievalHandler.java:80)
at org.apache.flink.runtime.rest.handler.LeaderRetrievalHandler.channelRead0(LeaderRetrievalHandler.java:49)
at org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.runtime.rest.handler.router.RouterHandler.routed(RouterHandler.java:115)
at org.apache.flink.runtime.rest.handler.router.RouterHandler.channelRead0(RouterHandler.java:94)
at org.apache.flink.runtime.rest.handler.router.RouterHandler.channelRead0(RouterHandler.java:55)
at org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.shaded.netty4.io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.runtime.rest.FileUploadHandler.channelRead0(FileUploadHandler.java:208)
at org.apache.flink.runtime.rest.FileUploadHandler.channelRead0(FileUploadHandler.java:69)
at org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.shaded.netty4.io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:436)
at org.apache.flink.shaded.netty4.io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:336)
at org.apache.flink.shaded.netty4.io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:308)
at org.apache.flink.shaded.netty4.io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:251)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.flink.shaded.netty4.io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.flink.shaded.netty4.io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:788)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:724)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:650)
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562)
at org.apache.flink.shaded.netty4.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
at org.apache.flink.shaded.netty4.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.flink.table.gateway.api.utils.SqlGatewayException: Failed to getSessionConfig.
at org.apache.flink.table.gateway.service.SqlGatewayServiceImpl.getSessionConfig(SqlGatewayServiceImpl.java:119)
at org.apache.flink.table.gateway.rest.handler.session.GetSessionConfigHandler.handleRequest(GetSessionConfigHandler.java:61)
... 48 more
Caused by: org.apache.flink.table.gateway.api.utils.SqlGatewayException: Session 'd436a4a9-904d-4455-90c5-ad753eda1798' does not exist.
at org.apache.flink.table.gateway.service.session.SessionManager.getSession(SessionManager.java:127)
at org.apache.flink.table.gateway.service.SqlGatewayServiceImpl.getSession(SqlGatewayServiceImpl.java:374)
at org.apache.flink.table.gateway.service.SqlGatewayServiceImpl.getSessionConfig(SqlGatewayServiceImpl.java:116)
... 49 more
]
at org.apache.flink.runtime.rest.RestClient.parseResponse(RestClient.java:536)
at org.apache.flink.runtime.rest.RestClient.lambda$submitRequest$3(RestClient.java:516)
at java.util.concurrent.CompletableFuture.uniCompose(CompletableFuture.java:952)
at java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:926)
at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Shutting down the session...
done.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org