[core] Skip manual-only tag groups in preview time travel

paimon-core/src/main/java/org/apache/paimon/tag/TagPreview.java

@@ -31,6 +31,7 @@
 import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 
 import static java.util.Collections.singletonMap;
@@ -87,6 +88,7 @@ public Map<String, String> timeTravel(DataTable table, String tag) {
         Optional<String> findTag =
                 tagManager.tags().values().stream()
                         .map(this::toOneAutoTag)
+                        .filter(Objects::nonNull)
                         .filter(t -> t.compareTo(tag) <= 0)
                         .max(Comparator.naturalOrder());
         if (findTag.isPresent()) {
@@ -103,6 +105,10 @@ private String toOneAutoTag(List<String> tags) {
                 autoTags.add(tag);
             }
         }
+        if (autoTags.isEmpty()) {
+            // a snapshot may carry only manually created tags: nothing to resolve here
+            return null;
+        }
         return TagAutoCreation.checkAndGetOneAutoTag(autoTags);
     }
 }

paimon-core/src/test/java/org/apache/paimon/tag/TagPreviewTest.java

@@ -38,6 +38,7 @@
 import static org.apache.paimon.CoreOptions.SNAPSHOT_NUM_RETAINED_MAX;
 import static org.apache.paimon.CoreOptions.SNAPSHOT_NUM_RETAINED_MIN;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** Test for {@link TagPreview}. */
 public class TagPreviewTest extends PrimaryKeyTableTestBase {
@@ -91,6 +92,58 @@ public void testTimeTravel() {
                 .containsAllEntriesOf(singletonMap(SCAN_TAG_NAME.key(), "2023-07-18"));
     }
 
+    @Test
+    public void testTimeTravelSkipsManualOnlySnapshots() throws Exception {
+        TagPreview preview = create();
+        TableCommitImpl commit = table.newCommit(commitUser).ignoreEmptyCommit(false);
+
+        // only manually named tags exist and no snapshot's preview time resolves the
+        // requested tag: the lookup must fail with the intended error instead of the
+        // misleading "more than 1 auto-created tags"
+        commit.commit(new ManifestCommittable(0, utcMills("2023-07-18T12:12:00")));
+        table.createTag("my-manual-tag", 1);
+
+        assertThatThrownBy(() -> preview.timeTravel(table, "2023-07-01"))
+                .isInstanceOf(RuntimeException.class)
+                .hasMessageContaining("Cannot find snapshot or tag");
+
+        // a date-shaped (auto-format) tag resolves normally through the short-circuit
+        table.createTag("2023-07-19", 1);
+        assertThat(preview.timeTravel(table, "2023-07-19"))
+                .containsAllEntriesOf(singletonMap(SCAN_TAG_NAME.key(), "2023-07-19"));
+
+        commit.close();
+    }
+
+    @Test
+    public void testTimeTravelResolvesAutoTagDespiteManualOnlyGroup() throws Exception {
+        TagPreview preview = create();
+        Map<String, String> dynamicOptions = new HashMap<>();
+        dynamicOptions.put(SNAPSHOT_NUM_RETAINED_MIN.key(), "3");
+        dynamicOptions.put(SNAPSHOT_NUM_RETAINED_MAX.key(), "3");
+        TableCommitImpl commit =
+                table.copy(dynamicOptions).newCommit(commitUser).ignoreEmptyCommit(false);
+
+        // snapshot 1 keeps an auto-format tag, snapshot 2 keeps only a manually named tag
+        commit.commit(new ManifestCommittable(0, utcMills("2023-07-18T12:12:00")));
+        commit.commit(new ManifestCommittable(0, utcMills("2023-07-19T12:12:00")));
+        table.createTag("2023-07-18", 1);
+        table.createTag("my-manual-tag", 2);
+
+        // push the watermark past the request and expire snapshots 1 and 2, so the snapshot
+        // traversal finds nothing and the tags() fallback runs. The surviving auto tag on
+        // snapshot 1 must resolve even though snapshot 2's group holds only a manual tag: on
+        // master that manual-only group made toOneAutoTag throw and poisoned the whole max().
+        for (int i = 0; i < 5; i++) {
+            commit.commit(new ManifestCommittable(0, utcMills("2023-07-21T12:12:00")));
+        }
+
+        assertThat(preview.timeTravel(table, "2023-07-19"))
+                .containsAllEntriesOf(singletonMap(SCAN_TAG_NAME.key(), "2023-07-18"));
+
+        commit.close();
+    }
+
     private TagPreview create() {
         Options options = new Options();
         options.set(METASTORE_TAG_TO_PARTITION_PREVIEW, TagCreationMode.WATERMARK);