[FLINK-31956][table] Extend the CompiledPlan to read from/write to Fl… #22539

xishuaidelin · 2023-05-06T16:21:30Z

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)

…ink's FileSystem

flinkbot · 2023-05-06T16:25:17Z

CI report:

28343fa Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

LadyForest

Hi, @xishuaidelin, thanks for your contribution.
I have briefly reviewed the code, which may be inaccurate, but the interface should remain the same. What do you think?

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/CompiledPlan.java

...table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/InternalPlan.java

…esystem

xishuaidelin · 2023-05-17T02:21:12Z

@flinkbot run azure

LadyForest

Hi, @xishuaidelin, thanks for your update.
I think we may reuse org.apache.flink.table.resource.ResourceManager to take over the URI check and resource download/cleanup. You can take the impl for the CREATE FUNCTION USING JAR statement as a reference.

...k-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java

…VisibleForTesting

…VisibleForTesting

xishuaidelin · 2023-05-24T06:33:20Z

@flinkbot run azure

LadyForest

Thanks for your update.

However, it doesn't work as expected.

I tried on the s3 & local file systems, but both failed; please check.

...ble-planner/src/main/java/org/apache/flink/table/planner/plan/ExecNodeGraphInternalPlan.java

LadyForest · 2023-05-27T10:48:50Z

...k-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java

                                filePath, TableConfigOptions.PLAN_FORCE_RECOMPILE.key()));
            }
        }
-


revert this back

LadyForest · 2023-05-27T13:07:15Z

...k-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java

            if (ifNotExists) {
-                return loadPlan(PlanReference.fromFile(filePath));
+                if (fs.isDistributedFS()) {
+                    URL localUrl = resourceManager.downloadResource(filePath);


It's a little unnatural to do so. Can we wrap up this logic?

LadyForest · 2023-05-28T07:06:52Z

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.

Nit: you can use minio to mock s3

…pies at local.

LadyForest · 2023-06-08T09:34:14Z

...k-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java


-        compiledPlan.writeToFile(file, false);
+        compiledPlan.writeToFile(localPath, false);
+        resourceManager.updateFilePath(filePath);


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.

LadyForest · 2023-06-08T09:37:33Z

...able/flink-table-api-java/src/main/java/org/apache/flink/table/resource/ResourceManager.java

+    }
+
+    /**
+     * register the filePath of flink filesystem. If it is remote filesystem and the file exists


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.

...able/flink-table-api-java/src/main/java/org/apache/flink/table/resource/ResourceManager.java

LadyForest

Thank you for your update. However, the code quality is not yet at a level where it can be merged. As time is tight, I will take over your work and continue to make improvements. Can we review the changes together at a later time?

LadyForest · 2023-06-14T02:15:32Z

...ble/flink-table-planner/src/test/scala/org/apache/flink/table/api/TableEnvironmentTest.scala

+    sql = String.format(
+      "COMPILE and EXECUTE plan '%s' FOR INSERT INTO MySink SELECT * FROM MyTable",
+      path)
+    tableEnv.executeSql(sql)


What's the purpose of this test? It does not assert anything.

LadyForest · 2023-06-14T02:16:54Z

...ble/flink-table-planner/src/test/scala/org/apache/flink/table/api/TableEnvironmentTest.scala

+    val execEnv = StreamExecutionEnvironment.getExecutionEnvironment
+    execEnv.setParallelism(1)
+    val settings = EnvironmentSettings.newInstance().inStreamingMode().build()
+    val tableEnv = StreamTableEnvironment.create(execEnv, settings)


What's the purpose of re-creating a tableEnv?

LadyForest · 2023-06-15T09:23:03Z

...api-java/src/main/java/org/apache/flink/table/operations/CompileAndExecutePlanOperation.java

                "Child operation of CompileAndExecuteOperation must be either a "
                        + "ModifyOperation or a StatementSetOperation.");
-        this.filePath = filePath;
+        this.filePath = new Path(filePath);


Perhaps we don't need to change this.

LadyForest · 2023-06-15T09:23:12Z

...e-api-java/src/main/java/org/apache/flink/table/operations/command/ExecutePlanOperation.java

 public class ExecutePlanOperation implements Operation {

-    private final String filePath;
+    private final Path filePath;


LadyForest · 2023-06-15T09:23:21Z

...table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CompilePlanOperation.java

                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);


LadyForest · 2023-06-15T09:27:54Z

...ink-end-to-end-tests-sql/src/test/java/org/apache/flink/table/sql/codegen/SqlITCaseBase.java

+    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);


Why duplicate the method and assert nothing?

LadyForest · 2023-06-15T09:29:09Z

...-sql/src/test/java/org/apache/flink/table/sql/codegen/CompileAndExecuteRemoteFileITCase.java

+import java.util.Map;
+
+/** End-to-End tests for compile and execute remote file. */
+public class CompileAndExecuteRemoteFileITCase extends SqlITCaseBase {


This test shares the same HDFS configuration with UsingRemoteJarITCase. Maybe deserve a common base class

LadyForest · 2023-06-15T09:31:18Z

...-sql/src/test/java/org/apache/flink/table/sql/codegen/CompileAndExecuteRemoteFileITCase.java

+            MiniDFSCluster.Builder builder = new MiniDFSCluster.Builder(hdConf);
+            hdfsCluster = builder.build();
+
+            hdPath = new org.apache.hadoop.fs.Path("/test.json");


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.

LadyForest · 2023-06-15T09:34:43Z

...-sql/src/test/java/org/apache/flink/table/sql/codegen/CompileAndExecuteRemoteFileITCase.java

+            hdfs = hdPath.getFileSystem(hdConf);
+
+        } catch (Throwable e) {
+            e.printStackTrace();


I don't think it's a good practice to do so...

LadyForest · 2023-06-15T09:36:45Z

...-sql/src/test/java/org/apache/flink/table/sql/codegen/CompileAndExecuteRemoteFileITCase.java

+
+    @Test
+    public void testCompilePlanRemoteFile() throws Exception {
+        runSQL("compile_plan_use_remote_file_e2e.sql", generateReplaceVars());


The test does not assert anything...

LadyForest · 2023-06-15T09:46:34Z

...end-tests/flink-end-to-end-tests-sql/src/test/resources/compile_plan_use_remote_file_e2e.sql

+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+SET execution.runtime-mode = $MODE;


COMPILE PLAN statement is not supported under batch mode.

[FLINK-31956][table] Extend the CompiledPlan to read from/write to Fl…

7847fbb

…ink's FileSystem

flinkbot added component=TableSQL/Planner component=TableSQL/Client labels May 6, 2023

LadyForest requested changes May 11, 2023

View reviewed changes

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/CompiledPlan.java Outdated Show resolved Hide resolved

...table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/InternalPlan.java Outdated Show resolved Hide resolved

xishuaidelin and others added 4 commits May 12, 2023 15:42

[FLINK-31956][table] fix the interface and add the test for flink fil…

c7f9b9b

…esystem

[FLINK-31956][table] format

8f0022f

[FLINK-31956][table] format

56337c4

Merge branch 'apache:master' into flink-31956

f0330ff

LadyForest requested changes May 23, 2023

View reviewed changes

...k-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java Outdated Show resolved Hide resolved

xishuaidelin added 2 commits May 24, 2023 10:36

[FLINK-31956][table] restore the original interface

69d4f41

[FLINK-31956][table] fix violation of calling methods annotated with @…

7a0512d

…VisibleForTesting

LadyForest requested changes May 27, 2023

View reviewed changes

xishuaidelin and others added 4 commits June 7, 2023 11:54

Merge branch 'apache:master' into flink-31956

151b082

[FLINK-31956][table] bug fix

e5e9c8c

[FLINK-31956][table] bug fix for a remote file could have multiple co…

52e0791

…pies at local.

[FLINK-31956][table] fix checkstyle.

02a42e9

LadyForest requested changes Jun 8, 2023

View reviewed changes

xishuaidelin added 2 commits June 8, 2023 23:29

[FLINK-31956][table] update

10ca973

[FLINK-31956][table] format

28343fa

LadyForest reviewed Jun 15, 2023

View reviewed changes

xishuaidelin closed this Jun 28, 2023

[FLINK-31956][table] Extend the CompiledPlan to read from/write to Fl… #22539

[FLINK-31956][table] Extend the CompiledPlan to read from/write to Fl… #22539

Uh oh!

Conversation

xishuaidelin commented May 6, 2023

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

flinkbot commented May 6, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

LadyForest left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

xishuaidelin commented May 17, 2023

Uh oh!

LadyForest left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

xishuaidelin commented May 24, 2023

Uh oh!

LadyForest left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

LadyForest commented May 28, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

LadyForest left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

flinkbot commented May 6, 2023 •

edited

Loading

LadyForest left a comment •

edited

Loading

LadyForest commented May 28, 2023 •

edited

Loading