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 2022/10/24 03:07:51 UTC
[GitHub] [flink] yuzelin opened a new pull request, #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
yuzelin opened a new pull request, #21133:
URL: https://github.com/apache/flink/pull/21133
…h sql statement.
<!--
*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
Implement `SqlGatewayService#configureSession` described in [FLIP-91](https://cwiki.apache.org/confluence/display/FLINK/FLIP-91%3A+Support+SQL+Gateway) .
## Brief change log
- Implement `SqlGatewayService#configureSession`.
## Verifying this change
This change added tests and can be verified as follows:
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): **no**
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: **yes**
- The serializers: **no**
- The runtime per-record code paths (performance sensitive): **no**
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: **no**
- The S3 file system connector: **no**
## Documentation
- Does this pull request introduce a new feature? **yes**
- If yes, how is the feature documented? **JavaDocs**
--
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] yuzelin commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
yuzelin commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1022294537
##########
flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/SqlGatewayServiceITCase.java:
##########
@@ -157,6 +169,80 @@ public void testOpenSessionWithEnvironment() throws Exception {
assertThat(tableEnv.listModules()).contains(moduleName);
}
+ @Test
+ public void testConfigureSessionWithLegalStatement(@TempDir java.nio.file.Path tmpDir)
+ throws Exception {
+ SessionHandle sessionHandle = service.openSession(defaultSessionEnvironment);
+
+ // SET & RESET
+ verifyConfigureSession(sessionHandle, "SET 'key1' = 'value1';");
+ Map<String, String> config = new HashMap<>();
+ config.put("key1", "value1");
+ assertThat(service.getSessionConfig(sessionHandle)).containsAllEntriesOf(config);
+
+ verifyConfigureSession(sessionHandle, "RESET 'key1';");
+ assertThat(service.getSessionConfig(sessionHandle)).doesNotContainEntry("key1", "value1");
+
+ // CREATE & USE & ALTER & DROP
+ verifyConfigureSession(
+ sessionHandle,
+ "CREATE CATALOG mycat with ('type' = 'generic_in_memory', 'default-database' = 'db');");
+
+ verifyConfigureSession(sessionHandle, "USE CATALOG mycat;");
+ assertThat(service.getCurrentCatalog(sessionHandle)).isEqualTo("mycat");
+
+ verifyConfigureSession(
+ sessionHandle, "CREATE TABLE db.tbl (score INT) WITH ('connector' = 'datagen');");
+
+ Set<TableKind> tableKinds = new HashSet<>();
+ tableKinds.add(TableKind.TABLE);
+ assertThat(service.listTables(sessionHandle, "mycat", "db", tableKinds))
+ .contains(
+ new TableInfo(ObjectIdentifier.of("mycat", "db", "tbl"), TableKind.TABLE));
+
+ verifyConfigureSession(sessionHandle, "ALTER TABLE db.tbl RENAME TO tbl1;");
+ assertThat(service.listTables(sessionHandle, "mycat", "db", tableKinds))
+ .doesNotContain(
+ new TableInfo(ObjectIdentifier.of("mycat", "db", "tbl"), TableKind.TABLE))
+ .contains(
+ new TableInfo(ObjectIdentifier.of("mycat", "db", "tbl1"), TableKind.TABLE));
+
+ service.configureSession(sessionHandle, "USE CATALOG default_catalog;", 0);
+ verifyConfigureSession(sessionHandle, "DROP CATALOG mycat;");
+ assertThat(service.listCatalogs(sessionHandle)).doesNotContain("mycat");
+
+ // LOAD & UNLOAD MODULE
+ verifyConfigureSession(sessionHandle, "LOAD MODULE dummy;");
+
+ TableEnvironmentInternal tableEnv =
+ service.getSession(sessionHandle).createExecutor().getTableEnvironment();
+ assertThat(
+ CollectionUtil.iteratorToList(
+ tableEnv.executeSql("SHOW FULL MODULES;").collect()))
+ .contains(Row.of("dummy", true));
+
+ verifyConfigureSession(sessionHandle, "UNLOAD MODULE dummy;");
+ assertThat(
+ CollectionUtil.iteratorToList(
+ tableEnv.executeSql("SHOW FULL MODULES;").collect()))
+ .isEqualTo(Collections.singletonList(Row.of("core", true)));
+
+ // ADD JAR
+ String udfClassName = GENERATED_LOWER_UDF_CLASS + new Random().nextInt(50);
+ String jarPath =
+ UserClassLoaderJarTestUtils.createJarFile(
+ new File(tmpDir.toUri()),
+ "test-add-jar.jar",
+ udfClassName,
+ String.format(GENERATED_LOWER_UDF_CODE, udfClassName))
+ .toURI()
+ .toString();
+ verifyConfigureSession(sessionHandle, String.format("ADD JAR '%s';", jarPath));
+
+ assertThat(CollectionUtil.iteratorToList(tableEnv.executeSql("SHOW JARS;").collect()))
+ .isEqualTo(Collections.singletonList(Row.of(new Path(jarPath).getPath())));
Review Comment:
Mentioned above.
--
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 #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
fsk119 commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1033065242
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java:
##########
@@ -79,6 +79,23 @@ public void closeSession(SessionHandle sessionHandle) throws SqlGatewayException
}
}
+ @Override
+ public void configureSession(
+ SessionHandle sessionHandle, String statement, long executionTimeoutMs)
+ throws SqlGatewayException {
+ try {
+ if (executionTimeoutMs > 0) {
+ // TODO: support the feature in FLINK-27838
+ throw new UnsupportedOperationException(
+ "SqlGatewayService doesn't support timeout mechanism now.");
+ }
+ getSession(sessionHandle).createExecutor().configureSession(statement);
Review Comment:
It's not thread-safe here.
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationExecutor.java:
##########
@@ -84,6 +91,39 @@ public OperationExecutor(SessionContext context, Configuration executionConfig)
this.executionConfig = executionConfig;
}
+ public void configureSession(String statement) throws ExecutionException, InterruptedException {
+ TableEnvironmentInternal tableEnv = getTableEnvironment();
+ List<Operation> parsedOperations = tableEnv.getParser().parse(statement);
+ if (parsedOperations.size() > 1) {
+ throw new UnsupportedOperationException(
+ "Unsupported SQL statement! Execute statement only accepts a single SQL statement or "
+ + "multiple 'INSERT INTO' statements wrapped in a 'STATEMENT SET' block.");
+ }
+ Operation op = parsedOperations.get(0);
Review Comment:
Can we reuse the code as we do in the CliClient? I think we can introduce an enum `ExecutionMode` and validate the statement kind in the `INITIALIZATION` mode.
##########
flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/SqlGatewayService.java:
##########
@@ -66,6 +66,14 @@ public interface SqlGatewayService {
*/
void closeSession(SessionHandle sessionHandle) throws SqlGatewayException;
+ /**
+ * Using the statement to initialize the Session. It's only allowed to execute
+ * SET/RESET/CREATE/DROP/USE/ALTER/LOAD MODULE/UNLOAD MODULE/ADD JAR. The execution should be
+ * finished before returning because jobs submitted later may depend on it.
+ */
Review Comment:
It's better if we can align with other java docs.
```
/**
* Configure the basic settings for the session, including configuring the settings, preparing the catalog and so
* on.
*
* @param sessionHandle handle to identify the Session needs to be closed.
* ....
*/
```
##########
flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/SqlGatewayServiceITCase.java:
##########
@@ -157,6 +168,83 @@ public void testOpenSessionWithEnvironment() throws Exception {
assertThat(tableEnv.listModules()).contains(moduleName);
}
+ @Test
+ public void testConfigureSessionWithLegalStatement(@TempDir java.nio.file.Path tmpDir)
+ throws Exception {
+ SessionHandle sessionHandle = service.openSession(defaultSessionEnvironment);
+
+ // SET & RESET
+ service.configureSession(sessionHandle, "SET 'key1' = 'value1';", 0);
+ Map<String, String> config = new HashMap<>();
+ config.put("key1", "value1");
+ assertThat(service.getSessionConfig(sessionHandle)).containsAllEntriesOf(config);
+
+ service.configureSession(sessionHandle, "RESET 'key1';", 0);
+ assertThat(service.getSessionConfig(sessionHandle)).doesNotContainEntry("key1", "value1");
+
+ // CREATE & USE & ALTER & DROP
+ service.configureSession(
+ sessionHandle,
+ "CREATE CATALOG mycat with ('type' = 'generic_in_memory', 'default-database' = 'db');",
+ 0);
+
+ service.configureSession(sessionHandle, "USE CATALOG mycat;", 0);
+ assertThat(service.getCurrentCatalog(sessionHandle)).isEqualTo("mycat");
+
+ service.configureSession(
+ sessionHandle,
+ "CREATE TABLE db.tbl (score INT) WITH ('connector' = 'datagen');",
+ 0);
+
+ Set<TableKind> tableKinds = new HashSet<>();
+ tableKinds.add(TableKind.TABLE);
+ assertThat(service.listTables(sessionHandle, "mycat", "db", tableKinds))
+ .contains(
+ new TableInfo(ObjectIdentifier.of("mycat", "db", "tbl"), TableKind.TABLE));
+
+ service.configureSession(sessionHandle, "ALTER TABLE db.tbl RENAME TO tbl1;", 0);
+ assertThat(service.listTables(sessionHandle, "mycat", "db", tableKinds))
+ .doesNotContain(
+ new TableInfo(ObjectIdentifier.of("mycat", "db", "tbl"), TableKind.TABLE))
+ .contains(
+ new TableInfo(ObjectIdentifier.of("mycat", "db", "tbl1"), TableKind.TABLE));
+
+ service.configureSession(sessionHandle, "USE CATALOG default_catalog;", 0);
+ service.configureSession(sessionHandle, "DROP CATALOG mycat;", 0);
+ assertThat(service.listCatalogs(sessionHandle)).doesNotContain("mycat");
+
+ // LOAD & UNLOAD MODULE
+ service.configureSession(sessionHandle, "LOAD MODULE dummy;", 0);
+
+ TableEnvironmentInternal tableEnv =
+ service.getSession(sessionHandle).createExecutor().getTableEnvironment();
+ assertThat(
+ CollectionUtil.iteratorToList(
Review Comment:
Can we unload module core then load module core? So we don't need to introduce another dependency?
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationExecutor.java:
##########
@@ -84,6 +91,39 @@ public OperationExecutor(SessionContext context, Configuration executionConfig)
this.executionConfig = executionConfig;
}
+ public void configureSession(String statement) throws ExecutionException, InterruptedException {
+ TableEnvironmentInternal tableEnv = getTableEnvironment();
+ List<Operation> parsedOperations = tableEnv.getParser().parse(statement);
+ if (parsedOperations.size() > 1) {
+ throw new UnsupportedOperationException(
+ "Unsupported SQL statement! Execute statement only accepts a single SQL statement or "
+ + "multiple 'INSERT INTO' statements wrapped in a 'STATEMENT SET' block.");
+ }
+ Operation op = parsedOperations.get(0);
+
+ if (!(op instanceof SetOperation)
+ && !(op instanceof ResetOperation)
+ && !(op instanceof CreateOperation)
+ && !(op instanceof DropOperation)
+ && !(op instanceof UseOperation)
+ && !(op instanceof AlterOperation)
+ && !(op instanceof LoadModuleOperation)
+ && !(op instanceof UnloadModuleOperation)
+ && !(op instanceof AddJarOperation)) {
+ throw new UnsupportedOperationException(
+ String.format("Unsupported statement for configuring session:\n%s", statement));
+ }
+
+ // Ignore the execution result
+ if (op instanceof SetOperation) {
+ callSetOperation(tableEnv, OperationHandle.create(), (SetOperation) op);
+ } else if (op instanceof ResetOperation) {
+ callResetOperation(OperationHandle.create(), (ResetOperation) op);
+ } else {
+ tableEnv.executeInternal(op).await();
Review Comment:
Actually, all Operations above are executed in sync mode. We don't need to await here.
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationExecutor.java:
##########
@@ -84,6 +91,39 @@ public OperationExecutor(SessionContext context, Configuration executionConfig)
this.executionConfig = executionConfig;
}
+ public void configureSession(String statement) throws ExecutionException, InterruptedException {
+ TableEnvironmentInternal tableEnv = getTableEnvironment();
+ List<Operation> parsedOperations = tableEnv.getParser().parse(statement);
+ if (parsedOperations.size() > 1) {
+ throw new UnsupportedOperationException(
+ "Unsupported SQL statement! Execute statement only accepts a single SQL statement or "
+ + "multiple 'INSERT INTO' statements wrapped in a 'STATEMENT SET' block.");
Review Comment:
The msg is not correct.
##########
flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/SqlGatewayServiceITCase.java:
##########
@@ -711,6 +799,17 @@ public void testReleaseLockWhenFailedToSubmitOperation() throws Exception {
// Negative tests
// --------------------------------------------------------------------------------------------
+ @Test
+ void testConfigureSessionWithIllegalStatement() {
+ SessionHandle sessionHandle = service.openSession(defaultSessionEnvironment);
+
+ assertThatThrownBy(() -> service.configureSession(sessionHandle, "SELECT 1;", 0))
+ .satisfies(
+ FlinkAssertions.anyCauseMatches(
+ UnsupportedOperationException.class,
+ "Unsupported statement for configuring session:\nSELECT 1;"));
Review Comment:
I prefer to use
```
Unsupported statement for configuring session: SELECT 1;
The configureSession API only supports to execute statement of tyep
CREATE TABLE, DROP TABLE, ALTER TABLE, CREATE DATABASE, DROP DATABASE, ALTER DATABASE,
CREATE FUNCTION, DROP FUNCTION, ALTER FUNCTION, CREATE CATALOG, DROP CATALOG,
USE CATALOG, USE [CATALOG.]DATABASE, CREATE VIEW, DROP VIEW, LOAD MODULE, UNLOAD MODULE
USE MODULE.
```
--
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] jnh5y commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
jnh5y commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1021876246
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java:
##########
@@ -79,6 +82,32 @@ public void closeSession(SessionHandle sessionHandle) throws SqlGatewayException
}
}
+ @Override
+ public ResultSet configureSession(
+ SessionHandle sessionHandle, String statement, long executionTimeoutMs)
+ throws SqlGatewayException {
+ try {
+ if (executionTimeoutMs > 0) {
+ // TODO: support the feature in FLINK-27838
+ throw new UnsupportedOperationException(
+ "SqlGatewayService doesn't support timeout mechanism now.");
+ }
+ OperationHandle operationHandle =
+ getSession(sessionHandle)
+ .getOperationManager()
+ .submitOperation(
+ handle ->
+ getSession(sessionHandle)
+ .createExecutor()
+ .configureSession(handle, statement));
+ return fetchConfigureSessionResult(sessionHandle, operationHandle);
Review Comment:
Are there any concurrency concerns here?
I'm new to the codebase; seems like it ought to be ok, but figured I'd ask.
To validate what I understand so far, I'm guessing that the SQL client will block on getting a response back. Also, it looks like the `fetchConfigureSessionResult` will block on getting the result back.
--
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] yuzelin commented on pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
yuzelin commented on PR #21133:
URL: https://github.com/apache/flink/pull/21133#issuecomment-1340844259
@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] jnh5y commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
jnh5y commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1021878225
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java:
##########
@@ -79,6 +82,32 @@ public void closeSession(SessionHandle sessionHandle) throws SqlGatewayException
}
}
+ @Override
+ public ResultSet configureSession(
+ SessionHandle sessionHandle, String statement, long executionTimeoutMs)
+ throws SqlGatewayException {
+ try {
+ if (executionTimeoutMs > 0) {
+ // TODO: support the feature in FLINK-27838
Review Comment:
As another thought, in the future, if a `configureSession` call *does timeout*, it seems like it could be challenging to manage rolling back the configuration change as new requests are coming in.
--
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] yuzelin commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
yuzelin commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1022288458
##########
flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/SqlGatewayServiceITCase.java:
##########
@@ -157,6 +169,80 @@ public void testOpenSessionWithEnvironment() throws Exception {
assertThat(tableEnv.listModules()).contains(moduleName);
}
+ @Test
+ public void testConfigureSessionWithLegalStatement(@TempDir java.nio.file.Path tmpDir)
+ throws Exception {
+ SessionHandle sessionHandle = service.openSession(defaultSessionEnvironment);
+
+ // SET & RESET
+ verifyConfigureSession(sessionHandle, "SET 'key1' = 'value1';");
+ Map<String, String> config = new HashMap<>();
+ config.put("key1", "value1");
+ assertThat(service.getSessionConfig(sessionHandle)).containsAllEntriesOf(config);
+
+ verifyConfigureSession(sessionHandle, "RESET 'key1';");
+ assertThat(service.getSessionConfig(sessionHandle)).doesNotContainEntry("key1", "value1");
+
+ // CREATE & USE & ALTER & DROP
+ verifyConfigureSession(
+ sessionHandle,
+ "CREATE CATALOG mycat with ('type' = 'generic_in_memory', 'default-database' = 'db');");
+
+ verifyConfigureSession(sessionHandle, "USE CATALOG mycat;");
+ assertThat(service.getCurrentCatalog(sessionHandle)).isEqualTo("mycat");
Review Comment:
The `type` is used for discovering the corresponding catalog factory. Here the `generic_in_memory` identify the `GenericInMemoryCatalogFactory`. I think we can verify it by `mycat instance of GenericInMemoryCatalog` but the `SqlGatewayService` doesn't expose the api to get the catalog class but only the name string. Otherwise, creating mycat successfully means the factory has been found, so I think it's not a problem.
Also, creating the catalog successfully means the properties in `with` clause can pass the validation, so I think it's not necessary to check the properties again.
--
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 #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
fsk119 commented on PR #21133:
URL: https://github.com/apache/flink/pull/21133#issuecomment-1340534858
@yuzelin I add a commit to imporve codes. Please cc
--
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] yuzelin commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
yuzelin commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1024355066
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java:
##########
@@ -311,8 +340,29 @@ public GatewayInfo getGatewayInfo() {
return GatewayInfo.INSTANCE;
}
+ // --------------------------------------------------------------------------------------------
+
@VisibleForTesting
public Session getSession(SessionHandle sessionHandle) {
return sessionManager.getSession(sessionHandle);
}
+
+ /** Fetch all results for configuring session. */
+ private ResultSet fetchConfigureSessionResult(
+ SessionHandle sessionHandle, OperationHandle operationHandle) {
+ ResultSet firstResult = fetchResults(sessionHandle, operationHandle, 0, Integer.MAX_VALUE);
+ while (firstResult == ResultSet.NOT_READY_RESULTS) {
+ firstResult = fetchResults(sessionHandle, operationHandle, 0, Integer.MAX_VALUE);
Review Comment:
> Let me ask this question: If one of the operations on the gateway took some time, say, 5 seconds. How many calls to the gateway would we expect to make here while waiting for the results to be ready?
I'm afraid that I have no answer to this question. I guess you are worried about frequent queries will make it a burden to the execution. However, I think it's not a problem. Let's see how the codes perform the fetching.
Fetching result will finally go to [fetchResultsInternal](https://github.com/apache/flink/blob/5b32b9defcbb2adf7a0ad0898a67c40e0e012ebb/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationManager.java#L331), here if the operation is not finished, the result will always be `NOT_READY`. At line 239, we can see the `ResultFetcher` is ready before the status is turned to finished, and the statement is executed [(in an `OperationExecutor`)](https://github.com/apache/flink/blob/5b32b9defcbb2adf7a0ad0898a67c40e0e012ebb/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java#L178). Although we still have to get data through `ResultFetcher`, but at least the execution is not bothered. Then the task of fetching the rest results is taken by `ResultStore`, a [thread](https://github.com/apache/flink/blob/5b32b9defcbb2adf7a0ad0898a67c40e0e012ebb/flink-table/flink-sql-gateway/src/main/java/or
g/apache/flink/table/gateway/service/result/ResultStore.java#L149) is fetching continuously based on the [`TableResultInternal#collectInternal`](https://github.com/apache/flink/blob/5b32b9defcbb2adf7a0ad0898a67c40e0e012ebb/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableResultInternal.java#L45).
Anyway, the first result is important. The [implementation](https://github.com/apache/flink/blob/5b32b9defcbb2adf7a0ad0898a67c40e0e012ebb/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableResultImpl.java#L105) is the proof. After getting the first result, we can control the interval of fetching rest results. For example, in `CliResultView` of SQL client for print query results, it has set several `REFRESH_INTERVALS`. However, I see that in remote mode, the cost of network communications in fetching the first ready result may be large. I think it's worth discussing when implementing the remote mode. I will also ask @fsk119 for advice.
--
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] jnh5y commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
jnh5y commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1021874117
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java:
##########
@@ -79,6 +82,32 @@ public void closeSession(SessionHandle sessionHandle) throws SqlGatewayException
}
}
+ @Override
+ public ResultSet configureSession(
+ SessionHandle sessionHandle, String statement, long executionTimeoutMs)
+ throws SqlGatewayException {
+ try {
+ if (executionTimeoutMs > 0) {
+ // TODO: support the feature in FLINK-27838
Review Comment:
As a small detail, FLINK-27838 says it is about a timeout for `executeStatement`. Should the ticket be updated to note that `configureSession` also has a timeout?
--
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] yuzelin commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
yuzelin commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1022313111
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java:
##########
@@ -311,8 +340,29 @@ public GatewayInfo getGatewayInfo() {
return GatewayInfo.INSTANCE;
}
+ // --------------------------------------------------------------------------------------------
+
@VisibleForTesting
public Session getSession(SessionHandle sessionHandle) {
return sessionManager.getSession(sessionHandle);
}
+
+ /** Fetch all results for configuring session. */
+ private ResultSet fetchConfigureSessionResult(
+ SessionHandle sessionHandle, OperationHandle operationHandle) {
+ ResultSet firstResult = fetchResults(sessionHandle, operationHandle, 0, Integer.MAX_VALUE);
+ while (firstResult == ResultSet.NOT_READY_RESULTS) {
+ firstResult = fetchResults(sessionHandle, operationHandle, 0, Integer.MAX_VALUE);
Review Comment:
I can't understand exactly what you mean.
In my opinion, if some operations are executed synchronously, the result would be ready after the execution, so fetchResults will wait for some time and then get a ready result, and I think here won't be busy wait. Otherwise, if the operation is executed asynchronously, the execution result may not be ready after invoking fetchResult, and here will be busy wait.
The current design of result fetching is asynchronous, so here will be busy wait.
--
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] yuzelin commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
yuzelin commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1022361091
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java:
##########
@@ -79,6 +82,32 @@ public void closeSession(SessionHandle sessionHandle) throws SqlGatewayException
}
}
+ @Override
+ public ResultSet configureSession(
+ SessionHandle sessionHandle, String statement, long executionTimeoutMs)
+ throws SqlGatewayException {
+ try {
+ if (executionTimeoutMs > 0) {
+ // TODO: support the feature in FLINK-27838
Review Comment:
> As a small detail, [FLINK-27838](https://issues.apache.org/jira/browse/FLINK-27838) says it is about a timeout for `executeStatement`. Should the ticket be updated to note that `configureSession` also has a timeout?
Yes, I will add description of `configureSession` to that ticket.
> As another thought, in the future, if a `configureSession` call _does timeout_, it seems like it could be challenging to manage rolling back the configuration change as new requests are coming in.
Yes, you are right. I have some thoughts about this problem. Let's see why we need `configureSession` since the `executeStatement` can submit the same statement. This method is for the client to execute initialization file (mentioned above), and the client will be closed if any error occurs before the execution of initialization file finished. In this scenario. If in other scenarios this method should be invoked in the future, I think the caller should handle the error, such as retry mechanism.
--
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 #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
fsk119 commented on PR #21133:
URL: https://github.com/apache/flink/pull/21133#issuecomment-1336779615
I have added a commit to improve the codes. Could you take a time to look, @yuzelin ?
--
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] yuzelin commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
yuzelin commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1039581297
##########
flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/SqlGatewayServiceITCase.java:
##########
@@ -157,6 +168,83 @@ public void testOpenSessionWithEnvironment() throws Exception {
assertThat(tableEnv.listModules()).contains(moduleName);
}
+ @Test
+ public void testConfigureSessionWithLegalStatement(@TempDir java.nio.file.Path tmpDir)
+ throws Exception {
+ SessionHandle sessionHandle = service.openSession(defaultSessionEnvironment);
+
+ // SET & RESET
+ service.configureSession(sessionHandle, "SET 'key1' = 'value1';", 0);
+ Map<String, String> config = new HashMap<>();
+ config.put("key1", "value1");
+ assertThat(service.getSessionConfig(sessionHandle)).containsAllEntriesOf(config);
+
+ service.configureSession(sessionHandle, "RESET 'key1';", 0);
+ assertThat(service.getSessionConfig(sessionHandle)).doesNotContainEntry("key1", "value1");
+
+ // CREATE & USE & ALTER & DROP
+ service.configureSession(
+ sessionHandle,
+ "CREATE CATALOG mycat with ('type' = 'generic_in_memory', 'default-database' = 'db');",
+ 0);
+
+ service.configureSession(sessionHandle, "USE CATALOG mycat;", 0);
+ assertThat(service.getCurrentCatalog(sessionHandle)).isEqualTo("mycat");
+
+ service.configureSession(
+ sessionHandle,
+ "CREATE TABLE db.tbl (score INT) WITH ('connector' = 'datagen');",
+ 0);
+
+ Set<TableKind> tableKinds = new HashSet<>();
+ tableKinds.add(TableKind.TABLE);
+ assertThat(service.listTables(sessionHandle, "mycat", "db", tableKinds))
+ .contains(
+ new TableInfo(ObjectIdentifier.of("mycat", "db", "tbl"), TableKind.TABLE));
+
+ service.configureSession(sessionHandle, "ALTER TABLE db.tbl RENAME TO tbl1;", 0);
+ assertThat(service.listTables(sessionHandle, "mycat", "db", tableKinds))
+ .doesNotContain(
+ new TableInfo(ObjectIdentifier.of("mycat", "db", "tbl"), TableKind.TABLE))
+ .contains(
+ new TableInfo(ObjectIdentifier.of("mycat", "db", "tbl1"), TableKind.TABLE));
+
+ service.configureSession(sessionHandle, "USE CATALOG default_catalog;", 0);
+ service.configureSession(sessionHandle, "DROP CATALOG mycat;", 0);
+ assertThat(service.listCatalogs(sessionHandle)).doesNotContain("mycat");
+
+ // LOAD & UNLOAD MODULE
+ service.configureSession(sessionHandle, "LOAD MODULE dummy;", 0);
+
+ TableEnvironmentInternal tableEnv =
+ service.getSession(sessionHandle).createExecutor().getTableEnvironment();
+ assertThat(
+ CollectionUtil.iteratorToList(
Review Comment:
Sure.
--
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] jnh5y commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
jnh5y commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1021817966
##########
flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/SqlGatewayServiceITCase.java:
##########
@@ -157,6 +169,80 @@ public void testOpenSessionWithEnvironment() throws Exception {
assertThat(tableEnv.listModules()).contains(moduleName);
}
+ @Test
+ public void testConfigureSessionWithLegalStatement(@TempDir java.nio.file.Path tmpDir)
+ throws Exception {
+ SessionHandle sessionHandle = service.openSession(defaultSessionEnvironment);
+
+ // SET & RESET
+ verifyConfigureSession(sessionHandle, "SET 'key1' = 'value1';");
+ Map<String, String> config = new HashMap<>();
+ config.put("key1", "value1");
+ assertThat(service.getSessionConfig(sessionHandle)).containsAllEntriesOf(config);
+
+ verifyConfigureSession(sessionHandle, "RESET 'key1';");
+ assertThat(service.getSessionConfig(sessionHandle)).doesNotContainEntry("key1", "value1");
+
+ // CREATE & USE & ALTER & DROP
+ verifyConfigureSession(
+ sessionHandle,
+ "CREATE CATALOG mycat with ('type' = 'generic_in_memory', 'default-database' = 'db');");
+
+ verifyConfigureSession(sessionHandle, "USE CATALOG mycat;");
+ assertThat(service.getCurrentCatalog(sessionHandle)).isEqualTo("mycat");
+
+ verifyConfigureSession(
+ sessionHandle, "CREATE TABLE db.tbl (score INT) WITH ('connector' = 'datagen');");
+
+ Set<TableKind> tableKinds = new HashSet<>();
+ tableKinds.add(TableKind.TABLE);
+ assertThat(service.listTables(sessionHandle, "mycat", "db", tableKinds))
+ .contains(
+ new TableInfo(ObjectIdentifier.of("mycat", "db", "tbl"), TableKind.TABLE));
+
+ verifyConfigureSession(sessionHandle, "ALTER TABLE db.tbl RENAME TO tbl1;");
+ assertThat(service.listTables(sessionHandle, "mycat", "db", tableKinds))
+ .doesNotContain(
+ new TableInfo(ObjectIdentifier.of("mycat", "db", "tbl"), TableKind.TABLE))
+ .contains(
+ new TableInfo(ObjectIdentifier.of("mycat", "db", "tbl1"), TableKind.TABLE));
+
+ service.configureSession(sessionHandle, "USE CATALOG default_catalog;", 0);
+ verifyConfigureSession(sessionHandle, "DROP CATALOG mycat;");
+ assertThat(service.listCatalogs(sessionHandle)).doesNotContain("mycat");
+
+ // LOAD & UNLOAD MODULE
+ verifyConfigureSession(sessionHandle, "LOAD MODULE dummy;");
+
+ TableEnvironmentInternal tableEnv =
+ service.getSession(sessionHandle).createExecutor().getTableEnvironment();
+ assertThat(
+ CollectionUtil.iteratorToList(
+ tableEnv.executeSql("SHOW FULL MODULES;").collect()))
+ .contains(Row.of("dummy", true));
+
+ verifyConfigureSession(sessionHandle, "UNLOAD MODULE dummy;");
+ assertThat(
+ CollectionUtil.iteratorToList(
+ tableEnv.executeSql("SHOW FULL MODULES;").collect()))
+ .isEqualTo(Collections.singletonList(Row.of("core", true)));
+
+ // ADD JAR
+ String udfClassName = GENERATED_LOWER_UDF_CLASS + new Random().nextInt(50);
+ String jarPath =
+ UserClassLoaderJarTestUtils.createJarFile(
+ new File(tmpDir.toUri()),
+ "test-add-jar.jar",
+ udfClassName,
+ String.format(GENERATED_LOWER_UDF_CODE, udfClassName))
+ .toURI()
+ .toString();
+ verifyConfigureSession(sessionHandle, String.format("ADD JAR '%s';", jarPath));
+
+ assertThat(CollectionUtil.iteratorToList(tableEnv.executeSql("SHOW JARS;").collect()))
+ .isEqualTo(Collections.singletonList(Row.of(new Path(jarPath).getPath())));
Review Comment:
Assuming that `REMOVE JAR` is added above, it'd be make sense to remove the jar here.
--
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] jnh5y commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
jnh5y commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1023338175
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java:
##########
@@ -311,8 +340,29 @@ public GatewayInfo getGatewayInfo() {
return GatewayInfo.INSTANCE;
}
+ // --------------------------------------------------------------------------------------------
+
@VisibleForTesting
public Session getSession(SessionHandle sessionHandle) {
return sessionManager.getSession(sessionHandle);
}
+
+ /** Fetch all results for configuring session. */
+ private ResultSet fetchConfigureSessionResult(
+ SessionHandle sessionHandle, OperationHandle operationHandle) {
+ ResultSet firstResult = fetchResults(sessionHandle, operationHandle, 0, Integer.MAX_VALUE);
+ while (firstResult == ResultSet.NOT_READY_RESULTS) {
+ firstResult = fetchResults(sessionHandle, operationHandle, 0, Integer.MAX_VALUE);
Review Comment:
Let me ask this question: If one of the operations on the gateway took some time, say, 5 seconds. How many calls to the gateway would we expect to make here while waiting for the results to be ready?
--
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 closed pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
fsk119 closed pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
URL: https://github.com/apache/flink/pull/21133
--
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] yuzelin commented on pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
yuzelin commented on PR #21133:
URL: https://github.com/apache/flink/pull/21133#issuecomment-1340279326
@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] jnh5y commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
jnh5y commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1021816197
##########
flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/SqlGatewayServiceITCase.java:
##########
@@ -157,6 +169,80 @@ public void testOpenSessionWithEnvironment() throws Exception {
assertThat(tableEnv.listModules()).contains(moduleName);
}
+ @Test
+ public void testConfigureSessionWithLegalStatement(@TempDir java.nio.file.Path tmpDir)
+ throws Exception {
+ SessionHandle sessionHandle = service.openSession(defaultSessionEnvironment);
+
+ // SET & RESET
+ verifyConfigureSession(sessionHandle, "SET 'key1' = 'value1';");
+ Map<String, String> config = new HashMap<>();
+ config.put("key1", "value1");
+ assertThat(service.getSessionConfig(sessionHandle)).containsAllEntriesOf(config);
+
+ verifyConfigureSession(sessionHandle, "RESET 'key1';");
+ assertThat(service.getSessionConfig(sessionHandle)).doesNotContainEntry("key1", "value1");
+
+ // CREATE & USE & ALTER & DROP
+ verifyConfigureSession(
+ sessionHandle,
+ "CREATE CATALOG mycat with ('type' = 'generic_in_memory', 'default-database' = 'db');");
+
+ verifyConfigureSession(sessionHandle, "USE CATALOG mycat;");
+ assertThat(service.getCurrentCatalog(sessionHandle)).isEqualTo("mycat");
Review Comment:
Is it worth checking the `type` and/or `default-database`?
--
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] yuzelin commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
yuzelin commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1022368927
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java:
##########
@@ -79,6 +82,32 @@ public void closeSession(SessionHandle sessionHandle) throws SqlGatewayException
}
}
+ @Override
+ public ResultSet configureSession(
+ SessionHandle sessionHandle, String statement, long executionTimeoutMs)
+ throws SqlGatewayException {
+ try {
+ if (executionTimeoutMs > 0) {
+ // TODO: support the feature in FLINK-27838
+ throw new UnsupportedOperationException(
+ "SqlGatewayService doesn't support timeout mechanism now.");
+ }
+ OperationHandle operationHandle =
+ getSession(sessionHandle)
+ .getOperationManager()
+ .submitOperation(
+ handle ->
+ getSession(sessionHandle)
+ .createExecutor()
+ .configureSession(handle, statement));
+ return fetchConfigureSessionResult(sessionHandle, operationHandle);
Review Comment:
For the Operation (not the flink table operation; it is the gateway service operation) submitted to `OperationManager`, it has to get lock (see [operationLock](https://github.com/apache/flink/blob/3c9a394c608125827d224180c44128718972a82c/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationManager.java#L66)) before running. Since we make sure that the configure session returns after the execution result is ready, concurrency problem won't happen here (for example, if two clients holds the same `SessionHandle` somehow and submit two conflict configuring statement, one should succeed and then another should fail).
--
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] yuzelin commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
yuzelin commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1022288458
##########
flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/SqlGatewayServiceITCase.java:
##########
@@ -157,6 +169,80 @@ public void testOpenSessionWithEnvironment() throws Exception {
assertThat(tableEnv.listModules()).contains(moduleName);
}
+ @Test
+ public void testConfigureSessionWithLegalStatement(@TempDir java.nio.file.Path tmpDir)
+ throws Exception {
+ SessionHandle sessionHandle = service.openSession(defaultSessionEnvironment);
+
+ // SET & RESET
+ verifyConfigureSession(sessionHandle, "SET 'key1' = 'value1';");
+ Map<String, String> config = new HashMap<>();
+ config.put("key1", "value1");
+ assertThat(service.getSessionConfig(sessionHandle)).containsAllEntriesOf(config);
+
+ verifyConfigureSession(sessionHandle, "RESET 'key1';");
+ assertThat(service.getSessionConfig(sessionHandle)).doesNotContainEntry("key1", "value1");
+
+ // CREATE & USE & ALTER & DROP
+ verifyConfigureSession(
+ sessionHandle,
+ "CREATE CATALOG mycat with ('type' = 'generic_in_memory', 'default-database' = 'db');");
+
+ verifyConfigureSession(sessionHandle, "USE CATALOG mycat;");
+ assertThat(service.getCurrentCatalog(sessionHandle)).isEqualTo("mycat");
Review Comment:
The `type` is used for discovering the corresponding catalog factory. Here the `generic_in_memory` identify the `GenericInMemoryCatalogFactory`. I think we can verify it by `mycat instance of GenericInMemoryCatalog` but the `SqlGatewayService` doesn't expose the api to get the catalog class but only the name string. Otherwise, creating mycat successfully means the factory has been found, so I think it's not a problem.
Checking `default-database` is added.
--
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] yuzelin commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
yuzelin commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1024355066
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java:
##########
@@ -311,8 +340,29 @@ public GatewayInfo getGatewayInfo() {
return GatewayInfo.INSTANCE;
}
+ // --------------------------------------------------------------------------------------------
+
@VisibleForTesting
public Session getSession(SessionHandle sessionHandle) {
return sessionManager.getSession(sessionHandle);
}
+
+ /** Fetch all results for configuring session. */
+ private ResultSet fetchConfigureSessionResult(
+ SessionHandle sessionHandle, OperationHandle operationHandle) {
+ ResultSet firstResult = fetchResults(sessionHandle, operationHandle, 0, Integer.MAX_VALUE);
+ while (firstResult == ResultSet.NOT_READY_RESULTS) {
+ firstResult = fetchResults(sessionHandle, operationHandle, 0, Integer.MAX_VALUE);
Review Comment:
> Let me ask this question: If one of the operations on the gateway took some time, say, 5 seconds. How many calls to the gateway would we expect to make here while waiting for the results to be ready?
I'm afraid that I have no answer to this question. I guess you are worried about frequent queries will make it a burden to the execution. However, I think it's not a problem. Let's see how the codes perform the fetching.
Fetching result will finally go to [fetchResultsInternal](https://github.com/apache/flink/blob/5b32b9defcbb2adf7a0ad0898a67c40e0e012ebb/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationManager.java#L331), here if the operation is not finished, the result will always be `NOT_READY`. At line 239, we can see the `ResultFetcher` is ready before the status is turned to finished, and the statement is executed [(in an `OperationExecutor`)](https://github.com/apache/flink/blob/5b32b9defcbb2adf7a0ad0898a67c40e0e012ebb/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java#L178). Although we still have to get data through `ResultFetcher`, at least the execution is not bothered. Then the task of fetching the rest results is taken by `ResultStore`, a [thread](https://github.com/apache/flink/blob/5b32b9defcbb2adf7a0ad0898a67c40e0e012ebb/flink-table/flink-sql-gateway/src/main/java/org/ap
ache/flink/table/gateway/service/result/ResultStore.java#L149) is fetching continuously based on the [`TableResultInternal#collectInternal`](https://github.com/apache/flink/blob/5b32b9defcbb2adf7a0ad0898a67c40e0e012ebb/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableResultInternal.java#L45).
Anyway, the first result is important. The [implementation](https://github.com/apache/flink/blob/5b32b9defcbb2adf7a0ad0898a67c40e0e012ebb/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableResultImpl.java#L105) is the proof. After getting the first result, we can control the interval of fetching rest results. For example, in `CliResultView` of SQL client for print query results, it has set several `REFRESH_INTERVALS`. However, I see that in remote mode, the cost of network communications in fetching the first ready result may be large. I think it's worth discussing when implementing the remote mode. I will also ask @fsk119 for advice.
--
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] jnh5y commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
jnh5y commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1024424268
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java:
##########
@@ -311,8 +340,29 @@ public GatewayInfo getGatewayInfo() {
return GatewayInfo.INSTANCE;
}
+ // --------------------------------------------------------------------------------------------
+
@VisibleForTesting
public Session getSession(SessionHandle sessionHandle) {
return sessionManager.getSession(sessionHandle);
}
+
+ /** Fetch all results for configuring session. */
+ private ResultSet fetchConfigureSessionResult(
+ SessionHandle sessionHandle, OperationHandle operationHandle) {
+ ResultSet firstResult = fetchResults(sessionHandle, operationHandle, 0, Integer.MAX_VALUE);
+ while (firstResult == ResultSet.NOT_READY_RESULTS) {
+ firstResult = fetchResults(sessionHandle, operationHandle, 0, Integer.MAX_VALUE);
Review Comment:
Sounds good and sounds like you and @fsk119 will figure out a good solution!
--
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] yuzelin commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
yuzelin commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1022368927
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java:
##########
@@ -79,6 +82,32 @@ public void closeSession(SessionHandle sessionHandle) throws SqlGatewayException
}
}
+ @Override
+ public ResultSet configureSession(
+ SessionHandle sessionHandle, String statement, long executionTimeoutMs)
+ throws SqlGatewayException {
+ try {
+ if (executionTimeoutMs > 0) {
+ // TODO: support the feature in FLINK-27838
+ throw new UnsupportedOperationException(
+ "SqlGatewayService doesn't support timeout mechanism now.");
+ }
+ OperationHandle operationHandle =
+ getSession(sessionHandle)
+ .getOperationManager()
+ .submitOperation(
+ handle ->
+ getSession(sessionHandle)
+ .createExecutor()
+ .configureSession(handle, statement));
+ return fetchConfigureSessionResult(sessionHandle, operationHandle);
Review Comment:
For the Operation (not the flink table operation; it is the gateway service operation) submitted to `OperationManager`, it has to get lock (see [operationLock](https://github.com/apache/flink/blob/3c9a394c608125827d224180c44128718972a82c/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationManager.java#L66)) before running. And we make sure that the configure session returns after the execution result is ready, so concurrency problem won't happen here (for example, if two clients holds the same `SessionHandle` somehow and submit two conflict configuring statement, one should succeed and then another should fail).
--
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] yuzelin commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
yuzelin commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1022353242
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java:
##########
@@ -311,8 +340,29 @@ public GatewayInfo getGatewayInfo() {
return GatewayInfo.INSTANCE;
}
+ // --------------------------------------------------------------------------------------------
+
@VisibleForTesting
public Session getSession(SessionHandle sessionHandle) {
return sessionManager.getSession(sessionHandle);
}
+
+ /** Fetch all results for configuring session. */
+ private ResultSet fetchConfigureSessionResult(
Review Comment:
I have discussed with @fsk119 about this and the `busy wait` problem above. The result is here:
As the [FLIP-91](https://cwiki.apache.org/confluence/display/FLINK/FLIP-91%3A+Support+SQL+Gateway) designed, the `configureSession` shouldn't return result (if it runs successfully means the result is OK; otherwise exception will be thrown). To make sure the execution has finished without error, we should use `TableResult#await` to wait if necessary until the data is ready.
--
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] yuzelin commented on pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
yuzelin commented on PR #21133:
URL: https://github.com/apache/flink/pull/21133#issuecomment-1289910411
@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] jnh5y commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
jnh5y commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1023336995
##########
flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/SqlGatewayServiceITCase.java:
##########
@@ -157,6 +169,80 @@ public void testOpenSessionWithEnvironment() throws Exception {
assertThat(tableEnv.listModules()).contains(moduleName);
}
+ @Test
+ public void testConfigureSessionWithLegalStatement(@TempDir java.nio.file.Path tmpDir)
+ throws Exception {
+ SessionHandle sessionHandle = service.openSession(defaultSessionEnvironment);
+
+ // SET & RESET
+ verifyConfigureSession(sessionHandle, "SET 'key1' = 'value1';");
+ Map<String, String> config = new HashMap<>();
+ config.put("key1", "value1");
+ assertThat(service.getSessionConfig(sessionHandle)).containsAllEntriesOf(config);
+
+ verifyConfigureSession(sessionHandle, "RESET 'key1';");
+ assertThat(service.getSessionConfig(sessionHandle)).doesNotContainEntry("key1", "value1");
+
+ // CREATE & USE & ALTER & DROP
+ verifyConfigureSession(
+ sessionHandle,
+ "CREATE CATALOG mycat with ('type' = 'generic_in_memory', 'default-database' = 'db');");
+
+ verifyConfigureSession(sessionHandle, "USE CATALOG mycat;");
+ assertThat(service.getCurrentCatalog(sessionHandle)).isEqualTo("mycat");
Review Comment:
Makes sense!
--
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] jnh5y commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
jnh5y commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1021869458
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java:
##########
@@ -311,8 +340,29 @@ public GatewayInfo getGatewayInfo() {
return GatewayInfo.INSTANCE;
}
+ // --------------------------------------------------------------------------------------------
+
@VisibleForTesting
public Session getSession(SessionHandle sessionHandle) {
return sessionManager.getSession(sessionHandle);
}
+
+ /** Fetch all results for configuring session. */
+ private ResultSet fetchConfigureSessionResult(
+ SessionHandle sessionHandle, OperationHandle operationHandle) {
+ ResultSet firstResult = fetchResults(sessionHandle, operationHandle, 0, Integer.MAX_VALUE);
+ while (firstResult == ResultSet.NOT_READY_RESULTS) {
+ firstResult = fetchResults(sessionHandle, operationHandle, 0, Integer.MAX_VALUE);
Review Comment:
It seems like operations like AddJar could be synchronous and take some period of time, right?
If that's the case, this will be a busy wait on any synchronous operations which take time, right?
--
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] yuzelin commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
yuzelin commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1022277824
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationExecutor.java:
##########
@@ -84,6 +90,30 @@ public OperationExecutor(SessionContext context, Configuration executionConfig)
this.executionConfig = executionConfig;
}
+ public ResultFetcher configureSession(OperationHandle handle, String statement) {
+ TableEnvironmentInternal tableEnv = getTableEnvironment();
+ List<Operation> parsedOperations = tableEnv.getParser().parse(statement);
+ if (parsedOperations.size() > 1) {
+ throw new UnsupportedOperationException(
+ "Unsupported SQL statement! Execute statement only accepts a single SQL statement or "
+ + "multiple 'INSERT INTO' statements wrapped in a 'STATEMENT SET' block.");
+ }
+ Operation op = parsedOperations.get(0);
+ if (!(op instanceof SetOperation)
+ && !(op instanceof ResetOperation)
+ && !(op instanceof CreateOperation)
+ && !(op instanceof DropOperation)
+ && !(op instanceof UseOperation)
+ && !(op instanceof AlterOperation)
+ && !(op instanceof LoadModuleOperation)
+ && !(op instanceof UnloadModuleOperation)
+ && !(op instanceof AddJarOperation)) {
Review Comment:
I think we should not support RemoveJar in gateway side. Let me explain.
The `configureSession` is mostly for the client to execute initialization file, and RemoveJar is one of the legal operations in initialization file. Currently, the implementation of RemoveJarOperation depends on `ClientWrapperClassLoader`. As it's comment said, I think the implementation has some technical problems, so
maybe it's better to let the RemoveJar implementation be in the client side now. The `Executor#removeJar` can do the 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] jnh5y commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
jnh5y commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1023336842
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationExecutor.java:
##########
@@ -84,6 +90,30 @@ public OperationExecutor(SessionContext context, Configuration executionConfig)
this.executionConfig = executionConfig;
}
+ public ResultFetcher configureSession(OperationHandle handle, String statement) {
+ TableEnvironmentInternal tableEnv = getTableEnvironment();
+ List<Operation> parsedOperations = tableEnv.getParser().parse(statement);
+ if (parsedOperations.size() > 1) {
+ throw new UnsupportedOperationException(
+ "Unsupported SQL statement! Execute statement only accepts a single SQL statement or "
+ + "multiple 'INSERT INTO' statements wrapped in a 'STATEMENT SET' block.");
+ }
+ Operation op = parsedOperations.get(0);
+ if (!(op instanceof SetOperation)
+ && !(op instanceof ResetOperation)
+ && !(op instanceof CreateOperation)
+ && !(op instanceof DropOperation)
+ && !(op instanceof UseOperation)
+ && !(op instanceof AlterOperation)
+ && !(op instanceof LoadModuleOperation)
+ && !(op instanceof UnloadModuleOperation)
+ && !(op instanceof AddJarOperation)) {
Review Comment:
Ok. That makes some sense; I may need to learn more about what the jars are doing here to understand it better.
As long as leaving out RemoveJar is a thoughtful decision, I'm good with it.
--
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] yuzelin commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
yuzelin commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1024355066
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java:
##########
@@ -311,8 +340,29 @@ public GatewayInfo getGatewayInfo() {
return GatewayInfo.INSTANCE;
}
+ // --------------------------------------------------------------------------------------------
+
@VisibleForTesting
public Session getSession(SessionHandle sessionHandle) {
return sessionManager.getSession(sessionHandle);
}
+
+ /** Fetch all results for configuring session. */
+ private ResultSet fetchConfigureSessionResult(
+ SessionHandle sessionHandle, OperationHandle operationHandle) {
+ ResultSet firstResult = fetchResults(sessionHandle, operationHandle, 0, Integer.MAX_VALUE);
+ while (firstResult == ResultSet.NOT_READY_RESULTS) {
+ firstResult = fetchResults(sessionHandle, operationHandle, 0, Integer.MAX_VALUE);
Review Comment:
> Let me ask this question: If one of the operations on the gateway took some time, say, 5 seconds. How many calls to the gateway would we expect to make here while waiting for the results to be ready?
I'm afraid that I have no answer to this question. I guess you are worried about frequent queries will make it a burden to the execution. However, I think it's not a problem. Let's see how the codes perform the fetching.
Fetching result will finally go to [fetchResultsInternal](https://github.com/apache/flink/blob/5b32b9defcbb2adf7a0ad0898a67c40e0e012ebb/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationManager.java#L331), here if the operation is not finished, the result will always be `NOT_READY`. At line 239, we can see the `ResultFetcher` is ready before the status is turned to finished, and the statement is executed [(in an `OperationExecutor`)](https://github.com/apache/flink/blob/5b32b9defcbb2adf7a0ad0898a67c40e0e012ebb/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java#L178). Although we still have to get data through `ResultFetcher`, at least the execution is not bothered. Then the task of fetching the rest results is taken by `ResultStore`, a [thread](https://github.com/apache/flink/blob/5b32b9defcbb2adf7a0ad0898a67c40e0e012ebb/flink-table/flink-sql-gateway/src/main/java/org/ap
ache/flink/table/gateway/service/result/ResultStore.java#L149) is fetching continuously based on the [`TableResultInternal#collectInternal`](https://github.com/apache/flink/blob/5b32b9defcbb2adf7a0ad0898a67c40e0e012ebb/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableResultInternal.java#L45).
Anyway, the first result is important. The [implementation](https://github.com/apache/flink/blob/5b32b9defcbb2adf7a0ad0898a67c40e0e012ebb/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableResultImpl.java#L105) of `TableResult#await` is the proof. After getting the first result, we can control the interval of fetching rest results. For example, in `CliResultView` of SQL client for print query results, it has set several `REFRESH_INTERVALS`. However, I see that in remote mode, the cost of network communications in fetching the first ready result may be large. I think it's worth discussing when implementing the remote mode. I will also ask @fsk119 for advice.
--
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 #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
flinkbot commented on PR #21133:
URL: https://github.com/apache/flink/pull/21133#issuecomment-1288356742
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c6c7fe02c0c3b17bd3dd8231c25cbe279b600a5a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "c6c7fe02c0c3b17bd3dd8231c25cbe279b600a5a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c6c7fe02c0c3b17bd3dd8231c25cbe279b600a5a 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] yuzelin commented on pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
yuzelin commented on PR #21133:
URL: https://github.com/apache/flink/pull/21133#issuecomment-1308294143
cc @fsk119
--
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] yuzelin commented on pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
yuzelin commented on PR #21133:
URL: https://github.com/apache/flink/pull/21133#issuecomment-1337326532
> I have added a commit to improve the codes. Could you take a time to look, @yuzelin ?
Quite cool. LGTM. But I also found some codes can be improved. Please check out, thanks.
--
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] yuzelin commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
yuzelin commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1022361091
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java:
##########
@@ -79,6 +82,32 @@ public void closeSession(SessionHandle sessionHandle) throws SqlGatewayException
}
}
+ @Override
+ public ResultSet configureSession(
+ SessionHandle sessionHandle, String statement, long executionTimeoutMs)
+ throws SqlGatewayException {
+ try {
+ if (executionTimeoutMs > 0) {
+ // TODO: support the feature in FLINK-27838
Review Comment:
> As a small detail, [FLINK-27838](https://issues.apache.org/jira/browse/FLINK-27838) says it is about a timeout for `executeStatement`. Should the ticket be updated to note that `configureSession` also has a timeout?
Yes, I will add description of `configureSession` to that ticket.
> As another thought, in the future, if a `configureSession` call _does timeout_, it seems like it could be challenging to manage rolling back the configuration change as new requests are coming in.
Yes, you are right. I have some thoughts about this problem. Let's see why we need `configureSession` since the `executeStatement` can submit the same statement. This method is for the client to execute initialization file (mentioned above), and the client will be closed if any error occurs before the execution of initialization file finished. In this scenario we doesn't need to consider the rolling back. If in other scenarios this method should be invoked in the future, I think the caller should handle the error, such as retry or restart mechanism.
--
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] jnh5y commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
jnh5y commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1021870936
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java:
##########
@@ -311,8 +340,29 @@ public GatewayInfo getGatewayInfo() {
return GatewayInfo.INSTANCE;
}
+ // --------------------------------------------------------------------------------------------
+
@VisibleForTesting
public Session getSession(SessionHandle sessionHandle) {
return sessionManager.getSession(sessionHandle);
}
+
+ /** Fetch all results for configuring session. */
+ private ResultSet fetchConfigureSessionResult(
Review Comment:
It seems like this function is general and could be used to fetch the results from any `OperationHandle`, right?
Should it be factor out or maybe named something more general?
--
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] jnh5y commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
jnh5y commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1021808704
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationExecutor.java:
##########
@@ -84,6 +90,30 @@ public OperationExecutor(SessionContext context, Configuration executionConfig)
this.executionConfig = executionConfig;
}
+ public ResultFetcher configureSession(OperationHandle handle, String statement) {
+ TableEnvironmentInternal tableEnv = getTableEnvironment();
+ List<Operation> parsedOperations = tableEnv.getParser().parse(statement);
+ if (parsedOperations.size() > 1) {
+ throw new UnsupportedOperationException(
+ "Unsupported SQL statement! Execute statement only accepts a single SQL statement or "
+ + "multiple 'INSERT INTO' statements wrapped in a 'STATEMENT SET' block.");
+ }
+ Operation op = parsedOperations.get(0);
+ if (!(op instanceof SetOperation)
+ && !(op instanceof ResetOperation)
+ && !(op instanceof CreateOperation)
+ && !(op instanceof DropOperation)
+ && !(op instanceof UseOperation)
+ && !(op instanceof AlterOperation)
+ && !(op instanceof LoadModuleOperation)
+ && !(op instanceof UnloadModuleOperation)
+ && !(op instanceof AddJarOperation)) {
Review Comment:
Should RemoveJarOperation be in the this list?
--
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] yuzelin commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
yuzelin commented on code in PR #21133:
URL: https://github.com/apache/flink/pull/21133#discussion_r1022368927
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java:
##########
@@ -79,6 +82,32 @@ public void closeSession(SessionHandle sessionHandle) throws SqlGatewayException
}
}
+ @Override
+ public ResultSet configureSession(
+ SessionHandle sessionHandle, String statement, long executionTimeoutMs)
+ throws SqlGatewayException {
+ try {
+ if (executionTimeoutMs > 0) {
+ // TODO: support the feature in FLINK-27838
+ throw new UnsupportedOperationException(
+ "SqlGatewayService doesn't support timeout mechanism now.");
+ }
+ OperationHandle operationHandle =
+ getSession(sessionHandle)
+ .getOperationManager()
+ .submitOperation(
+ handle ->
+ getSession(sessionHandle)
+ .createExecutor()
+ .configureSession(handle, statement));
+ return fetchConfigureSessionResult(sessionHandle, operationHandle);
Review Comment:
For the Operation (not the flink table operation; it is the gateway service operation) submitted to `OperationManager`, it has to get lock (see [operationLock](https://github.com/apache/flink/blob/3c9a394c608125827d224180c44128718972a82c/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationManager.java#L66) before running. And we make sure that the configure session returns after the execution result is ready, so concurrency problem won't happen here (for example, if two clients holds the same `SessionHandle` somehow and submit two conflict configuring statement, one should succeed and then another should fail).
--
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] yuzelin commented on pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
Posted by GitBox <gi...@apache.org>.
yuzelin commented on PR #21133:
URL: https://github.com/apache/flink/pull/21133#issuecomment-1340843902
> @yuzelin I add a commit to imporve codes. Please cc
LGTM.
--
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