[Feature] Support Flink read / write data branch #3029

sunxiaojian · 2024-03-16T15:57:15Z

Purpose

Linked issue: close (#2861)

When writing and reading branch data for Flink, it was found that if the "TableCommitImpl newcommit (String commitUser, String branchName)" interface is added like this The parameter "branchName" has a wide range of impacts, and when adding features in the future, we will also consider whether to support the branch, which is a challenge for developers. Therefore, in order to make developers not need to worry about the branch function and not need to consider branch writing and reading when adding interfaces, I have made the following refactoring and optimization

Refactoring TagManager, SnapshotManager, and SchemeManager to support specifying default branch parameters,

The implementation logic of 'SnapshotManager snapshotManager()', 'TagManager tagManager()', and 'BranchManager branchManager()' directly specifies the default written branch, so that we no longer need to specify the 'branch' parameter when declaring new interfaces for subsequent interfaces.

Add necessary unit test tests, such as CDC testing, to support Flink CDC writing to branches
After this PR, remove interfaces with the branchName parameter such as' TableCommitImpl newcommit (String commitUser, String branchName) '

Tests

API and Format

Documentation

sunxiaojian · 2024-03-16T16:02:18Z

@FangYongs @schnappi17 PTAL cc @JingsongLi

sunxiaojian · 2024-03-27T06:31:24Z

@FangYongs @schnappi17 When you have time, please take a look at this PR

sunxiaojian · 2024-04-07T12:03:35Z

@FangYongs @schnappi17 Do we need to send an email to discuss the part of this optimization ?

schnappi17 · 2024-04-24T08:12:03Z

paimon-common/src/main/java/org/apache/paimon/CoreOptions.java

+    }
+
+    public static String branch(Map<String, String> options) {
+        if (options.containsKey(BRANCH.key())) {


Just use options.get() will handle with the default value

I don't get why...

Sorry, the previous description was incorrect, Most of the parameters for this method come from "org.apache.paimon.table.FileStoreTable #options"， 'branch' parameter may be empty.

schnappi17 · 2024-04-24T09:09:53Z

paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java

@@ -78,61 +79,75 @@ public class SchemaManager implements Serializable {

    private final FileIO fileIO;
    private final Path tableRoot;
-


No need to do redundant modifications.

schnappi17 · 2024-04-24T09:12:38Z

paimon-core/src/main/java/org/apache/paimon/schema/TableSchema.java

@@ -104,7 +104,8 @@ public TableSchema(
        this.highestFieldId = highestFieldId;
        this.partitionKeys = partitionKeys;
        this.primaryKeys = primaryKeys;
-        this.options = Collections.unmodifiableMap(options);
+        Objects.requireNonNull(options);


Why need this? The @param options is not marked as @NotNull

Why need this? The @param options is not marked as @NotNull

removed

schnappi17 · 2024-04-24T09:15:37Z

paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java

@@ -139,13 +138,13 @@ public RowKeyExtractor createRowKeyExtractor() {

    @Override
    public SnapshotReader newSnapshotReader() {
-        return newSnapshotReader(DEFAULT_MAIN_BRANCH);
+        return newSnapshotReader(CoreOptions.branch(options()));
    }

    @Override
    public SnapshotReader newSnapshotReader(String branchName) {


Seems the branchName here is not used.

Seems the branchName here is not used.

Yes, I will contribute a PR and delete it uniformly

schnappi17 · 2024-04-24T10:38:01Z

@JingsongLi Please help to take a look at this, thanks a lot!

schnappi17 · 2024-04-25T06:26:32Z

@FangYongs Please also help to take a look at~Thanks!

JingsongLi · 2024-05-14T06:12:56Z

paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java

@@ -83,6 +86,7 @@ protected AbstractCatalog(FileIO fileIO, Options options) {
        this.tableDefaultOptions =
                convertToPropertiesPrefixKey(options.toMap(), TABLE_DEFAULT_OPTION_PREFIX);
        this.catalogOptions = options;
+        this.branchName = options.get(CoreOptions.BRANCH);


What is the scope of branch option?
Catalog? Table? Why catalog need to know?

What is the scope of branch option? Catalog? Table? Why catalog need to know?

The scope is Table, but when using getTable() or manipulating the schema in the catalog, it is necessary to specify the branch.

JingsongLi · 2024-05-14T06:15:32Z

paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java

@@ -139,13 +138,13 @@ public RowKeyExtractor createRowKeyExtractor() {

    @Override
    public SnapshotReader newSnapshotReader() {
-        return newSnapshotReader(DEFAULT_MAIN_BRANCH);
+        return newSnapshotReader(CoreOptions.branch(options()));


just use coreOptions()?

just use coreOptions()?

Yes, the table already specified the branch during initialization

JingsongLi · 2024-05-14T06:17:27Z

paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java

@@ -85,55 +92,74 @@ public Path changelogDirectory() {
        return new Path(tablePath + "/changelog");
    }

-    public Path longLivedChangelogPath(long snapshotId) {
-        return new Path(tablePath + "/changelog/" + CHANGELOG_PREFIX + snapshotId);
+    public Path changelogDirectory(String branchName) {


What is the design? Why do need to pass branch here when there is already a branch in the constructor?

I have mostly opened up methods with and without branch parameters. The methods with branch parameters are mainly for BranchManager and test preparation

sunxiaojian · 2024-05-14T15:21:41Z

@JingsongLi PTAL

sunxiaojian force-pushed the support-flink-write-branch-v2 branch from 2a0fd59 to 2ab54f9 Compare March 16, 2024 15:59

sunxiaojian mentioned this pull request Mar 16, 2024

[Feature]Support flink write/read data branch #3026

Closed

sunxiaojian force-pushed the support-flink-write-branch-v2 branch from a395178 to 31894ce Compare March 16, 2024 16:13

sunxiaojian mentioned this pull request Mar 16, 2024

[core] Fix collecting wrong sorted runs when compacting on custom branch #2950

Open

sunxiaojian force-pushed the support-flink-write-branch-v2 branch 12 times, most recently from 60ec1f8 to 2796791 Compare March 23, 2024 15:17

sunxiaojian force-pushed the support-flink-write-branch-v2 branch from 2796791 to e16fa74 Compare March 25, 2024 02:22

sunxiaojian force-pushed the support-flink-write-branch-v2 branch from e16fa74 to b4da7b7 Compare March 27, 2024 11:53

sunxiaojian force-pushed the support-flink-write-branch-v2 branch 4 times, most recently from 36c1484 to c9069e6 Compare April 15, 2024 15:13

sunxiaojian force-pushed the support-flink-write-branch-v2 branch 5 times, most recently from e9005ef to c98f80b Compare April 18, 2024 13:18

sunxiaojian force-pushed the support-flink-write-branch-v2 branch 10 times, most recently from a98a702 to ed4a620 Compare April 19, 2024 09:03

schnappi17 reviewed Apr 24, 2024

View reviewed changes

sunxiaojian force-pushed the support-flink-write-branch-v2 branch 2 times, most recently from 01402fa to 5fb55e0 Compare April 24, 2024 09:52

sunxiaojian force-pushed the support-flink-write-branch-v2 branch from 5fb55e0 to 7714477 Compare April 25, 2024 01:59

sunxiaojian force-pushed the support-flink-write-branch-v2 branch 2 times, most recently from e7fbe3e to 734ecec Compare May 8, 2024 10:39

JingsongLi reviewed May 14, 2024

View reviewed changes

sunxiaojian force-pushed the support-flink-write-branch-v2 branch 3 times, most recently from 363691d to 1642152 Compare May 14, 2024 11:27

support flink write branch

ecbc3bc

sunxiaojian force-pushed the support-flink-write-branch-v2 branch from 1642152 to ecbc3bc Compare May 14, 2024 11:38

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Feature] Support Flink read / write data branch #3029

[Feature] Support Flink read / write data branch #3029

sunxiaojian commented Mar 16, 2024 •

edited

sunxiaojian commented Mar 16, 2024

sunxiaojian commented Mar 27, 2024

sunxiaojian commented Apr 7, 2024

schnappi17 Apr 24, 2024

JingsongLi May 14, 2024

sunxiaojian May 14, 2024

schnappi17 Apr 24, 2024

JingsongLi May 14, 2024

sunxiaojian May 14, 2024

schnappi17 Apr 24, 2024

sunxiaojian Apr 24, 2024

schnappi17 Apr 24, 2024

sunxiaojian Apr 24, 2024

schnappi17 commented Apr 24, 2024

schnappi17 commented Apr 25, 2024

JingsongLi May 14, 2024

sunxiaojian May 14, 2024

JingsongLi May 14, 2024

sunxiaojian May 14, 2024

JingsongLi May 14, 2024

sunxiaojian May 14, 2024

sunxiaojian commented May 14, 2024

		@@ -78,61 +79,75 @@ public class SchemaManager implements Serializable {

		private final FileIO fileIO;
		private final Path tableRoot;

[Feature] Support Flink read / write data branch #3029

Are you sure you want to change the base?

[Feature] Support Flink read / write data branch #3029

Conversation

sunxiaojian commented Mar 16, 2024 • edited

Purpose

Tests

API and Format

Documentation

sunxiaojian commented Mar 16, 2024

sunxiaojian commented Mar 27, 2024

sunxiaojian commented Apr 7, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

schnappi17 commented Apr 24, 2024

schnappi17 commented Apr 25, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sunxiaojian commented May 14, 2024

sunxiaojian commented Mar 16, 2024 •

edited