Skip to content

Commit

Permalink
fix
Browse files Browse the repository at this point in the history
  • Loading branch information
yuzelin committed Jan 2, 2025
1 parent 016f616 commit 7bc5feb
Show file tree
Hide file tree
Showing 6 changed files with 20 additions and 20 deletions.
19 changes: 9 additions & 10 deletions ...mmon/src/test/java/org/apache/paimon/flink/action/CreateTagFromWatermarkActionITTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -80,8 +80,8 @@ public void testCreateTagsFromSnapshotsWatermark() throws Exception {
Long.toString(watermark2 - 1))
.run();
assertThat(table.tagManager().tagExists("tag2")).isTrue();
assertThat(table.tagManager().taggedSnapshot("tag2").watermark()).isEqualTo(watermark2);
assertThat(table.tagManager().taggedSnapshot("tag2").timeMillis()).isEqualTo(commitTime2);
assertThat(table.tagManager().getOrThrow("tag2").watermark()).isEqualTo(watermark2);
assertThat(table.tagManager().getOrThrow("tag2").timeMillis()).isEqualTo(commitTime2);

createAction(
CreateTagFromWatermarkAction.class,
Expand All @@ -98,8 +98,8 @@ public void testCreateTagsFromSnapshotsWatermark() throws Exception {
Long.toString(watermark2 + 1))
.run();
assertThat(table.tagManager().tagExists("tag3")).isTrue();
assertThat(table.tagManager().taggedSnapshot("tag3").watermark()).isEqualTo(watermark3);
assertThat(table.tagManager().taggedSnapshot("tag3").timeMillis()).isEqualTo(commitTime3);
assertThat(table.tagManager().getOrThrow("tag3").watermark()).isEqualTo(watermark3);
assertThat(table.tagManager().getOrThrow("tag3").timeMillis()).isEqualTo(commitTime3);
}

@Test
Expand Down Expand Up @@ -131,7 +131,7 @@ public void testCreateTagsFromTagsWatermark() throws Exception {

assertThat(table.snapshotManager().snapshotExists(1)).isFalse();

Snapshot tagSnapshot1 = table.tagManager().taggedSnapshot("tag1");
Snapshot tagSnapshot1 = table.tagManager().getOrThrow("tag1");

long tagsCommitTime = tagSnapshot1.timeMillis();
long tagsWatermark = tagSnapshot1.watermark();
Expand All @@ -158,9 +158,8 @@ public void testCreateTagsFromTagsWatermark() throws Exception {
Long.toString(tagsWatermark - 1))
.run();
assertThat(table.tagManager().tagExists("tag2")).isTrue();
assertThat(table.tagManager().taggedSnapshot("tag2").watermark()).isEqualTo(tagsWatermark);
assertThat(table.tagManager().taggedSnapshot("tag2").timeMillis())
.isEqualTo(tagsCommitTime);
assertThat(table.tagManager().getOrThrow("tag2").watermark()).isEqualTo(tagsWatermark);
assertThat(table.tagManager().getOrThrow("tag2").timeMillis()).isEqualTo(tagsCommitTime);

createAction(
CreateTagFromWatermarkAction.class,
Expand All @@ -177,7 +176,7 @@ public void testCreateTagsFromTagsWatermark() throws Exception {
Long.toString(watermark2 - 1))
.run();
assertThat(table.tagManager().tagExists("tag3")).isTrue();
assertThat(table.tagManager().taggedSnapshot("tag3").watermark()).isEqualTo(watermark2);
assertThat(table.tagManager().taggedSnapshot("tag3").timeMillis()).isEqualTo(commitTime2);
assertThat(table.tagManager().getOrThrow("tag3").watermark()).isEqualTo(watermark2);
assertThat(table.tagManager().getOrThrow("tag3").timeMillis()).isEqualTo(commitTime2);
}
}
10 changes: 5 additions & 5 deletions ...aimon-flink-common/src/test/java/org/apache/paimon/flink/action/ReplaceTagActionTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -59,8 +59,8 @@ public void testReplaceTag() throws Exception {

bEnv.executeSql("CALL sys.create_tag(`table` => 'default.T', tag => 'test_tag')");
assertThat(tagManager.tagExists("test_tag")).isEqualTo(true);
assertThat(tagManager.tag("test_tag").trimToSnapshot().id()).isEqualTo(2);
assertThat(tagManager.tag("test_tag").getTagTimeRetained()).isEqualTo(null);
assertThat(tagManager.getOrThrow("test_tag").id()).isEqualTo(2);
assertThat(tagManager.getOrThrow("test_tag").getTagTimeRetained()).isEqualTo(null);

// replace tag with new time_retained
createAction(
Expand All @@ -77,7 +77,7 @@ public void testReplaceTag() throws Exception {
"--time_retained",
"1 d")
.run();
assertThat(tagManager.tag("test_tag").getTagTimeRetained().toHours()).isEqualTo(24);
assertThat(tagManager.getOrThrow("test_tag").getTagTimeRetained().toHours()).isEqualTo(24);

// replace tag with new snapshot and time_retained
createAction(
Expand All @@ -96,7 +96,7 @@ public void testReplaceTag() throws Exception {
"--time_retained",
"2 d")
.run();
assertThat(tagManager.tag("test_tag").trimToSnapshot().id()).isEqualTo(1);
assertThat(tagManager.tag("test_tag").getTagTimeRetained().toHours()).isEqualTo(48);
assertThat(tagManager.getOrThrow("test_tag").id()).isEqualTo(1);
assertThat(tagManager.getOrThrow("test_tag").getTagTimeRetained().toHours()).isEqualTo(48);
}
}
2 changes: 1 addition & 1 deletion ...rc/test/java/org/apache/paimon/flink/procedure/CreateTagFromTimestampProcedureITCase.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -133,7 +133,7 @@ public void testCreateTagsFromTagsCommitTime() throws Exception {

FileStoreTable table = paimonTable("T");
long earliestCommitTime = table.snapshotManager().earliestSnapshot().timeMillis();
long tagSnapshotCommitTime = table.tagManager().taggedSnapshot("tag1").timeMillis();
long tagSnapshotCommitTime = table.tagManager().getOrThrow("tag1").timeMillis();

assertThat(tagSnapshotCommitTime < earliestCommitTime).isTrue();

Expand Down
2 changes: 1 addition & 1 deletion ...rc/test/java/org/apache/paimon/flink/procedure/CreateTagFromWatermarkProcedureITCase.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -147,7 +147,7 @@ public void testCreateTagsFromTagsWatermark() throws Exception {

assertThat(table.snapshotManager().snapshotExists(1)).isFalse();

Snapshot tagSnapshot1 = table.tagManager().taggedSnapshot("tag1");
Snapshot tagSnapshot1 = table.tagManager().getOrThrow("tag1");

long tagsCommitTime = tagSnapshot1.timeMillis();
long tagsWatermark = tagSnapshot1.watermark();
Expand Down
4 changes: 2 additions & 2 deletions ...ink-common/src/test/java/org/apache/paimon/flink/procedure/ExpireTagsProcedureITCase.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,7 @@ public void testExpireTagsByOlderThanTime() throws Exception {

// tag-2 as the base older_than time.
// tag-1 expired by its file creation time.
LocalDateTime olderThanTime1 = table.tagManager().tag("tag-2").getTagCreateTime();
LocalDateTime olderThanTime1 = table.tagManager().getOrThrow("tag-2").getTagCreateTime();
java.sql.Timestamp timestamp1 =
new java.sql.Timestamp(
Timestamp.fromLocalDateTime(olderThanTime1).getMillisecond());
Expand All @@ -119,7 +119,7 @@ public void testExpireTagsByOlderThanTime() throws Exception {

// tag-4 as the base older_than time.
// tag-2,tag-3,tag-5 expired, tag-5 reached its tagTimeRetained.
LocalDateTime olderThanTime2 = table.tagManager().tag("tag-4").getTagCreateTime();
LocalDateTime olderThanTime2 = table.tagManager().getOrThrow("tag-4").getTagCreateTime();
java.sql.Timestamp timestamp2 =
new java.sql.Timestamp(
Timestamp.fromLocalDateTime(olderThanTime2).getMillisecond());
Expand Down
3 changes: 2 additions & 1 deletion ...common/src/test/java/org/apache/paimon/flink/sink/BatchWriteGeneratorTagOperatorTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -97,7 +97,8 @@ public void testBatchWriteGeneratorTag() throws Exception {
// Get tagName from tagManager.
String tagName = tagManager.allTagNames().get(0);
// The tag is consistent with the latest snapshot
assertThat(tagManager.taggedSnapshot(tagName)).isEqualTo(snapshotManager.latestSnapshot());
assertThat(tagManager.getOrThrow(tagName).trimToSnapshot())
.isEqualTo(snapshotManager.latestSnapshot());

// test tag expiration
table.createTag("many-tags-test1");
Expand Down

0 comments on commit 7bc5feb

Please sign in to comment.