You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "xishuaidelin (via GitHub)" <gi...@apache.org> on 2023/05/06 16:21:32 UTC
[GitHub] [flink] xishuaidelin opened a new pull request, #22539: [FLINK-31956][table] Extend the CompiledPlan to read from/write to Fl…
xishuaidelin opened a new pull request, #22539:
URL: https://github.com/apache/flink/pull/22539
<!--
*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
* This pull request Extend the CompiledPlan to read from/write to Flink's FileSystem which contributes to the FLINK-31791
## Brief change log
- *Modify compiledPlanOperation and executePlanOperation to support flink's filesystem which could be verified by sqlddl(other)OperationConverterTest *
- *Modify Planreference and InternalPlan to support the path of flink's filesystem*
- *Modify TableEnv#compilePlanAndWrite and verify this for TableEnvTest#testCompileAndExecutePlanWithFlinkFilesystem*
## Verifying this change
This change could be verified by the testSqlExecutePlanPath in SqlOtherToOperationConverterTest, testSqlcompilePlanPath in SqlOtherToOperationConverterTest and testCompileAndExecutePlanWithFlinkFilesystem in TableEnvironmentTest.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / **no**)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**)
- The serializers: (yes / **no** / don't know)
- The runtime per-record code paths (performance sensitive): (yes / **no** / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / **no** / don't know)
- The S3 file system connector: (yes / **no** / don't know)
## Documentation
- Does this pull request introduce a new feature? (yes / **no**)
- If yes, how is the feature documented? (**not applicable** / docs / JavaDocs / not documented)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] LadyForest commented on a diff in pull request #22539: [FLINK-31956][table] Extend the CompiledPlan to read from/write to Fl…
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #22539:
URL: https://github.com/apache/flink/pull/22539#discussion_r1207961398
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java:
##########
@@ -731,11 +732,22 @@ public TableResultInternal executePlan(InternalPlan plan) {
}
private CompiledPlan compilePlanAndWrite(
- String filePath, boolean ifNotExists, Operation operation) {
- File file = Paths.get(filePath).toFile();
- if (file.exists()) {
+ Path filePath, boolean ifNotExists, Operation operation) throws IOException {
+ FileSystem fs;
+ boolean exists;
+ try {
+ fs = filePath.getFileSystem();
+ exists = fs.exists(filePath);
+ } catch (IOException e) {
+ throw new TableException(e.getMessage());
+ }
+ if (exists) {
if (ifNotExists) {
- return loadPlan(PlanReference.fromFile(filePath));
+ if (fs.isDistributedFS()) {
+ URL localUrl = resourceManager.downloadResource(filePath);
Review Comment:
It's a little unnatural to do so. Can we wrap up this logic?
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java:
##########
@@ -748,7 +760,6 @@ private CompiledPlan compilePlanAndWrite(
filePath, TableConfigOptions.PLAN_FORCE_RECOMPILE.key()));
}
}
-
Review Comment:
revert this back
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/ExecNodeGraphInternalPlan.java:
##########
@@ -63,7 +63,6 @@ public void writeToFile(File file, boolean ignoreIfExists, boolean failIfExists)
if (ignoreIfExists) {
return;
}
-
Review Comment:
Please revert this change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #22539: [FLINK-31956][table] Extend the CompiledPlan to read from/write to Fl…
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #22539:
URL: https://github.com/apache/flink/pull/22539#issuecomment-1537175882
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7847fbbb95ea29f8c9a13360de49f72953dc5f14",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "7847fbbb95ea29f8c9a13360de49f72953dc5f14",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7847fbbb95ea29f8c9a13360de49f72953dc5f14 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] LadyForest commented on a diff in pull request #22539: [FLINK-31956][table] Extend the CompiledPlan to read from/write to Fl…
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #22539:
URL: https://github.com/apache/flink/pull/22539#discussion_r1191301752
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/CompiledPlan.java:
##########
@@ -109,6 +109,8 @@ default void writeToFile(File file) {
*/
void writeToFile(File file, boolean ignoreIfExists);
+ void writeToFile(org.apache.flink.core.fs.Path path, boolean ignoreIfExists);
Review Comment:
I think there's no need to add an extra interface method to achieve the goal.
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/InternalPlan.java:
##########
@@ -44,6 +45,13 @@ public interface InternalPlan {
*/
void writeToFile(File file, boolean ignoreIfExists, boolean failIfExists);
+ /**
+ * Note that {@code ignoreIfExists} has precedence over {@code failIfExists}.
+ *
+ * @see CompiledPlan#writeToFile(Path, boolean)
+ */
+ void writeToFile(Path path, boolean ignoreIfExists, boolean failIfExists);
Review Comment:
ditto.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xishuaidelin commented on pull request #22539: [FLINK-31956][table] Extend the CompiledPlan to read from/write to Fl…
Posted by "xishuaidelin (via GitHub)" <gi...@apache.org>.
xishuaidelin commented on PR #22539:
URL: https://github.com/apache/flink/pull/22539#issuecomment-1560529796
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] LadyForest commented on a diff in pull request #22539: [FLINK-31956][table] Extend the CompiledPlan to read from/write to Fl…
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #22539:
URL: https://github.com/apache/flink/pull/22539#discussion_r1230748686
##########
flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/resources/compile_plan_use_remote_file_e2e.sql:
##########
@@ -0,0 +1,44 @@
+/*
+ * 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.
+ */
+SET execution.runtime-mode = $MODE;
Review Comment:
COMPILE PLAN statement is not supported under batch mode.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] LadyForest commented on a diff in pull request #22539: [FLINK-31956][table] Extend the CompiledPlan to read from/write to Fl…
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #22539:
URL: https://github.com/apache/flink/pull/22539#discussion_r1201419123
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java:
##########
@@ -731,9 +731,16 @@ public TableResultInternal executePlan(InternalPlan plan) {
}
private CompiledPlan compilePlanAndWrite(
- String filePath, boolean ifNotExists, Operation operation) {
- File file = Paths.get(filePath).toFile();
- if (file.exists()) {
+ Path filePath, boolean ifNotExists, Operation operation) {
+ FileSystem fs;
+ boolean exists;
+ try {
+ fs = filePath.getFileSystem();
+ exists = fs.exists(filePath);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
Review Comment:
I think we'd better throw `TableException` instead of `RuntimeException`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xishuaidelin closed pull request #22539: [FLINK-31956][table] Extend the CompiledPlan to read from/write to Fl…
Posted by "xishuaidelin (via GitHub)" <gi...@apache.org>.
xishuaidelin closed pull request #22539: [FLINK-31956][table] Extend the CompiledPlan to read from/write to Fl…
URL: https://github.com/apache/flink/pull/22539
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] LadyForest commented on a diff in pull request #22539: [FLINK-31956][table] Extend the CompiledPlan to read from/write to Fl…
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #22539:
URL: https://github.com/apache/flink/pull/22539#discussion_r1230718029
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/CompileAndExecutePlanOperation.java:
##########
@@ -19,25 +19,26 @@
package org.apache.flink.table.operations;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.core.fs.Path;
import org.apache.flink.util.Preconditions;
/** Operation to describe an {@code COMPILE AND EXECUTE PLAN} statement. */
@Internal
public class CompileAndExecutePlanOperation implements Operation {
- private final String filePath;
+ private final Path filePath;
private final Operation operation;
public CompileAndExecutePlanOperation(String filePath, Operation operation) {
Preconditions.checkArgument(
operation instanceof StatementSetOperation || operation instanceof ModifyOperation,
"Child operation of CompileAndExecuteOperation must be either a "
+ "ModifyOperation or a StatementSetOperation.");
- this.filePath = filePath;
+ this.filePath = new Path(filePath);
Review Comment:
Perhaps we don't need to change this.
##########
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/TableEnvironmentTest.scala:
##########
@@ -2155,6 +2158,47 @@ class TableEnvironmentTest {
testUnsupportedExplain("explain plan as json for select * from MyTable")
}
+ @Test
+ def testCompileAndExecutePlan(): Unit = {
+ val execEnv = StreamExecutionEnvironment.getExecutionEnvironment
+ execEnv.setParallelism(1)
+ val settings = EnvironmentSettings.newInstance().inStreamingMode().build()
+ val tableEnv = StreamTableEnvironment.create(execEnv, settings)
+ tableEnv.getConfig.set(TableConfigOptions.PLAN_FORCE_RECOMPILE, Boolean.box(true))
+
+ val srcTableDdl =
+ "CREATE TABLE MyTable (\n" + " a bigint,\n" + " b int,\n" + " c varchar\n" + ") with (\n" + " 'connector' = 'values',\n" + " 'bounded' = 'false')"
+ tableEnv.executeSql(srcTableDdl)
+
+ val sinkTableDdl =
+ "CREATE TABLE MySink (\n" + " a bigint,\n" + " b int,\n" + " c varchar\n" + ") with (\n" + " 'connector' = 'values',\n" + " 'table-sink-class' = 'DEFAULT')"
+ tableEnv.executeSql(sinkTableDdl)
+
+ val planfile = tempFolder.newFile("compile1.json")
+ var file = "file://" + planfile.toPath
+
+ var sql = String.format("COMPILE PLAN '%s' FOR INSERT INTO MySink SELECT * FROM MyTable", file)
+ tableEnv.executeSql(sql)
+
+ sql = String.format("EXECUTE PLAN '%s'", file)
+ tableEnv.executeSql(sql)
+
+ val planPath = tempFolder.getRoot.getPath
+ var path = "file://" + planPath + "/compile2.json"
+
+ sql = String.format("COMPILE PLAN '%s' FOR INSERT INTO MySink SELECT * FROM MyTable", path)
+ tableEnv.executeSql(sql)
+
+ sql = String.format("EXECUTE PLAN '%s'", path)
+ tableEnv.executeSql(sql)
+
+ path = "file://" + planPath + "/compile3.json"
+ sql = String.format(
+ "COMPILE and EXECUTE plan '%s' FOR INSERT INTO MySink SELECT * FROM MyTable",
+ path)
+ tableEnv.executeSql(sql)
Review Comment:
What's the purpose of this test? It does not assert anything.
##########
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/TableEnvironmentTest.scala:
##########
@@ -2155,6 +2158,47 @@ class TableEnvironmentTest {
testUnsupportedExplain("explain plan as json for select * from MyTable")
}
+ @Test
+ def testCompileAndExecutePlan(): Unit = {
+ val execEnv = StreamExecutionEnvironment.getExecutionEnvironment
+ execEnv.setParallelism(1)
+ val settings = EnvironmentSettings.newInstance().inStreamingMode().build()
+ val tableEnv = StreamTableEnvironment.create(execEnv, settings)
Review Comment:
What's the purpose of re-creating a `tableEnv`?
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/resource/ResourceManager.java:
##########
@@ -83,6 +83,56 @@ public ResourceManager(ReadableConfig config, MutableURLClassLoader userClassLoa
this.userClassLoader = userClassLoader;
}
+ public boolean exists(Path filePath) throws IOException {
+ return FileSystem.getUnguardedFileSystem(filePath.toUri()).exists(filePath);
+ }
+
+ /**
+ * Register the filePath of flink filesystem. If it is remote filesystem and the file exists
+ * then download the file at local. register the filePath map to localURL.
+ */
+ public void registerFsResources(Path filePath) throws IOException {
Review Comment:
Meanwhile, this method lacks test coverage.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java:
##########
@@ -2165,6 +2168,25 @@ public void testCreateViewWithDynamicTableOptions() {
assertThat(operation).isInstanceOf(CreateViewOperation.class);
}
+ @ParameterizedTest
Review Comment:
We could add the test at the syntax level since the operation level requires no change.
##########
flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/SqlITCaseBase.java:
##########
@@ -113,6 +113,13 @@ public void runAndCheckSQL(
}
}
+ public void runSQL(String sqlPath, Map<String, String> varsMap) throws Exception {
+ try (ClusterController clusterController = flink.startCluster(1)) {
+ List<String> sqlLines = initializeSqlLines(sqlPath, varsMap);
+ executeSqlStatements(clusterController, sqlLines);
Review Comment:
Why duplicate the method and assert nothing?
##########
flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/CompileAndExecuteRemoteFileITCase.java:
##########
@@ -0,0 +1,112 @@
+/*
+ * 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.sql.codegen;
+
+import org.apache.flink.test.resources.ResourceTestUtils;
+import org.apache.flink.test.util.SQLJobSubmission;
+import org.apache.flink.tests.util.flink.ClusterController;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileUtil;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+
+/** End-to-End tests for compile and execute remote file. */
+public class CompileAndExecuteRemoteFileITCase extends SqlITCaseBase {
Review Comment:
This test shares the same HDFS configuration with `UsingRemoteJarITCase.` Maybe deserve a common base class
##########
flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/CompileAndExecuteRemoteFileITCase.java:
##########
@@ -0,0 +1,112 @@
+/*
+ * 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.sql.codegen;
+
+import org.apache.flink.test.resources.ResourceTestUtils;
+import org.apache.flink.test.util.SQLJobSubmission;
+import org.apache.flink.tests.util.flink.ClusterController;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileUtil;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+
+/** End-to-End tests for compile and execute remote file. */
+public class CompileAndExecuteRemoteFileITCase extends SqlITCaseBase {
+ private static final Path HADOOP_CLASSPATH =
+ ResourceTestUtils.getResource(".*hadoop.classpath");
+
+ private MiniDFSCluster hdfsCluster;
+ private org.apache.hadoop.fs.Path hdPath;
+ private org.apache.hadoop.fs.FileSystem hdfs;
+
+ public CompileAndExecuteRemoteFileITCase(String executionMode) {
+ super(executionMode);
+ }
+
+ @Before
+ public void before() throws Exception {
+ super.before();
+ createHDFS();
+ }
+
+ private void createHDFS() {
+ try {
+ Configuration hdConf = new Configuration();
+
+ File baseDir = new File("./target/hdfs/hdfsTest").getAbsoluteFile();
+ FileUtil.fullyDelete(baseDir);
+ hdConf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, baseDir.getAbsolutePath());
+ MiniDFSCluster.Builder builder = new MiniDFSCluster.Builder(hdConf);
+ hdfsCluster = builder.build();
+
+ hdPath = new org.apache.hadoop.fs.Path("/test.json");
+ hdfs = hdPath.getFileSystem(hdConf);
+
+ } catch (Throwable e) {
+ e.printStackTrace();
Review Comment:
I don't think it's a good practice to do so...
##########
flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/CompileAndExecuteRemoteFileITCase.java:
##########
@@ -0,0 +1,112 @@
+/*
+ * 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.sql.codegen;
+
+import org.apache.flink.test.resources.ResourceTestUtils;
+import org.apache.flink.test.util.SQLJobSubmission;
+import org.apache.flink.tests.util.flink.ClusterController;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileUtil;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+
+/** End-to-End tests for compile and execute remote file. */
+public class CompileAndExecuteRemoteFileITCase extends SqlITCaseBase {
+ private static final Path HADOOP_CLASSPATH =
+ ResourceTestUtils.getResource(".*hadoop.classpath");
+
+ private MiniDFSCluster hdfsCluster;
+ private org.apache.hadoop.fs.Path hdPath;
+ private org.apache.hadoop.fs.FileSystem hdfs;
+
+ public CompileAndExecuteRemoteFileITCase(String executionMode) {
+ super(executionMode);
+ }
+
+ @Before
+ public void before() throws Exception {
+ super.before();
+ createHDFS();
+ }
+
+ private void createHDFS() {
+ try {
+ Configuration hdConf = new Configuration();
+
+ File baseDir = new File("./target/hdfs/hdfsTest").getAbsoluteFile();
+ FileUtil.fullyDelete(baseDir);
+ hdConf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, baseDir.getAbsolutePath());
+ MiniDFSCluster.Builder builder = new MiniDFSCluster.Builder(hdConf);
+ hdfsCluster = builder.build();
+
+ hdPath = new org.apache.hadoop.fs.Path("/test.json");
Review Comment:
If the purpose of declaring this variable is to delete this path in the after method, then it can be turned into a local variable in `after`.
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/resource/ResourceManager.java:
##########
@@ -83,6 +83,56 @@ public ResourceManager(ReadableConfig config, MutableURLClassLoader userClassLoa
this.userClassLoader = userClassLoader;
}
+ public boolean exists(Path filePath) throws IOException {
+ return FileSystem.getUnguardedFileSystem(filePath.toUri()).exists(filePath);
+ }
+
+ /**
+ * Register the filePath of flink filesystem. If it is remote filesystem and the file exists
+ * then download the file at local. register the filePath map to localURL.
+ */
+ public void registerFsResources(Path filePath) throws IOException {
Review Comment:
File resource shares almost the same logic like jar resource, maybe we can extract the common utils to avoid repeatation.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/ExecNodeGraphInternalPlan.java:
##########
@@ -75,6 +75,9 @@ public void writeToFile(File file, boolean ignoreIfExists, boolean failIfExists)
}
}
try {
+ if (!file.getParentFile().exists()) {
Review Comment:
I don't think it's appropriate to make parent dirs here. The logic spreads across multiple classes...
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/command/ExecutePlanOperation.java:
##########
@@ -19,19 +19,20 @@
package org.apache.flink.table.operations.command;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.core.fs.Path;
import org.apache.flink.table.operations.Operation;
/** Operation to describe an EXECUTE PLAN statement. */
@Internal
public class ExecutePlanOperation implements Operation {
- private final String filePath;
+ private final Path filePath;
Review Comment:
ditto
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlOtherOperationConverterTest.java:
##########
@@ -321,4 +326,23 @@ private void assertShowFunctions(
assertThat(showFunctionsOperation.getFunctionScope()).isEqualTo(expectedScope);
assertThat(showFunctionsOperation.asSummaryString()).isEqualTo(expectedSummary);
}
+
+ @ParameterizedTest
Review Comment:
ditto
##########
flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/CompileAndExecuteRemoteFileITCase.java:
##########
@@ -0,0 +1,112 @@
+/*
+ * 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.sql.codegen;
+
+import org.apache.flink.test.resources.ResourceTestUtils;
+import org.apache.flink.test.util.SQLJobSubmission;
+import org.apache.flink.tests.util.flink.ClusterController;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileUtil;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+
+/** End-to-End tests for compile and execute remote file. */
+public class CompileAndExecuteRemoteFileITCase extends SqlITCaseBase {
+ private static final Path HADOOP_CLASSPATH =
+ ResourceTestUtils.getResource(".*hadoop.classpath");
+
+ private MiniDFSCluster hdfsCluster;
+ private org.apache.hadoop.fs.Path hdPath;
+ private org.apache.hadoop.fs.FileSystem hdfs;
+
+ public CompileAndExecuteRemoteFileITCase(String executionMode) {
+ super(executionMode);
+ }
+
+ @Before
+ public void before() throws Exception {
+ super.before();
+ createHDFS();
+ }
+
+ private void createHDFS() {
+ try {
+ Configuration hdConf = new Configuration();
+
+ File baseDir = new File("./target/hdfs/hdfsTest").getAbsoluteFile();
+ FileUtil.fullyDelete(baseDir);
+ hdConf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, baseDir.getAbsolutePath());
+ MiniDFSCluster.Builder builder = new MiniDFSCluster.Builder(hdConf);
+ hdfsCluster = builder.build();
+
+ hdPath = new org.apache.hadoop.fs.Path("/test.json");
+ hdfs = hdPath.getFileSystem(hdConf);
+
+ } catch (Throwable e) {
+ e.printStackTrace();
+ Assert.fail("Test failed " + e.getMessage());
+ }
+ }
+
+ @After
+ public void destroyHDFS() {
+ try {
+ hdfs.delete(hdPath, false);
+ hdfsCluster.shutdown();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ protected Map<String, String> generateReplaceVars() {
+ String remoteJsonPath =
+ String.format(
+ "hdfs://%s:%s/%s",
+ hdfsCluster.getURI().getHost(), hdfsCluster.getNameNodePort(), hdPath);
+
+ Map<String, String> map = super.generateReplaceVars();
+ map.put("$HDFS_Json_Plan_PATH", remoteJsonPath);
+ return map;
+ }
+
+ @Test
+ public void testCompilePlanRemoteFile() throws Exception {
+ runSQL("compile_plan_use_remote_file_e2e.sql", generateReplaceVars());
Review Comment:
The test does not assert anything...
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CompilePlanOperation.java:
##########
@@ -26,20 +27,20 @@
/** Operation to describe an {@code COMPILE PLAN} statement. */
public class CompilePlanOperation implements Operation {
- private final String filePath;
+ private final Path filePath;
private final boolean ifNotExists;
private final Operation operation;
public CompilePlanOperation(String filePath, boolean ifNotExists, Operation operation) {
Preconditions.checkArgument(
operation instanceof StatementSetOperation || operation instanceof ModifyOperation,
"child operation of CompileOperation must be either a ModifyOperation or a StatementSetOperation");
- this.filePath = filePath;
+ this.filePath = new Path(filePath);
Review Comment:
ditto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xishuaidelin commented on a diff in pull request #22539: [FLINK-31956][table] Extend the CompiledPlan to read from/write to Fl…
Posted by "xishuaidelin (via GitHub)" <gi...@apache.org>.
xishuaidelin commented on code in PR #22539:
URL: https://github.com/apache/flink/pull/22539#discussion_r1222759885
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/resource/ResourceManager.java:
##########
@@ -83,6 +83,60 @@ public ResourceManager(ReadableConfig config, MutableURLClassLoader userClassLoa
this.userClassLoader = userClassLoader;
}
+ public boolean exists(Path filePath) throws IOException {
+ boolean exist;
+ FileSystem fs = FileSystem.getUnguardedFileSystem(filePath.toUri());
+ exist = fs.exists(filePath);
+ return exist;
Review Comment:
Yes, this looks better. I would update it soon.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xishuaidelin commented on pull request #22539: [FLINK-31956][table] Extend the CompiledPlan to read from/write to Fl…
Posted by "xishuaidelin (via GitHub)" <gi...@apache.org>.
xishuaidelin commented on PR #22539:
URL: https://github.com/apache/flink/pull/22539#issuecomment-1550592634
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] LadyForest commented on a diff in pull request #22539: [FLINK-31956][table] Extend the CompiledPlan to read from/write to Fl…
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #22539:
URL: https://github.com/apache/flink/pull/22539#discussion_r1230722665
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/ExecNodeGraphInternalPlan.java:
##########
@@ -75,6 +75,9 @@ public void writeToFile(File file, boolean ignoreIfExists, boolean failIfExists)
}
}
try {
+ if (!file.getParentFile().exists()) {
Review Comment:
I don't think it's appropriate to make parent dirs here. The logic spreads across multiple classes...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xishuaidelin commented on a diff in pull request #22539: [FLINK-31956][table] Extend the CompiledPlan to read from/write to Fl…
Posted by "xishuaidelin (via GitHub)" <gi...@apache.org>.
xishuaidelin commented on code in PR #22539:
URL: https://github.com/apache/flink/pull/22539#discussion_r1192028319
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/CompiledPlan.java:
##########
@@ -109,6 +109,8 @@ default void writeToFile(File file) {
*/
void writeToFile(File file, boolean ignoreIfExists);
+ void writeToFile(org.apache.flink.core.fs.Path path, boolean ignoreIfExists);
Review Comment:
Okey. I have removed this interface and modify other interface to achieve the goal.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xishuaidelin commented on a diff in pull request #22539: [FLINK-31956][table] Extend the CompiledPlan to read from/write to Fl…
Posted by "xishuaidelin (via GitHub)" <gi...@apache.org>.
xishuaidelin commented on code in PR #22539:
URL: https://github.com/apache/flink/pull/22539#discussion_r1192027937
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/CompiledPlan.java:
##########
@@ -109,6 +109,8 @@ default void writeToFile(File file) {
*/
void writeToFile(File file, boolean ignoreIfExists);
+ void writeToFile(org.apache.flink.core.fs.Path path, boolean ignoreIfExists);
Review Comment:
Okey. I have removed this interface and modify other interface to achieve the goal.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] LadyForest commented on a diff in pull request #22539: [FLINK-31956][table] Extend the CompiledPlan to read from/write to Fl…
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #22539:
URL: https://github.com/apache/flink/pull/22539#discussion_r1222720612
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java:
##########
@@ -761,7 +760,9 @@ private CompiledPlan compilePlanAndWrite(
+ ". This is a bug, please file an issue.");
}
- compiledPlan.writeToFile(file, false);
+ compiledPlan.writeToFile(localPath, false);
+ resourceManager.updateFilePath(filePath);
Review Comment:
Correct me if I'm wrong, but I think we should not and probably cannot write files to a remote FileSystem.
Although users have configured related conf in flink-conf.yaml, such as accessKey and accessSecret, having permission to read files under a bucket does not mean having the same write permission.
If users feel it is necessary, they can upload files manually. Implementing this through the framework may have many limitations.
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/resource/ResourceManager.java:
##########
@@ -83,6 +83,60 @@ public ResourceManager(ReadableConfig config, MutableURLClassLoader userClassLoa
this.userClassLoader = userClassLoader;
}
+ public boolean exists(Path filePath) throws IOException {
+ boolean exist;
+ FileSystem fs = FileSystem.getUnguardedFileSystem(filePath.toUri());
+ exist = fs.exists(filePath);
+ return exist;
+ }
+
+ /**
+ * register the filePath of flink filesystem. If it is remote filesystem and the file exists
Review Comment:
Nit: JavaDoc comments should start with the capitalized sentence. You can take https://www.oracle.com/technical-resources/articles/java/javadoc-tool.html
as a ref.
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/resource/ResourceManager.java:
##########
@@ -83,6 +83,60 @@ public ResourceManager(ReadableConfig config, MutableURLClassLoader userClassLoa
this.userClassLoader = userClassLoader;
}
+ public boolean exists(Path filePath) throws IOException {
+ boolean exist;
+ FileSystem fs = FileSystem.getUnguardedFileSystem(filePath.toUri());
+ exist = fs.exists(filePath);
+ return exist;
Review Comment:
Can we inline this?
```java
return FileSystem.getUnguardedFileSystem(filePath.toUri()).exists(filePath);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xishuaidelin commented on a diff in pull request #22539: [FLINK-31956][table] Extend the CompiledPlan to read from/write to Fl…
Posted by "xishuaidelin (via GitHub)" <gi...@apache.org>.
xishuaidelin commented on code in PR #22539:
URL: https://github.com/apache/flink/pull/22539#discussion_r1203332236
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java:
##########
@@ -731,9 +731,16 @@ public TableResultInternal executePlan(InternalPlan plan) {
}
private CompiledPlan compilePlanAndWrite(
- String filePath, boolean ifNotExists, Operation operation) {
- File file = Paths.get(filePath).toFile();
- if (file.exists()) {
+ Path filePath, boolean ifNotExists, Operation operation) {
+ FileSystem fs;
+ boolean exists;
+ try {
+ fs = filePath.getFileSystem();
+ exists = fs.exists(filePath);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
Review Comment:
Okey. I have modified this in the latest commit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] LadyForest commented on pull request #22539: [FLINK-31956][table] Extend the CompiledPlan to read from/write to Fl…
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on PR #22539:
URL: https://github.com/apache/flink/pull/22539#issuecomment-1565947626
By the way, you can evaluate the param `table.resources.download-dir` by the command `SET;` to verify that the remote plan is downloaded under this dir and is cleaned up when the session closes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org