[GitHub] [flink] jnh5y commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
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 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-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] jnh5y commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
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 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-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] jnh5y commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
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 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 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-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] jnh5y commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
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-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] jnh5y commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
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-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] jnh5y commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
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-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] jnh5y commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
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-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] jnh5y commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
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-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] jnh5y commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
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 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-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] jnh5y commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
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 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-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] jnh5y commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
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-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] jnh5y commented on a diff in pull request #21133: [FLINK-29732][sql-gateway] support configuring session with SQL statement.
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-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org