diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java index 4c238781b9b1..4fdb146fdf2e 100644 --- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java +++ b/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); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/SchemasTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/SchemasTable.java index b5093c3aeb9f..9c3fd9ca6107 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/system/SchemasTable.java +++ b/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 schemasWithId( SchemaManager schemaManager, List schemaIds) { - return schemaIds.stream().map(schemaManager::schema).collect(Collectors.toList()); + List schemas = new ArrayList<>(); + for (long schemaId : schemaIds) { + try { + schemas.add(schemaManager.tryGetSchema(schemaId)); + } catch (FileNotFoundException ignored) { + } + } + return schemas; } private static List 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 latest = schemaManager.latest(); diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java b/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java index 23694934e804..75be8fe921b4 100644 --- a/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java +++ b/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java @@ -550,14 +550,24 @@ public Stream snapshotIdStream() throws IOException { } public Iterator snapshotsWithId(List snapshotIds) { - return snapshotIds.stream() - .map(this::snapshot) - .sorted(Comparator.comparingLong(Snapshot::id)) - .iterator(); + List 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 snapshotsWithinRange( Optional optionalMaxSnapshotId, Optional optionalMinSnapshotId) { + if (optionalMaxSnapshotId.isPresent() + && optionalMaxSnapshotId.equals(optionalMinSnapshotId)) { + return snapshotsWithId(Collections.singletonList(optionalMaxSnapshotId.get())); + } + Long lowerBoundSnapshotId = earliestSnapshotId(); Long upperBoundSnapshotId = latestSnapshotId(); Long lowerId; diff --git a/paimon-core/src/test/java/org/apache/paimon/table/system/SchemasTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/system/SchemasTableTest.java index c7c0ca5cf872..36bf05788c59 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/system/SchemasTableTest.java +++ b/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 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 result = new ArrayList<>(); + readBuilder + .newRead() + .createReader(readBuilder.newScan().plan()) + .forEachRemaining(result::add); + + assertThat(result).isEmpty(); + } + private List getExpectedResult() { List tableSchemas = schemaManager.listAll(); diff --git a/paimon-core/src/test/java/org/apache/paimon/table/system/SnapshotsTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/system/SnapshotsTableTest.java index c1a6b6f7140a..34d634e6420b 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/system/SnapshotsTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/system/SnapshotsTableTest.java @@ -25,9 +25,12 @@ 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; @@ -35,6 +38,7 @@ 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 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 result = new ArrayList<>(); + readBuilder + .newRead() + .createReader(readBuilder.newScan().plan()) + .forEachRemaining(result::add); + + assertThat(result).isEmpty(); + } + private List getExpectedResult(long[] snapshotIds) { List expectedRow = new ArrayList<>(); for (long snapshotId : snapshotIds) { diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java b/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java index 6c09b9a327be..2750feb9029c 100644 --- a/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java +++ b/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 { diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java index 802bca58f561..dda34a05654d 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java +++ b/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( diff --git a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/FlinkGenericCatalogITCase.java b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/FlinkGenericCatalogITCase.java index 0ac539224c7c..b48404fb1c5a 100644 --- a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/FlinkGenericCatalogITCase.java +++ b/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 result3 =