This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 29549c8084 [core] Fix snapshots table failing IN filter for unknown id
(#9134)
29549c8084 is described below
commit 29549c8084e04f326fe66b650956344a4c9ed1f3
Author: Arnav Balyan <[email protected]>
AuthorDate: Mon Aug 17 11:05:01 2026 +0530
[core] Fix snapshots table failing IN filter for unknown id (#9134)
---
.../org/apache/paimon/schema/SchemaManager.java | 4 +++
.../apache/paimon/table/system/SchemasTable.java | 14 +++++++-
.../org/apache/paimon/utils/SnapshotManager.java | 18 +++++++---
.../paimon/table/system/SchemasTableTest.java | 40 ++++++++++++++++++++++
.../paimon/table/system/SnapshotsTableTest.java | 40 ++++++++++++++++++++++
.../apache/paimon/utils/SnapshotManagerTest.java | 14 ++++++++
.../apache/paimon/flink/CatalogTableITCase.java | 12 +++----
.../paimon/hive/FlinkGenericCatalogITCase.java | 11 +++---
8 files changed, 134 insertions(+), 19 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 28175245af..2c83cbcb57 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
@@ -1227,6 +1227,10 @@ public class SchemaManager implements Serializable {
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 b5093c3aeb..9c3fd9ca61 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
org.apache.paimon.shade.guava30.com.google.common.collect.Iterators;
import javax.annotation.Nullable;
+import java.io.FileNotFoundException;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
@@ -310,13 +311,24 @@ public class SchemasTable implements ReadonlyTable {
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();
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 23694934e8..75be8fe921 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 class SnapshotManager implements Serializable {
}
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;
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 c7c0ca5cf8..36bf05788c 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.BinaryString;
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.Instant;
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 class SchemasTableTest extends TableTestBase {
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();
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 c1a6b6f714..34d634e642 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.BinaryString;
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.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.Instant;
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 class SnapshotsTableTest extends TableTestBase {
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) {
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 6c09b9a327..2750feb902 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 org.mockito.Mockito;
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 class SnapshotManagerTest {
}
}
+ @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 3dc809c5ef..9024edeca5 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 class CatalogTableITCase extends CatalogITCaseBase
{
+ "\"manifest.compression\":\"snappy\"}, ]]");
// 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 0ac539224c..b48404fb1c 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 class FlinkGenericCatalogITCase extends
AbstractTestBaseJUnit4 {
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 =
