You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by yu...@apache.org on 2023/06/25 09:35:18 UTC
[flink] branch master updated (65217c35e7c -> ed36b2062dd)
This is an automated email from the ASF dual-hosted git repository.
yuxia pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from 65217c35e7c [FLINK-20887][table-planner] Fix incorrect aggregate project merge for LogicalWindowAggregate
new c83eb5cd5a6 [FLINK-31956][table-planner] Fix COMPILE PLAN statement fails to write to file when parent directories do not exist
new ed36b2062dd [FLINK-31956][table] Extend COMPILE PLAN statement to read/write from/to Flink FileSystem This closes #22818
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../codegen/CompileAndExecuteRemotePlanITCase.java | 94 +++++++
.../table/sql/codegen/CreateTableAsITCase.java | 6 +-
.../flink/table/sql/codegen/HdfsITCaseBase.java | 128 ++++++++++
.../table/sql/codegen/PlannerScalaFreeITCase.java | 4 +-
.../flink/table/sql/codegen/SqlITCaseBase.java | 49 ++--
.../table/sql/codegen/UsingRemoteJarITCase.java | 132 ++--------
.../resources/compile_and_execute_plan_e2e.sql | 50 ++++
.../client/resource/ClientResourceManager.java | 2 +-
.../gateway/AbstractSqlGatewayStatementITCase.java | 7 +
.../src/test/resources/sql/insert.q | 7 +-
.../flink/sql/parser/FlinkSqlParserImplTest.java | 26 ++
.../table/api/internal/TableEnvironmentImpl.java | 81 +++---
.../flink/table/resource/ResourceManager.java | 271 +++++++++++++++------
.../flink/table/resource/ResourceManagerTest.java | 142 ++++++++++-
.../apache/flink/table/resource/ResourceUri.java | 2 +-
.../planner/plan/ExecNodeGraphInternalPlan.java | 1 +
.../apache/flink/table/api/CompiledPlanITCase.java | 27 +-
17 files changed, 765 insertions(+), 264 deletions(-)
create mode 100644 flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/CompileAndExecuteRemotePlanITCase.java
create mode 100644 flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/HdfsITCaseBase.java
create mode 100644 flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/resources/compile_and_execute_plan_e2e.sql
[flink] 02/02: [FLINK-31956][table] Extend COMPILE PLAN statement to read/write from/to Flink FileSystem This closes #22818
Posted by yu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yuxia pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit ed36b2062dd04ccf5cdc7f9bdde239c41cc6cab2
Author: Jane Chan <qi...@gmail.com>
AuthorDate: Sun Jun 18 13:01:45 2023 +0800
[FLINK-31956][table] Extend COMPILE PLAN statement to read/write from/to Flink FileSystem
This closes #22818
---
.../codegen/CompileAndExecuteRemotePlanITCase.java | 94 +++++++
.../table/sql/codegen/CreateTableAsITCase.java | 6 +-
.../flink/table/sql/codegen/HdfsITCaseBase.java | 128 ++++++++++
.../table/sql/codegen/PlannerScalaFreeITCase.java | 4 +-
.../flink/table/sql/codegen/SqlITCaseBase.java | 49 ++--
.../table/sql/codegen/UsingRemoteJarITCase.java | 132 ++--------
.../resources/compile_and_execute_plan_e2e.sql | 50 ++++
.../client/resource/ClientResourceManager.java | 2 +-
.../gateway/AbstractSqlGatewayStatementITCase.java | 7 +
.../src/test/resources/sql/insert.q | 7 +-
.../flink/sql/parser/FlinkSqlParserImplTest.java | 26 ++
.../table/api/internal/TableEnvironmentImpl.java | 81 +++---
.../flink/table/resource/ResourceManager.java | 271 +++++++++++++++------
.../flink/table/resource/ResourceManagerTest.java | 142 ++++++++++-
.../apache/flink/table/resource/ResourceUri.java | 2 +-
.../apache/flink/table/api/CompiledPlanITCase.java | 18 ++
16 files changed, 765 insertions(+), 254 deletions(-)
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/CompileAndExecuteRemotePlanITCase.java b/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/CompileAndExecuteRemotePlanITCase.java
new file mode 100644
index 00000000000..b74c70794ab
--- /dev/null
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/CompileAndExecuteRemotePlanITCase.java
@@ -0,0 +1,94 @@
+/*
+ * 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.hadoop.fs.LocatedFileStatus;
+import org.apache.hadoop.fs.RemoteIterator;
+import org.junit.Assume;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** End-to-End tests for COMPILE AND EXECUTE PLAN statement with hdfs as remote uri. */
+public class CompileAndExecuteRemotePlanITCase extends HdfsITCaseBase {
+
+ private static final String TABLE1 = "message";
+ private static final String TABLE2 = "employee";
+
+ private String planDir;
+
+ public CompileAndExecuteRemotePlanITCase(String executionMode) {
+ super(executionMode);
+ }
+
+ @Override
+ protected void createHDFS() {
+ super.createHDFS();
+ planDir = getRemotePlanDir();
+ }
+
+ @Override
+ protected Map<String, String> generateReplaceVars() {
+ Map<String, String> varsMap = super.generateReplaceVars();
+ varsMap.put("$REMOTE_PLAN_DIR", planDir);
+ varsMap.put("$TABLE1", TABLE1);
+ varsMap.put("$TABLE2", TABLE2);
+ return varsMap;
+ }
+
+ @Test
+ public void testCompileAndExecutePlan() throws Exception {
+ // COMPILE AND EXECUTE PLAN is not supported under batch mode
+ Assume.assumeTrue(executionMode.equals("streaming"));
+ Map<Path, List<String>> resultItems = new HashMap<>();
+ resultItems.put(result.resolve(TABLE1), Arrays.asList("1,Meow", "2,Purr"));
+ resultItems.put(result.resolve(TABLE2), Arrays.asList("1,Tom", "2,Jerry"));
+ runAndCheckSQL("compile_and_execute_plan_e2e.sql", resultItems);
+
+ assertPlanExists();
+ }
+
+ private String getRemotePlanDir() {
+ return String.format(
+ "hdfs://%s:%s/foo/bar",
+ hdfsCluster.getURI().getHost(), hdfsCluster.getNameNodePort());
+ }
+
+ private void assertPlanExists() throws IOException {
+ org.apache.hadoop.fs.Path dirPath = new org.apache.hadoop.fs.Path(planDir);
+ final RemoteIterator<LocatedFileStatus> iterable =
+ dirPath.getFileSystem(hdConf).listFiles(dirPath, false);
+ List<org.apache.hadoop.fs.Path> files = new ArrayList<>();
+ while (iterable.hasNext()) {
+ files.add(iterable.next().getPath());
+ }
+ assertThat(files)
+ .containsExactlyInAnyOrder(
+ new org.apache.hadoop.fs.Path(dirPath, TABLE1 + ".json"),
+ new org.apache.hadoop.fs.Path(dirPath, TABLE2 + ".json"));
+ }
+}
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/CreateTableAsITCase.java b/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/CreateTableAsITCase.java
index 1204842547b..11ef84708d0 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/CreateTableAsITCase.java
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/CreateTableAsITCase.java
@@ -27,7 +27,7 @@ import java.time.Duration;
import java.util.Arrays;
import java.util.List;
-/** End to End tests for create table as select syntax. */
+/** End-to-End tests for create table as select syntax. */
public class CreateTableAsITCase extends SqlITCaseBase {
public CreateTableAsITCase(String executionMode) {
@@ -38,8 +38,6 @@ public class CreateTableAsITCase extends SqlITCaseBase {
public void testCreateTableAs() throws Exception {
runAndCheckSQL(
"create_table_as_e2e.sql",
- generateReplaceVars(),
- 2,
Arrays.asList(
"{\"before\":null,\"after\":{\"user_name\":\"Alice\",\"order_cnt\":1},\"op\":\"c\"}",
"{\"before\":null,\"after\":{\"user_name\":\"Bob\",\"order_cnt\":2},\"op\":\"c\"}"));
@@ -49,8 +47,6 @@ public class CreateTableAsITCase extends SqlITCaseBase {
public void testCreateTableAsInStatementSet() throws Exception {
runAndCheckSQL(
"create_table_as_statementset_e2e.sql",
- generateReplaceVars(),
- 2,
Arrays.asList(
"{\"before\":null,\"after\":{\"user_name\":\"Alice\",\"order_cnt\":1},\"op\":\"c\"}",
"{\"before\":null,\"after\":{\"user_name\":\"Bob\",\"order_cnt\":2},\"op\":\"c\"}"));
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/HdfsITCaseBase.java b/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/HdfsITCaseBase.java
new file mode 100644
index 00000000000..a3f83ddd95a
--- /dev/null
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/HdfsITCaseBase.java
@@ -0,0 +1,128 @@
+/*
+ * 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.flink.util.FileUtils;
+import org.apache.flink.util.OperatingSystem;
+
+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.Assume;
+import org.junit.Before;
+import org.junit.BeforeClass;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.List;
+
+/** Base class for sql ITCase which depends on HDFS env. */
+public abstract class HdfsITCaseBase extends SqlITCaseBase {
+
+ private static final Path HADOOP_CLASSPATH =
+ ResourceTestUtils.getResource(".*hadoop.classpath");
+
+ protected Configuration hdConf;
+ protected MiniDFSCluster hdfsCluster;
+
+ public HdfsITCaseBase(String executionMode) {
+ super(executionMode);
+ }
+
+ @BeforeClass
+ public static void verifyOS() {
+ Assume.assumeTrue(
+ "HDFS cluster cannot be started on Windows without extensions.",
+ !OperatingSystem.isWindows());
+ }
+
+ @Before
+ public void before() throws Exception {
+ super.before();
+ createHDFS();
+ }
+
+ @After
+ public void after() {
+ destroyHDFS();
+ }
+
+ protected void createHDFS() {
+ try {
+ hdConf = new Configuration();
+ File baseDir =
+ new File(String.format("./target/hdfs/%s", getClass().getSimpleName()))
+ .getAbsoluteFile();
+ FileUtil.fullyDelete(baseDir);
+ hdConf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, baseDir.getAbsolutePath());
+ MiniDFSCluster.Builder builder = new MiniDFSCluster.Builder(hdConf);
+ hdfsCluster = builder.build();
+ } catch (Throwable e) {
+ Assert.fail("Failed to create HDFS env" + e.getMessage());
+ }
+ }
+
+ protected void destroyHDFS() {
+ hdfsCluster.shutdown();
+ }
+
+ @Override
+ protected void executeSqlStatements(ClusterController clusterController, List<String> sqlLines)
+ throws Exception {
+ clusterController.submitSQLJob(
+ new SQLJobSubmission.SQLJobSubmissionBuilder(sqlLines)
+ .setEnvProcessor(
+ map -> map.put("HADOOP_CLASSPATH", getHadoopClassPathContent()))
+ .build(),
+ Duration.ofMinutes(2L));
+ }
+
+ private String getHadoopClassPathContent() {
+ // Prepare all hadoop jars to mock HADOOP_CLASSPATH, use hadoop.classpath which contains all
+ // hadoop jars
+ File hadoopClasspathFile = new File(HADOOP_CLASSPATH.toAbsolutePath().toString());
+
+ if (!hadoopClasspathFile.exists()) {
+ try {
+ throw new FileNotFoundException(
+ String.format(
+ "File that contains hadoop classpath %s does not exist.",
+ HADOOP_CLASSPATH));
+ } catch (FileNotFoundException e) {
+ Assert.fail("Test failed " + e.getMessage());
+ }
+ }
+
+ String classPathContent = null;
+ try {
+ classPathContent = FileUtils.readFileUtf8(hadoopClasspathFile);
+ } catch (IOException e) {
+ Assert.fail("Test failed " + e.getMessage());
+ }
+ return classPathContent;
+ }
+}
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/PlannerScalaFreeITCase.java b/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/PlannerScalaFreeITCase.java
index 52b09aa6826..8b03cb54920 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/PlannerScalaFreeITCase.java
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/PlannerScalaFreeITCase.java
@@ -28,7 +28,7 @@ import java.util.Arrays;
import java.util.List;
/**
- * End to End tests for table planner scala-free since 1.15. Due to scala-free of table planner
+ * End-to-End tests for table planner scala-free since 1.15. Due to scala-free of table planner
* introduced, the class in table planner is not visible in distribution runtime, if we use these
* class in execution time, ClassNotFound exception will be thrown. ITCase in table planner can not
* cover it, so we should add E2E test for these case.
@@ -42,8 +42,6 @@ public class PlannerScalaFreeITCase extends SqlITCaseBase {
public void testImperativeUdaf() throws Exception {
runAndCheckSQL(
"scala_free_e2e.sql",
- generateReplaceVars(),
- 2,
Arrays.asList(
"{\"before\":null,\"after\":{\"user_name\":\"Alice\",\"order_cnt\":1},\"op\":\"c\"}",
"{\"before\":null,\"after\":{\"user_name\":\"Bob\",\"order_cnt\":2},\"op\":\"c\"}"));
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/SqlITCaseBase.java b/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/SqlITCaseBase.java
index aadbb4b154d..683799281a7 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/SqlITCaseBase.java
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/SqlITCaseBase.java
@@ -45,6 +45,7 @@ import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -73,9 +74,9 @@ public abstract class SqlITCaseBase extends TestLogger {
@Rule public final TemporaryFolder tmp = new TemporaryFolder();
- private final String executionMode;
+ protected final String executionMode;
- private Path result;
+ protected Path result;
protected static final Path SQL_TOOL_BOX_JAR =
ResourceTestUtils.getResource(".*SqlToolbox.jar");
@@ -98,17 +99,22 @@ public abstract class SqlITCaseBase extends TestLogger {
this.result = tmpPath.resolve(String.format("result-%s", UUID.randomUUID()));
}
- public void runAndCheckSQL(
- String sqlPath, Map<String, String> varsMap, int resultSize, List<String> resultItems)
+ public void runAndCheckSQL(String sqlPath, List<String> resultItems) throws Exception {
+ runAndCheckSQL(sqlPath, Collections.singletonMap(result, resultItems));
+ }
+
+ public void runAndCheckSQL(String sqlPath, Map<Path, List<String>> resultItems)
throws Exception {
try (ClusterController clusterController = flink.startCluster(1)) {
- List<String> sqlLines = initializeSqlLines(sqlPath, varsMap);
+ List<String> sqlLines = initializeSqlLines(sqlPath);
executeSqlStatements(clusterController, sqlLines);
// Wait until all the results flushed to the json file.
- LOG.info("Verify the json result.");
- checkJsonResultFile(resultSize, resultItems);
+ LOG.info("Verify the result.");
+ for (Map.Entry<Path, List<String>> entry : resultItems.entrySet()) {
+ checkResultFile(entry.getKey(), entry.getValue());
+ }
LOG.info("The SQL client test run successfully.");
}
}
@@ -123,13 +129,12 @@ public abstract class SqlITCaseBase extends TestLogger {
protected abstract void executeSqlStatements(
ClusterController clusterController, List<String> sqlLines) throws Exception;
- private List<String> initializeSqlLines(String sqlPath, Map<String, String> vars)
- throws IOException {
+ private List<String> initializeSqlLines(String sqlPath) throws IOException {
URL url = SqlITCaseBase.class.getClassLoader().getResource(sqlPath);
if (url == null) {
throw new FileNotFoundException(sqlPath);
}
-
+ Map<String, String> vars = generateReplaceVars();
List<String> lines = Files.readAllLines(new File(url.getFile()).toPath());
List<String> result = new ArrayList<>();
for (String line : lines) {
@@ -142,26 +147,28 @@ public abstract class SqlITCaseBase extends TestLogger {
return result;
}
- private void checkJsonResultFile(int resultSize, List<String> items) throws Exception {
+ private static void checkResultFile(Path resultPath, List<String> expectedItems)
+ throws Exception {
boolean success = false;
final Deadline deadline = Deadline.fromNow(Duration.ofSeconds(20));
List<String> lines = null;
+ int resultSize = expectedItems.size();
while (deadline.hasTimeLeft()) {
- if (Files.exists(result)) {
- lines = readJsonResultFiles(result);
+ if (Files.exists(resultPath)) {
+ lines = readResultFiles(resultPath);
if (lines.size() == resultSize) {
success = true;
- assertThat(lines).hasSameElementsAs(items);
+ assertThat(lines).hasSameElementsAs(expectedItems);
break;
} else {
LOG.info(
- "The target Json {} does not contain enough records, current {} records, left time: {}s",
- result,
+ "The target result {} does not contain enough records, current {} records, left time: {}s",
+ resultPath,
lines.size(),
deadline.timeLeft().getSeconds());
}
} else {
- LOG.info("The target Json {} does not exist now", result);
+ LOG.info("The target result {} does not exist now", resultPath);
}
Thread.sleep(500);
}
@@ -171,13 +178,13 @@ public abstract class SqlITCaseBase extends TestLogger {
"Did not get expected results before timeout, actual result: %s.", lines));
}
- private static List<String> readJsonResultFiles(Path path) throws IOException {
+ private static List<String> readResultFiles(Path path) throws IOException {
File filePath = path.toFile();
// list all the non-hidden files
- File[] csvFiles = filePath.listFiles((dir, name) -> !name.startsWith("."));
+ File[] files = filePath.listFiles((dir, name) -> !name.startsWith("."));
List<String> result = new ArrayList<>();
- if (csvFiles != null) {
- for (File file : csvFiles) {
+ if (files != null) {
+ for (File file : files) {
result.addAll(Files.readAllLines(file.toPath()));
}
}
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/UsingRemoteJarITCase.java b/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/UsingRemoteJarITCase.java
index 81018b279c7..6ca68b7cfd9 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/UsingRemoteJarITCase.java
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/UsingRemoteJarITCase.java
@@ -18,38 +18,20 @@
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.flink.util.FileUtils;
-import org.apache.flink.util.OperatingSystem;
-
-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.Assume;
-import org.junit.Before;
-import org.junit.BeforeClass;
+import org.junit.Rule;
import org.junit.Test;
+import org.junit.rules.TestName;
-import java.io.File;
-import java.io.FileNotFoundException;
import java.io.IOException;
-import java.nio.file.Path;
-import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
-import java.util.List;
import java.util.Map;
-/** End to End tests for using remote jar. */
-public class UsingRemoteJarITCase extends SqlITCaseBase {
- private static final Path HADOOP_CLASSPATH =
- ResourceTestUtils.getResource(".*hadoop.classpath");
+/** End-to-End tests for using remote jar. */
+public class UsingRemoteJarITCase extends HdfsITCaseBase {
- private MiniDFSCluster hdfsCluster;
+ @Rule public TestName name = new TestName();
private org.apache.hadoop.fs.Path hdPath;
private org.apache.hadoop.fs.FileSystem hdfs;
@@ -57,56 +39,34 @@ public class UsingRemoteJarITCase extends SqlITCaseBase {
super(executionMode);
}
- @BeforeClass
- public static void verifyOS() {
- Assume.assumeTrue(
- "HDFS cluster cannot be started on Windows without extensions.",
- !OperatingSystem.isWindows());
- }
-
- @Before
- public void before() throws Exception {
- super.before();
- createHDFS();
- }
-
- private void createHDFS() {
+ @Override
+ protected void createHDFS() {
+ super.createHDFS();
+ hdPath = new org.apache.hadoop.fs.Path("/test.jar");
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.jar");
hdfs = hdPath.getFileSystem(hdConf);
hdfs.copyFromLocalFile(
new org.apache.hadoop.fs.Path(SQL_TOOL_BOX_JAR.toString()), hdPath);
- } catch (Throwable e) {
- e.printStackTrace();
- Assert.fail("Test failed " + e.getMessage());
+ } catch (IOException e) {
+ Assert.fail("Failed to copy local test.jar to HDFS env" + e.getMessage());
}
}
- @After
- public void destroyHDFS() {
+ @Override
+ protected void destroyHDFS() {
try {
hdfs.delete(hdPath, false);
- hdfsCluster.shutdown();
} catch (IOException e) {
- throw new RuntimeException(e);
+ Assert.fail("Failed to cleanup HDFS path" + e.getMessage());
}
+ super.destroyHDFS();
}
@Test
public void testUdfInRemoteJar() throws Exception {
runAndCheckSQL(
"remote_jar_e2e.sql",
- generateReplaceVars(),
- 2,
Arrays.asList(
"{\"before\":null,\"after\":{\"user_name\":\"Alice\",\"order_cnt\":1},\"op\":\"c\"}",
"{\"before\":null,\"after\":{\"user_name\":\"Bob\",\"order_cnt\":2},\"op\":\"c\"}"));
@@ -116,20 +76,14 @@ public class UsingRemoteJarITCase extends SqlITCaseBase {
public void testScalarUdfWhenCheckpointEnable() throws Exception {
runAndCheckSQL(
"scalar_udf_e2e.sql",
- generateReplaceVars(),
- 1,
Collections.singletonList(
"{\"before\":null,\"after\":{\"id\":1,\"str\":\"Hello Flink\"},\"op\":\"c\"}"));
}
@Test
public void testCreateTemporarySystemFunctionUsingRemoteJar() throws Exception {
- Map<String, String> replaceVars = generateReplaceVars();
- replaceVars.put("$TEMPORARY", "TEMPORARY SYSTEM");
runAndCheckSQL(
"create_function_using_remote_jar_e2e.sql",
- replaceVars,
- 2,
Arrays.asList(
"{\"before\":null,\"after\":{\"user_name\":\"Alice\",\"order_cnt\":1},\"op\":\"c\"}",
"{\"before\":null,\"after\":{\"user_name\":\"Bob\",\"order_cnt\":2},\"op\":\"c\"}"));
@@ -137,12 +91,8 @@ public class UsingRemoteJarITCase extends SqlITCaseBase {
@Test
public void testCreateCatalogFunctionUsingRemoteJar() throws Exception {
- Map<String, String> replaceVars = generateReplaceVars();
- replaceVars.put("$TEMPORARY", "");
runAndCheckSQL(
"create_function_using_remote_jar_e2e.sql",
- replaceVars,
- 2,
Arrays.asList(
"{\"before\":null,\"after\":{\"user_name\":\"Alice\",\"order_cnt\":1},\"op\":\"c\"}",
"{\"before\":null,\"after\":{\"user_name\":\"Bob\",\"order_cnt\":2},\"op\":\"c\"}"));
@@ -154,8 +104,6 @@ public class UsingRemoteJarITCase extends SqlITCaseBase {
replaceVars.put("$TEMPORARY", "TEMPORARY");
runAndCheckSQL(
"create_function_using_remote_jar_e2e.sql",
- replaceVars,
- 2,
Arrays.asList(
"{\"before\":null,\"after\":{\"user_name\":\"Alice\",\"order_cnt\":1},\"op\":\"c\"}",
"{\"before\":null,\"after\":{\"user_name\":\"Bob\",\"order_cnt\":2},\"op\":\"c\"}"));
@@ -168,46 +116,16 @@ public class UsingRemoteJarITCase extends SqlITCaseBase {
"hdfs://%s:%s/%s",
hdfsCluster.getURI().getHost(), hdfsCluster.getNameNodePort(), hdPath);
- Map<String, String> map = super.generateReplaceVars();
- map.put("$JAR_PATH", remoteJarPath);
- return map;
- }
-
- @Override
- protected void executeSqlStatements(ClusterController clusterController, List<String> sqlLines)
- throws Exception {
- clusterController.submitSQLJob(
- new SQLJobSubmission.SQLJobSubmissionBuilder(sqlLines)
- .setEnvProcessor(
- map -> map.put("HADOOP_CLASSPATH", getHadoopClassPathContent()))
- .build(),
- Duration.ofMinutes(2L));
- }
-
- private String getHadoopClassPathContent() {
- // Prepare all hadoop jars to mock HADOOP_CLASSPATH, use hadoop.classpath which contains all
- // hadoop jars
- File hadoopClasspathFile = new File(HADOOP_CLASSPATH.toAbsolutePath().toString());
-
- if (!hadoopClasspathFile.exists()) {
- try {
- throw new FileNotFoundException(
- String.format(
- "File that contains hadoop classpath %s does not exist.",
- HADOOP_CLASSPATH));
- } catch (FileNotFoundException e) {
- e.printStackTrace();
- Assert.fail("Test failed " + e.getMessage());
- }
- }
-
- String classPathContent = null;
- try {
- classPathContent = FileUtils.readFileUtf8(hadoopClasspathFile);
- } catch (IOException e) {
- e.printStackTrace();
- Assert.fail("Test failed " + e.getMessage());
+ Map<String, String> varsMap = super.generateReplaceVars();
+ varsMap.put("$JAR_PATH", remoteJarPath);
+ String methodName = name.getMethodName();
+ if (methodName.startsWith("testCreateTemporarySystemFunction")) {
+ varsMap.put("$TEMPORARY", "TEMPORARY SYSTEM");
+ } else if (methodName.startsWith("testCreateTemporaryCatalogFunction")) {
+ varsMap.put("$TEMPORARY", "TEMPORARY");
+ } else if (methodName.startsWith("testCreateCatalogFunction")) {
+ varsMap.put("$TEMPORARY", "");
}
- return classPathContent;
+ return varsMap;
}
}
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/resources/compile_and_execute_plan_e2e.sql b/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/resources/compile_and_execute_plan_e2e.sql
new file mode 100644
index 00000000000..733a7aae0a7
--- /dev/null
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-sql/src/test/resources/compile_and_execute_plan_e2e.sql
@@ -0,0 +1,50 @@
+/*
+ * 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' = 'streaming';
+SET 'table.dml-sync' = 'true';
+
+CREATE TABLE $TABLE1 (
+ msg_id INTEGER,
+ msg STRING
+) WITH (
+ 'connector' = 'filesystem',
+ 'sink.rolling-policy.rollover-interval' = '2s',
+ 'sink.rolling-policy.check-interval' = '2s',
+ 'path' = '$RESULT/$TABLE1',
+ 'format' = 'csv');
+
+CREATE TABLE $TABLE2 (
+ emp_id INTEGER,
+ emp_name STRING
+) WITH (
+ 'connector' = 'filesystem',
+ 'sink.rolling-policy.rollover-interval' = '2s',
+ 'sink.rolling-policy.check-interval' = '2s',
+ 'path' = '$RESULT/$TABLE2',
+ 'format' = 'csv');
+
+COMPILE PLAN '$REMOTE_PLAN_DIR/$TABLE1.json' FOR INSERT INTO $TABLE1 SELECT * FROM (VALUES (1, 'Hello'), (2, 'Flink'), (3, 'Bye')) T(msg_id, msg);
+
+SET 'table.plan.force-recompile' = 'true';
+
+COMPILE PLAN '$REMOTE_PLAN_DIR/$TABLE1.json' FOR INSERT INTO $TABLE1 SELECT * FROM (VALUES (1, 'Meow'), (2, 'Purr')) T(msg_id, msg);
+
+EXECUTE PLAN '$REMOTE_PLAN_DIR/$TABLE1.json';
+
+COMPILE AND EXECUTE PLAN'$REMOTE_PLAN_DIR/$TABLE2.json' FOR INSERT INTO $TABLE2 SELECT * FROM (VALUES (1, 'Tom'), (2, 'Jerry')) T(emp_id, emp_name);
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/resource/ClientResourceManager.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/resource/ClientResourceManager.java
index d57ff08f31e..13e64b6f499 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/resource/ClientResourceManager.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/resource/ClientResourceManager.java
@@ -51,7 +51,7 @@ public class ClientResourceManager extends ResourceManager {
public URL unregisterJarResource(String jarPath) {
Path path = new Path(jarPath);
try {
- checkJarPath(path);
+ checkPath(path, ResourceType.JAR);
return resourceInfos.remove(
new ResourceUri(ResourceType.JAR, getURLFromPath(path).getPath()));
} catch (IOException e) {
diff --git a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/AbstractSqlGatewayStatementITCase.java b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/AbstractSqlGatewayStatementITCase.java
index f6c35e9ff7c..45474390faa 100644
--- a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/AbstractSqlGatewayStatementITCase.java
+++ b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/AbstractSqlGatewayStatementITCase.java
@@ -132,6 +132,13 @@ public abstract class AbstractSqlGatewayStatementITCase extends AbstractTestBase
Files.createDirectory(temporaryFolder.resolve("streaming_compiled_plan"))
.toFile()
.getPath());
+ replaceVars.put(
+ "$VAR_STREAMING_PLAN_RELATIVE_PATH",
+ new File(".")
+ .getCanonicalFile()
+ .toPath()
+ .relativize(Paths.get(replaceVars.get("$VAR_STREAMING_PLAN_PATH")))
+ .toString());
}
@TestTemplate
diff --git a/flink-table/flink-sql-gateway/src/test/resources/sql/insert.q b/flink-table/flink-sql-gateway/src/test/resources/sql/insert.q
index 1314e4d31bd..ddaff5e1f34 100644
--- a/flink-table/flink-sql-gateway/src/test/resources/sql/insert.q
+++ b/flink-table/flink-sql-gateway/src/test/resources/sql/insert.q
@@ -108,7 +108,8 @@ create table StreamingTableForPlan (
1 row in set
!ok
-COMPILE AND EXECUTE PLAN '$VAR_STREAMING_PLAN_PATH/plan1.json' FOR INSERT INTO StreamingTableForPlan SELECT * FROM (VALUES (1, 'Hello'));
+# test path with scheme
+COMPILE AND EXECUTE PLAN 'file://$VAR_STREAMING_PLAN_PATH/plan1.json' FOR INSERT INTO StreamingTableForPlan SELECT * FROM (VALUES (1, 'Hello'));
!output
Job ID:
!info
@@ -134,6 +135,7 @@ SELECT * FROM StreamingTableForPlan;
!ok
+# test absolute path without scheme
COMPILE PLAN '$VAR_STREAMING_PLAN_PATH/plan2.json' FOR INSERT INTO StreamingTableForPlan SELECT * FROM (VALUES (1, 'Hello'));
!output
+--------+
@@ -144,7 +146,8 @@ COMPILE PLAN '$VAR_STREAMING_PLAN_PATH/plan2.json' FOR INSERT INTO StreamingTabl
1 row in set
!ok
-EXECUTE PLAN '$VAR_STREAMING_PLAN_PATH/plan2.json';
+# test relative path without scheme
+EXECUTE PLAN '$VAR_STREAMING_PLAN_RELATIVE_PATH/plan2.json';
!output
Job ID:
!info
diff --git a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java
index 425e41d0abe..c4ce3cac29c 100644
--- a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java
+++ b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java
@@ -2214,6 +2214,8 @@ class FlinkSqlParserImplTest extends SqlParserTest {
sql("execute plan './test.json'").ok("EXECUTE PLAN './test.json'");
sql("execute plan '/some/absolute/dir/plan.json'")
.ok("EXECUTE PLAN '/some/absolute/dir/plan.json'");
+ sql("execute plan 'file:///foo/bar/test.json'")
+ .ok("EXECUTE PLAN 'file:///foo/bar/test.json'");
}
@Test
@@ -2228,6 +2230,11 @@ class FlinkSqlParserImplTest extends SqlParserTest {
"COMPILE PLAN './test.json' IF NOT EXISTS FOR INSERT INTO `T1`\n"
+ "(SELECT *\n"
+ "FROM `T2`)");
+ sql("compile plan 'file:///foo/bar/test.json' if not exists for insert into t1 select * from t2")
+ .ok(
+ "COMPILE PLAN 'file:///foo/bar/test.json' IF NOT EXISTS FOR INSERT INTO `T1`\n"
+ + "(SELECT *\n"
+ + "FROM `T2`)");
sql("compile plan './test.json' for statement set "
+ "begin insert into t1 select * from t2; insert into t2 select * from t3; end")
@@ -2255,6 +2262,20 @@ class FlinkSqlParserImplTest extends SqlParserTest {
+ "FROM `T3`)\n"
+ ";\n"
+ "END");
+
+ sql("compile plan 'file:///foo/bar/test.json' if not exists for statement set "
+ + "begin insert into t1 select * from t2; insert into t2 select * from t3; end")
+ .ok(
+ "COMPILE PLAN 'file:///foo/bar/test.json' IF NOT EXISTS FOR STATEMENT SET BEGIN\n"
+ + "INSERT INTO `T1`\n"
+ + "(SELECT *\n"
+ + "FROM `T2`)\n"
+ + ";\n"
+ + "INSERT INTO `T2`\n"
+ + "(SELECT *\n"
+ + "FROM `T3`)\n"
+ + ";\n"
+ + "END");
}
@Test
@@ -2278,6 +2299,11 @@ class FlinkSqlParserImplTest extends SqlParserTest {
+ "FROM `T3`)\n"
+ ";\n"
+ "END");
+ sql("compile and execute plan 'file:///foo/bar/test.json' for insert into t1 select * from t2")
+ .ok(
+ "COMPILE AND EXECUTE PLAN 'file:///foo/bar/test.json' FOR INSERT INTO `T1`\n"
+ + "(SELECT *\n"
+ + "FROM `T2`)");
}
@Test
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
index 2132e70acd8..75198df1c8a 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
@@ -24,6 +24,7 @@ import org.apache.flink.api.dag.Pipeline;
import org.apache.flink.api.dag.Transformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.core.fs.Path;
import org.apache.flink.table.api.CompiledPlan;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.EnvironmentSettings;
@@ -91,6 +92,7 @@ import org.apache.flink.table.operations.ddl.AnalyzeTableOperation;
import org.apache.flink.table.operations.ddl.CompilePlanOperation;
import org.apache.flink.table.operations.utils.OperationTreeBuilder;
import org.apache.flink.table.resource.ResourceManager;
+import org.apache.flink.table.resource.ResourceType;
import org.apache.flink.table.resource.ResourceUri;
import org.apache.flink.table.sinks.TableSink;
import org.apache.flink.table.sources.TableSource;
@@ -104,10 +106,8 @@ import org.apache.flink.util.FlinkUserCodeClassLoaders;
import org.apache.flink.util.MutableURLClassLoader;
import org.apache.flink.util.Preconditions;
-import java.io.File;
import java.io.IOException;
import java.net.URL;
-import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -731,38 +731,47 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
}
private CompiledPlan compilePlanAndWrite(
- String filePath, boolean ifNotExists, Operation operation) {
- File file = Paths.get(filePath).toFile();
- if (file.exists()) {
- if (ifNotExists) {
- return loadPlan(PlanReference.fromFile(filePath));
+ String pathString, boolean ignoreIfExists, Operation operation) {
+ try {
+ ResourceUri planResource = new ResourceUri(ResourceType.FILE, pathString);
+ Path planPath = new Path(pathString);
+ if (resourceManager.exists(planPath)) {
+ if (ignoreIfExists) {
+ return loadPlan(
+ PlanReference.fromFile(
+ resourceManager.registerFileResource(planResource)));
+ }
+
+ if (!tableConfig.get(TableConfigOptions.PLAN_FORCE_RECOMPILE)) {
+ throw new TableException(
+ String.format(
+ "Cannot overwrite the plan file '%s'. "
+ + "Either manually remove the file or, "
+ + "if you're debugging your job, "
+ + "set the option '%s' to true.",
+ pathString, TableConfigOptions.PLAN_FORCE_RECOMPILE.key()));
+ }
}
- if (!tableConfig.get(TableConfigOptions.PLAN_FORCE_RECOMPILE)) {
+ CompiledPlan compiledPlan;
+ if (operation instanceof StatementSetOperation) {
+ compiledPlan = compilePlan(((StatementSetOperation) operation).getOperations());
+ } else if (operation instanceof ModifyOperation) {
+ compiledPlan = compilePlan(Collections.singletonList((ModifyOperation) operation));
+ } else {
throw new TableException(
- String.format(
- "Cannot overwrite the plan file '%s'. "
- + "Either manually remove the file or, "
- + "if you're debugging your job, "
- + "set the option '%s' to true.",
- filePath, TableConfigOptions.PLAN_FORCE_RECOMPILE.key()));
+ "Unsupported operation to compile: "
+ + operation.getClass()
+ + ". This is a bug, please file an issue.");
}
- }
-
- CompiledPlan compiledPlan;
- if (operation instanceof StatementSetOperation) {
- compiledPlan = compilePlan(((StatementSetOperation) operation).getOperations());
- } else if (operation instanceof ModifyOperation) {
- compiledPlan = compilePlan(Collections.singletonList((ModifyOperation) operation));
- } else {
+ resourceManager.syncFileResource(
+ planResource, path -> compiledPlan.writeToFile(path, false));
+ return compiledPlan;
+ } catch (IOException e) {
throw new TableException(
- "Unsupported operation to compile: "
- + operation.getClass()
- + ". This is a bug, please file an issue.");
+ String.format("Failed to execute %s statement.", operation.asSummaryString()),
+ e);
}
-
- compiledPlan.writeToFile(file, false);
- return compiledPlan;
}
@Override
@@ -938,8 +947,20 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
return executeQueryOperation((QueryOperation) operation);
} else if (operation instanceof ExecutePlanOperation) {
ExecutePlanOperation executePlanOperation = (ExecutePlanOperation) operation;
- return (TableResultInternal)
- executePlan(PlanReference.fromFile(executePlanOperation.getFilePath()));
+ try {
+ return (TableResultInternal)
+ executePlan(
+ PlanReference.fromFile(
+ resourceManager.registerFileResource(
+ new ResourceUri(
+ ResourceType.FILE,
+ executePlanOperation.getFilePath()))));
+ } catch (IOException e) {
+ throw new TableException(
+ String.format(
+ "Failed to execute %s statement.", operation.asSummaryString()),
+ e);
+ }
} else if (operation instanceof CompilePlanOperation) {
CompilePlanOperation compilePlanOperation = (CompilePlanOperation) operation;
compilePlanAndWrite(
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/resource/ResourceManager.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/resource/ResourceManager.java
index 7af0239b412..04e789b07ee 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/resource/ResourceManager.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/resource/ResourceManager.java
@@ -52,6 +52,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
+import java.util.function.Consumer;
import java.util.stream.Collectors;
/** A manager for dealing with all user defined resource. */
@@ -89,52 +90,43 @@ public class ResourceManager implements Closeable {
* register them into the {@link ResourceManager}.
*/
public void registerJarResources(List<ResourceUri> resourceUris) throws IOException {
- // check jar resource before register
- checkJarResources(resourceUris);
-
- Map<ResourceUri, URL> stagingResourceLocalURLs = new HashMap<>();
- for (ResourceUri resourceUri : resourceUris) {
- // check whether the resource has been registered
- if (resourceInfos.containsKey(resourceUri) && resourceInfos.get(resourceUri) != null) {
- LOG.info(
- "Resource [{}] has been registered, overwriting of registered resource is not supported "
- + "in the current version, skipping.",
- resourceUri.getUri());
- continue;
- }
-
- // here can check whether the resource path is valid
- Path path = new Path(resourceUri.getUri());
- URL localUrl;
- // check resource scheme
- String scheme = StringUtils.lowerCase(path.toUri().getScheme());
- // download resource to local path firstly if in remote
- if (scheme != null && !FILE_SCHEME.equals(scheme)) {
- localUrl = downloadResource(path);
- } else {
- localUrl = getURLFromPath(path);
- // if the local jar resource is a relative path, here convert it to absolute path
- // before register
- resourceUri = new ResourceUri(ResourceType.JAR, localUrl.getPath());
- }
-
- // check the local jar file
- JarUtils.checkJarFile(localUrl);
-
- // add it to staging map
- stagingResourceLocalURLs.put(resourceUri, localUrl);
- }
-
- // register resource in batch
- stagingResourceLocalURLs.forEach(
- (resourceUri, url) -> {
- // jar resource need add to classloader
- userClassLoader.addURL(url);
- LOG.info("Added jar resource [{}] to class path.", url);
+ registerResources(
+ prepareStagingResources(
+ resourceUris,
+ ResourceType.JAR,
+ true,
+ url -> {
+ try {
+ JarUtils.checkJarFile(url);
+ } catch (IOException e) {
+ throw new ValidationException(
+ String.format("Failed to register jar resource [%s]", url),
+ e);
+ }
+ }),
+ true);
+ }
- resourceInfos.put(resourceUri, url);
- LOG.info("Register resource [{}] successfully.", resourceUri.getUri());
- });
+ /**
+ * Register a file resource into {@link ResourceManager} and return the absolute local file path
+ * without the scheme.
+ *
+ * <p>If the file is remote, it will be copied to a local file, with file name suffixed with a
+ * UUID.
+ *
+ * @param resourceUri resource with type as {@link ResourceType#FILE}, the resource uri might or
+ * might not contain the uri scheme, or it could be a relative path.
+ * @return the absolute local file path.
+ */
+ public String registerFileResource(ResourceUri resourceUri) throws IOException {
+ Map<ResourceUri, URL> stagingResources =
+ prepareStagingResources(
+ Collections.singletonList(resourceUri),
+ ResourceType.FILE,
+ false,
+ url -> {});
+ registerResources(stagingResources, false);
+ return resourceInfos.get(new ArrayList<>(stagingResources.keySet()).get(0)).getPath();
}
public URLClassLoader getUserClassLoader() {
@@ -203,54 +195,80 @@ public class ResourceManager implements Closeable {
}
}
- private void checkJarResources(List<ResourceUri> resourceUris) throws IOException {
- // only support register jar resource
- if (resourceUris.stream()
- .anyMatch(resourceUri -> !ResourceType.JAR.equals(resourceUri.getResourceType()))) {
- throw new ValidationException(
- String.format(
- "Only support to register jar resource, resource info:\n %s.",
- resourceUris.stream()
- .map(ResourceUri::getUri)
- .collect(Collectors.joining(",\n"))));
- }
+ /** Check whether the {@link Path} exists. */
+ public boolean exists(Path filePath) throws IOException {
+ return filePath.getFileSystem().exists(filePath);
+ }
- for (ResourceUri resourceUri : resourceUris) {
- checkJarPath(new Path(resourceUri.getUri()));
+ /**
+ * Generate a local file resource by the given resource generator and then synchronize to the
+ * path identified by the given {@link ResourceUri}. The path passed to resource generator
+ * should be a local path retrieved from the given {@link ResourceUri}.
+ *
+ * <p>NOTE: if the given {@link ResourceUri} represents a remote file path like
+ * "hdfs://path/to/file.json", then the retrieved local path will be
+ * "/localResourceDir/file-${uuid}.json"
+ *
+ * @param resourceUri the file resource uri to synchronize to
+ * @param resourceGenerator a consumer that generates a local copy of the file resource
+ */
+ public void syncFileResource(ResourceUri resourceUri, Consumer<String> resourceGenerator)
+ throws IOException {
+ Path targetPath = new Path(resourceUri.getUri());
+ String localPath;
+ boolean remote = isRemotePath(targetPath);
+ if (remote) {
+ localPath = getResourceLocalPath(targetPath).getPath();
+ } else {
+ localPath = getURLFromPath(targetPath).getPath();
+ }
+ resourceGenerator.accept(localPath);
+ if (remote) {
+ if (exists(targetPath)) {
+ // FileUtils#copy will not do copy if targetPath already exists
+ targetPath.getFileSystem().delete(targetPath, false);
+ }
+ FileUtils.copy(new Path(localPath), targetPath, false);
}
}
- protected void checkJarPath(Path path) throws IOException {
- // file name should end with .jar suffix
- String fileExtension = Files.getFileExtension(path.getName());
- if (!fileExtension.toLowerCase().endsWith(JAR_SUFFIX)) {
- throw new ValidationException(
- String.format(
- "The registering or unregistering jar resource [%s] must ends with '.jar' suffix.",
- path));
- }
+ // ------------------------------------------------------------------------
+ protected void checkPath(Path path, ResourceType expectedType) throws IOException {
FileSystem fs = FileSystem.getUnguardedFileSystem(path.toUri());
// check resource exists firstly
if (!fs.exists(path)) {
- throw new FileNotFoundException(String.format("Jar resource [%s] not found.", path));
+ throw new FileNotFoundException(
+ String.format(
+ "%s resource [%s] not found.",
+ expectedType.name().toLowerCase(), path));
}
-
// register directory is not allowed for resource
if (fs.getFileStatus(path).isDir()) {
throw new ValidationException(
String.format(
- "The registering or unregistering jar resource [%s] is a directory that is not allowed.",
- path));
+ "The registering or unregistering %s resource [%s] is a directory that is not allowed.",
+ expectedType.name().toLowerCase(), path));
+ }
+
+ if (expectedType == ResourceType.JAR) {
+ // file name should end with .jar suffix
+ String fileExtension = Files.getFileExtension(path.getName());
+ if (!fileExtension.toLowerCase().endsWith(JAR_SUFFIX)) {
+ throw new ValidationException(
+ String.format(
+ "The registering or unregistering jar resource [%s] must ends with '.jar' suffix.",
+ path));
+ }
}
}
@VisibleForTesting
- URL downloadResource(Path remotePath) throws IOException {
- // get local resource path
+ URL downloadResource(Path remotePath, boolean executable) throws IOException {
+ // get a local resource path
Path localPath = getResourceLocalPath(remotePath);
try {
- FileUtils.copy(remotePath, localPath, true);
+ FileUtils.copy(remotePath, localPath, executable);
LOG.info(
"Download resource [{}] to local path [{}] successfully.",
remotePath,
@@ -279,6 +297,18 @@ public class ResourceManager implements Closeable {
return localResourceDir;
}
+ @VisibleForTesting
+ boolean isRemotePath(Path path) {
+ String scheme = path.toUri().getScheme();
+ if (scheme == null) {
+ // whether the default fs is configured as remote mode,
+ // see FileSystem#getDefaultFsUri()
+ return !FILE_SCHEME.equalsIgnoreCase(FileSystem.getDefaultFsUri().getScheme());
+ } else {
+ return !FILE_SCHEME.equalsIgnoreCase(scheme);
+ }
+ }
+
private Path getResourceLocalPath(Path remotePath) {
String fileName = remotePath.getName();
String fileExtension = Files.getFileExtension(fileName);
@@ -296,4 +326,97 @@ public class ResourceManager implements Closeable {
}
return new Path(localResourceDir, fileNameWithUUID);
}
+
+ private void checkResources(List<ResourceUri> resourceUris, ResourceType expectedType)
+ throws IOException {
+ // check the resource type
+ if (resourceUris.stream()
+ .anyMatch(resourceUri -> expectedType != resourceUri.getResourceType())) {
+ throw new ValidationException(
+ String.format(
+ "Expect the resource type to be %s, but encounter a resource %s.",
+ expectedType.name().toLowerCase(),
+ resourceUris.stream()
+ .filter(
+ resourceUri ->
+ expectedType != resourceUri.getResourceType())
+ .findFirst()
+ .map(
+ resourceUri ->
+ String.format(
+ "[%s] with type %s",
+ resourceUri.getUri(),
+ resourceUri
+ .getResourceType()
+ .name()
+ .toLowerCase()))
+ .get()));
+ }
+
+ // check the resource path
+ for (ResourceUri resourceUri : resourceUris) {
+ checkPath(new Path(resourceUri.getUri()), expectedType);
+ }
+ }
+
+ private Map<ResourceUri, URL> prepareStagingResources(
+ List<ResourceUri> resourceUris,
+ ResourceType expectedType,
+ boolean executable,
+ Consumer<URL> resourceChecker)
+ throws IOException {
+ checkResources(resourceUris, expectedType);
+
+ Map<ResourceUri, URL> stagingResourceLocalURLs = new HashMap<>();
+ boolean supportOverwrite = !executable;
+ for (ResourceUri resourceUri : resourceUris) {
+ // check whether the resource has been registered
+ if (resourceInfos.containsKey(resourceUri) && resourceInfos.get(resourceUri) != null) {
+ if (!supportOverwrite) {
+ LOG.info(
+ "Resource [{}] has been registered, overwriting of registered resource is not supported "
+ + "in the current version, skipping.",
+ resourceUri.getUri());
+ continue;
+ }
+ }
+
+ // here can check whether the resource path is valid
+ Path path = new Path(resourceUri.getUri());
+ URL localUrl;
+ // download resource to a local path firstly if in remote
+ if (isRemotePath(path)) {
+ localUrl = downloadResource(path, executable);
+ } else {
+ localUrl = getURLFromPath(path);
+ // if the local resource is a relative path, here convert it to an absolute path
+ // before register
+ resourceUri = new ResourceUri(expectedType, localUrl.getPath());
+ }
+
+ // check the local file
+ resourceChecker.accept(localUrl);
+
+ // add it to a staging map
+ stagingResourceLocalURLs.put(resourceUri, localUrl);
+ }
+ return stagingResourceLocalURLs;
+ }
+
+ private void registerResources(
+ Map<ResourceUri, URL> stagingResources, boolean addToClassLoader) {
+ // register resource in batch
+ stagingResources.forEach(
+ (resourceUri, url) -> {
+ if (addToClassLoader) {
+ userClassLoader.addURL(url);
+ LOG.info(
+ "Added {} resource [{}] to class path.",
+ resourceUri.getResourceType().name(),
+ url);
+ }
+ resourceInfos.put(resourceUri, url);
+ LOG.info("Register resource [{}] successfully.", resourceUri.getUri());
+ });
+ }
}
diff --git a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/resource/ResourceManagerTest.java b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/resource/ResourceManagerTest.java
index a8d72cb8de6..bc18ce9bfd1 100644
--- a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/resource/ResourceManagerTest.java
+++ b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/resource/ResourceManagerTest.java
@@ -19,6 +19,7 @@
package org.apache.flink.table.resource;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
import org.apache.flink.core.testutils.CommonTestUtils;
@@ -31,21 +32,31 @@ import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.MethodSource;
import java.io.File;
+import java.io.IOException;
import java.net.URL;
import java.net.URLClassLoader;
+import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
+import java.nio.file.StandardOpenOption;
import java.util.Arrays;
import java.util.Collections;
import java.util.Map;
+import java.util.stream.Stream;
import static org.apache.flink.table.utils.UserDefinedFunctions.GENERATED_LOWER_UDF_CLASS;
import static org.apache.flink.table.utils.UserDefinedFunctions.GENERATED_LOWER_UDF_CODE;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.fail;
/** Tests for {@link ResourceManager}. */
public class ResourceManagerTest {
@@ -53,6 +64,8 @@ public class ResourceManagerTest {
@TempDir private static File tempFolder;
private static File udfJar;
+ private static File file;
+
private ResourceManager resourceManager;
@BeforeAll
@@ -63,6 +76,8 @@ public class ResourceManagerTest {
"test-classloader-udf.jar",
GENERATED_LOWER_UDF_CLASS,
String.format(GENERATED_LOWER_UDF_CODE, GENERATED_LOWER_UDF_CLASS));
+ file = File.createTempFile("ResourceManagerTest", ".txt", tempFolder);
+ FileUtils.writeFileUtf8(file, "Hello World");
}
@BeforeEach
@@ -75,10 +90,11 @@ public class ResourceManagerTest {
@AfterEach
public void after() throws Exception {
resourceManager.close();
+ FileSystem.initialize(new Configuration(), null);
}
@Test
- public void testRegisterResource() throws Exception {
+ public void testRegisterJarResource() throws Exception {
URLClassLoader userClassLoader = resourceManager.getUserClassLoader();
// test class loading before register resource
@@ -105,6 +121,45 @@ public class ResourceManagerTest {
assertEquals(clazz1, clazz2);
}
+ @Test
+ public void testRegisterFileResource() throws Exception {
+ ResourceUri normalizedResource =
+ new ResourceUri(
+ ResourceType.FILE,
+ resourceManager.getURLFromPath(new Path(file.getPath())).getPath());
+
+ // register file resource, uri is formatted with "file" scheme prefix
+ String localFilePath =
+ resourceManager.registerFileResource(
+ new ResourceUri(ResourceType.FILE, "file://" + file.getPath()));
+ assertEquals(file.getPath(), localFilePath);
+ Map<ResourceUri, URL> actualResource =
+ Collections.singletonMap(
+ normalizedResource,
+ resourceManager.getURLFromPath(new Path(localFilePath)));
+ assertThat(resourceManager.getResources()).containsExactlyEntriesOf(actualResource);
+
+ // register the same file resource repeatedly, but without scheme
+ assertThat(
+ resourceManager.registerFileResource(
+ new ResourceUri(ResourceType.FILE, file.getPath())))
+ .isEqualTo(localFilePath);
+ assertThat(resourceManager.getResources()).containsExactlyEntriesOf(actualResource);
+
+ // register the same file resource repeatedly, use relative path as uri
+ assertThat(
+ resourceManager.registerFileResource(
+ new ResourceUri(
+ ResourceType.FILE,
+ new File(".")
+ .getCanonicalFile()
+ .toPath()
+ .relativize(file.toPath())
+ .toString())))
+ .isEqualTo(localFilePath);
+ assertThat(resourceManager.getResources()).containsExactlyEntriesOf(actualResource);
+ }
+
@Test
public void testRegisterResourceWithRelativePath() throws Exception {
URLClassLoader userClassLoader = resourceManager.getUserClassLoader();
@@ -142,12 +197,13 @@ public class ResourceManagerTest {
}
@Test
- public void testRegisterInvalidResource() throws Exception {
- final String fileUri = tempFolder.getAbsolutePath() + Path.SEPARATOR + "test-file";
+ public void testRegisterInvalidJarResource() throws Exception {
+ final String fileUri = file.getPath();
CommonTestUtils.assertThrows(
String.format(
- "Only support to register jar resource, resource info:\n %s.", fileUri),
+ "Expect the resource type to be jar, but encounter a resource [%s] with type %s.",
+ fileUri, ResourceType.FILE.name().toLowerCase()),
ValidationException.class,
() -> {
resourceManager.registerJarResources(
@@ -155,12 +211,24 @@ public class ResourceManagerTest {
return null;
});
+ // test register jar resource with invalid suffix
+ CommonTestUtils.assertThrows(
+ String.format(
+ "The registering or unregistering jar resource [%s] must ends with '.jar' suffix.",
+ fileUri),
+ ValidationException.class,
+ () -> {
+ resourceManager.registerJarResources(
+ Collections.singletonList(new ResourceUri(ResourceType.JAR, fileUri)));
+ return null;
+ });
+
// test register directory for jar resource
final String jarDir = tempFolder.getPath();
CommonTestUtils.assertThrows(
String.format(
- "The registering or unregistering jar resource [%s] must ends with '.jar' suffix.",
+ "The registering or unregistering jar resource [%s] is a directory that is not allowed.",
jarDir),
ValidationException.class,
() -> {
@@ -185,22 +253,76 @@ public class ResourceManagerTest {
});
}
- @Test
- public void testDownloadResource() throws Exception {
- Path srcPath = new Path(udfJar.getPath());
+ @MethodSource("provideResource")
+ @ParameterizedTest
+ public void testDownloadResource(String pathString, boolean executable) throws Exception {
+ Path srcPath = new Path(pathString);
// test download resource to local path
- URL localUrl = resourceManager.downloadResource(srcPath);
+ URL localUrl = resourceManager.downloadResource(srcPath, executable);
- byte[] expected = FileUtils.readAllBytes(udfJar.toPath());
+ byte[] expected = FileUtils.readAllBytes(Paths.get(pathString));
byte[] actual = FileUtils.readAllBytes(Paths.get(localUrl.toURI()));
assertArrayEquals(expected, actual);
}
+ @CsvSource({
+ "file://path/to/file,hdfs://foo/bar:9000/,false",
+ "/path/to/file,file://foo/bar/,false",
+ "../path/to/file,file://foo/bar/,false",
+ "/path/to/file,hdfs://foo/bar:9000/,true",
+ "../path/to/file,hdfs://foo/bar:9000/,true",
+ "hdfs://path/to/file,file://foo/bar/,true",
+ })
+ @ParameterizedTest
+ public void testIsRemotePath(String pathString, String defaultFsScheme, boolean remote) {
+ Configuration conf = new Configuration();
+ conf.set(CoreOptions.DEFAULT_FILESYSTEM_SCHEME, defaultFsScheme);
+ FileSystem.initialize(conf, null);
+ assertThat(resourceManager.isRemotePath(new Path(pathString))).isEqualTo(remote);
+ }
+
+ @Test
+ public void testSyncFileResource() throws Exception {
+ String targetUri = tempFolder.getAbsolutePath() + "foo/bar.txt";
+ ResourceUri resource = new ResourceUri(ResourceType.FILE, targetUri);
+ resourceManager.syncFileResource(
+ resource,
+ path -> {
+ try {
+ FileUtils.copy(new Path(file.getPath()), new Path(path), false);
+ } catch (IOException e) {
+ fail("Test failed.", e);
+ }
+ });
+ assertThat(FileUtils.readFileUtf8(new File(targetUri))).isEqualTo("Hello World");
+
+ // test overwrite
+ resourceManager.syncFileResource(
+ resource,
+ path -> {
+ try {
+ Files.write(
+ new File(targetUri).toPath(),
+ "Bye Bye".getBytes(StandardCharsets.UTF_8),
+ StandardOpenOption.CREATE,
+ StandardOpenOption.TRUNCATE_EXISTING,
+ StandardOpenOption.WRITE);
+ } catch (IOException e) {
+ fail("Test failed.", e);
+ }
+ });
+ assertThat(FileUtils.readFileUtf8(new File(targetUri))).isEqualTo("Bye Bye");
+ }
+
@Test
public void testCloseResourceManagerCleanDownloadedResources() throws Exception {
resourceManager.close();
FileSystem fileSystem = FileSystem.getLocalFileSystem();
assertFalse(fileSystem.exists(resourceManager.getLocalResourceDir()));
}
+
+ public static Stream<Arguments> provideResource() {
+ return Stream.of(Arguments.of(udfJar.getPath(), true), Arguments.of(file.getPath(), false));
+ }
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/resource/ResourceUri.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/resource/ResourceUri.java
index 56c1277cf31..34eaedcaf70 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/resource/ResourceUri.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/resource/ResourceUri.java
@@ -22,7 +22,7 @@ import org.apache.flink.annotation.PublicEvolving;
import java.util.Objects;
-/** Description of function resource information. */
+/** Description of resource information. */
@PublicEvolving
public class ResourceUri {
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/CompiledPlanITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/CompiledPlanITCase.java
index 91cd2d9fdf5..4ca0a5d038d 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/CompiledPlanITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/CompiledPlanITCase.java
@@ -26,6 +26,7 @@ import org.apache.flink.table.planner.utils.JsonPlanTestBase;
import org.apache.flink.table.planner.utils.JsonTestUtils;
import org.apache.flink.table.planner.utils.TableTestUtil;
+import org.apache.commons.io.FileUtils;
import org.junit.Before;
import org.junit.Test;
@@ -146,6 +147,23 @@ public class CompiledPlanITCase extends JsonPlanTestBase {
assertResult(DATA, sinkPath);
}
+ @Test
+ public void testCompileWriteToFilePathWithSchemeAndThenExecuteSql() throws Exception {
+ Path planPath = Paths.get(URI.create(getTempDirPath("plan")).getPath(), "plan.json");
+
+ File sinkPath = createSourceSinkTables();
+
+ tableEnv.executeSql(
+ String.format(
+ "COMPILE PLAN 'file://%s' FOR INSERT INTO sink SELECT * FROM src",
+ planPath.toAbsolutePath()));
+
+ tableEnv.executeSql(String.format("EXECUTE PLAN 'file://%s'", planPath.toAbsolutePath()))
+ .await();
+
+ assertResult(DATA, sinkPath);
+ }
+
@Test
public void testCompilePlan() throws Exception {
Path planPath =
[flink] 01/02: [FLINK-31956][table-planner] Fix COMPILE PLAN statement fails to write to file when parent directories do not exist
Posted by yu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yuxia pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit c83eb5cd5a632e1e25ee5cda897b6acf68b72e53
Author: Jane Chan <qi...@gmail.com>
AuthorDate: Sat Jun 17 19:01:03 2023 +0800
[FLINK-31956][table-planner] Fix COMPILE PLAN statement fails to write to file when parent directories do not exist
---
.../flink/table/planner/plan/ExecNodeGraphInternalPlan.java | 1 +
.../java/org/apache/flink/table/api/CompiledPlanITCase.java | 11 -----------
2 files changed, 1 insertion(+), 11 deletions(-)
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/ExecNodeGraphInternalPlan.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/ExecNodeGraphInternalPlan.java
index 71c9b13b04e..7f138c6517a 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/ExecNodeGraphInternalPlan.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/ExecNodeGraphInternalPlan.java
@@ -75,6 +75,7 @@ public class ExecNodeGraphInternalPlan implements InternalPlan {
}
}
try {
+ Files.createDirectories(file.toPath().getParent());
Files.write(
file.toPath(),
serializedPlan.getBytes(StandardCharsets.UTF_8),
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/CompiledPlanITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/CompiledPlanITCase.java
index b5485097aa6..91cd2d9fdf5 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/CompiledPlanITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/CompiledPlanITCase.java
@@ -26,7 +26,6 @@ import org.apache.flink.table.planner.utils.JsonPlanTestBase;
import org.apache.flink.table.planner.utils.JsonTestUtils;
import org.apache.flink.table.planner.utils.TableTestUtil;
-import org.apache.commons.io.FileUtils;
import org.junit.Before;
import org.junit.Test;
@@ -136,7 +135,6 @@ public class CompiledPlanITCase extends JsonPlanTestBase {
@Test
public void testCompileWriteToFileAndThenExecuteSql() throws Exception {
Path planPath = Paths.get(URI.create(getTempDirPath("plan")).getPath(), "plan.json");
- FileUtils.createParentDirectories(planPath.toFile());
File sinkPath = createSourceSinkTables();
@@ -153,7 +151,6 @@ public class CompiledPlanITCase extends JsonPlanTestBase {
Path planPath =
Paths.get(URI.create(getTempDirPath("plan")).getPath(), "plan.json")
.toAbsolutePath();
- FileUtils.createParentDirectories(planPath.toFile());
File sinkPath = createSourceSinkTables();
@@ -184,7 +181,6 @@ public class CompiledPlanITCase extends JsonPlanTestBase {
Path planPath =
Paths.get(URI.create(getTempDirPath("plan")).getPath(), "plan.json")
.toAbsolutePath();
- FileUtils.createParentDirectories(planPath.toFile());
createTestCsvSourceTable("src", DATA, COLUMNS_DEFINITION);
File sinkAPath = createTestCsvSinkTable("sinkA", COLUMNS_DEFINITION);
@@ -216,7 +212,6 @@ public class CompiledPlanITCase extends JsonPlanTestBase {
Path planPath =
Paths.get(URI.create(getTempDirPath("plan")).getPath(), "plan.json")
.toAbsolutePath();
- FileUtils.createParentDirectories(planPath.toFile());
File sinkPath = createSourceSinkTables();
@@ -249,7 +244,6 @@ public class CompiledPlanITCase extends JsonPlanTestBase {
Path planPath =
Paths.get(URI.create(getTempDirPath("plan")).getPath(), "plan.json")
.toAbsolutePath();
- FileUtils.createParentDirectories(planPath.toFile());
List<String> expectedData =
Arrays.asList(
@@ -289,7 +283,6 @@ public class CompiledPlanITCase extends JsonPlanTestBase {
Path planPath =
Paths.get(URI.create(getTempDirPath("plan")).getPath(), "plan.json")
.toAbsolutePath();
- FileUtils.createParentDirectories(planPath.toFile());
File sinkPath = createSourceSinkTables();
@@ -309,7 +302,6 @@ public class CompiledPlanITCase extends JsonPlanTestBase {
Path planPath =
Paths.get(URI.create(getTempDirPath("plan")).getPath(), "plan.json")
.toAbsolutePath();
- FileUtils.createParentDirectories(planPath.toFile());
createTestCsvSourceTable("src", DATA, COLUMNS_DEFINITION);
File sinkAPath = createTestCsvSinkTable("sinkA", COLUMNS_DEFINITION);
@@ -351,9 +343,6 @@ public class CompiledPlanITCase extends JsonPlanTestBase {
@Test
public void testPersistedConfigOption() throws Exception {
- Path planPath = Paths.get(URI.create(getTempDirPath("plan")).getPath(), "plan.json");
- FileUtils.createParentDirectories(planPath.toFile());
-
List<String> data =
Stream.concat(
DATA.stream(),