Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[Flink]Datastream sink supports dynamic bucketing #520

Open
wants to merge 21 commits into
base: main
Choose a base branch
from
Open

[Flink]Datastream sink supports dynamic bucketing #520

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
46181fe
parse csair log
Jul 26, 2024
6622b11
enlarge ci checkpoint interval
Jul 26, 2024
4cac816
datastream sink support dynamic bucketing
Jul 30, 2024
871013d
remove redundant log
Jul 30, 2024
cd41b1d
fix clippy
Jul 30, 2024
56185b0
update verify result format
Jul 30, 2024
c71d1f1
update verify result format
Jul 30, 2024
6b94ceb
fix case of schema migration
Jul 31, 2024
6347ae0
fix case of schema migration
Jul 31, 2024
8ab7caf
run ci with DYNAMIC_BUCKETING=false
Jul 31, 2024
5fbdf95
update ci log info
Jul 31, 2024
096bac8
merge LakeSoulMultiTableSinkGlobalCommittable with sorted createtime
Jul 31, 2024
7b9c9ee
merge LakeSoulMultiTableSinkGlobalCommittable with sorted createtime
Jul 31, 2024
b885474
fix schema migration
Jul 31, 2024
fe1ca65
fix schema migration
Jul 31, 2024
4235e04
use debezium ts_ms as latestUpdateTime
Aug 1, 2024
26e43f8
update schema with record of largest debezium ts_ms
Aug 1, 2024
ff9b3c1
update schema with record of largest debezium ts_ms
Aug 1, 2024
81024d0
add debug info
Aug 1, 2024
9b64b0c
add debug info
Aug 1, 2024
0e63939
test dynamic_bucket failover recover
Aug 2, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
test dynamic_bucket failover recover 
Signed-off-by: zenghua <huazeng@dmetasoul.com>
  • Loading branch information
zenghua committed Aug 2, 2024
commit 0e63939f783f78ed57d5af33f0c22dc06d3e7788
12 changes: 6 additions & 6 deletions ...k/src/main/java/org/apache/flink/lakesoul/sink/committer/LakeSoulSinkGlobalCommitter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -169,10 +169,10 @@ public List<LakeSoulMultiTableSinkGlobalCommittable> commit(
.setTableSchema(msgSchema.toJson())
.setPartitions(partition)
.setProperties(properties.toJSONString());
Tuple2<Long, TableInfo> last = finalTableInfoMap.get(tablePath);
if (last == null) {
Tuple2<Long, TableInfo> latest = finalTableInfoMap.get(tablePath);
if (latest == null) {
finalTableInfoMap.put(tablePath, Tuple2.of(latestSrcTsMs, builder.build()));
} else if (last.f0 < latestSrcTsMs) {
} else if (latest.f0 < latestSrcTsMs) {
finalTableInfoMap.put(tablePath, Tuple2.of(latestSrcTsMs, builder.build()));
}
} else {
Expand All @@ -187,11 +187,11 @@ public List<LakeSoulMultiTableSinkGlobalCommittable> commit(
}

String tablePath = tableInfo.getTablePath();
Tuple2<Long, TableInfo> last = finalTableInfoMap.get(tablePath);
Tuple2<Long, TableInfo> latest = finalTableInfoMap.get(tablePath);
TableInfo.Builder builder = tableInfo.toBuilder().setTableSchema(msgSchema.toJson());
if (last == null) {
if (latest == null) {
finalTableInfoMap.put(tablePath, Tuple2.of(latestSrcTsMs, builder.build()));
} else if (last.f0 < latestSrcTsMs) {
} else if (latest.f0 < latestSrcTsMs) {
finalTableInfoMap.put(tablePath, Tuple2.of(latestSrcTsMs, builder.build()));
}

Expand Down
2 changes: 1 addition & 1 deletion ...c/test/java/org/apache/flink/lakesoul/test/connector/sink/LakeSoulDataStreamSinkCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,7 @@ public void testLakeSoulDataStreamSink() throws Exception {
TableId tableId = new TableId(LakeSoulCatalog.CATALOG_NAME, "default", tableName);
LakeSoulRowDataWrapper data = mockInsertLakeSoulRowDataWrapper(1, useCDC, tableId);
recordCollection.add(new BinarySourceRecord(topic, primaryKeys, tableId, path, partitionKeys, false, data, ""));

Thread.sleep(100);
// LakeSoulRowDataWrapper data1 = mockUpdateLakeSoulRowDataWrapper(2, useCDC, tableId);
LakeSoulRowDataWrapper data1 = mockInsertLakeSoulRowDataWrapper(4, useCDC, tableId);
recordCollection.add(new BinarySourceRecord(topic, primaryKeys, tableId, path, partitionKeys, false, data1, ""));
Expand Down
10 changes: 5 additions & 5 deletions lakesoul-spark/src/test/scala/org/apache/spark/sql/lakesoul/DDLSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -486,9 +486,9 @@ abstract class DDLTestBase extends QueryTest with SQLTestUtils {
}
}

// test("read") {
// sql("show create table TestBinarySourceRecordSink").show(false)
// sql("desc TestBinarySourceRecordSink").show()
// sql("select * from TestBinarySourceRecordSink").show()
// }
test("read") {
sql("show create table TestBinarySourceRecordSink").show(false)
sql("desc TestBinarySourceRecordSink").show()
sql("select * from TestBinarySourceRecordSink").show()
}
}
Loading