You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "herefree (via GitHub)" <gi...@apache.org> on 2024/08/29 13:17:44 UTC
[PR] [flink] add CreateTagFromWatermarkAction [paimon]
herefree opened a new pull request, #4101:
URL: https://github.com/apache/paimon/pull/4101
<!-- Please specify the module before the PR name: [core] ... or [flink] ... -->
### Purpose
add CreateTagFromWatermarkAction for old flink version
<!-- What is the purpose of the change -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] add CreateTagFromWatermarkAction [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #4101:
URL: https://github.com/apache/paimon/pull/4101
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] add CreateTagFromWatermarkAction [paimon]
Posted by "herefree (via GitHub)" <gi...@apache.org>.
herefree commented on PR #4101:
URL: https://github.com/apache/paimon/pull/4101#issuecomment-2320554898
Thanks for the comment @LinMingQiang ,I have finished the modification.Flink UT case passed in my workflow.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] add CreateTagFromWatermarkAction [paimon]
Posted by "LinMingQiang (via GitHub)" <gi...@apache.org>.
LinMingQiang commented on code in PR #4101:
URL: https://github.com/apache/paimon/pull/4101#discussion_r1737619230
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CreateTagFromWatermarkProcedureITCase.java:
##########
@@ -160,28 +202,72 @@ public void testCreatTagsFromTagsWatermark() throws Exception {
assertThat(watermark2 == 2000).isTrue();
// create tag from tag1 that snapshot is 1.
- assertThat(
- sql(
- "CALL sys.create_tag_from_watermark("
- + "`table` => 'default.T',"
- + "`tag` => 'tag2',"
- + "`watermark` => %s)",
- tagsWatermark - 1)
- .stream()
- .map(Row::toString))
- .containsExactlyInAnyOrder(
- String.format("+I[tag2, 1, %s, %s]", tagsCommitTime, tagsWatermark));
-
- assertThat(
- sql(
- "CALL sys.create_tag_from_watermark("
- + "`table` => 'default.T',"
- + "`tag` => 'tag3',"
- + "`watermark` => %s)",
- watermark2 - 1)
- .stream()
- .map(Row::toString))
- .containsExactlyInAnyOrder(
- String.format("+I[tag3, 2, %s, %s]", commitTime2, watermark2));
+ if (ThreadLocalRandom.current().nextBoolean()) {
+ assertThat(
+ sql(
+ "CALL sys.create_tag_from_watermark("
+ + "`table` => 'default.T',"
+ + "`tag` => 'tag2',"
+ + "`watermark` => %s)",
+ tagsWatermark - 1)
+ .stream()
+ .map(Row::toString))
+ .containsExactlyInAnyOrder(
+ String.format("+I[tag2, 1, %s, %s]", tagsCommitTime, tagsWatermark));
+ } else {
+ createAction(
Review Comment:
ditto
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CreateTagFromWatermarkProcedureITCase.java:
##########
@@ -160,28 +202,72 @@ public void testCreatTagsFromTagsWatermark() throws Exception {
assertThat(watermark2 == 2000).isTrue();
// create tag from tag1 that snapshot is 1.
- assertThat(
- sql(
- "CALL sys.create_tag_from_watermark("
- + "`table` => 'default.T',"
- + "`tag` => 'tag2',"
- + "`watermark` => %s)",
- tagsWatermark - 1)
- .stream()
- .map(Row::toString))
- .containsExactlyInAnyOrder(
- String.format("+I[tag2, 1, %s, %s]", tagsCommitTime, tagsWatermark));
-
- assertThat(
- sql(
- "CALL sys.create_tag_from_watermark("
- + "`table` => 'default.T',"
- + "`tag` => 'tag3',"
- + "`watermark` => %s)",
- watermark2 - 1)
- .stream()
- .map(Row::toString))
- .containsExactlyInAnyOrder(
- String.format("+I[tag3, 2, %s, %s]", commitTime2, watermark2));
+ if (ThreadLocalRandom.current().nextBoolean()) {
+ assertThat(
+ sql(
+ "CALL sys.create_tag_from_watermark("
+ + "`table` => 'default.T',"
+ + "`tag` => 'tag2',"
+ + "`watermark` => %s)",
+ tagsWatermark - 1)
+ .stream()
+ .map(Row::toString))
+ .containsExactlyInAnyOrder(
+ String.format("+I[tag2, 1, %s, %s]", tagsCommitTime, tagsWatermark));
+ } else {
+ createAction(
+ CreateTagFromWatermarkAction.class,
+ "create_tag_from_watermark",
+ "--warehouse",
+ path,
+ "--table",
+ "default.T",
+ "--tag",
+ "tag2",
+ "--watermark",
+ 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);
+
+ if (ThreadLocalRandom.current().nextBoolean()) {
+ assertThat(
+ sql(
+ "CALL sys.create_tag_from_watermark("
+ + "`table` => 'default.T',"
+ + "`tag` => 'tag3',"
+ + "`watermark` => %s)",
+ watermark2 - 1)
+ .stream()
+ .map(Row::toString))
+ .containsExactlyInAnyOrder(
+ String.format("+I[tag3, 2, %s, %s]", commitTime2, watermark2));
+ } else {
+ createAction(
+ CreateTagFromWatermarkAction.class,
+ "create_tag_from_watermark",
+ "--warehouse",
+ path,
+ "--table",
+ "default.T",
+ "--tag",
+ "tag3",
+ "--watermark",
+ 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);
+ }
+
+ private <T extends ActionBase> T createAction(Class<T> clazz, String... args) {
Review Comment:
ditto
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CreateTagFromWatermarkProcedureITCase.java:
##########
@@ -80,30 +85,67 @@ public void testCreatTagsFromSnapshotsWatermark() throws Exception {
assertThat(watermark1 == Long.MIN_VALUE).isTrue();
assertThat(watermark2 == 1000).isTrue();
assertThat(watermark3 == 2000).isTrue();
+ if (ThreadLocalRandom.current().nextBoolean()) {
+ assertThat(
+ sql(
+ "CALL sys.create_tag_from_watermark("
+ + "`table` => 'default.T',"
+ + "`tag` => 'tag2',"
+ + "`watermark` => %s)",
+ watermark2 - 1)
+ .stream()
+ .map(Row::toString))
+ .containsExactlyInAnyOrder(
+ String.format("+I[tag2, 2, %s, %s]", commitTime2, watermark2));
+ } else {
+ createAction(
Review Comment:
You shouldn't add tests in ProcedureITCase.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org