Support multiple tables sink for Flink CDC #86

dmetasoul01 · 2022-09-25T10:43:34Z

Support read from multiple tables with different schemas into corresponding lakesoul tables.

Close #84
Close #87

…_cdc_multitable

…bles with different schemas

…_cdc_multitable

sonatype-lift · 2022-09-25T10:49:07Z

lakesoul-flink/src/main/java/org/apache/flink/lakesoul/sink/bucket/BulkFormatBuilder.java

+        return self();
+    }
+
+    public T withRollingPolicy(CheckpointRollingPolicy<RowData, String> rollingPolicy) {


CanIgnoreReturnValueSuggester: Methods that always 'return this' should be annotated with @CanIgnoreReturnValue

Suggested change

public T withRollingPolicy(CheckpointRollingPolicy<RowData, String> rollingPolicy) {

@CanIgnoreReturnValue

ℹ️ Learn about @sonatype-lift commands

You can reply with the following commands. For example, reply with @sonatype-lift ignoreall to leave out all findings.

Command Usage

@sonatype-lift ignore Leave out the above finding from this PR

@sonatype-lift ignoreall Leave out all the existing findings from this PR

@sonatype-lift exclude <file|issue|path|tool> Exclude specified file|issue|path|tool from Lift findings by updating your config.toml file

Note: When talking to LiftBot, you need to refresh the page to see its response.
_{Click here to add LiftBot to another repo.}

Was this a good recommendation?
[ 🙁 Not relevant ] - [ 😕 Won't fix ] - [ 😑 Not critical, will fix ] - [ 🙂 Critical, will fix ] - [ 😊 Critical, fixing now ]

sonatype-lift · 2022-09-25T10:49:08Z

lakesoul-flink/src/main/java/org/apache/flink/lakesoul/sink/bucket/BulkFormatBuilder.java

+        return RowType.of(true, new IntType(true));
+    }
+
+    public T withBucketCheckInterval(final long interval) {


CanIgnoreReturnValueSuggester: Methods that always 'return this' should be annotated with @CanIgnoreReturnValue

Suggested change

public T withBucketCheckInterval(final long interval) {

@CanIgnoreReturnValue

ℹ️ Learn about @sonatype-lift commands

You can reply with the following commands. For example, reply with @sonatype-lift ignoreall to leave out all findings.

Command Usage

@sonatype-lift ignore Leave out the above finding from this PR

@sonatype-lift ignoreall Leave out all the existing findings from this PR

@sonatype-lift exclude <file|issue|path|tool> Exclude specified file|issue|path|tool from Lift findings by updating your config.toml file

Note: When talking to LiftBot, you need to refresh the page to see its response.
_{Click here to add LiftBot to another repo.}

Was this a good recommendation?
[ 🙁 Not relevant ] - [ 😕 Won't fix ] - [ 😑 Not critical, will fix ] - [ 🙂 Critical, will fix ] - [ 😊 Critical, fixing now ]

sonatype-lift · 2022-09-25T10:49:09Z

lakesoul-flink/src/main/java/org/apache/flink/lakesoul/sink/bucket/BulkFormatBuilder.java

+        return self();
+    }
+
+    public T withOutputFileConfig(final OutputFileConfig outputFileConfig) {


CanIgnoreReturnValueSuggester: Methods that always 'return this' should be annotated with @CanIgnoreReturnValue

Suggested change

public T withOutputFileConfig(final OutputFileConfig outputFileConfig) {

@CanIgnoreReturnValue

ℹ️ Learn about @sonatype-lift commands

You can reply with the following commands. For example, reply with @sonatype-lift ignoreall to leave out all findings.

Command Usage

@sonatype-lift ignore Leave out the above finding from this PR

@sonatype-lift ignoreall Leave out all the existing findings from this PR

@sonatype-lift exclude <file|issue|path|tool> Exclude specified file|issue|path|tool from Lift findings by updating your config.toml file

Note: When talking to LiftBot, you need to refresh the page to see its response.
_{Click here to add LiftBot to another repo.}

Was this a good recommendation?
[ 🙁 Not relevant ] - [ 😕 Won't fix ] - [ 😑 Not critical, will fix ] - [ 🙂 Critical, will fix ] - [ 😊 Critical, fixing now ]

sonatype-lift · 2022-09-25T10:49:10Z

...rc/main/java/org/apache/flink/lakesoul/sink/writer/AbstractLakeSoulMultiTableSinkWriter.java

+            TableSchemaIdentity tableId,
+            String bucketId,
+            TableSchemaWriterCreator creator) throws IOException {
+        LakeSoulWriterBucket bucket = activeBuckets.get(Tuple2.of(creator.identity.tableId, bucketId));


CollectionIncompatibleType: Argument 'Tuple2.of(creator.identity.tableId, bucketId)' should not be passed to this method; its type Tuple2<TableId, String> is not compatible with its collection's type argument Tuple2<TableSchemaIdentity, String>

ℹ️ Learn about @sonatype-lift commands

You can reply with the following commands. For example, reply with @sonatype-lift ignoreall to leave out all findings.

Command Usage

@sonatype-lift ignore Leave out the above finding from this PR

@sonatype-lift ignoreall Leave out all the existing findings from this PR

@sonatype-lift exclude <file|issue|path|tool> Exclude specified file|issue|path|tool from Lift findings by updating your config.toml file

Note: When talking to LiftBot, you need to refresh the page to see its response.
_{Click here to add LiftBot to another repo.}

Was this a good recommendation?
[ 🙁 Not relevant ] - [ 😕 Won't fix ] - [ 😑 Not critical, will fix ] - [ 🙂 Critical, will fix ] - [ 😊 Critical, fixing now ]

sonatype-lift · 2022-09-25T10:49:11Z

lakesoul-flink/src/main/java/org/apache/flink/lakesoul/sink/HashPartitioner.java


 import org.apache.flink.api.common.functions.Partitioner;

-public class DataPartitioner<Long> implements Partitioner<Long> {
+public class HashPartitioner<Long> implements Partitioner<Long> {


JavaLangClash: Long clashes with java.lang.Long

ℹ️ Learn about @sonatype-lift commands

You can reply with the following commands. For example, reply with @sonatype-lift ignoreall to leave out all findings.

Command Usage

@sonatype-lift ignore Leave out the above finding from this PR

@sonatype-lift ignoreall Leave out all the existing findings from this PR

@sonatype-lift exclude <file|issue|path|tool> Exclude specified file|issue|path|tool from Lift findings by updating your config.toml file

Note: When talking to LiftBot, you need to refresh the page to see its response.
_{Click here to add LiftBot to another repo.}

Was this a good recommendation?
[ 🙁 Not relevant ] - [ 😕 Won't fix ] - [ 😑 Not critical, will fix ] - [ 🙂 Critical, will fix ] - [ 😊 Critical, fixing now ]

sonatype-lift · 2022-09-25T10:49:13Z

lakesoul-flink/src/main/java/org/apache/flink/lakesoul/types/TableId.java

+        return id.hashCode();
+    }
+
+    public int compareTo(TableId that) {


MissingImplementsComparable: Classes implementing valid compareTo function should implement Comparable interface

ℹ️ Learn about @sonatype-lift commands

You can reply with the following commands. For example, reply with @sonatype-lift ignoreall to leave out all findings.

Command Usage

@sonatype-lift ignore Leave out the above finding from this PR

@sonatype-lift ignoreall Leave out all the existing findings from this PR

@sonatype-lift exclude <file|issue|path|tool> Exclude specified file|issue|path|tool from Lift findings by updating your config.toml file

Note: When talking to LiftBot, you need to refresh the page to see its response.
_{Click here to add LiftBot to another repo.}

Was this a good recommendation?
[ 🙁 Not relevant ] - [ 😕 Won't fix ] - [ 😑 Not critical, will fix ] - [ 🙂 Critical, will fix ] - [ 😊 Critical, fixing now ]

sonatype-lift · 2022-09-25T10:49:14Z

lakesoul-flink/src/main/java/org/apache/flink/lakesoul/types/TableId.java

+    }
+
+    public int compareTo(TableId that) {
+        if (this == that) {


ReferenceEquality: Comparison using reference equality instead of value equality

Suggested change

if (this == that) {

if (this.equals(that)) {

ℹ️ Learn about @sonatype-lift commands

You can reply with the following commands. For example, reply with @sonatype-lift ignoreall to leave out all findings.

Command Usage

@sonatype-lift ignore Leave out the above finding from this PR

@sonatype-lift ignoreall Leave out all the existing findings from this PR

@sonatype-lift exclude <file|issue|path|tool> Exclude specified file|issue|path|tool from Lift findings by updating your config.toml file

Note: When talking to LiftBot, you need to refresh the page to see its response.
_{Click here to add LiftBot to another repo.}

Was this a good recommendation?
[ 🙁 Not relevant ] - [ 😕 Won't fix ] - [ 😑 Not critical, will fix ] - [ 🙂 Critical, will fix ] - [ 😊 Critical, fixing now ]

sonatype-lift · 2022-09-25T10:49:15Z

lakesoul-flink/src/main/java/org/apache/flink/lakesoul/tool/FlinkUtil.java

+    FileSystem fileSystem = p.getFileSystem();
+    return p.makeQualified(fileSystem);
+  }
+
  public static String getDatabaseName(String fullDatabaseName) {
    String[] splited = fullDatabaseName.split("\\.");


StringSplitter: String.split(String) has surprising behavior

Suggested change

String[] splited = fullDatabaseName.split("\\.");

List<String> splited = Splitter.on('.').splitToList(fullDatabaseName);

ℹ️ Learn about @sonatype-lift commands

You can reply with the following commands. For example, reply with @sonatype-lift ignoreall to leave out all findings.

Command Usage

@sonatype-lift ignore Leave out the above finding from this PR

@sonatype-lift ignoreall Leave out all the existing findings from this PR

@sonatype-lift exclude <file|issue|path|tool> Exclude specified file|issue|path|tool from Lift findings by updating your config.toml file

Note: When talking to LiftBot, you need to refresh the page to see its response.
_{Click here to add LiftBot to another repo.}

Was this a good recommendation?
[ 🙁 Not relevant ] - [ 😕 Won't fix ] - [ 😑 Not critical, will fix ] - [ 🙂 Critical, will fix ] - [ 😊 Critical, fixing now ]

sonatype-lift · 2022-09-25T10:49:17Z

lakesoul-flink/src/main/java/org/apache/flink/lakesoul/sink/HashPartitioner.java


 import org.apache.flink.api.common.functions.Partitioner;

-public class DataPartitioner<Long> implements Partitioner<Long> {
+public class HashPartitioner<Long> implements Partitioner<Long> {


TypeNameShadowing: Found type parameters shadowing other declared types:

ℹ️ Learn about @sonatype-lift commands

You can reply with the following commands. For example, reply with @sonatype-lift ignoreall to leave out all findings.

Command Usage

@sonatype-lift ignore Leave out the above finding from this PR

@sonatype-lift ignoreall Leave out all the existing findings from this PR

@sonatype-lift exclude <file|issue|path|tool> Exclude specified file|issue|path|tool from Lift findings by updating your config.toml file

Note: When talking to LiftBot, you need to refresh the page to see its response.
_{Click here to add LiftBot to another repo.}

Was this a good recommendation?
[ 🙁 Not relevant ] - [ 😕 Won't fix ] - [ 😑 Not critical, will fix ] - [ 🙂 Critical, will fix ] - [ 😊 Critical, fixing now ]

sonatype-lift · 2022-09-25T10:49:18Z

lakesoul-flink/src/main/java/org/apache/flink/lakesoul/table/LakeSoulDynamicTableFactory.java

UnusedVariable: The field 'TABLE_NAME' is never read.

ℹ️ Learn about @sonatype-lift commands

You can reply with the following commands. For example, reply with @sonatype-lift ignoreall to leave out all findings.

Command Usage

@sonatype-lift ignore Leave out the above finding from this PR

@sonatype-lift ignoreall Leave out all the existing findings from this PR

@sonatype-lift exclude <file|issue|path|tool> Exclude specified file|issue|path|tool from Lift findings by updating your config.toml file

Note: When talking to LiftBot, you need to refresh the page to see its response.
_{Click here to add LiftBot to another repo.}

Was this a good recommendation?
[ 🙁 Not relevant ] - [ 😕 Won't fix ] - [ 😑 Not critical, will fix ] - [ 🙂 Critical, will fix ] - [ 😊 Critical, fixing now ]

sonatype-lift · 2022-09-25T10:49:19Z

lakesoul-flink/src/main/java/org/apache/flink/lakesoul/table/LakeSoulTableSink.java

+   * DataStream sink fileSystem and upload metadata
+   */
+  private DataStreamSink<?> createStreamingSink(DataStream<RowData> dataStream,
+                                                Context sinkContext) throws IOException {


UnusedVariable: The parameter 'sinkContext' is never read.

Suggested change

Context sinkContext) throws IOException {

return createStreamingSink(dataStream);

ℹ️ Learn about @sonatype-lift commands

You can reply with the following commands. For example, reply with @sonatype-lift ignoreall to leave out all findings.

Command Usage

@sonatype-lift ignore Leave out the above finding from this PR

@sonatype-lift ignoreall Leave out all the existing findings from this PR

@sonatype-lift exclude <file|issue|path|tool> Exclude specified file|issue|path|tool from Lift findings by updating your config.toml file

Note: When talking to LiftBot, you need to refresh the page to see its response.
_{Click here to add LiftBot to another repo.}

Was this a good recommendation?
[ 🙁 Not relevant ] - [ 😕 Won't fix ] - [ 😑 Not critical, will fix ] - [ 🙂 Critical, will fix ] - [ 😊 Critical, fixing now ]

…_cdc_multitable

dmetasoul-opensource and others added 13 commits September 18, 2022 14:12

refactor: separate writer and stream operator

18b5ce5

add multi table stream sink operator

fbc2262

Merge remote-tracking branch 'origin/flink_cdc_multitable' into flink…

26d7265

…_cdc_multitable

refactor: implement Sink and SinkWriter for LakeSoul with multiple ta…

822a1bf

…bles with different schemas

Merge remote-tracking branch 'origin/flink_cdc_multitable' into flink…

ae1a981

…_cdc_multitable

add method for judging whether commitId exists

119ce95

supports flink metadata idempotent submission interface

80524f9

implement lakesoul sink committer and states recovery

602dbb4

Merge remote-tracking branch 'origin/flink_cdc_multitable' into flink…

298dfc3

…_cdc_multitable

fix: committable serde. simple write passed

9922ca4

remove unused options

a5215ae

Merge branch 'main' into flink_cdc_multitable

f7c992a

cleanup code after merge from main

7f57dca

dmetasoul01 added enhancement New feature or request flink flink support into lakesoul labels Sep 25, 2022

dmetasoul01 requested review from Ceng23333, moresun and F-PHantam September 25, 2022 10:43

dmetasoul01 self-assigned this Sep 25, 2022

sonatype-lift bot reviewed Sep 25, 2022

View reviewed changes

dmetasoul-opensource and others added 27 commits October 8, 2022 11:44

fix some code

2564946

Merge remote-tracking branch 'origin/flink_cdc_multitable' into flink…

1fe1c43

…_cdc_multitable

fix more test errors related to catalog change

7f7823b

fix wrong partition filter for parquet scan

f0aa6e2

fix lakesoul-spark TableCreationTests

180e84e

fix some create table test cases

efd976b

fix some path equal assertion fails

e38766c

use default namespace for test

325cb07

Merge remote-tracking branch 'origin/flink_cdc_multitable' into flink…

3f1dafc

…_cdc_multitable

fix lakesoul-spark TableCreationTests

58af56d

Merge remote-tracking branch 'origin/flink_cdc_multitable' into flink…

4af594c

…_cdc_multitable

fix lakesoul-spark DDLUsingPathSuite.scala

5dc658c

fix some test suite for spark

6e990bd

Merge remote-tracking branch 'origin/flink_cdc_multitable' into flink…

1274b9e

…_cdc_multitable

fix spark dev docker file

092d5ef

modify parameter name wrong

8e57910

fix deserialize ClassNotFoundException

b0264c5

Merge remote-tracking branch 'origin/flink_cdc_multitable' into flink…

d684177

…_cdc_multitable

fix mysql-cdc end2end timezone offset

c6cca09

fix compaction and add to external catalog

3e752d0

Merge remote-tracking branch 'origin/flink_cdc_multitable' into flink…

bfc2991

…_cdc_multitable

JsonRecordForLakeSoulTest add serverTimeZone

303d383

fix classloader for kryo in table schema serializer

e869f92

Merge remote-tracking branch 'origin/flink_cdc_multitable' into flink…

729ab91

…_cdc_multitable

remove redundant code

beb19f2

Merge remote-tracking branch 'origin/flink_cdc_multitable' into flink…

fa28173

…_cdc_multitable

Merge remote-tracking branch 'origin/main' into flink_cdc_multitable

11e5ecc

moresun approved these changes Oct 12, 2022

View reviewed changes

moresun merged commit f73089a into main Oct 12, 2022

dmetasoul01 deleted the flink_cdc_multitable branch October 12, 2022 04:22

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Support multiple tables sink for Flink CDC #86

Support multiple tables sink for Flink CDC #86

dmetasoul01 commented Sep 25, 2022 •

edited

sonatype-lift bot Sep 25, 2022

sonatype-lift bot Sep 25, 2022

sonatype-lift bot Sep 25, 2022

sonatype-lift bot Sep 25, 2022

sonatype-lift bot Sep 25, 2022

sonatype-lift bot Sep 25, 2022

sonatype-lift bot Sep 25, 2022

sonatype-lift bot Sep 25, 2022

sonatype-lift bot Sep 25, 2022

sonatype-lift bot Sep 25, 2022

sonatype-lift bot Sep 25, 2022

	public T withRollingPolicy(CheckpointRollingPolicy<RowData, String> rollingPolicy) {
	@CanIgnoreReturnValue

Command	Usage
`@sonatype-lift ignore`	Leave out the above finding from this PR
`@sonatype-lift ignoreall`	Leave out all the existing findings from this PR
`@sonatype-lift exclude <file\|issue\|path\|tool>`	Exclude specified `file\|issue\|path\|tool` from Lift findings by updating your config.toml file

	public T withBucketCheckInterval(final long interval) {
	@CanIgnoreReturnValue

	public T withOutputFileConfig(final OutputFileConfig outputFileConfig) {
	@CanIgnoreReturnValue

	String[] splited = fullDatabaseName.split("\\.");
	List<String> splited = Splitter.on('.').splitToList(fullDatabaseName);

	Context sinkContext) throws IOException {
	return createStreamingSink(dataStream);

Support multiple tables sink for Flink CDC #86

Support multiple tables sink for Flink CDC #86

Conversation

dmetasoul01 commented Sep 25, 2022 • edited

sonatype-lift bot Sep 25, 2022

Choose a reason for hiding this comment

sonatype-lift bot Sep 25, 2022

Choose a reason for hiding this comment

sonatype-lift bot Sep 25, 2022

Choose a reason for hiding this comment

sonatype-lift bot Sep 25, 2022

Choose a reason for hiding this comment

sonatype-lift bot Sep 25, 2022

Choose a reason for hiding this comment

sonatype-lift bot Sep 25, 2022

Choose a reason for hiding this comment

sonatype-lift bot Sep 25, 2022

Choose a reason for hiding this comment

sonatype-lift bot Sep 25, 2022

Choose a reason for hiding this comment

sonatype-lift bot Sep 25, 2022

Choose a reason for hiding this comment

sonatype-lift bot Sep 25, 2022

Choose a reason for hiding this comment

sonatype-lift bot Sep 25, 2022

Choose a reason for hiding this comment

dmetasoul01 commented Sep 25, 2022 •

edited