[GitHub] [incubator-paimon] JingsongLi commented on a diff in pull request #1949: [flink] Auto create new tag for flink checkpoint

["JingsongLi (via GitHub)" <gi...@apache.org>]

JingsongLi commented on code in PR #1949:
URL: https://github.com/apache/incubator-paimon/pull/1949#discussion_r1318169900


##########
docs/content/maintenance/manage-tags.md:
##########
@@ -257,6 +257,11 @@ Paimon's snapshot is similar to flink's checkpoint, and both will automatically
 allows snapshots to be retained for a long time. Therefore, we can combine the two features of paimon's tag and flink's 
 savepoint to achieve incremental recovery of job from the specified savepoint.
 
+At the same time to open the change parameter will also promote the automatic creation of the tag, now although open the "tag.automatic-creation" parameter but,

Review Comment:
   remove this sentence.. there is no `sink.savepoint.auto-tag` here.
   
   Document after `You can set `sink.savepoint.auto-tag` to `true` to enable the feature of automatically creating tags for savepoint.`



##########
docs/content/maintenance/manage-tags.md:
##########
@@ -257,6 +257,11 @@ Paimon's snapshot is similar to flink's checkpoint, and both will automatically
 allows snapshots to be retained for a long time. Therefore, we can combine the two features of paimon's tag and flink's 
 savepoint to achieve incremental recovery of job from the specified savepoint.
 
+At the same time to open the change parameter will also promote the automatic creation of the tag, now although open the "tag.automatic-creation" parameter but,

Review Comment:
   Just simplify documentation. For example: `This option will also force creating a Tag for the tag.automatic-creation=process-time mode at an appropriate time even when no data is written.`



##########
paimon-core/src/main/java/org/apache/paimon/utils/TagManager.java:
##########
@@ -61,7 +61,9 @@ public Path tagPath(String tagName) {
     /** Create a tag from given snapshot and save it in the storage. */
     public void createTag(Snapshot snapshot, String tagName) {
         checkArgument(!StringUtils.isBlank(tagName), "Tag name '%s' is blank.", tagName);
-        checkArgument(!tagExists(tagName), "Tag name '%s' already exists.", tagName);
+        if (tagExists(tagName)) {

Review Comment:
   Why add this?



##########
paimon-core/src/main/java/org/apache/paimon/tag/TagAutoCreation.java:
##########
@@ -115,10 +114,20 @@ private TagAutoCreation(
         }
     }
 
+    public void refreshTags() {

Review Comment:
   We should not add new method for `TagAutoCreation`.
   We should just invoke `run`.
   
   We can create a method:
   ```
   private void tryToTag(Snapshot snapshot, LocalDateTime time) {
           if (nextTag == null
                   || isAfterOrEqual(time.minus(delay), periodHandler.nextTagTime(nextTag))) {
               LocalDateTime thisTag = periodHandler.normalizeToTagTime(time);
               String tagName = periodHandler.timeToTag(thisTag);
               tagManager.createTag(snapshot, tagName);
               nextTag = periodHandler.nextTagTime(thisTag);
   
               if (numRetainedMax != null) {
                   SortedMap<Snapshot, String> tags = tagManager.tags();
                   if (tags.size() > numRetainedMax) {
                       int toDelete = tags.size() - numRetainedMax;
                       int i = 0;
                       for (String tag : tags.values()) {
                           tagManager.deleteTag(tag, tagDeletion, snapshotManager);
                           i++;
                           if (i == toDelete) {
                               break;
                           }
                       }
                   }
               }
           }
       }
   ```
   
   And in `run`:
   ```
       public void run() {
           while (true) {
               if (snapshotManager.snapshotExists(nextSnapshot)) {
                   tryToTag(snapshotManager.snapshot(nextSnapshot));
                   nextSnapshot++;
               } else {
                   // avoid snapshot has been expired
                   Long earliest = snapshotManager.earliestSnapshotId();
                   if (earliest != null && earliest > nextSnapshot) {
                       nextSnapshot = earliest;
                   } else {
                       break;
                   }
               }
           }
   
           if (timeExtractor instanceof ProcessTimeExtractor) {
               tryToTag(snapshotManager.snapshot(nextSnapshot - 1), LocalDateTime.now());
           }
       }
   ```
   
   Something like above codes.



##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AutoTagForSavepointCommitterOperator.java:
##########
@@ -58,31 +60,27 @@ public class AutoTagForSavepointCommitterOperator<CommitT, GlobalCommitT>
         implements OneInputStreamOperator<CommitT, CommitT>,
                 SetupableStreamOperator,
                 BoundedOneInput {
-    public static final String SAVEPOINT_TAG_PREFIX = "savepoint-";
-
+    @VisibleForTesting public static final String SAVEPOINT_TAG_PREFIX = "savepoint-";
     private static final long serialVersionUID = 1L;
-

Review Comment:
   Don't remove this empty line, this is just personal code style



-- 
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