From 997e20f4d05f8e6697497badcc79c022186c5c31 Mon Sep 17 00:00:00 2001 From: Arnav Balyan Date: Sun, 9 Aug 2026 21:11:23 +0530 Subject: [PATCH 1/4] update --- .../apache/paimon/utils/SnapshotManager.java | 1 + .../table/system/SnapshotsTableTest.java | 41 +++++++++++++++++++ 2 files changed, 42 insertions(+) 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..1e15431d1bff 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 @@ -551,6 +551,7 @@ public Stream snapshotIdStream() throws IOException { public Iterator snapshotsWithId(List snapshotIds) { return snapshotIds.stream() + .filter(this::snapshotExists) .map(this::snapshot) .sorted(Comparator.comparingLong(Snapshot::id)) .iterator(); 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..b4fc530c7eb6 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,11 +49,13 @@ 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; import static org.apache.paimon.catalog.Identifier.SYSTEM_TABLE_SPLITTER; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Tests for {@link SnapshotsTable}. */ public class SnapshotsTableTest extends TableTestBase { @@ -96,6 +102,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 testReadSnapshotsWithEqualFilterOnUnknownIdStillFails() { + PredicateBuilder builder = new PredicateBuilder(snapshotsTable.rowType()); + Predicate predicate = + builder.equal(snapshotsTable.rowType().getFieldNames().indexOf("snapshot_id"), 99L); + + ReadBuilder readBuilder = snapshotsTable.newReadBuilder().withFilter(predicate); + assertThatThrownBy( + () -> + readBuilder + .newRead() + .createReader(readBuilder.newScan().plan()) + .forEachRemaining(row -> {})) + .isInstanceOf(RuntimeException.class); + } + private List getExpectedResult(long[] snapshotIds) { List expectedRow = new ArrayList<>(); for (long snapshotId : snapshotIds) { From 7820999f244114425b2222f438844a1e8be84831 Mon Sep 17 00:00:00 2001 From: Arnav Balyan Date: Sun, 9 Aug 2026 23:09:28 +0530 Subject: [PATCH 2/4] update --- .../paimon/table/system/SchemasTable.java | 5 +++- .../paimon/table/system/SchemasTableTest.java | 24 +++++++++++++++++++ 2 files changed, 28 insertions(+), 1 deletion(-) 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..af6f2f718127 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 @@ -310,7 +310,10 @@ private static BinaryString toJson(Object obj) { private static List schemasWithId( SchemaManager schemaManager, List schemaIds) { - return schemaIds.stream().map(schemaManager::schema).collect(Collectors.toList()); + return schemaIds.stream() + .filter(schemaManager::schemaExists) + .map(schemaManager::schema) + .collect(Collectors.toList()); } private static List listWithRange( 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..dbb64c356bfb 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,25 @@ 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()); + } + private List getExpectedResult() { List tableSchemas = schemaManager.listAll(); From 80f96dcc511969498ce672aeb114bc206383a791 Mon Sep 17 00:00:00 2001 From: Arnav Balyan Date: Mon, 10 Aug 2026 12:35:39 +0530 Subject: [PATCH 3/4] update --- .../apache/paimon/schema/SchemaManager.java | 4 ++++ .../paimon/table/system/SchemasTable.java | 17 +++++++++++++---- .../apache/paimon/utils/SnapshotManager.java | 19 ++++++++++++++----- .../paimon/table/system/SchemasTableTest.java | 16 ++++++++++++++++ .../table/system/SnapshotsTableTest.java | 17 ++++++++--------- .../paimon/utils/SnapshotManagerTest.java | 14 ++++++++++++++ 6 files changed, 69 insertions(+), 18 deletions(-) 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 af6f2f718127..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,16 +311,24 @@ private static BinaryString toJson(Object obj) { private static List schemasWithId( SchemaManager schemaManager, List schemaIds) { - return schemaIds.stream() - .filter(schemaManager::schemaExists) - .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 1e15431d1bff..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,15 +550,24 @@ public Stream snapshotIdStream() throws IOException { } public Iterator snapshotsWithId(List snapshotIds) { - return snapshotIds.stream() - .filter(this::snapshotExists) - .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 dbb64c356bfb..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 @@ -102,6 +102,22 @@ public void testReadSchemasWithInFilterContainingUnknownId() throws Exception { 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 b4fc530c7eb6..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 @@ -55,7 +55,6 @@ import static org.apache.paimon.SnapshotTest.newSnapshotManager; import static org.apache.paimon.catalog.Identifier.SYSTEM_TABLE_SPLITTER; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Tests for {@link SnapshotsTable}. */ public class SnapshotsTableTest extends TableTestBase { @@ -122,19 +121,19 @@ public void testReadSnapshotsWithInFilterContainingUnknownId() throws Exception } @Test - public void testReadSnapshotsWithEqualFilterOnUnknownIdStillFails() { + 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); - assertThatThrownBy( - () -> - readBuilder - .newRead() - .createReader(readBuilder.newScan().plan()) - .forEachRemaining(row -> {})) - .isInstanceOf(RuntimeException.class); + List result = new ArrayList<>(); + readBuilder + .newRead() + .createReader(readBuilder.newScan().plan()) + .forEachRemaining(result::add); + + assertThat(result).isEmpty(); } private List getExpectedResult(long[] 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 { From 91974d86d20f2e1dc5e0fde69ff39f5210a348ad Mon Sep 17 00:00:00 2001 From: Arnav Balyan Date: Mon, 10 Aug 2026 16:33:19 +0530 Subject: [PATCH 4/4] update --- .../org/apache/paimon/flink/CatalogTableITCase.java | 12 +++++------- .../paimon/hive/FlinkGenericCatalogITCase.java | 11 ++++------- 2 files changed, 9 insertions(+), 14 deletions(-) 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 =