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 2021/02/24 13:57:40 UTC
[GitHub] [flink] fsk119 opened a new pull request #15006: [Flink-21485][sql-client] Simplify the ExecutionContext
fsk119 opened a new pull request #15006:
URL: https://github.com/apache/flink/pull/15006
<!--
*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
*(For example: This pull request makes task deployment go through the blob server, rather than through RPC. That way we avoid re-transferring them on each deployment (during recovery).)*
## Brief change log
*(for example:)*
- *The TaskInfo is stored in the blob store on job creation time as a persistent artifact*
- *Deployments RPC transmits only the blob storage reference*
- *TaskManagers retrieve the TaskInfo from the blob cache*
## Verifying this change
*(Please pick either of the following options)*
This change is a trivial rework / code cleanup without any test coverage.
*(or)*
This change is already covered by existing tests, such as *(please describe tests)*.
*(or)*
This change added tests and can be verified as follows:
*(example:)*
- *Added integration tests for end-to-end deployment with large payloads (100MB)*
- *Extended integration test for recovery after master (JobManager) failure*
- *Added test that validates that TaskInfo is transferred only once across recoveries*
- *Manually verified the change by running a 4 node cluser with 2 JobManagers and 4 TaskManagers, a stateful streaming program, and killing one JobManager and two TaskManagers during the execution, verifying that recovery happens correctly.*
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / no)
- The serializers: (yes / no / don't know)
- The runtime per-record code paths (performance sensitive): (yes / no / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: (yes / no / don't know)
- The S3 file system connector: (yes / no / don't know)
## Documentation
- Does this pull request introduce a new feature? (yes / no)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
}, {
"hash" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"triggerType" : "PUSH"
}, {
"hash" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736",
"triggerID" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4cd5563d44a62aec2b8db80cfe8da650fddf4ffa UNKNOWN
* 4cc39b300203dd122980e6ef22080746834fcc9b Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
}, {
"hash" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"triggerType" : "PUSH"
}, {
"hash" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736",
"triggerID" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"triggerType" : "PUSH"
}, {
"hash" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13813",
"triggerID" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"triggerType" : "PUSH"
}, {
"hash" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14060",
"triggerID" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"triggerType" : "PUSH"
}, {
"hash" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14126",
"triggerID" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"triggerType" : "PUSH"
}, {
"hash" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14133",
"triggerID" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"triggerType" : "PUSH"
}, {
"hash" : "83446f3a0e26473892f97f68de12b5f4a8e50cc5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14434",
"triggerID" : "83446f3a0e26473892f97f68de12b5f4a8e50cc5",
"triggerType" : "PUSH"
}, {
"hash" : "5fb5449d10b57a84f6da97e0be6fc875a46344cf",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14442",
"triggerID" : "5fb5449d10b57a84f6da97e0be6fc875a46344cf",
"triggerType" : "PUSH"
}, {
"hash" : "9e86f4b592c7699fc0fade586d99daa907b9bffc",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14465",
"triggerID" : "9e86f4b592c7699fc0fade586d99daa907b9bffc",
"triggerType" : "PUSH"
}, {
"hash" : "741114ff56881d722e10008529ec3e60a94d8431",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "741114ff56881d722e10008529ec3e60a94d8431",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4cd5563d44a62aec2b8db80cfe8da650fddf4ffa UNKNOWN
* 5fb5449d10b57a84f6da97e0be6fc875a46344cf Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14442)
* 9e86f4b592c7699fc0fade586d99daa907b9bffc Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14465)
* 741114ff56881d722e10008529ec3e60a94d8431 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fsk119 commented on a change in pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
fsk119 commented on a change in pull request #15006:
URL: https://github.com/apache/flink/pull/15006#discussion_r592033547
##########
File path: flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/context/ExecutionContext.java
##########
@@ -0,0 +1,233 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.client.gateway.context;
+
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.TimeCharacteristic;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.bridge.java.internal.BatchTableEnvironmentImpl;
+import org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl;
+import org.apache.flink.table.catalog.CatalogManager;
+import org.apache.flink.table.catalog.FunctionCatalog;
+import org.apache.flink.table.client.config.Environment;
+import org.apache.flink.table.client.gateway.SqlExecutionException;
+import org.apache.flink.table.delegation.Executor;
+import org.apache.flink.table.delegation.ExecutorFactory;
+import org.apache.flink.table.delegation.Planner;
+import org.apache.flink.table.delegation.PlannerFactory;
+import org.apache.flink.table.factories.ComponentFactoryService;
+import org.apache.flink.table.module.ModuleManager;
+import org.apache.flink.util.TemporaryClassLoaderContext;
+
+import java.lang.reflect.Method;
+import java.net.URLClassLoader;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.Supplier;
+
+import static org.apache.flink.table.client.gateway.context.SessionContext.SessionState;
+
+/**
+ * Context for executing table programs. This class caches everything that can be cached across
+ * multiple queries as long as the session context does not change. This must be thread-safe as it
+ * might be reused across different query submissions.
+ */
+public class ExecutionContext {
Review comment:
Add a TODO for future 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
}, {
"hash" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"triggerType" : "PUSH"
}, {
"hash" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736",
"triggerID" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"triggerType" : "PUSH"
}, {
"hash" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13813",
"triggerID" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"triggerType" : "PUSH"
}, {
"hash" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14060",
"triggerID" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"triggerType" : "PUSH"
}, {
"hash" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14126",
"triggerID" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"triggerType" : "PUSH"
}, {
"hash" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14133",
"triggerID" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"triggerType" : "PUSH"
}, {
"hash" : "83446f3a0e26473892f97f68de12b5f4a8e50cc5",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14434",
"triggerID" : "83446f3a0e26473892f97f68de12b5f4a8e50cc5",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4cd5563d44a62aec2b8db80cfe8da650fddf4ffa UNKNOWN
* 83446f3a0e26473892f97f68de12b5f4a8e50cc5 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14434)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [Flink-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
}, {
"hash" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"triggerType" : "PUSH"
}, {
"hash" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736",
"triggerID" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"triggerType" : "PUSH"
}, {
"hash" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13813",
"triggerID" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"triggerType" : "PUSH"
}, {
"hash" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14060",
"triggerID" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"triggerType" : "PUSH"
}, {
"hash" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14126",
"triggerID" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"triggerType" : "PUSH"
}, {
"hash" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14133",
"triggerID" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"triggerType" : "PUSH"
}, {
"hash" : "83446f3a0e26473892f97f68de12b5f4a8e50cc5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14434",
"triggerID" : "83446f3a0e26473892f97f68de12b5f4a8e50cc5",
"triggerType" : "PUSH"
}, {
"hash" : "5fb5449d10b57a84f6da97e0be6fc875a46344cf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14442",
"triggerID" : "5fb5449d10b57a84f6da97e0be6fc875a46344cf",
"triggerType" : "PUSH"
}, {
"hash" : "9e86f4b592c7699fc0fade586d99daa907b9bffc",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14465",
"triggerID" : "9e86f4b592c7699fc0fade586d99daa907b9bffc",
"triggerType" : "PUSH"
}, {
"hash" : "741114ff56881d722e10008529ec3e60a94d8431",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14481",
"triggerID" : "741114ff56881d722e10008529ec3e60a94d8431",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4cd5563d44a62aec2b8db80cfe8da650fddf4ffa UNKNOWN
* 9e86f4b592c7699fc0fade586d99daa907b9bffc Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14465)
* 741114ff56881d722e10008529ec3e60a94d8431 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14481)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] wuchong commented on a change in pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
wuchong commented on a change in pull request #15006:
URL: https://github.com/apache/flink/pull/15006#discussion_r591160927
##########
File path: flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/context/SessionContext.java
##########
@@ -0,0 +1,269 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.client.gateway.context;
+
+import org.apache.flink.client.ClientUtils;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogManager;
+import org.apache.flink.table.catalog.FunctionCatalog;
+import org.apache.flink.table.catalog.GenericInMemoryCatalog;
+import org.apache.flink.table.client.config.Environment;
+import org.apache.flink.table.client.gateway.Executor;
+import org.apache.flink.table.client.gateway.SqlExecutionException;
+import org.apache.flink.table.module.ModuleManager;
+import org.apache.flink.util.TemporaryClassLoaderContext;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URLClassLoader;
+import java.util.Collections;
+import java.util.Objects;
+
+/**
+ * Context describing a session, it's mainly used for user to open a new session in the backend. If
+ * client request to open a new session, the backend {@link Executor} will maintain the session
+ * context map util users close it.
+ */
+public class SessionContext {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SessionContext.class);
+
+ private final String sessionId;
+ private final DefaultContext defaultContext;
+
+ private Environment sessionEnv;
+ private final Configuration sessionConfiguration;
+
+ private final SessionState sessionState;
+ private final URLClassLoader classLoader;
+ private ExecutionContext executionContext;
+
+ public SessionContext(
+ DefaultContext defaultContext,
+ String sessionId,
+ Environment sessionEnv,
+ Configuration sessionConfiguration,
+ URLClassLoader classLoader,
+ SessionState sessionState,
+ ExecutionContext executionContext) {
+ this.defaultContext = defaultContext;
+ this.sessionId = sessionId;
+ this.sessionEnv = sessionEnv;
+ this.sessionConfiguration = sessionConfiguration;
+ this.classLoader = classLoader;
+ this.sessionState = sessionState;
+ this.executionContext = executionContext;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Getter method
+ // --------------------------------------------------------------------------------------------
+
+ public String getSessionId() {
+ return this.sessionId;
+ }
+
+ public Environment getSessionEnvironment() {
+ return this.sessionEnv;
+ }
+
+ public ExecutionContext getExecutionContext() {
+ return this.executionContext;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Method to execute commands
+ // --------------------------------------------------------------------------------------------
+
+ /** Reset properties to default. It will rebuild a new {@link ExecutionContext}. */
+ public void reset() {
+ sessionEnv = defaultContext.getDefaultEnv().clone();
+ for (String key : sessionConfiguration.toMap().keySet()) {
+ // Don't care the type of the option
+ ConfigOption<String> keyToDelete = ConfigOptions.key(key).stringType().noDefaultValue();
+ sessionConfiguration.removeConfig(keyToDelete);
+ }
+ sessionConfiguration.addAll(defaultContext.getFlinkConfig());
Review comment:
I see, could you add comment to mention this?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] wuchong commented on a change in pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
wuchong commented on a change in pull request #15006:
URL: https://github.com/apache/flink/pull/15006#discussion_r592268616
##########
File path: flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/context/SessionContextTest.java
##########
@@ -119,20 +129,27 @@ public void testSetAndResetOption() throws Exception {
.asMap()
.get("pipeline.name"));
Assert.assertEquals(
- (long) DEFAULT_PARALLELISM.defaultValue(),
+ 16,
sessionContext
.getExecutionContext()
.getTableEnvironment()
.getConfig()
.getConfiguration()
- .getInteger(DEFAULT_PARALLELISM));
+ .getInteger(MAX_PARALLELISM));
Assert.assertNull(
sessionContext
.getExecutionContext()
.getTableEnvironment()
.getConfig()
.getConfiguration()
.getString(NAME, null));
+ Assert.assertTrue(
Review comment:
nit: ditto.
##########
File path: flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/context/SessionContextTest.java
##########
@@ -119,20 +129,27 @@ public void testSetAndResetOption() throws Exception {
.asMap()
.get("pipeline.name"));
Assert.assertEquals(
- (long) DEFAULT_PARALLELISM.defaultValue(),
+ 16,
Review comment:
nit: would be better to add comment to explain why it fallback to 16.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
}, {
"hash" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"triggerType" : "PUSH"
}, {
"hash" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736",
"triggerID" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"triggerType" : "PUSH"
}, {
"hash" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13813",
"triggerID" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"triggerType" : "PUSH"
}, {
"hash" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14060",
"triggerID" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"triggerType" : "PUSH"
}, {
"hash" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14126",
"triggerID" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"triggerType" : "PUSH"
}, {
"hash" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14133",
"triggerID" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4cd5563d44a62aec2b8db80cfe8da650fddf4ffa UNKNOWN
* f13e54215fb6fbf668f8a51ccbec7f62d835a542 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14060)
* 840b4cf15689b6bf7b7561172a66d4b3571c0745 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14126)
* f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14133)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [Flink-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fsk119 edited a comment on pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
fsk119 edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-797312429
The failed test is because of the network and it passes the test in the test before.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] wuchong merged pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
wuchong merged pull request #15006:
URL: https://github.com/apache/flink/pull/15006
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] wuchong commented on a change in pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
wuchong commented on a change in pull request #15006:
URL: https://github.com/apache/flink/pull/15006#discussion_r589163832
##########
File path: flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/context/DefaultContext.java
##########
@@ -0,0 +1,296 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.client.gateway.context;
+
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.client.cli.CliArgsException;
+import org.apache.flink.client.cli.CliFrontendParser;
+import org.apache.flink.client.cli.CustomCommandLine;
+import org.apache.flink.client.cli.ExecutionConfigAccessor;
+import org.apache.flink.client.cli.ProgramOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.configuration.PipelineOptions;
+import org.apache.flink.configuration.RestartStrategyOptions;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.plugin.PluginUtils;
+import org.apache.flink.streaming.api.TimeCharacteristic;
+import org.apache.flink.streaming.api.environment.StreamPipelineOptions;
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
+import org.apache.flink.table.client.config.Environment;
+import org.apache.flink.table.client.config.entries.DeploymentEntry;
+import org.apache.flink.table.client.config.entries.ExecutionEntry;
+import org.apache.flink.table.client.gateway.Executor;
+import org.apache.flink.table.client.gateway.SqlExecutionException;
+import org.apache.flink.table.descriptors.FunctionDescriptorValidator;
+import org.apache.flink.util.FlinkException;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.Options;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.nio.file.Paths;
+import java.time.Duration;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Context describing default environment, command line options, flink config, etc.
+ *
+ * <p>When the {@link Executor} execute `reset` commands, the session can restore from the "default"
+ * context.
+ */
+public class DefaultContext {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultContext.class);
+
+ private final Environment defaultEnv;
+ private final List<URL> dependencies;
+ private final Configuration flinkConfig;
+
+ public DefaultContext(
+ Environment defaultEnv,
+ List<URL> dependencies,
+ Configuration flinkConfig,
+ List<CustomCommandLine> commandLines) {
+ this.defaultEnv = defaultEnv;
+ this.dependencies = dependencies;
+ this.flinkConfig = flinkConfig;
+ Options commandLineOptions = collectCommandLineOptions(commandLines);
+
+ // initialize default file system
+ FileSystem.initialize(
+ flinkConfig, PluginUtils.createPluginManagerFromRootFolder(flinkConfig));
+
+ // add python dependencies
+ if (containsPythonFunction(defaultEnv)) {
+ addPythonDependency();
+ }
+
+ // put environment entry into Configuration
+ initConfigurationFromConfigurationEntry();
Review comment:
`initConfigurationFromEnvironment();`
##########
File path: flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableConfig.java
##########
@@ -75,7 +75,15 @@
private MathContext decimalContext = MathContext.DECIMAL128;
/** A configuration object to hold all key/value configuration. */
- private Configuration configuration = new Configuration();
+ private Configuration configuration;
Review comment:
revert changes to this file.
##########
File path: flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/context/LegacyTableEnvironmentInitializer.java
##########
@@ -0,0 +1,388 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.client.gateway.context;
+
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.internal.TableEnvironmentInternal;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogTableImpl;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.client.config.Environment;
+import org.apache.flink.table.client.config.entries.SinkTableEntry;
+import org.apache.flink.table.client.config.entries.SourceSinkTableEntry;
+import org.apache.flink.table.client.config.entries.SourceTableEntry;
+import org.apache.flink.table.client.config.entries.TemporalTableEntry;
+import org.apache.flink.table.client.config.entries.ViewEntry;
+import org.apache.flink.table.client.gateway.SqlExecutionException;
+import org.apache.flink.table.descriptors.CoreModuleDescriptorValidator;
+import org.apache.flink.table.factories.BatchTableSinkFactory;
+import org.apache.flink.table.factories.BatchTableSourceFactory;
+import org.apache.flink.table.factories.CatalogFactory;
+import org.apache.flink.table.factories.ModuleFactory;
+import org.apache.flink.table.factories.TableFactoryService;
+import org.apache.flink.table.factories.TableSinkFactory;
+import org.apache.flink.table.factories.TableSinkFactoryContextImpl;
+import org.apache.flink.table.factories.TableSourceFactory;
+import org.apache.flink.table.factories.TableSourceFactoryContextImpl;
+import org.apache.flink.table.functions.AggregateFunction;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.functions.FunctionService;
+import org.apache.flink.table.functions.ScalarFunction;
+import org.apache.flink.table.functions.TableFunction;
+import org.apache.flink.table.functions.UserDefinedFunction;
+import org.apache.flink.table.module.Module;
+import org.apache.flink.table.sinks.TableSink;
+import org.apache.flink.table.sources.TableSource;
+import org.apache.flink.util.TemporaryClassLoaderContext;
+
+import java.net.URLClassLoader;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Utils to initialize {@link TableEnvironment} from {@link Environment}. */
+public class LegacyTableEnvironmentInitializer {
Review comment:
Add a comment that this file can be removed once we don't support `sql-client.yaml`.
##########
File path: flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/context/SessionContextTest.java
##########
@@ -0,0 +1,122 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.client.gateway.context;
+
+import org.apache.flink.client.cli.DefaultCLI;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.client.config.Environment;
+import org.apache.flink.table.client.gateway.utils.EnvironmentFileUtil;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.flink.table.api.config.TableConfigOptions.TABLE_SQL_DIALECT;
+
+/** Test {@link SessionContext}. */
+public class SessionContextTest {
+
+ private static final String DEFAULTS_ENVIRONMENT_FILE = "test-sql-client-defaults.yaml";
+
+ @Test
+ public void testSetAndResetYamlKey() throws Exception {
+ SessionContext sessionContext = createSessionContext();
+ sessionContext.set("execution.max-table-result-rows", "100000");
+ Assert.assertEquals(
+ "100000",
+ sessionContext
+ .getSessionEnvironment()
+ .getExecution()
+ .asMap()
+ .get("max-table-result-rows"));
+
+ sessionContext.reset();
+
+ Assert.assertEquals(
+ "100",
Review comment:
What's the influence scope of `reset` ? Should it reset all configuration to default value, or rollback to initial value defined in `sql-client-defaults.yaml` or `init.sql`?
If it reset to default value, then the expected value of `max-table-result-rows` should be `1000000`.
##########
File path: flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/context/SessionContext.java
##########
@@ -0,0 +1,269 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.client.gateway.context;
+
+import org.apache.flink.client.ClientUtils;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogManager;
+import org.apache.flink.table.catalog.FunctionCatalog;
+import org.apache.flink.table.catalog.GenericInMemoryCatalog;
+import org.apache.flink.table.client.config.Environment;
+import org.apache.flink.table.client.gateway.Executor;
+import org.apache.flink.table.client.gateway.SqlExecutionException;
+import org.apache.flink.table.module.ModuleManager;
+import org.apache.flink.util.TemporaryClassLoaderContext;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URLClassLoader;
+import java.util.Collections;
+import java.util.Objects;
+
+/**
+ * Context describing a session, it's mainly used for user to open a new session in the backend. If
+ * client request to open a new session, the backend {@link Executor} will maintain the session
+ * context map util users close it.
+ */
+public class SessionContext {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SessionContext.class);
+
+ private final String sessionId;
+ private final DefaultContext defaultContext;
+
+ private Environment sessionEnv;
+ private final Configuration sessionConfiguration;
+
+ private final SessionState sessionState;
+ private final URLClassLoader classLoader;
+ private ExecutionContext executionContext;
+
+ public SessionContext(
+ DefaultContext defaultContext,
+ String sessionId,
+ Environment sessionEnv,
+ Configuration sessionConfiguration,
+ URLClassLoader classLoader,
+ SessionState sessionState,
+ ExecutionContext executionContext) {
+ this.defaultContext = defaultContext;
+ this.sessionId = sessionId;
+ this.sessionEnv = sessionEnv;
+ this.sessionConfiguration = sessionConfiguration;
+ this.classLoader = classLoader;
+ this.sessionState = sessionState;
+ this.executionContext = executionContext;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Getter method
+ // --------------------------------------------------------------------------------------------
+
+ public String getSessionId() {
+ return this.sessionId;
+ }
+
+ public Environment getSessionEnvironment() {
+ return this.sessionEnv;
+ }
+
+ public ExecutionContext getExecutionContext() {
+ return this.executionContext;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Method to execute commands
+ // --------------------------------------------------------------------------------------------
+
+ /** Reset properties to default. It will rebuild a new {@link ExecutionContext}. */
+ public void reset() {
+ sessionEnv = defaultContext.getDefaultEnv().clone();
+ for (String key : sessionConfiguration.toMap().keySet()) {
+ // Don't care the type of the option
+ ConfigOption<String> keyToDelete = ConfigOptions.key(key).stringType().noDefaultValue();
+ sessionConfiguration.removeConfig(keyToDelete);
+ }
+ sessionConfiguration.addAll(defaultContext.getFlinkConfig());
Review comment:
What about changing `sessionConfiguration` to not-final? We can create a new configuration for the session here, `new Configuration(defaultContext.getFlinkConfig())`. This would be much more straightforward. I think it's fine to declare not-final here, as other session varaible (e.g. `executionContext`) are also not-final.
##########
File path: flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/context/ExecutionContext.java
##########
@@ -0,0 +1,233 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.client.gateway.context;
+
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.TimeCharacteristic;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.bridge.java.internal.BatchTableEnvironmentImpl;
+import org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl;
+import org.apache.flink.table.catalog.CatalogManager;
+import org.apache.flink.table.catalog.FunctionCatalog;
+import org.apache.flink.table.client.config.Environment;
+import org.apache.flink.table.client.gateway.SqlExecutionException;
+import org.apache.flink.table.delegation.Executor;
+import org.apache.flink.table.delegation.ExecutorFactory;
+import org.apache.flink.table.delegation.Planner;
+import org.apache.flink.table.delegation.PlannerFactory;
+import org.apache.flink.table.factories.ComponentFactoryService;
+import org.apache.flink.table.module.ModuleManager;
+import org.apache.flink.util.TemporaryClassLoaderContext;
+
+import java.lang.reflect.Method;
+import java.net.URLClassLoader;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.Supplier;
+
+import static org.apache.flink.table.client.gateway.context.SessionContext.SessionState;
+
+/**
+ * Context for executing table programs. This class caches everything that can be cached across
+ * multiple queries as long as the session context does not change. This must be thread-safe as it
+ * might be reused across different query submissions.
+ */
+public class ExecutionContext {
Review comment:
I'm wondering `ExecutionContext` is also not needed anymore. The only thing in `ExecutionContext` been used by other components is the `classloader` and the `tableEnv`. However, IMO, `classloader` is a more session variable which should be get from `SessionContext` (e.g. we may support `ADD JAR` to add jar dependency to the session in the future). Then `ExecutionContext` would be just a wrapper of `TableEnvironment`.
If this is the case, what do you think about createing a new `TableEnvironment` instead of `ExecutionContext` in `SessionContext`.?
##########
File path: flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/context/SessionContext.java
##########
@@ -0,0 +1,269 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.client.gateway.context;
+
+import org.apache.flink.client.ClientUtils;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogManager;
+import org.apache.flink.table.catalog.FunctionCatalog;
+import org.apache.flink.table.catalog.GenericInMemoryCatalog;
+import org.apache.flink.table.client.config.Environment;
+import org.apache.flink.table.client.gateway.Executor;
+import org.apache.flink.table.client.gateway.SqlExecutionException;
+import org.apache.flink.table.module.ModuleManager;
+import org.apache.flink.util.TemporaryClassLoaderContext;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URLClassLoader;
+import java.util.Collections;
+import java.util.Objects;
+
+/**
+ * Context describing a session, it's mainly used for user to open a new session in the backend. If
+ * client request to open a new session, the backend {@link Executor} will maintain the session
+ * context map util users close it.
+ */
+public class SessionContext {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SessionContext.class);
+
+ private final String sessionId;
+ private final DefaultContext defaultContext;
+
+ private Environment sessionEnv;
+ private final Configuration sessionConfiguration;
+
+ private final SessionState sessionState;
+ private final URLClassLoader classLoader;
+ private ExecutionContext executionContext;
+
+ public SessionContext(
Review comment:
We can make it `private`, so users can all use `create` method for initialization.
##########
File path: flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/context/SessionContext.java
##########
@@ -0,0 +1,269 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.client.gateway.context;
+
+import org.apache.flink.client.ClientUtils;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogManager;
+import org.apache.flink.table.catalog.FunctionCatalog;
+import org.apache.flink.table.catalog.GenericInMemoryCatalog;
+import org.apache.flink.table.client.config.Environment;
+import org.apache.flink.table.client.gateway.Executor;
+import org.apache.flink.table.client.gateway.SqlExecutionException;
+import org.apache.flink.table.module.ModuleManager;
+import org.apache.flink.util.TemporaryClassLoaderContext;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URLClassLoader;
+import java.util.Collections;
+import java.util.Objects;
+
+/**
+ * Context describing a session, it's mainly used for user to open a new session in the backend. If
+ * client request to open a new session, the backend {@link Executor} will maintain the session
+ * context map util users close it.
+ */
+public class SessionContext {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SessionContext.class);
+
+ private final String sessionId;
+ private final DefaultContext defaultContext;
+
+ private Environment sessionEnv;
+ private final Configuration sessionConfiguration;
+
+ private final SessionState sessionState;
+ private final URLClassLoader classLoader;
+ private ExecutionContext executionContext;
+
+ public SessionContext(
+ DefaultContext defaultContext,
+ String sessionId,
+ Environment sessionEnv,
+ Configuration sessionConfiguration,
+ URLClassLoader classLoader,
+ SessionState sessionState,
+ ExecutionContext executionContext) {
+ this.defaultContext = defaultContext;
+ this.sessionId = sessionId;
+ this.sessionEnv = sessionEnv;
+ this.sessionConfiguration = sessionConfiguration;
+ this.classLoader = classLoader;
+ this.sessionState = sessionState;
+ this.executionContext = executionContext;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Getter method
+ // --------------------------------------------------------------------------------------------
+
+ public String getSessionId() {
+ return this.sessionId;
+ }
+
+ public Environment getSessionEnvironment() {
+ return this.sessionEnv;
+ }
+
+ public ExecutionContext getExecutionContext() {
+ return this.executionContext;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Method to execute commands
+ // --------------------------------------------------------------------------------------------
+
+ /** Reset properties to default. It will rebuild a new {@link ExecutionContext}. */
+ public void reset() {
+ sessionEnv = defaultContext.getDefaultEnv().clone();
+ for (String key : sessionConfiguration.toMap().keySet()) {
+ // Don't care the type of the option
+ ConfigOption<String> keyToDelete = ConfigOptions.key(key).stringType().noDefaultValue();
+ sessionConfiguration.removeConfig(keyToDelete);
+ }
+ sessionConfiguration.addAll(defaultContext.getFlinkConfig());
+ executionContext = new ExecutionContext(sessionEnv, executionContext);
+ }
+
+ /** Set properties. It will rebuild a new {@link ExecutionContext} */
+ public void set(String key, String value) {
+ // put key-value into the Environment
+ try {
+ this.sessionEnv = Environment.enrich(sessionEnv, Collections.singletonMap(key, value));
+ sessionConfiguration.setString(key, value);
+ } catch (Throwable t) {
+ throw new SqlExecutionException("Could not set session property.", t);
+ }
+
+ // Renew the ExecutionContext by new environment.
+ // Book keep all the session states of current ExecutionContext then
+ // re-register them into the new one.
+ this.executionContext = new ExecutionContext(sessionEnv, executionContext);
+ }
+
+ /** Close resources, e.g. catalogs. */
+ public void close() {
+ try (TemporaryClassLoaderContext ignored = TemporaryClassLoaderContext.of(classLoader)) {
+ for (String name : sessionState.catalogManager.listCatalogs()) {
+ sessionState.catalogManager.getCatalog(name).ifPresent(Catalog::close);
+ }
+ }
+ try {
+ classLoader.close();
+ } catch (IOException e) {
+ LOG.debug("Error while closing class loader.", e);
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Helper method to create
+ // --------------------------------------------------------------------------------------------
+
+ public static SessionContext create(DefaultContext defaultContext, String sessionId) {
+ // --------------------------------------------------------------------------------------------------------------
+ // Init config
+ // --------------------------------------------------------------------------------------------------------------
+
+ Environment sessionEnv = defaultContext.getDefaultEnv().clone();
+ Configuration configuration = defaultContext.getFlinkConfig().clone();
+
+ // --------------------------------------------------------------------------------------------------------------
+ // Init classloader
+ // --------------------------------------------------------------------------------------------------------------
+
+ URLClassLoader classLoader =
+ ClientUtils.buildUserCodeClassLoader(
+ defaultContext.getDependencies(),
+ Collections.emptyList(),
+ SessionContext.class.getClassLoader(),
+ configuration);
+
+ // --------------------------------------------------------------------------------------------------------------
+ // Init session state
+ // --------------------------------------------------------------------------------------------------------------
+
+ ModuleManager moduleManager = new ModuleManager();
+
+ final EnvironmentSettings settings = sessionEnv.getExecution().getEnvironmentSettings();
+
+ CatalogManager catalogManager =
+ CatalogManager.newBuilder()
+ .classLoader(classLoader)
+ .config(configuration)
+ .defaultCatalog(
+ settings.getBuiltInCatalogName(),
+ new GenericInMemoryCatalog(
+ settings.getBuiltInCatalogName(),
+ settings.getBuiltInDatabaseName()))
+ .build();
+
+ FunctionCatalog functionCatalog =
+ new FunctionCatalog(configuration, catalogManager, moduleManager);
+ SessionState sessionState =
+ new SessionState(catalogManager, moduleManager, functionCatalog);
+
+ // --------------------------------------------------------------------------------------------------------------
+ // Init
+ // --------------------------------------------------------------------------------------------------------------
+
+ return new SessionContext(
+ defaultContext,
+ sessionId,
+ sessionEnv,
+ configuration,
+ classLoader,
+ sessionState,
+ new ExecutionContext(sessionEnv, configuration, classLoader, sessionState));
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof SessionContext)) {
+ return false;
+ }
+ SessionContext context = (SessionContext) o;
+ return Objects.equals(sessionId, context.sessionId)
+ && Objects.equals(defaultContext, context.defaultContext)
+ && Objects.equals(sessionEnv, context.sessionEnv)
+ && Objects.equals(sessionConfiguration, context.sessionConfiguration)
+ && Objects.equals(classLoader, context.classLoader)
+ && Objects.equals(sessionState, context.sessionState)
+ && Objects.equals(executionContext, context.executionContext);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ sessionId,
+ defaultContext,
+ sessionEnv,
+ sessionConfiguration,
+ classLoader,
+ sessionState,
+ executionContext);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Inner class
+ // --------------------------------------------------------------------------------------------
+
+ /** session state. */
+ public static class SessionState {
+
+ public final CatalogManager catalogManager;
+ public final FunctionCatalog functionCatalog;
+ public final ModuleManager moduleManager;
+
+ public SessionState(
+ CatalogManager catalogManager,
+ ModuleManager moduleManager,
+ FunctionCatalog functionCatalog) {
+ this.catalogManager = catalogManager;
+ this.moduleManager = moduleManager;
+ this.functionCatalog = functionCatalog;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
Review comment:
Do we need this?
##########
File path: flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/context/SessionContextTest.java
##########
@@ -0,0 +1,122 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.client.gateway.context;
+
+import org.apache.flink.client.cli.DefaultCLI;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.client.config.Environment;
+import org.apache.flink.table.client.gateway.utils.EnvironmentFileUtil;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.flink.table.api.config.TableConfigOptions.TABLE_SQL_DIALECT;
+
+/** Test {@link SessionContext}. */
+public class SessionContextTest {
+
+ private static final String DEFAULTS_ENVIRONMENT_FILE = "test-sql-client-defaults.yaml";
+
+ @Test
+ public void testSetAndResetYamlKey() throws Exception {
+ SessionContext sessionContext = createSessionContext();
+ sessionContext.set("execution.max-table-result-rows", "100000");
+ Assert.assertEquals(
+ "100000",
+ sessionContext
+ .getSessionEnvironment()
+ .getExecution()
+ .asMap()
+ .get("max-table-result-rows"));
+
+ sessionContext.reset();
+
+ Assert.assertEquals(
+ "100",
+ sessionContext
+ .getSessionEnvironment()
+ .getExecution()
+ .asMap()
+ .get("max-table-result-rows"));
+ }
+
+ @Test
+ public void testSetAndResetOption() throws Exception {
Review comment:
Could you try to set and reset `CoreOptions#DEFAULT_PARALLELISM`?
##########
File path: flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/SqlClient.java
##########
@@ -91,17 +102,17 @@ private void start() {
} else {
libDirs = Collections.emptyList();
}
- final Executor executor = new LocalExecutor(options.getDefaults(), jars, libDirs);
+ final Executor executor = new LocalExecutor();
executor.start();
- // create CLI client with session environment
- final Environment sessionEnv = readSessionEnvironment(options.getEnvironment());
- appendPythonConfig(sessionEnv, options.getPythonConfiguration());
+ // build default context
+ DefaultContext defaultContext = buildEmbeddedDefaultContext(jars, libDirs);
Review comment:
Personally, I think `buildEmbeddedDefaultContext` should belong to `LocalExecutor`. `SqlClient` shouldn't care about how to build context, it just need to setup Executor (`LocalExecutor`, or `GatewayExecutor` in the future).
I think we can also improve the interface of `Executor`, changing `openSession` to `openSession(String sessionId)` and create `SessionContext` in it.
##########
File path: flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/context/SessionContext.java
##########
@@ -0,0 +1,269 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.client.gateway.context;
+
+import org.apache.flink.client.ClientUtils;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogManager;
+import org.apache.flink.table.catalog.FunctionCatalog;
+import org.apache.flink.table.catalog.GenericInMemoryCatalog;
+import org.apache.flink.table.client.config.Environment;
+import org.apache.flink.table.client.gateway.Executor;
+import org.apache.flink.table.client.gateway.SqlExecutionException;
+import org.apache.flink.table.module.ModuleManager;
+import org.apache.flink.util.TemporaryClassLoaderContext;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URLClassLoader;
+import java.util.Collections;
+import java.util.Objects;
+
+/**
+ * Context describing a session, it's mainly used for user to open a new session in the backend. If
+ * client request to open a new session, the backend {@link Executor} will maintain the session
+ * context map util users close it.
+ */
+public class SessionContext {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SessionContext.class);
+
+ private final String sessionId;
+ private final DefaultContext defaultContext;
+
+ private Environment sessionEnv;
+ private final Configuration sessionConfiguration;
+
+ private final SessionState sessionState;
+ private final URLClassLoader classLoader;
+ private ExecutionContext executionContext;
+
+ public SessionContext(
+ DefaultContext defaultContext,
+ String sessionId,
+ Environment sessionEnv,
+ Configuration sessionConfiguration,
+ URLClassLoader classLoader,
+ SessionState sessionState,
+ ExecutionContext executionContext) {
+ this.defaultContext = defaultContext;
+ this.sessionId = sessionId;
+ this.sessionEnv = sessionEnv;
+ this.sessionConfiguration = sessionConfiguration;
+ this.classLoader = classLoader;
+ this.sessionState = sessionState;
+ this.executionContext = executionContext;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Getter method
+ // --------------------------------------------------------------------------------------------
+
+ public String getSessionId() {
+ return this.sessionId;
+ }
+
+ public Environment getSessionEnvironment() {
+ return this.sessionEnv;
+ }
+
+ public ExecutionContext getExecutionContext() {
+ return this.executionContext;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Method to execute commands
+ // --------------------------------------------------------------------------------------------
+
+ /** Reset properties to default. It will rebuild a new {@link ExecutionContext}. */
+ public void reset() {
+ sessionEnv = defaultContext.getDefaultEnv().clone();
+ for (String key : sessionConfiguration.toMap().keySet()) {
+ // Don't care the type of the option
+ ConfigOption<String> keyToDelete = ConfigOptions.key(key).stringType().noDefaultValue();
+ sessionConfiguration.removeConfig(keyToDelete);
+ }
+ sessionConfiguration.addAll(defaultContext.getFlinkConfig());
+ executionContext = new ExecutionContext(sessionEnv, executionContext);
+ }
+
+ /** Set properties. It will rebuild a new {@link ExecutionContext} */
+ public void set(String key, String value) {
+ // put key-value into the Environment
+ try {
+ this.sessionEnv = Environment.enrich(sessionEnv, Collections.singletonMap(key, value));
+ sessionConfiguration.setString(key, value);
+ } catch (Throwable t) {
+ throw new SqlExecutionException("Could not set session property.", t);
+ }
+
+ // Renew the ExecutionContext by new environment.
+ // Book keep all the session states of current ExecutionContext then
+ // re-register them into the new one.
+ this.executionContext = new ExecutionContext(sessionEnv, executionContext);
+ }
+
+ /** Close resources, e.g. catalogs. */
+ public void close() {
+ try (TemporaryClassLoaderContext ignored = TemporaryClassLoaderContext.of(classLoader)) {
+ for (String name : sessionState.catalogManager.listCatalogs()) {
+ sessionState.catalogManager.getCatalog(name).ifPresent(Catalog::close);
+ }
+ }
+ try {
+ classLoader.close();
+ } catch (IOException e) {
+ LOG.debug("Error while closing class loader.", e);
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Helper method to create
+ // --------------------------------------------------------------------------------------------
+
+ public static SessionContext create(DefaultContext defaultContext, String sessionId) {
+ // --------------------------------------------------------------------------------------------------------------
+ // Init config
+ // --------------------------------------------------------------------------------------------------------------
+
+ Environment sessionEnv = defaultContext.getDefaultEnv().clone();
+ Configuration configuration = defaultContext.getFlinkConfig().clone();
+
+ // --------------------------------------------------------------------------------------------------------------
+ // Init classloader
+ // --------------------------------------------------------------------------------------------------------------
+
+ URLClassLoader classLoader =
+ ClientUtils.buildUserCodeClassLoader(
+ defaultContext.getDependencies(),
+ Collections.emptyList(),
+ SessionContext.class.getClassLoader(),
+ configuration);
+
+ // --------------------------------------------------------------------------------------------------------------
+ // Init session state
+ // --------------------------------------------------------------------------------------------------------------
+
+ ModuleManager moduleManager = new ModuleManager();
+
+ final EnvironmentSettings settings = sessionEnv.getExecution().getEnvironmentSettings();
+
+ CatalogManager catalogManager =
+ CatalogManager.newBuilder()
+ .classLoader(classLoader)
+ .config(configuration)
+ .defaultCatalog(
+ settings.getBuiltInCatalogName(),
+ new GenericInMemoryCatalog(
+ settings.getBuiltInCatalogName(),
+ settings.getBuiltInDatabaseName()))
+ .build();
+
+ FunctionCatalog functionCatalog =
+ new FunctionCatalog(configuration, catalogManager, moduleManager);
+ SessionState sessionState =
+ new SessionState(catalogManager, moduleManager, functionCatalog);
+
+ // --------------------------------------------------------------------------------------------------------------
+ // Init
+ // --------------------------------------------------------------------------------------------------------------
+
+ return new SessionContext(
+ defaultContext,
+ sessionId,
+ sessionEnv,
+ configuration,
+ classLoader,
+ sessionState,
+ new ExecutionContext(sessionEnv, configuration, classLoader, sessionState));
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof SessionContext)) {
+ return false;
+ }
+ SessionContext context = (SessionContext) o;
+ return Objects.equals(sessionId, context.sessionId)
+ && Objects.equals(defaultContext, context.defaultContext)
+ && Objects.equals(sessionEnv, context.sessionEnv)
+ && Objects.equals(sessionConfiguration, context.sessionConfiguration)
+ && Objects.equals(classLoader, context.classLoader)
Review comment:
Do we need `hashCode` and `equals` for `SessionContext`? I'm afraid it doesn't work to compare classloader directly.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fsk119 commented on a change in pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
fsk119 commented on a change in pull request #15006:
URL: https://github.com/apache/flink/pull/15006#discussion_r591109797
##########
File path: flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/context/SessionContext.java
##########
@@ -0,0 +1,269 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.client.gateway.context;
+
+import org.apache.flink.client.ClientUtils;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogManager;
+import org.apache.flink.table.catalog.FunctionCatalog;
+import org.apache.flink.table.catalog.GenericInMemoryCatalog;
+import org.apache.flink.table.client.config.Environment;
+import org.apache.flink.table.client.gateway.Executor;
+import org.apache.flink.table.client.gateway.SqlExecutionException;
+import org.apache.flink.table.module.ModuleManager;
+import org.apache.flink.util.TemporaryClassLoaderContext;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URLClassLoader;
+import java.util.Collections;
+import java.util.Objects;
+
+/**
+ * Context describing a session, it's mainly used for user to open a new session in the backend. If
+ * client request to open a new session, the backend {@link Executor} will maintain the session
+ * context map util users close it.
+ */
+public class SessionContext {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SessionContext.class);
+
+ private final String sessionId;
+ private final DefaultContext defaultContext;
+
+ private Environment sessionEnv;
+ private final Configuration sessionConfiguration;
+
+ private final SessionState sessionState;
+ private final URLClassLoader classLoader;
+ private ExecutionContext executionContext;
+
+ public SessionContext(
+ DefaultContext defaultContext,
+ String sessionId,
+ Environment sessionEnv,
+ Configuration sessionConfiguration,
+ URLClassLoader classLoader,
+ SessionState sessionState,
+ ExecutionContext executionContext) {
+ this.defaultContext = defaultContext;
+ this.sessionId = sessionId;
+ this.sessionEnv = sessionEnv;
+ this.sessionConfiguration = sessionConfiguration;
+ this.classLoader = classLoader;
+ this.sessionState = sessionState;
+ this.executionContext = executionContext;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Getter method
+ // --------------------------------------------------------------------------------------------
+
+ public String getSessionId() {
+ return this.sessionId;
+ }
+
+ public Environment getSessionEnvironment() {
+ return this.sessionEnv;
+ }
+
+ public ExecutionContext getExecutionContext() {
+ return this.executionContext;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Method to execute commands
+ // --------------------------------------------------------------------------------------------
+
+ /** Reset properties to default. It will rebuild a new {@link ExecutionContext}. */
+ public void reset() {
+ sessionEnv = defaultContext.getDefaultEnv().clone();
+ for (String key : sessionConfiguration.toMap().keySet()) {
+ // Don't care the type of the option
+ ConfigOption<String> keyToDelete = ConfigOptions.key(key).stringType().noDefaultValue();
+ sessionConfiguration.removeConfig(keyToDelete);
+ }
Review comment:
The main reasons is the `CatalogManager`, `FunctionManager` use `sessionConfiguration` . If we rebuild a new `Configuration`, we can 't modify the values of the option for `CatalogManager`, `FunctionManager` .
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
}, {
"hash" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"triggerType" : "PUSH"
}, {
"hash" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736",
"triggerID" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"triggerType" : "PUSH"
}, {
"hash" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13813",
"triggerID" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"triggerType" : "PUSH"
}, {
"hash" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14060",
"triggerID" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"triggerType" : "PUSH"
}, {
"hash" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14126",
"triggerID" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"triggerType" : "PUSH"
}, {
"hash" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4cd5563d44a62aec2b8db80cfe8da650fddf4ffa UNKNOWN
* f13e54215fb6fbf668f8a51ccbec7f62d835a542 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14060)
* 840b4cf15689b6bf7b7561172a66d4b3571c0745 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14126)
* f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
}, {
"hash" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"triggerType" : "PUSH"
}, {
"hash" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736",
"triggerID" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"triggerType" : "PUSH"
}, {
"hash" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13813",
"triggerID" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"triggerType" : "PUSH"
}, {
"hash" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14060",
"triggerID" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4cd5563d44a62aec2b8db80cfe8da650fddf4ffa UNKNOWN
* f13e54215fb6fbf668f8a51ccbec7f62d835a542 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14060)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
}, {
"hash" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"triggerType" : "PUSH"
}, {
"hash" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736",
"triggerID" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"triggerType" : "PUSH"
}, {
"hash" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4cd5563d44a62aec2b8db80cfe8da650fddf4ffa UNKNOWN
* 4cc39b300203dd122980e6ef22080746834fcc9b Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736)
* bc56c56211c178c4179bb850315eb47db5b76fc0 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
}, {
"hash" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"triggerType" : "PUSH"
}, {
"hash" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736",
"triggerID" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"triggerType" : "PUSH"
}, {
"hash" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13813",
"triggerID" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"triggerType" : "PUSH"
}, {
"hash" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4cd5563d44a62aec2b8db80cfe8da650fddf4ffa UNKNOWN
* bc56c56211c178c4179bb850315eb47db5b76fc0 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13813)
* f13e54215fb6fbf668f8a51ccbec7f62d835a542 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
}, {
"hash" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"triggerType" : "PUSH"
}, {
"hash" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736",
"triggerID" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"triggerType" : "PUSH"
}, {
"hash" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13813",
"triggerID" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"triggerType" : "PUSH"
}, {
"hash" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14060",
"triggerID" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"triggerType" : "PUSH"
}, {
"hash" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14126",
"triggerID" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4cd5563d44a62aec2b8db80cfe8da650fddf4ffa UNKNOWN
* f13e54215fb6fbf668f8a51ccbec7f62d835a542 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14060)
* 840b4cf15689b6bf7b7561172a66d4b3571c0745 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14126)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
}, {
"hash" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"triggerType" : "PUSH"
}, {
"hash" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736",
"triggerID" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"triggerType" : "PUSH"
}, {
"hash" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13813",
"triggerID" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"triggerType" : "PUSH"
}, {
"hash" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14060",
"triggerID" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"triggerType" : "PUSH"
}, {
"hash" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14126",
"triggerID" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"triggerType" : "PUSH"
}, {
"hash" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14133",
"triggerID" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"triggerType" : "PUSH"
}, {
"hash" : "83446f3a0e26473892f97f68de12b5f4a8e50cc5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14434",
"triggerID" : "83446f3a0e26473892f97f68de12b5f4a8e50cc5",
"triggerType" : "PUSH"
}, {
"hash" : "5fb5449d10b57a84f6da97e0be6fc875a46344cf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14442",
"triggerID" : "5fb5449d10b57a84f6da97e0be6fc875a46344cf",
"triggerType" : "PUSH"
}, {
"hash" : "9e86f4b592c7699fc0fade586d99daa907b9bffc",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14465",
"triggerID" : "9e86f4b592c7699fc0fade586d99daa907b9bffc",
"triggerType" : "PUSH"
}, {
"hash" : "741114ff56881d722e10008529ec3e60a94d8431",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14481",
"triggerID" : "741114ff56881d722e10008529ec3e60a94d8431",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4cd5563d44a62aec2b8db80cfe8da650fddf4ffa UNKNOWN
* 741114ff56881d722e10008529ec3e60a94d8431 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14481)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [Flink-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
}, {
"hash" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"triggerType" : "PUSH"
}, {
"hash" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704)
* 4cd5563d44a62aec2b8db80cfe8da650fddf4ffa UNKNOWN
* 4cc39b300203dd122980e6ef22080746834fcc9b UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
}, {
"hash" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"triggerType" : "PUSH"
}, {
"hash" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736",
"triggerID" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"triggerType" : "PUSH"
}, {
"hash" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13813",
"triggerID" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"triggerType" : "PUSH"
}, {
"hash" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14060",
"triggerID" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"triggerType" : "PUSH"
}, {
"hash" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14126",
"triggerID" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"triggerType" : "PUSH"
}, {
"hash" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14133",
"triggerID" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"triggerType" : "PUSH"
}, {
"hash" : "83446f3a0e26473892f97f68de12b5f4a8e50cc5",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14434",
"triggerID" : "83446f3a0e26473892f97f68de12b5f4a8e50cc5",
"triggerType" : "PUSH"
}, {
"hash" : "5fb5449d10b57a84f6da97e0be6fc875a46344cf",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14442",
"triggerID" : "5fb5449d10b57a84f6da97e0be6fc875a46344cf",
"triggerType" : "PUSH"
}, {
"hash" : "9e86f4b592c7699fc0fade586d99daa907b9bffc",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "9e86f4b592c7699fc0fade586d99daa907b9bffc",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4cd5563d44a62aec2b8db80cfe8da650fddf4ffa UNKNOWN
* 83446f3a0e26473892f97f68de12b5f4a8e50cc5 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14434)
* 5fb5449d10b57a84f6da97e0be6fc875a46344cf Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14442)
* 9e86f4b592c7699fc0fade586d99daa907b9bffc UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
}, {
"hash" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"triggerType" : "PUSH"
}, {
"hash" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736",
"triggerID" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"triggerType" : "PUSH"
}, {
"hash" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13813",
"triggerID" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"triggerType" : "PUSH"
}, {
"hash" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14060",
"triggerID" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"triggerType" : "PUSH"
}, {
"hash" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4cd5563d44a62aec2b8db80cfe8da650fddf4ffa UNKNOWN
* f13e54215fb6fbf668f8a51ccbec7f62d835a542 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14060)
* 840b4cf15689b6bf7b7561172a66d4b3571c0745 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #15006: [Flink-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785095017
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41 (Wed Feb 24 13:59:50 UTC 2021)
**Warnings:**
* **1 pom.xml files were touched**: Check for build and licensing issues.
* **This pull request references an unassigned [Jira ticket](https://issues.apache.org/jira/browse/FLINK-21485).** According to the [code contribution guide](https://flink.apache.org/contributing/contribute-code.html), tickets need to be assigned before starting with the implementation work.
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] godfreyhe commented on a change in pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
godfreyhe commented on a change in pull request #15006:
URL: https://github.com/apache/flink/pull/15006#discussion_r589133210
##########
File path: flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/context/SessionContext.java
##########
@@ -0,0 +1,269 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.client.gateway.context;
+
+import org.apache.flink.client.ClientUtils;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogManager;
+import org.apache.flink.table.catalog.FunctionCatalog;
+import org.apache.flink.table.catalog.GenericInMemoryCatalog;
+import org.apache.flink.table.client.config.Environment;
+import org.apache.flink.table.client.gateway.Executor;
+import org.apache.flink.table.client.gateway.SqlExecutionException;
+import org.apache.flink.table.module.ModuleManager;
+import org.apache.flink.util.TemporaryClassLoaderContext;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URLClassLoader;
+import java.util.Collections;
+import java.util.Objects;
+
+/**
+ * Context describing a session, it's mainly used for user to open a new session in the backend. If
+ * client request to open a new session, the backend {@link Executor} will maintain the session
+ * context map util users close it.
+ */
+public class SessionContext {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SessionContext.class);
+
+ private final String sessionId;
+ private final DefaultContext defaultContext;
+
+ private Environment sessionEnv;
+ private final Configuration sessionConfiguration;
+
+ private final SessionState sessionState;
+ private final URLClassLoader classLoader;
+ private ExecutionContext executionContext;
+
+ public SessionContext(
+ DefaultContext defaultContext,
+ String sessionId,
+ Environment sessionEnv,
+ Configuration sessionConfiguration,
+ URLClassLoader classLoader,
+ SessionState sessionState,
+ ExecutionContext executionContext) {
+ this.defaultContext = defaultContext;
+ this.sessionId = sessionId;
+ this.sessionEnv = sessionEnv;
+ this.sessionConfiguration = sessionConfiguration;
+ this.classLoader = classLoader;
+ this.sessionState = sessionState;
+ this.executionContext = executionContext;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Getter method
+ // --------------------------------------------------------------------------------------------
+
+ public String getSessionId() {
+ return this.sessionId;
+ }
+
+ public Environment getSessionEnvironment() {
+ return this.sessionEnv;
+ }
+
+ public ExecutionContext getExecutionContext() {
+ return this.executionContext;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Method to execute commands
+ // --------------------------------------------------------------------------------------------
+
+ /** Reset properties to default. It will rebuild a new {@link ExecutionContext}. */
+ public void reset() {
+ sessionEnv = defaultContext.getDefaultEnv().clone();
+ for (String key : sessionConfiguration.toMap().keySet()) {
+ // Don't care the type of the option
+ ConfigOption<String> keyToDelete = ConfigOptions.key(key).stringType().noDefaultValue();
+ sessionConfiguration.removeConfig(keyToDelete);
+ }
Review comment:
just re-initialize the `sessionConfiguration` based on `defaultContext` ?
`sessionConfiguration = new Configuration(defaultContext.getFlinkConfig())`
##########
File path: flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java
##########
@@ -409,20 +263,19 @@ public ResultDescriptor executeQuery(String sessionId, String query)
@Override
public void cancelQuery(String sessionId, String resultId) throws SqlExecutionException {
- final ExecutionContext<?> context = getExecutionContext(sessionId);
- cancelQueryInternal(context, resultId);
+ cancelQueryInternal(resultId);
}
@VisibleForTesting
List<ModuleEntry> listFullModules(String sessionId) throws SqlExecutionException {
- final ExecutionContext<?> context = getExecutionContext(sessionId);
+ final ExecutionContext context = getExecutionContext(sessionId);
final TableEnvironment tableEnv = context.getTableEnvironment();
return context.wrapClassLoader(() -> Arrays.asList(tableEnv.listFullModules()));
}
// --------------------------------------------------------------------------------------------
- private <T> void cancelQueryInternal(ExecutionContext<T> context, String resultId) {
+ private <T> void cancelQueryInternal(String resultId) {
Review comment:
`<T>` should be removed
##########
File path: flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/context/SessionContext.java
##########
@@ -0,0 +1,269 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.client.gateway.context;
+
+import org.apache.flink.client.ClientUtils;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogManager;
+import org.apache.flink.table.catalog.FunctionCatalog;
+import org.apache.flink.table.catalog.GenericInMemoryCatalog;
+import org.apache.flink.table.client.config.Environment;
+import org.apache.flink.table.client.gateway.Executor;
+import org.apache.flink.table.client.gateway.SqlExecutionException;
+import org.apache.flink.table.module.ModuleManager;
+import org.apache.flink.util.TemporaryClassLoaderContext;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URLClassLoader;
+import java.util.Collections;
+import java.util.Objects;
+
+/**
+ * Context describing a session, it's mainly used for user to open a new session in the backend. If
+ * client request to open a new session, the backend {@link Executor} will maintain the session
+ * context map util users close it.
+ */
+public class SessionContext {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SessionContext.class);
+
+ private final String sessionId;
+ private final DefaultContext defaultContext;
+
+ private Environment sessionEnv;
+ private final Configuration sessionConfiguration;
Review comment:
give some explanation why we need them both
##########
File path: flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/SqlClient.java
##########
@@ -121,6 +132,114 @@ private void start() {
}
}
+ private List<URL> discoverDependencies(List<URL> jars, List<URL> libraries) {
Review comment:
can we move these new methods into a utility class, this class has been too complex.
##########
File path: flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableConfig.java
##########
@@ -75,7 +75,15 @@
private MathContext decimalContext = MathContext.DECIMAL128;
/** A configuration object to hold all key/value configuration. */
- private Configuration configuration = new Configuration();
+ private Configuration configuration;
+
+ public TableConfig() {
+ this.configuration = new Configuration();
Review comment:
nit: call `this(new Configuration())`
##########
File path: flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/context/ExecutionContext.java
##########
@@ -0,0 +1,233 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.client.gateway.context;
+
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.TimeCharacteristic;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.bridge.java.internal.BatchTableEnvironmentImpl;
+import org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl;
+import org.apache.flink.table.catalog.CatalogManager;
+import org.apache.flink.table.catalog.FunctionCatalog;
+import org.apache.flink.table.client.config.Environment;
+import org.apache.flink.table.client.gateway.SqlExecutionException;
+import org.apache.flink.table.delegation.Executor;
+import org.apache.flink.table.delegation.ExecutorFactory;
+import org.apache.flink.table.delegation.Planner;
+import org.apache.flink.table.delegation.PlannerFactory;
+import org.apache.flink.table.factories.ComponentFactoryService;
+import org.apache.flink.table.module.ModuleManager;
+import org.apache.flink.util.TemporaryClassLoaderContext;
+
+import java.lang.reflect.Method;
+import java.net.URLClassLoader;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.Supplier;
+
+import static org.apache.flink.table.client.gateway.context.SessionContext.SessionState;
+
+/**
+ * Context for executing table programs. This class caches everything that can be cached across
+ * multiple queries as long as the session context does not change. This must be thread-safe as it
+ * might be reused across different query submissions.
+ */
+public class ExecutionContext {
+
+ // Members that should be reused in the same session.
+ private final Environment environment;
+ private final Configuration flinkConfig;
+ private final SessionState sessionState;
+ private final URLClassLoader classLoader;
+
+ private final TableEnvironment tableEnv;
+
+ public ExecutionContext(
+ Environment environment,
+ Configuration flinkConfig,
+ URLClassLoader classLoader,
+ SessionState sessionState) {
+ this.environment = environment;
+ this.flinkConfig = flinkConfig;
+ this.sessionState = sessionState;
+ this.classLoader = classLoader;
+ // Initialize the TableEnvironment.
+ this.tableEnv = createTableEnvironment();
+ LegacyTableEnvironmentInitializer.initializeTableEnvironmentFromEnvironment(
+ tableEnv, environment, classLoader);
+ }
+
+ /** Create a new {@link ExecutionContext} without initialization. */
+ public ExecutionContext(Environment environment, ExecutionContext context) {
Review comment:
I would like to remove this method, because the above method is more generic and the `initializeTableEnvironmentFromEnvironment` method will be called in `SessionContext#create`.
Actually, we can rename `initializeTableEnvironmentFromEnvironment` to `initializeSessionState`, because TableEnvironment instance will change with ExecutionContext
##########
File path: flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java
##########
@@ -441,7 +294,7 @@ public void cancelQuery(String sessionId, String resultId) throws SqlExecutionEx
}
private <C> ResultDescriptor executeQueryInternal(
Review comment:
`<C>` should be removed
##########
File path: flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/context/LegacyTableEnvironmentInitializer.java
##########
@@ -0,0 +1,388 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.client.gateway.context;
+
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.internal.TableEnvironmentInternal;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogTableImpl;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.client.config.Environment;
+import org.apache.flink.table.client.config.entries.SinkTableEntry;
+import org.apache.flink.table.client.config.entries.SourceSinkTableEntry;
+import org.apache.flink.table.client.config.entries.SourceTableEntry;
+import org.apache.flink.table.client.config.entries.TemporalTableEntry;
+import org.apache.flink.table.client.config.entries.ViewEntry;
+import org.apache.flink.table.client.gateway.SqlExecutionException;
+import org.apache.flink.table.descriptors.CoreModuleDescriptorValidator;
+import org.apache.flink.table.factories.BatchTableSinkFactory;
+import org.apache.flink.table.factories.BatchTableSourceFactory;
+import org.apache.flink.table.factories.CatalogFactory;
+import org.apache.flink.table.factories.ModuleFactory;
+import org.apache.flink.table.factories.TableFactoryService;
+import org.apache.flink.table.factories.TableSinkFactory;
+import org.apache.flink.table.factories.TableSinkFactoryContextImpl;
+import org.apache.flink.table.factories.TableSourceFactory;
+import org.apache.flink.table.factories.TableSourceFactoryContextImpl;
+import org.apache.flink.table.functions.AggregateFunction;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.functions.FunctionService;
+import org.apache.flink.table.functions.ScalarFunction;
+import org.apache.flink.table.functions.TableFunction;
+import org.apache.flink.table.functions.UserDefinedFunction;
+import org.apache.flink.table.module.Module;
+import org.apache.flink.table.sinks.TableSink;
+import org.apache.flink.table.sources.TableSource;
+import org.apache.flink.util.TemporaryClassLoaderContext;
+
+import java.net.URLClassLoader;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Utils to initialize {@link TableEnvironment} from {@link Environment}. */
+public class LegacyTableEnvironmentInitializer {
Review comment:
will we introduce `TableEnvironmentInitializer` for ddl ? if that, it's better we could introduce an interface for LegacyTableEnvironmentInitializer and TableEnvironmentInitializer
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fsk119 commented on a change in pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
fsk119 commented on a change in pull request #15006:
URL: https://github.com/apache/flink/pull/15006#discussion_r592024400
##########
File path: flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/context/LegacyTableEnvironmentInitializer.java
##########
@@ -0,0 +1,388 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.client.gateway.context;
+
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.internal.TableEnvironmentInternal;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogTableImpl;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.client.config.Environment;
+import org.apache.flink.table.client.config.entries.SinkTableEntry;
+import org.apache.flink.table.client.config.entries.SourceSinkTableEntry;
+import org.apache.flink.table.client.config.entries.SourceTableEntry;
+import org.apache.flink.table.client.config.entries.TemporalTableEntry;
+import org.apache.flink.table.client.config.entries.ViewEntry;
+import org.apache.flink.table.client.gateway.SqlExecutionException;
+import org.apache.flink.table.descriptors.CoreModuleDescriptorValidator;
+import org.apache.flink.table.factories.BatchTableSinkFactory;
+import org.apache.flink.table.factories.BatchTableSourceFactory;
+import org.apache.flink.table.factories.CatalogFactory;
+import org.apache.flink.table.factories.ModuleFactory;
+import org.apache.flink.table.factories.TableFactoryService;
+import org.apache.flink.table.factories.TableSinkFactory;
+import org.apache.flink.table.factories.TableSinkFactoryContextImpl;
+import org.apache.flink.table.factories.TableSourceFactory;
+import org.apache.flink.table.factories.TableSourceFactoryContextImpl;
+import org.apache.flink.table.functions.AggregateFunction;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.functions.FunctionService;
+import org.apache.flink.table.functions.ScalarFunction;
+import org.apache.flink.table.functions.TableFunction;
+import org.apache.flink.table.functions.UserDefinedFunction;
+import org.apache.flink.table.module.Module;
+import org.apache.flink.table.sinks.TableSink;
+import org.apache.flink.table.sources.TableSource;
+import org.apache.flink.util.TemporaryClassLoaderContext;
+
+import java.net.URLClassLoader;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Utils to initialize {@link TableEnvironment} from {@link Environment}. */
+public class LegacyTableEnvironmentInitializer {
Review comment:
We will not introduce `TableEnvironmentInitializer` for ddl right now. When useing sql file to init, it will just work like put commands in the terminal.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fsk119 commented on a change in pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
fsk119 commented on a change in pull request #15006:
URL: https://github.com/apache/flink/pull/15006#discussion_r592031634
##########
File path: flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/context/SessionContextTest.java
##########
@@ -0,0 +1,122 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.client.gateway.context;
+
+import org.apache.flink.client.cli.DefaultCLI;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.client.config.Environment;
+import org.apache.flink.table.client.gateway.utils.EnvironmentFileUtil;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.flink.table.api.config.TableConfigOptions.TABLE_SQL_DIALECT;
+
+/** Test {@link SessionContext}. */
+public class SessionContextTest {
+
+ private static final String DEFAULTS_ENVIRONMENT_FILE = "test-sql-client-defaults.yaml";
+
+ @Test
+ public void testSetAndResetYamlKey() throws Exception {
+ SessionContext sessionContext = createSessionContext();
+ sessionContext.set("execution.max-table-result-rows", "100000");
+ Assert.assertEquals(
+ "100000",
+ sessionContext
+ .getSessionEnvironment()
+ .getExecution()
+ .asMap()
+ .get("max-table-result-rows"));
+
+ sessionContext.reset();
+
+ Assert.assertEquals(
+ "100",
Review comment:
After offline discussion,
- reset should reset value to the value set in the yaml.
- it should reset to the default value if yaml doesn't set the value.
- it should clear the value if it doesn't have default value.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
}, {
"hash" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"triggerType" : "PUSH"
}, {
"hash" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736",
"triggerID" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"triggerType" : "PUSH"
}, {
"hash" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13813",
"triggerID" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"triggerType" : "PUSH"
}, {
"hash" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14060",
"triggerID" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"triggerType" : "PUSH"
}, {
"hash" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14126",
"triggerID" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"triggerType" : "PUSH"
}, {
"hash" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14133",
"triggerID" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4cd5563d44a62aec2b8db80cfe8da650fddf4ffa UNKNOWN
* f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14133)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fsk119 commented on a change in pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
fsk119 commented on a change in pull request #15006:
URL: https://github.com/apache/flink/pull/15006#discussion_r591110332
##########
File path: flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/context/SessionContext.java
##########
@@ -0,0 +1,269 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.client.gateway.context;
+
+import org.apache.flink.client.ClientUtils;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogManager;
+import org.apache.flink.table.catalog.FunctionCatalog;
+import org.apache.flink.table.catalog.GenericInMemoryCatalog;
+import org.apache.flink.table.client.config.Environment;
+import org.apache.flink.table.client.gateway.Executor;
+import org.apache.flink.table.client.gateway.SqlExecutionException;
+import org.apache.flink.table.module.ModuleManager;
+import org.apache.flink.util.TemporaryClassLoaderContext;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URLClassLoader;
+import java.util.Collections;
+import java.util.Objects;
+
+/**
+ * Context describing a session, it's mainly used for user to open a new session in the backend. If
+ * client request to open a new session, the backend {@link Executor} will maintain the session
+ * context map util users close it.
+ */
+public class SessionContext {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SessionContext.class);
+
+ private final String sessionId;
+ private final DefaultContext defaultContext;
+
+ private Environment sessionEnv;
+ private final Configuration sessionConfiguration;
+
+ private final SessionState sessionState;
+ private final URLClassLoader classLoader;
+ private ExecutionContext executionContext;
+
+ public SessionContext(
+ DefaultContext defaultContext,
+ String sessionId,
+ Environment sessionEnv,
+ Configuration sessionConfiguration,
+ URLClassLoader classLoader,
+ SessionState sessionState,
+ ExecutionContext executionContext) {
+ this.defaultContext = defaultContext;
+ this.sessionId = sessionId;
+ this.sessionEnv = sessionEnv;
+ this.sessionConfiguration = sessionConfiguration;
+ this.classLoader = classLoader;
+ this.sessionState = sessionState;
+ this.executionContext = executionContext;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Getter method
+ // --------------------------------------------------------------------------------------------
+
+ public String getSessionId() {
+ return this.sessionId;
+ }
+
+ public Environment getSessionEnvironment() {
+ return this.sessionEnv;
+ }
+
+ public ExecutionContext getExecutionContext() {
+ return this.executionContext;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Method to execute commands
+ // --------------------------------------------------------------------------------------------
+
+ /** Reset properties to default. It will rebuild a new {@link ExecutionContext}. */
+ public void reset() {
+ sessionEnv = defaultContext.getDefaultEnv().clone();
+ for (String key : sessionConfiguration.toMap().keySet()) {
+ // Don't care the type of the option
+ ConfigOption<String> keyToDelete = ConfigOptions.key(key).stringType().noDefaultValue();
+ sessionConfiguration.removeConfig(keyToDelete);
+ }
+ sessionConfiguration.addAll(defaultContext.getFlinkConfig());
Review comment:
The main reasons is the CatalogManager, FunctionManager use sessionConfiguration . If we rebuild a new Configuration, we can 't modify the values of the option for CatalogManager, FunctionManager .
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
}, {
"hash" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"triggerType" : "PUSH"
}, {
"hash" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736",
"triggerID" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"triggerType" : "PUSH"
}, {
"hash" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13813",
"triggerID" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"triggerType" : "PUSH"
}, {
"hash" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14060",
"triggerID" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"triggerType" : "PUSH"
}, {
"hash" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14126",
"triggerID" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"triggerType" : "PUSH"
}, {
"hash" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14133",
"triggerID" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"triggerType" : "PUSH"
}, {
"hash" : "83446f3a0e26473892f97f68de12b5f4a8e50cc5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14434",
"triggerID" : "83446f3a0e26473892f97f68de12b5f4a8e50cc5",
"triggerType" : "PUSH"
}, {
"hash" : "5fb5449d10b57a84f6da97e0be6fc875a46344cf",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14442",
"triggerID" : "5fb5449d10b57a84f6da97e0be6fc875a46344cf",
"triggerType" : "PUSH"
}, {
"hash" : "9e86f4b592c7699fc0fade586d99daa907b9bffc",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14465",
"triggerID" : "9e86f4b592c7699fc0fade586d99daa907b9bffc",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4cd5563d44a62aec2b8db80cfe8da650fddf4ffa UNKNOWN
* 5fb5449d10b57a84f6da97e0be6fc875a46344cf Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14442)
* 9e86f4b592c7699fc0fade586d99daa907b9bffc Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14465)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
}, {
"hash" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"triggerType" : "PUSH"
}, {
"hash" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736",
"triggerID" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"triggerType" : "PUSH"
}, {
"hash" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13813",
"triggerID" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"triggerType" : "PUSH"
}, {
"hash" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14060",
"triggerID" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"triggerType" : "PUSH"
}, {
"hash" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14126",
"triggerID" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"triggerType" : "PUSH"
}, {
"hash" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14133",
"triggerID" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"triggerType" : "PUSH"
}, {
"hash" : "83446f3a0e26473892f97f68de12b5f4a8e50cc5",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14434",
"triggerID" : "83446f3a0e26473892f97f68de12b5f4a8e50cc5",
"triggerType" : "PUSH"
}, {
"hash" : "5fb5449d10b57a84f6da97e0be6fc875a46344cf",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5fb5449d10b57a84f6da97e0be6fc875a46344cf",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4cd5563d44a62aec2b8db80cfe8da650fddf4ffa UNKNOWN
* 83446f3a0e26473892f97f68de12b5f4a8e50cc5 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14434)
* 5fb5449d10b57a84f6da97e0be6fc875a46344cf UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] wuchong commented on a change in pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
wuchong commented on a change in pull request #15006:
URL: https://github.com/apache/flink/pull/15006#discussion_r592119863
##########
File path: flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/context/DefaultContext.java
##########
@@ -0,0 +1,296 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.client.gateway.context;
+
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.client.cli.CliArgsException;
+import org.apache.flink.client.cli.CliFrontendParser;
+import org.apache.flink.client.cli.CustomCommandLine;
+import org.apache.flink.client.cli.ExecutionConfigAccessor;
+import org.apache.flink.client.cli.ProgramOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.configuration.PipelineOptions;
+import org.apache.flink.configuration.RestartStrategyOptions;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.plugin.PluginUtils;
+import org.apache.flink.streaming.api.TimeCharacteristic;
+import org.apache.flink.streaming.api.environment.StreamPipelineOptions;
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
+import org.apache.flink.table.client.config.Environment;
+import org.apache.flink.table.client.config.entries.DeploymentEntry;
+import org.apache.flink.table.client.config.entries.ExecutionEntry;
+import org.apache.flink.table.client.gateway.Executor;
+import org.apache.flink.table.client.gateway.SqlExecutionException;
+import org.apache.flink.table.descriptors.FunctionDescriptorValidator;
+import org.apache.flink.util.FlinkException;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.Options;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.nio.file.Paths;
+import java.time.Duration;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Context describing default environment, command line options, flink config, etc.
+ *
+ * <p>When the {@link Executor} execute `reset` commands, the session can restore from the "default"
+ * context.
+ */
+public class DefaultContext {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultContext.class);
+
+ private final Environment defaultEnv;
+ private final List<URL> dependencies;
+ private final Configuration flinkConfig;
+
+ public DefaultContext(
+ Environment defaultEnv,
+ List<URL> dependencies,
+ Configuration flinkConfig,
+ List<CustomCommandLine> commandLines) {
+ this.defaultEnv = defaultEnv;
+ this.dependencies = dependencies;
+ this.flinkConfig = flinkConfig;
+ Options commandLineOptions = collectCommandLineOptions(commandLines);
+
+ // initialize default file system
+ FileSystem.initialize(
+ flinkConfig, PluginUtils.createPluginManagerFromRootFolder(flinkConfig));
+
+ // add python dependencies
+ if (containsPythonFunction(defaultEnv)) {
+ addPythonDependency();
+ }
+
+ // put environment entry into Configuration
+ initConfigurationFromEnvironment();
+ try {
+ CommandLine deploymentCommandLine =
+ createCommandLine(defaultEnv.getDeployment(), commandLineOptions);
+ flinkConfig.addAll(
+ createExecutionConfig(
+ deploymentCommandLine, commandLineOptions, commandLines, dependencies));
+ } catch (Exception e) {
+ throw new SqlExecutionException(
+ "Could not load available CLI with Environment Deployment entry.", e);
+ }
+ }
+
+ public Configuration getFlinkConfig() {
+ return flinkConfig;
+ }
+
+ public Environment getDefaultEnv() {
+ return defaultEnv;
+ }
+
+ public List<URL> getDependencies() {
+ return dependencies;
+ }
+
+ private Options collectCommandLineOptions(List<CustomCommandLine> commandLines) {
+ final Options customOptions = new Options();
+ for (CustomCommandLine customCommandLine : commandLines) {
+ customCommandLine.addGeneralOptions(customOptions);
+ customCommandLine.addRunOptions(customOptions);
+ }
+ return CliFrontendParser.mergeOptions(
+ CliFrontendParser.getRunCommandOptions(), customOptions);
+ }
+
+ private void initConfigurationFromEnvironment() {
+ defaultEnv.getConfiguration().asMap().forEach(flinkConfig::setString);
+ ExecutionEntry execution = defaultEnv.getExecution();
+ flinkConfig.set(
+ ExecutionConfigOptions.IDLE_STATE_RETENTION,
+ Duration.ofMillis(execution.getMinStateRetention()));
+
+ if (execution.getParallelism().isPresent()) {
+ flinkConfig.set(CoreOptions.DEFAULT_PARALLELISM, execution.getParallelism().get());
+ }
+ flinkConfig.set(PipelineOptions.MAX_PARALLELISM, execution.getMaxParallelism());
+ flinkConfig.set(
+ StreamPipelineOptions.TIME_CHARACTERISTIC, execution.getTimeCharacteristic());
+ if (execution.getTimeCharacteristic() == TimeCharacteristic.EventTime) {
+ flinkConfig.set(
+ PipelineOptions.AUTO_WATERMARK_INTERVAL,
+ Duration.ofMillis(execution.getPeriodicWatermarksInterval()));
+ }
+
+ setRestartStrategy();
+ }
+
+ private void setRestartStrategy() {
+ RestartStrategies.RestartStrategyConfiguration restartStrategy =
+ defaultEnv.getExecution().getRestartStrategy();
+ if (restartStrategy instanceof RestartStrategies.NoRestartStrategyConfiguration) {
+ flinkConfig.set(RestartStrategyOptions.RESTART_STRATEGY, "none");
+ } else if (restartStrategy
+ instanceof RestartStrategies.FixedDelayRestartStrategyConfiguration) {
+ flinkConfig.set(RestartStrategyOptions.RESTART_STRATEGY, "fixed-delay");
+ RestartStrategies.FixedDelayRestartStrategyConfiguration fixedDelay =
+ ((RestartStrategies.FixedDelayRestartStrategyConfiguration) restartStrategy);
+ flinkConfig.set(
+ RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_ATTEMPTS,
+ fixedDelay.getRestartAttempts());
+ flinkConfig.set(
+ RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_DELAY,
+ Duration.ofMillis(
+ fixedDelay.getDelayBetweenAttemptsInterval().toMilliseconds()));
+ } else if (restartStrategy
+ instanceof RestartStrategies.FailureRateRestartStrategyConfiguration) {
+ flinkConfig.set(RestartStrategyOptions.RESTART_STRATEGY, "failure-rate");
+ RestartStrategies.FailureRateRestartStrategyConfiguration failureRate =
+ (RestartStrategies.FailureRateRestartStrategyConfiguration) restartStrategy;
+ flinkConfig.set(
+ RestartStrategyOptions.RESTART_STRATEGY_FAILURE_RATE_MAX_FAILURES_PER_INTERVAL,
+ failureRate.getMaxFailureRate());
+ flinkConfig.set(
+ RestartStrategyOptions.RESTART_STRATEGY_FAILURE_RATE_FAILURE_RATE_INTERVAL,
+ Duration.ofMillis(failureRate.getFailureInterval().toMilliseconds()));
+ flinkConfig.set(
+ RestartStrategyOptions.RESTART_STRATEGY_FAILURE_RATE_DELAY,
+ Duration.ofMillis(
+ failureRate.getDelayBetweenAttemptsInterval().toMilliseconds()));
+ } else if (restartStrategy
+ instanceof RestartStrategies.FallbackRestartStrategyConfiguration) {
+ // default is FallbackRestartStrategyConfiguration
+ // see ExecutionConfig.restartStrategyConfiguration
+ flinkConfig.removeConfig(RestartStrategyOptions.RESTART_STRATEGY);
+ }
+ }
+
+ private static Configuration createExecutionConfig(
+ CommandLine commandLine,
+ Options commandLineOptions,
+ List<CustomCommandLine> availableCommandLines,
+ List<URL> dependencies)
+ throws FlinkException {
+ LOG.debug("Available commandline options: {}", commandLineOptions);
+ List<String> options =
+ Stream.of(commandLine.getOptions())
+ .map(o -> o.getOpt() + "=" + o.getValue())
+ .collect(Collectors.toList());
+ LOG.debug(
+ "Instantiated commandline args: {}, options: {}",
+ commandLine.getArgList(),
+ options);
+
+ final CustomCommandLine activeCommandLine =
+ findActiveCommandLine(availableCommandLines, commandLine);
+ LOG.debug(
+ "Available commandlines: {}, active commandline: {}",
+ availableCommandLines,
+ activeCommandLine);
+
+ Configuration executionConfig = activeCommandLine.toConfiguration(commandLine);
+
+ try {
+ final ProgramOptions programOptions = ProgramOptions.create(commandLine);
+ final ExecutionConfigAccessor executionConfigAccessor =
+ ExecutionConfigAccessor.fromProgramOptions(programOptions, dependencies);
+ executionConfigAccessor.applyToConfiguration(executionConfig);
+ } catch (CliArgsException e) {
+ throw new SqlExecutionException("Invalid deployment run options.", e);
+ }
+
+ LOG.info("Executor config: {}", executionConfig);
+ return executionConfig;
+ }
+
+ private static CustomCommandLine findActiveCommandLine(
+ List<CustomCommandLine> availableCommandLines, CommandLine commandLine) {
+ for (CustomCommandLine cli : availableCommandLines) {
+ if (cli.isActive(commandLine)) {
+ return cli;
+ }
+ }
+ throw new SqlExecutionException("Could not find a matching deployment.");
+ }
+
+ private static CommandLine createCommandLine(
+ DeploymentEntry deployment, Options commandLineOptions) {
+ try {
+ return deployment.getCommandLine(commandLineOptions);
+ } catch (Exception e) {
+ throw new SqlExecutionException("Invalid deployment options.", e);
+ }
+ }
+
+ private boolean containsPythonFunction(Environment environment) {
+ return environment.getFunctions().values().stream()
+ .anyMatch(
+ f ->
+ FunctionDescriptorValidator.FROM_VALUE_PYTHON.equals(
+ f.getDescriptor()
+ .toProperties()
+ .get(FunctionDescriptorValidator.FROM)));
+ }
+
+ private void addPythonDependency() {
+ try {
+ URL location =
+ Class.forName(
+ "org.apache.flink.python.PythonFunctionRunner",
+ false,
+ Thread.currentThread().getContextClassLoader())
+ .getProtectionDomain()
+ .getCodeSource()
+ .getLocation();
+ if (Paths.get(location.toURI()).toFile().isFile()) {
+ this.dependencies.add(location);
+ }
+ } catch (URISyntaxException | ClassNotFoundException e) {
+ throw new SqlExecutionException(
+ "Python UDF detected but flink-python jar not found. "
+ + "If you starts SQL-Client via `sql-client.sh`, please add the flink-python jar "
+ + "via `-j` command option manually.",
+ e);
+ }
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof DefaultContext)) {
+ return false;
+ }
+ DefaultContext context = (DefaultContext) o;
+ return Objects.equals(defaultEnv, context.defaultEnv)
+ && Objects.equals(dependencies, context.dependencies)
+ && Objects.equals(flinkConfig, context.flinkConfig);
+ }
+
+ @Override
+ public int hashCode() {
Review comment:
Do we need the `hashCode` and `equals` for `DefaultContext`?
##########
File path: flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/context/SessionContextTest.java
##########
@@ -0,0 +1,160 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.client.gateway.context;
+
+import org.apache.flink.client.cli.DefaultCLI;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.client.config.Environment;
+import org.apache.flink.table.client.gateway.utils.EnvironmentFileUtil;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.flink.configuration.CoreOptions.DEFAULT_PARALLELISM;
+import static org.apache.flink.configuration.PipelineOptions.NAME;
+import static org.apache.flink.table.api.config.TableConfigOptions.TABLE_SQL_DIALECT;
+
+/** Test {@link SessionContext}. */
+public class SessionContextTest {
+
+ private static final String DEFAULTS_ENVIRONMENT_FILE = "test-sql-client-defaults.yaml";
+
+ @Test
+ public void testSetAndResetYamlKey() throws Exception {
+ SessionContext sessionContext = createSessionContext();
+ sessionContext.set("execution.max-table-result-rows", "100000");
+ Assert.assertEquals(
+ "100000",
+ sessionContext
+ .getSessionEnvironment()
+ .getExecution()
+ .asMap()
+ .get("max-table-result-rows"));
+
+ sessionContext.reset();
+
+ Assert.assertEquals(
+ "100",
+ sessionContext
+ .getSessionEnvironment()
+ .getExecution()
+ .asMap()
+ .get("max-table-result-rows"));
+ }
+
+ @Test
+ public void testSetAndResetOption() throws Exception {
+ SessionContext sessionContext = createSessionContext();
+ // table config option
+ sessionContext.set(TABLE_SQL_DIALECT.key(), "hive");
+ // runtime config option and has default value
+ sessionContext.set(DEFAULT_PARALLELISM.key(), "128");
+ // runtime config option and doesn't have default value
+ sessionContext.set(NAME.key(), "test");
+ Assert.assertEquals(
+ "hive",
+ sessionContext
+ .getSessionEnvironment()
+ .getConfiguration()
+ .asMap()
+ .get(TABLE_SQL_DIALECT.key()));
+ Assert.assertEquals(
+ "hive",
+ sessionContext
+ .getExecutionContext()
+ .getTableEnvironment()
+ .getConfig()
+ .getConfiguration()
+ .getString(TABLE_SQL_DIALECT));
+ Assert.assertEquals(
+ 128,
+ sessionContext
+ .getExecutionContext()
+ .getTableEnvironment()
+ .getConfig()
+ .getConfiguration()
+ .getInteger(DEFAULT_PARALLELISM));
+ Assert.assertEquals(
+ "test",
+ sessionContext
+ .getExecutionContext()
+ .getTableEnvironment()
+ .getConfig()
+ .getConfiguration()
+ .getString(NAME));
+
+ sessionContext.reset();
Review comment:
Could you try to reset configuration in `flink-conf.yaml` (i.e. the `Configuration` in `DefaultContext`)? According to the desgin, the configuration can't be reset.
##########
File path: flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/agg/WindowAggregateTest.scala
##########
@@ -44,8 +44,8 @@ class WindowAggregateTest extends TableTestBase {
| d DECIMAL(10, 3),
| e BIGINT,
| rowtime TIMESTAMP(3),
- | proctime as PROCTIME(),
- | WATERMARK FOR rowtime AS rowtime - INTERVAL '1' SECOND
+ | ts as rowtime,
Review comment:
revert changes.
##########
File path: flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/Executor.java
##########
@@ -32,13 +32,13 @@
void start() throws SqlExecutionException;
/**
- * Open a new session by using the given {@link SessionContext}.
+ * Open a new session by using the given session id.
*
- * @param session context to create new session.
- * @return session identifier to track the session.
+ * @param sessionId session identifier.
+ * @return used session identifier to track the session.
* @throws SqlExecutionException if any error happen
*/
- String openSession(SessionContext session) throws SqlExecutionException;
+ String openSession(String sessionId) throws SqlExecutionException;
Review comment:
Currently, we call it in `SqlClient` like this `executor.openSession(options.getSessionId())`, but `options.getSessionId()` might be `null`. I suggest either add `@Nullable` annotation to the parameter, or construct a not-null `sessionId` in `SqlClient` and pass it to `openSession(..)`.
##########
File path: flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/context/ExecutionContext.java
##########
@@ -0,0 +1,219 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.client.gateway.context;
+
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.TimeCharacteristic;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.bridge.java.internal.BatchTableEnvironmentImpl;
+import org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl;
+import org.apache.flink.table.catalog.CatalogManager;
+import org.apache.flink.table.catalog.FunctionCatalog;
+import org.apache.flink.table.client.config.Environment;
+import org.apache.flink.table.client.gateway.SqlExecutionException;
+import org.apache.flink.table.delegation.Executor;
+import org.apache.flink.table.delegation.ExecutorFactory;
+import org.apache.flink.table.delegation.Planner;
+import org.apache.flink.table.delegation.PlannerFactory;
+import org.apache.flink.table.factories.ComponentFactoryService;
+import org.apache.flink.table.module.ModuleManager;
+import org.apache.flink.util.TemporaryClassLoaderContext;
+
+import java.lang.reflect.Method;
+import java.net.URLClassLoader;
+import java.util.Map;
+import java.util.function.Supplier;
+
+import static org.apache.flink.table.client.gateway.context.SessionContext.SessionState;
+
+/**
+ * Context for executing table programs. This class caches everything that can be cached across
+ * multiple queries as long as the session context does not change. This must be thread-safe as it
+ * might be reused across different query submissions.
+ */
+public class ExecutionContext {
+
+ // TODO: merge the ExecutionContext into the SessionContext.
+ // Members that should be reused in the same session.
+ private final Environment environment;
+ private final Configuration flinkConfig;
+ private final SessionState sessionState;
+ private final URLClassLoader classLoader;
+
+ private final TableEnvironment tableEnv;
+
+ public ExecutionContext(
+ Environment environment,
+ Configuration flinkConfig,
+ URLClassLoader classLoader,
+ SessionState sessionState) {
+ this.environment = environment;
+ this.flinkConfig = flinkConfig;
+ this.sessionState = sessionState;
+ this.classLoader = classLoader;
+
+ this.tableEnv = createTableEnvironment();
+ }
+
+ /**
+ * Create a new {@link ExecutionContext}.
+ *
+ * <p>It just copies from the {@link ExecutionContext} and rebuild a new {@link
+ * TableEnvironment}. But it still needs {@link Environment} because {@link Environment} doesn't
+ * allow modification.
+ *
+ * <p>When FLINK-21462 finishes, the constructor only uses {@link ExecutionContext} as input.
+ */
+ public ExecutionContext(Environment environment, ExecutionContext context) {
+ this.environment = environment;
+ this.flinkConfig = context.flinkConfig;
+ this.sessionState = context.sessionState;
+ this.classLoader = context.classLoader;
+ // create a new table env
+ this.tableEnv = createTableEnvironment();
+ }
+
+ /**
+ * Executes the given supplier using the execution context's classloader as thread classloader.
+ */
+ public <R> R wrapClassLoader(Supplier<R> supplier) {
+ try (TemporaryClassLoaderContext ignored = TemporaryClassLoaderContext.of(classLoader)) {
+ return supplier.get();
+ }
+ }
+
+ public TableEnvironment getTableEnvironment() {
+ return tableEnv;
+ }
+
+ // ------------------------------------------------------------------------------------------------------------------
+ // Helper to create Table Environment
+ // ------------------------------------------------------------------------------------------------------------------
+
+ private TableEnvironment createTableEnvironment() {
+ EnvironmentSettings settings = environment.getExecution().getEnvironmentSettings();
+ TableConfig config = new TableConfig();
+ config.addConfiguration(flinkConfig);
+ // Override the value in configuration.
+ // TODO: use `table.planner` and `execution.runtime-mode` to configure the TableEnvironment
+ config.addConfiguration(settings.toConfiguration());
+
+ if (environment.getExecution().isStreamingPlanner()) {
Review comment:
It seems the new introduced `table.planner` and `execution.mode` option doesn't take effect here. Currently, the implementation only depends on the SQL Client configuration.
I think we should use `table.planner` and `execution.mode` to decide how to create TableEnv.
Besides, please add tests for this.
##########
File path: flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientTest.java
##########
@@ -199,7 +197,6 @@ private void verifyUpdateSubmission(
} else {
assertTrue(client.submitUpdate(statement));
assertEquals(statement, mockExecutor.receivedStatement);
- assertEquals(context, mockExecutor.receivedContext);
Review comment:
`receivedContext` in `MockExecutor` is never used, can be removed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
}, {
"hash" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"triggerType" : "PUSH"
}, {
"hash" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736",
"triggerID" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"triggerType" : "PUSH"
}, {
"hash" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13813",
"triggerID" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"triggerType" : "PUSH"
}, {
"hash" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14060",
"triggerID" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"triggerType" : "PUSH"
}, {
"hash" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14126",
"triggerID" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"triggerType" : "PUSH"
}, {
"hash" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14133",
"triggerID" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"triggerType" : "PUSH"
}, {
"hash" : "83446f3a0e26473892f97f68de12b5f4a8e50cc5",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14434",
"triggerID" : "83446f3a0e26473892f97f68de12b5f4a8e50cc5",
"triggerType" : "PUSH"
}, {
"hash" : "5fb5449d10b57a84f6da97e0be6fc875a46344cf",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14442",
"triggerID" : "5fb5449d10b57a84f6da97e0be6fc875a46344cf",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4cd5563d44a62aec2b8db80cfe8da650fddf4ffa UNKNOWN
* 83446f3a0e26473892f97f68de12b5f4a8e50cc5 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14434)
* 5fb5449d10b57a84f6da97e0be6fc875a46344cf Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14442)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
}, {
"hash" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"triggerType" : "PUSH"
}, {
"hash" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736",
"triggerID" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"triggerType" : "PUSH"
}, {
"hash" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13813",
"triggerID" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"triggerType" : "PUSH"
}, {
"hash" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14060",
"triggerID" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"triggerType" : "PUSH"
}, {
"hash" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14126",
"triggerID" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"triggerType" : "PUSH"
}, {
"hash" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14133",
"triggerID" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4cd5563d44a62aec2b8db80cfe8da650fddf4ffa UNKNOWN
* 840b4cf15689b6bf7b7561172a66d4b3571c0745 Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14126)
* f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14133)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
}, {
"hash" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"triggerType" : "PUSH"
}, {
"hash" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736",
"triggerID" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"triggerType" : "PUSH"
}, {
"hash" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13813",
"triggerID" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"triggerType" : "PUSH"
}, {
"hash" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14060",
"triggerID" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"triggerType" : "PUSH"
}, {
"hash" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14126",
"triggerID" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"triggerType" : "PUSH"
}, {
"hash" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14133",
"triggerID" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"triggerType" : "PUSH"
}, {
"hash" : "83446f3a0e26473892f97f68de12b5f4a8e50cc5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14434",
"triggerID" : "83446f3a0e26473892f97f68de12b5f4a8e50cc5",
"triggerType" : "PUSH"
}, {
"hash" : "5fb5449d10b57a84f6da97e0be6fc875a46344cf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14442",
"triggerID" : "5fb5449d10b57a84f6da97e0be6fc875a46344cf",
"triggerType" : "PUSH"
}, {
"hash" : "9e86f4b592c7699fc0fade586d99daa907b9bffc",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14465",
"triggerID" : "9e86f4b592c7699fc0fade586d99daa907b9bffc",
"triggerType" : "PUSH"
}, {
"hash" : "741114ff56881d722e10008529ec3e60a94d8431",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "741114ff56881d722e10008529ec3e60a94d8431",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4cd5563d44a62aec2b8db80cfe8da650fddf4ffa UNKNOWN
* 9e86f4b592c7699fc0fade586d99daa907b9bffc Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14465)
* 741114ff56881d722e10008529ec3e60a94d8431 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
}, {
"hash" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"triggerType" : "PUSH"
}, {
"hash" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736",
"triggerID" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"triggerType" : "PUSH"
}, {
"hash" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13813",
"triggerID" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4cd5563d44a62aec2b8db80cfe8da650fddf4ffa UNKNOWN
* 4cc39b300203dd122980e6ef22080746834fcc9b Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736)
* bc56c56211c178c4179bb850315eb47db5b76fc0 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13813)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #15006: [Flink-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [Flink-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
}, {
"hash" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704)
* 4cd5563d44a62aec2b8db80cfe8da650fddf4ffa UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
}, {
"hash" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"triggerType" : "PUSH"
}, {
"hash" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736",
"triggerID" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"triggerType" : "PUSH"
}, {
"hash" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13813",
"triggerID" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4cd5563d44a62aec2b8db80cfe8da650fddf4ffa UNKNOWN
* bc56c56211c178c4179bb850315eb47db5b76fc0 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13813)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
}, {
"hash" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"triggerType" : "PUSH"
}, {
"hash" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736",
"triggerID" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"triggerType" : "PUSH"
}, {
"hash" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13813",
"triggerID" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"triggerType" : "PUSH"
}, {
"hash" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14060",
"triggerID" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4cd5563d44a62aec2b8db80cfe8da650fddf4ffa UNKNOWN
* bc56c56211c178c4179bb850315eb47db5b76fc0 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13813)
* f13e54215fb6fbf668f8a51ccbec7f62d835a542 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14060)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fsk119 commented on pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
fsk119 commented on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-797312429
The failed test is not related to my code.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
}, {
"hash" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"triggerType" : "PUSH"
}, {
"hash" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736",
"triggerID" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"triggerType" : "PUSH"
}, {
"hash" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13813",
"triggerID" : "bc56c56211c178c4179bb850315eb47db5b76fc0",
"triggerType" : "PUSH"
}, {
"hash" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14060",
"triggerID" : "f13e54215fb6fbf668f8a51ccbec7f62d835a542",
"triggerType" : "PUSH"
}, {
"hash" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14126",
"triggerID" : "840b4cf15689b6bf7b7561172a66d4b3571c0745",
"triggerType" : "PUSH"
}, {
"hash" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14133",
"triggerID" : "f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d",
"triggerType" : "PUSH"
}, {
"hash" : "83446f3a0e26473892f97f68de12b5f4a8e50cc5",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "83446f3a0e26473892f97f68de12b5f4a8e50cc5",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4cd5563d44a62aec2b8db80cfe8da650fddf4ffa UNKNOWN
* f5721f1c1aedfb0c344a14dfbecdbc8f6dff665d Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14133)
* 83446f3a0e26473892f97f68de12b5f4a8e50cc5 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15006: [Flink-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15006:
URL: https://github.com/apache/flink/pull/15006#issuecomment-785114095
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704",
"triggerID" : "f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41",
"triggerType" : "PUSH"
}, {
"hash" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cd5563d44a62aec2b8db80cfe8da650fddf4ffa",
"triggerType" : "PUSH"
}, {
"hash" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736",
"triggerID" : "4cc39b300203dd122980e6ef22080746834fcc9b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f6a9d79eb7ef2cc93837e0370ba91b62eb17ba41 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13704)
* 4cd5563d44a62aec2b8db80cfe8da650fddf4ffa UNKNOWN
* 4cc39b300203dd122980e6ef22080746834fcc9b Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13736)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fsk119 commented on a change in pull request #15006: [FLINK-21485][sql-client] Simplify the ExecutionContext
Posted by GitBox <gi...@apache.org>.
fsk119 commented on a change in pull request #15006:
URL: https://github.com/apache/flink/pull/15006#discussion_r592143080
##########
File path: flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/context/ExecutionContext.java
##########
@@ -0,0 +1,219 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.client.gateway.context;
+
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.TimeCharacteristic;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.bridge.java.internal.BatchTableEnvironmentImpl;
+import org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl;
+import org.apache.flink.table.catalog.CatalogManager;
+import org.apache.flink.table.catalog.FunctionCatalog;
+import org.apache.flink.table.client.config.Environment;
+import org.apache.flink.table.client.gateway.SqlExecutionException;
+import org.apache.flink.table.delegation.Executor;
+import org.apache.flink.table.delegation.ExecutorFactory;
+import org.apache.flink.table.delegation.Planner;
+import org.apache.flink.table.delegation.PlannerFactory;
+import org.apache.flink.table.factories.ComponentFactoryService;
+import org.apache.flink.table.module.ModuleManager;
+import org.apache.flink.util.TemporaryClassLoaderContext;
+
+import java.lang.reflect.Method;
+import java.net.URLClassLoader;
+import java.util.Map;
+import java.util.function.Supplier;
+
+import static org.apache.flink.table.client.gateway.context.SessionContext.SessionState;
+
+/**
+ * Context for executing table programs. This class caches everything that can be cached across
+ * multiple queries as long as the session context does not change. This must be thread-safe as it
+ * might be reused across different query submissions.
+ */
+public class ExecutionContext {
+
+ // TODO: merge the ExecutionContext into the SessionContext.
+ // Members that should be reused in the same session.
+ private final Environment environment;
+ private final Configuration flinkConfig;
+ private final SessionState sessionState;
+ private final URLClassLoader classLoader;
+
+ private final TableEnvironment tableEnv;
+
+ public ExecutionContext(
+ Environment environment,
+ Configuration flinkConfig,
+ URLClassLoader classLoader,
+ SessionState sessionState) {
+ this.environment = environment;
+ this.flinkConfig = flinkConfig;
+ this.sessionState = sessionState;
+ this.classLoader = classLoader;
+
+ this.tableEnv = createTableEnvironment();
+ }
+
+ /**
+ * Create a new {@link ExecutionContext}.
+ *
+ * <p>It just copies from the {@link ExecutionContext} and rebuild a new {@link
+ * TableEnvironment}. But it still needs {@link Environment} because {@link Environment} doesn't
+ * allow modification.
+ *
+ * <p>When FLINK-21462 finishes, the constructor only uses {@link ExecutionContext} as input.
+ */
+ public ExecutionContext(Environment environment, ExecutionContext context) {
+ this.environment = environment;
+ this.flinkConfig = context.flinkConfig;
+ this.sessionState = context.sessionState;
+ this.classLoader = context.classLoader;
+ // create a new table env
+ this.tableEnv = createTableEnvironment();
+ }
+
+ /**
+ * Executes the given supplier using the execution context's classloader as thread classloader.
+ */
+ public <R> R wrapClassLoader(Supplier<R> supplier) {
+ try (TemporaryClassLoaderContext ignored = TemporaryClassLoaderContext.of(classLoader)) {
+ return supplier.get();
+ }
+ }
+
+ public TableEnvironment getTableEnvironment() {
+ return tableEnv;
+ }
+
+ // ------------------------------------------------------------------------------------------------------------------
+ // Helper to create Table Environment
+ // ------------------------------------------------------------------------------------------------------------------
+
+ private TableEnvironment createTableEnvironment() {
+ EnvironmentSettings settings = environment.getExecution().getEnvironmentSettings();
+ TableConfig config = new TableConfig();
+ config.addConfiguration(flinkConfig);
+ // Override the value in configuration.
+ // TODO: use `table.planner` and `execution.runtime-mode` to configure the TableEnvironment
+ config.addConfiguration(settings.toConfiguration());
+
+ if (environment.getExecution().isStreamingPlanner()) {
Review comment:
Sure. But I think we can leave the work to FLINK-21462. It will do the work about Configuration and Yaml compatibility.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org