[core] Fix snapshots table failing IN filter for unknown id

paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java

@@ -1212,6 +1212,10 @@ public TableSchema schema(long id) {
         return fromPath(fileIO, toSchemaPath(id));
     }
 
+    public TableSchema tryGetSchema(long id) throws FileNotFoundException {
+        return tryFromPath(fileIO, toSchemaPath(id));
+    }
+
     /** Check if a schema exists. */
     public boolean schemaExists(long id) {
         Path path = toSchemaPath(id);

paimon-core/src/main/java/org/apache/paimon/table/system/SchemasTable.java

@@ -62,6 +62,7 @@
 
 import javax.annotation.Nullable;
 
+import java.io.FileNotFoundException;
 import java.time.Instant;
 import java.time.LocalDateTime;
 import java.time.ZoneId;
@@ -310,13 +311,24 @@ private static BinaryString toJson(Object obj) {
 
     private static List<TableSchema> schemasWithId(
             SchemaManager schemaManager, List<Long> schemaIds) {
-        return schemaIds.stream().map(schemaManager::schema).collect(Collectors.toList());
+        List<TableSchema> schemas = new ArrayList<>();
+        for (long schemaId : schemaIds) {
+            try {
+                schemas.add(schemaManager.tryGetSchema(schemaId));
+            } catch (FileNotFoundException ignored) {
+            }
+        }
+        return schemas;
     }
 
     private static List<TableSchema> listWithRange(
             SchemaManager schemaManager,
             @Nullable Long optionalMinSchemaId,
             @Nullable Long optionalMaxSchemaId) {
+        if (optionalMinSchemaId != null && optionalMinSchemaId.equals(optionalMaxSchemaId)) {
+            return schemasWithId(schemaManager, Collections.singletonList(optionalMinSchemaId));
+        }
+
         long lowerBoundSchemaId = 0L;
 
         Optional<TableSchema> latest = schemaManager.latest();

paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java

@@ -550,14 +550,24 @@ public Stream<Long> snapshotIdStream() throws IOException {
     }
 
     public Iterator<Snapshot> snapshotsWithId(List<Long> snapshotIds) {
-        return snapshotIds.stream()
-                .map(this::snapshot)
-                .sorted(Comparator.comparingLong(Snapshot::id))
-                .iterator();
+        List<Snapshot> snapshots = new ArrayList<>();
+        for (long snapshotId : snapshotIds) {
+            try {
+                snapshots.add(tryGetSnapshot(snapshotId));
+            } catch (FileNotFoundException ignored) {
+            }
+        }
+        snapshots.sort(Comparator.comparingLong(Snapshot::id));
+        return snapshots.iterator();
     }
 
     public Iterator<Snapshot> snapshotsWithinRange(
             Optional<Long> optionalMaxSnapshotId, Optional<Long> optionalMinSnapshotId) {
+        if (optionalMaxSnapshotId.isPresent()
+                && optionalMaxSnapshotId.equals(optionalMinSnapshotId)) {
+            return snapshotsWithId(Collections.singletonList(optionalMaxSnapshotId.get()));
+        }
+
         Long lowerBoundSnapshotId = earliestSnapshotId();
         Long upperBoundSnapshotId = latestSnapshotId();
         Long lowerId;

paimon-core/src/test/java/org/apache/paimon/table/system/SchemasTableTest.java

@@ -24,13 +24,17 @@
 import org.apache.paimon.data.GenericRow;
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.data.serializer.InternalRowSerializer;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
 import org.apache.paimon.schema.Schema;
 import org.apache.paimon.schema.SchemaManager;
 import org.apache.paimon.schema.TableSchema;
 import org.apache.paimon.table.TableTestBase;
+import org.apache.paimon.table.source.ReadBuilder;
 import org.apache.paimon.types.DataTypes;
 
 import org.junit.jupiter.api.BeforeEach;
@@ -40,6 +44,7 @@
 import java.time.LocalDateTime;
 import java.time.ZoneId;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 
 import static org.apache.paimon.utils.JsonSerdeUtil.toFlatJson;
@@ -78,6 +83,41 @@ public void testSchemasTable() throws Exception {
         assertThat(result).containsExactlyElementsOf(expectRow);
     }
 
+    @Test
+    public void testReadSchemasWithInFilterContainingUnknownId() throws Exception {
+        PredicateBuilder builder = new PredicateBuilder(schemasTable.rowType());
+        Predicate predicate =
+                builder.in(
+                        schemasTable.rowType().getFieldNames().indexOf("schema_id"),
+                        Arrays.asList(0L, 99L));
+
+        ReadBuilder readBuilder = schemasTable.newReadBuilder().withFilter(predicate);
+        List<InternalRow> result = new ArrayList<>();
+        InternalRowSerializer serializer = new InternalRowSerializer(schemasTable.rowType());
+        readBuilder
+                .newRead()
+                .createReader(readBuilder.newScan().plan())
+                .forEachRemaining(row -> result.add(serializer.copy(row)));
+
+        assertThat(result).containsExactlyElementsOf(getExpectedResult());
+    }
+
+    @Test
+    public void testReadSchemasWithEqualFilterOnUnknownId() throws Exception {
+        PredicateBuilder builder = new PredicateBuilder(schemasTable.rowType());
+        Predicate predicate =
+                builder.equal(schemasTable.rowType().getFieldNames().indexOf("schema_id"), 99L);
+
+        ReadBuilder readBuilder = schemasTable.newReadBuilder().withFilter(predicate);
+        List<InternalRow> result = new ArrayList<>();
+        readBuilder
+                .newRead()
+                .createReader(readBuilder.newScan().plan())
+                .forEachRemaining(result::add);
+
+        assertThat(result).isEmpty();
+    }
+
     private List<InternalRow> getExpectedResult() {
         List<TableSchema> tableSchemas = schemaManager.listAll();
 

paimon-core/src/test/java/org/apache/paimon/table/system/SnapshotsTableTest.java

@@ -25,16 +25,20 @@
 import org.apache.paimon.data.GenericRow;
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.data.serializer.InternalRowSerializer;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
 import org.apache.paimon.schema.Schema;
 import org.apache.paimon.schema.SchemaManager;
 import org.apache.paimon.schema.SchemaUtils;
 import org.apache.paimon.schema.TableSchema;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.FileStoreTableFactory;
 import org.apache.paimon.table.TableTestBase;
+import org.apache.paimon.table.source.ReadBuilder;
 import org.apache.paimon.types.DataTypes;
 import org.apache.paimon.utils.SnapshotManager;
 
@@ -45,6 +49,7 @@
 import java.time.LocalDateTime;
 import java.time.ZoneId;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 
 import static org.apache.paimon.SnapshotTest.newSnapshotManager;
@@ -96,6 +101,41 @@ public void testReadSnapshotsFromLatest() throws Exception {
         assertThat(result).containsExactlyInAnyOrderElementsOf(expectedRow);
     }
 
+    @Test
+    public void testReadSnapshotsWithInFilterContainingUnknownId() throws Exception {
+        PredicateBuilder builder = new PredicateBuilder(snapshotsTable.rowType());
+        Predicate predicate =
+                builder.in(
+                        snapshotsTable.rowType().getFieldNames().indexOf("snapshot_id"),
+                        Arrays.asList(1L, 99L));
+
+        ReadBuilder readBuilder = snapshotsTable.newReadBuilder().withFilter(predicate);
+        List<InternalRow> result = new ArrayList<>();
+        InternalRowSerializer serializer = new InternalRowSerializer(snapshotsTable.rowType());
+        readBuilder
+                .newRead()
+                .createReader(readBuilder.newScan().plan())
+                .forEachRemaining(row -> result.add(serializer.copy(row)));
+
+        assertThat(result).containsExactlyInAnyOrderElementsOf(getExpectedResult(new long[] {1}));
+    }
+
+    @Test
+    public void testReadSnapshotsWithEqualFilterOnUnknownId() throws Exception {
+        PredicateBuilder builder = new PredicateBuilder(snapshotsTable.rowType());
+        Predicate predicate =
+                builder.equal(snapshotsTable.rowType().getFieldNames().indexOf("snapshot_id"), 99L);
+
+        ReadBuilder readBuilder = snapshotsTable.newReadBuilder().withFilter(predicate);
+        List<InternalRow> result = new ArrayList<>();
+        readBuilder
+                .newRead()
+                .createReader(readBuilder.newScan().plan())
+                .forEachRemaining(result::add);
+
+        assertThat(result).isEmpty();
+    }
+
     private List<InternalRow> getExpectedResult(long[] snapshotIds) {
         List<InternalRow> expectedRow = new ArrayList<>();
         for (long snapshotId : snapshotIds) {

paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java

@@ -36,8 +36,10 @@
 import javax.annotation.Nullable;
 
 import java.io.File;
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -81,6 +83,18 @@ public void testSnapshotPath() {
         }
     }
 
+    @Test
+    public void testSnapshotsWithIdSkipsExpiredSnapshot() throws Exception {
+        FileIO fileIO = Mockito.mock(FileIO.class);
+        Mockito.when(fileIO.exists(Mockito.any(Path.class))).thenReturn(true);
+        Mockito.when(fileIO.readFileUtf8(Mockito.any(Path.class)))
+                .thenThrow(new FileNotFoundException());
+        SnapshotManager snapshotManager = newSnapshotManager(fileIO, new Path(tempDir.toString()));
+
+        assertThat(snapshotManager.snapshotsWithId(Collections.singletonList(1L)).hasNext())
+                .isFalse();
+    }
+
     @ParameterizedTest
     @ValueSource(booleans = {true, false})
     public void testEarliestSnapshot(boolean isRaceCondition) throws IOException {

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java

@@ -379,13 +379,11 @@ public void testSchemasTable() {
                                 + "\"snapshot.num-retained.min\":\"18\"}, ]]");
 
         // check with not exist schema id
-        assertThatThrownBy(
-                        () ->
-                                sql(
-                                        "SELECT schema_id, fields, partition_keys, "
-                                                + "primary_keys, options, `comment` FROM T$schemas where schema_id = 5"))
-                .hasCauseInstanceOf(RuntimeException.class)
-                .hasRootCauseMessage("schema id: 5 should not greater than max schema id: 4");
+        assertThat(
+                        sql(
+                                "SELECT schema_id, fields, partition_keys, "
+                                        + "primary_keys, options, `comment` FROM T$schemas where schema_id = 5"))
+                .isEmpty();
 
         // check with not exist schema id
         assertThatThrownBy(

paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/FlinkGenericCatalogITCase.java

@@ -163,13 +163,10 @@ public void testReadPaimonSystemTable() {
         assertThat(result2).containsExactly(Row.of(2L, 0L, "APPEND"));
 
         // check leaf predicate query with exist snapshot_id
-        assertThatThrownBy(
-                        () ->
-                                sql(
-                                        "SELECT snapshot_id, schema_id, commit_kind FROM paimon_t$snapshots where snapshot_id=6"))
-                .hasCauseInstanceOf(RuntimeException.class)
-                .hasRootCauseMessage(
-                        "snapshot upper id:6 should not greater than latestSnapshotId:4");
+        assertThat(
+                        sql(
+                                "SELECT snapshot_id, schema_id, commit_kind FROM paimon_t$snapshots where snapshot_id=6"))
+                .isEmpty();
 
         // check compound predicate query with right range
         List<Row> result3 =