[Feature] Auto complete missing tags

[luowanghaoyun]

Search before asking

• [X] I searched in the issues and found nothing similar.

Motivation

In auto tag creation mode, if the Flink write task is stopped for more than one creation period, there will be a loss of tag occurrence. For example:

        Options options = new Options();
        options.set(TAG_AUTOMATIC_CREATION, TagCreationMode.WATERMARK);
        options.set(TAG_CREATION_PERIOD, TagCreationPeriod.HOURLY);
        options.set(TAG_NUM_RETAINED_MAX, 3);
        FileStoreTable table = this.table.copy(options.toMap());
        TableCommitImpl commit = table.newCommit(commitUser).ignoreEmptyCommit(false);
        TagManager tagManager = table.store().newTagManager();

        // test normal creation
        commit.commit(new ManifestCommittable(0, utcMills("2023-07-18T12:12:00")));
        assertThat(tagManager.allTagNames()).containsOnly("2023-07-18 11");

        // task stop long time

        // task restart
        commit.commit(new ManifestCommittable(1, utcMills("2023-07-18T15:12:00")));
        // tag '2023-07-18 12' and '2023-07-18 13' are missing
        assertThat(tagManager.allTagNames()).containsOnly("2023-07-18 11", "2023-07-18 14");  

In some cases, missing tags may affect downstream tasks.

Solution

No response

Anything else?

No response

Are you willing to submit a PR?

• [X] I'm willing to submit a PR!