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 aea054381c [core] Return empty schemas/snapshots table result for
out-of-range id filters (#10206)
aea054381c is described below
commit aea054381c3646159b8c2e9152c21c4f400dfed5
Author: YangJie <[email protected]>
AuthorDate: Mon Oct 5 10:21:24 2026 -0400
[core] Return empty schemas/snapshots table result for out-of-range id
filters (#10206)
---
.../apache/paimon/table/system/SchemasTable.java | 12 ++--
.../org/apache/paimon/utils/SnapshotManager.java | 12 ++--
.../paimon/table/system/SchemasTableTest.java | 69 ++++++++++++++++++----
.../paimon/table/system/SnapshotsTableTest.java | 48 +++++++++++++++
.../apache/paimon/flink/CatalogTableITCase.java | 15 +++--
.../paimon/hive/FlinkGenericCatalogITCase.java | 15 ++---
.../paimon/spark/sql/PaimonSystemTableTest.scala | 12 ++++
7 files changed, 138 insertions(+), 45 deletions(-)
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 9c3fd9ca61..7e09f55b80 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
@@ -345,10 +345,8 @@ public class SchemasTable implements ReadonlyTable {
if (optionalMaxSchemaId != null) {
if (optionalMaxSchemaId < lowerBoundSchemaId) {
- throw new RuntimeException(
- String.format(
- "schema id: %s should not lower than min
schema id: %s",
- optionalMaxSchemaId, lowerBoundSchemaId));
+ // schema ids lie in [0, latest], so the range matches none of
them
+ return Collections.emptyList();
}
upperBoundSchematId =
optionalMaxSchemaId > upperBoundSchematId
@@ -358,10 +356,8 @@ public class SchemasTable implements ReadonlyTable {
if (optionalMinSchemaId != null) {
if (optionalMinSchemaId > upperBoundSchematId) {
- throw new RuntimeException(
- String.format(
- "schema id: %s should not greater than max
schema id: %s",
- optionalMinSchemaId, upperBoundSchematId));
+ // schema ids lie in [0, latest], so the range matches none of
them
+ return Collections.emptyList();
}
lowerBoundSchemaId =
optionalMinSchemaId > lowerBoundSchemaId
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 18740a9a3f..5f90d2db91 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
@@ -628,10 +628,8 @@ public class SnapshotManager implements Serializable {
if (optionalMaxSnapshotId.isPresent()) {
upperId = optionalMaxSnapshotId.get();
if (upperId < lowerBoundSnapshotId) {
- throw new RuntimeException(
- String.format(
- "snapshot upper id:%s should not greater than
earliestSnapshotId:%s",
- upperId, lowerBoundSnapshotId));
+ // no snapshot id in [earliest, latest] falls in the range
+ return Collections.emptyIterator();
}
upperBoundSnapshotId = upperId < upperBoundSnapshotId ? upperId :
upperBoundSnapshotId;
}
@@ -639,10 +637,8 @@ public class SnapshotManager implements Serializable {
if (optionalMinSnapshotId.isPresent()) {
lowerId = optionalMinSnapshotId.get();
if (lowerId > upperBoundSnapshotId) {
- throw new RuntimeException(
- String.format(
- "snapshot upper id:%s should not greater than
latestSnapshotId:%s",
- lowerId, upperBoundSnapshotId));
+ // no snapshot id in [earliest, latest] falls in the range
+ return Collections.emptyIterator();
}
lowerBoundSnapshotId = lowerId > lowerBoundSnapshotId ? lowerId :
lowerBoundSnapshotId;
}
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 5df2d5d8e6..dada6f00b2 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
@@ -32,6 +32,7 @@ import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.schema.FileSystemSchemaManager;
import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.TableTestBase;
@@ -119,24 +120,68 @@ public class SchemasTableTest extends TableTestBase {
assertThat(result).isEmpty();
}
+ @Test
+ public void testReadSchemasWithOutOfRangeFilterReturnsEmpty() throws
Exception {
+ // commit a second schema so the latest schema id is 1
+ catalog.alterTable(identifier("T"), SchemaChange.setOption("k", "v"),
false);
+ assertThat(read(schemasTable)).hasSize(2);
+
+ PredicateBuilder builder = new
PredicateBuilder(schemasTable.rowType());
+ int schemaIdIdx =
schemasTable.rowType().getFieldNames().indexOf("schema_id");
+
+ List<Predicate> outOfRange = new ArrayList<>();
+ // lower bound above the latest schema id
+ outOfRange.add(builder.greaterThan(schemaIdIdx, 99L));
+ // upper bound below the first schema id
+ outOfRange.add(builder.lessThan(schemaIdIdx, -1L));
+ // lower bound above the upper bound, both within [0, latest]
+ outOfRange.add(
+ PredicateBuilder.and(
+ builder.greaterThan(schemaIdIdx, 0L),
builder.lessThan(schemaIdIdx, 1L)));
+
+ for (Predicate predicate : outOfRange) {
+ assertThat(readWithFilter(predicate)).isEmpty();
+ }
+
+ // the bounds of [0, latest] still return their schema
+ assertThat(readWithFilter(builder.lessOrEqual(schemaIdIdx, 0L)))
+ .containsExactly(toExpectedRow(schemaManager.schema(0L)));
+ assertThat(readWithFilter(builder.greaterOrEqual(schemaIdIdx, 1L)))
+ .containsExactly(toExpectedRow(schemaManager.schema(1L)));
+ }
+
+ private List<InternalRow> readWithFilter(Predicate predicate) throws
Exception {
+ 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)));
+ return result;
+ }
+
private List<InternalRow> getExpectedResult() {
List<TableSchema> tableSchemas = schemaManager.listAll();
List<InternalRow> expectedRow = new ArrayList<>();
for (TableSchema schema : tableSchemas) {
- expectedRow.add(
- GenericRow.of(
- schema.id(),
-
BinaryString.fromString(toFlatJson(schema.fields())),
-
BinaryString.fromString(toFlatJson(schema.partitionKeys())),
-
BinaryString.fromString(toFlatJson(schema.primaryKeys())),
-
BinaryString.fromString(toFlatJson(schema.options())),
- BinaryString.fromString(schema.comment()),
- Timestamp.fromLocalDateTime(
- LocalDateTime.ofInstant(
-
Instant.ofEpochMilli(schema.timeMillis()),
- ZoneId.systemDefault()))));
+ expectedRow.add(toExpectedRow(schema));
}
return expectedRow;
}
+
+ private static InternalRow toExpectedRow(TableSchema schema) {
+ return GenericRow.of(
+ schema.id(),
+ BinaryString.fromString(toFlatJson(schema.fields())),
+ BinaryString.fromString(toFlatJson(schema.partitionKeys())),
+ BinaryString.fromString(toFlatJson(schema.primaryKeys())),
+ BinaryString.fromString(toFlatJson(schema.options())),
+ BinaryString.fromString(schema.comment()),
+ Timestamp.fromLocalDateTime(
+ LocalDateTime.ofInstant(
+ Instant.ofEpochMilli(schema.timeMillis()),
+ ZoneId.systemDefault())));
+ }
}
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 648344b2b8..e04da130b6 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
@@ -29,6 +29,7 @@ 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.options.ExpireConfig;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.schema.FileSystemSchemaManager;
@@ -45,6 +46,7 @@ import org.apache.paimon.utils.SnapshotManager;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.time.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
@@ -136,6 +138,52 @@ public class SnapshotsTableTest extends TableTestBase {
assertThat(result).isEmpty();
}
+ @Test
+ public void testReadSnapshotsWithOutOfRangeFilterReturnsEmpty() throws
Exception {
+ PredicateBuilder builder = new
PredicateBuilder(snapshotsTable.rowType());
+ int snapshotIdIdx =
snapshotsTable.rowType().getFieldNames().indexOf("snapshot_id");
+
+ // snapshots 1 and 2 exist
+ // lower bound above the latest snapshot id
+ assertThat(readWithFilter(builder.greaterThan(snapshotIdIdx,
99L))).isEmpty();
+ // upper bound below the earliest snapshot id
+ assertThat(readWithFilter(builder.lessThan(snapshotIdIdx,
1L))).isEmpty();
+ // lower bound above the upper bound, both within [earliest, latest]
+ assertThat(
+ readWithFilter(
+ PredicateBuilder.and(
+ builder.greaterThan(snapshotIdIdx, 1L),
+ builder.lessThan(snapshotIdIdx, 2L))))
+ .isEmpty();
+
+ // expire snapshot 1, then query a range that ends below the new
earliest snapshot
+ FileStoreTable table = (FileStoreTable)
catalog.getTable(identifier(tableName));
+ table.newExpireSnapshots()
+ .config(
+ ExpireConfig.builder()
+ .snapshotRetainMax(1)
+ .snapshotRetainMin(1)
+ .snapshotTimeRetain(Duration.ZERO)
+ .snapshotMaxDeletes(Integer.MAX_VALUE)
+ .build())
+ .expire();
+ assertThat(snapshotManager.earliestSnapshotId()).isEqualTo(2L);
+ assertThat(readWithFilter(builder.lessOrEqual(snapshotIdIdx,
1L))).isEmpty();
+ assertThat(readWithFilter(builder.lessOrEqual(snapshotIdIdx, 2L)))
+ .containsExactlyElementsOf(getExpectedResult(new long[] {2}));
+ }
+
+ private List<InternalRow> readWithFilter(Predicate predicate) throws
Exception {
+ 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)));
+ return result;
+ }
+
private List<InternalRow> getExpectedResult(long[] snapshotIds) {
List<InternalRow> expectedRow = new ArrayList<>();
for (long snapshotId : snapshotIds) {
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 9024edeca5..92cf7e5385 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
@@ -385,14 +385,13 @@ public class CatalogTableITCase extends CatalogITCaseBase
{
+ "primary_keys, options, `comment`
FROM T$schemas where schema_id = 5"))
.isEmpty();
- // check with not exist schema id
- assertThatThrownBy(
- () ->
- sql(
- "SELECT schema_id, fields,
partition_keys, "
- + "primary_keys, options,
`comment` FROM T$schemas where schema_id>=6"))
- .hasCauseInstanceOf(RuntimeException.class)
- .hasRootCauseMessage("schema id: 6 should not greater than max
schema id: 4");
+ // check with out-of-range schema id: no schema has id >= 6 (latest is
4), so the
+ // query returns zero rows instead of failing
+ assertThat(
+ sql(
+ "SELECT schema_id, fields, partition_keys, "
+ + "primary_keys, options, `comment`
FROM T$schemas where schema_id>=6"))
+ .isEmpty();
}
@Test
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 b48404fb1c..3bd7792243 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
@@ -46,7 +46,6 @@ import java.util.Collections;
import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
/** IT cases for using Flink {@code FlinkGenericCatalog}. */
@RunWith(PaimonEmbeddedHiveRunner.class)
@@ -179,14 +178,12 @@ public class FlinkGenericCatalogITCase extends
AbstractTestBaseJUnit4 {
Row.of(3L, 0L, "APPEND"),
Row.of(4L, 0L, "APPEND"));
- // check with wrong range
- assertThatThrownBy(
- () ->
- sql(
- "SELECT snapshot_id, schema_id,
commit_kind FROM paimon_t$snapshots where snapshot_id>9"))
- .hasCauseInstanceOf(RuntimeException.class)
- .hasRootCauseMessage(
- "snapshot upper id:10 should not greater than
latestSnapshotId:4");
+ // check with out-of-range snapshot id: the filter matches no
snapshot, so the query
+ // returns zero rows instead of failing
+ assertThat(
+ sql(
+ "SELECT snapshot_id, schema_id, commit_kind
FROM paimon_t$snapshots where snapshot_id>9"))
+ .isEmpty();
}
@Test
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonSystemTableTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonSystemTableTest.scala
index 12a18d3761..48f8c3b269 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonSystemTableTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonSystemTableTest.scala
@@ -125,4 +125,16 @@ class PaimonSystemTableTest extends PaimonSparkTestBase {
Seq(Row("+I", Array(3)), Row("+I", Array(2)))
)
}
+
+ test("system table: snapshots and schemas tables with out-of-range id
filter") {
+ sql("CREATE TABLE T (a INT, b STRING)")
+ sql("INSERT INTO T VALUES (1, 'a')")
+ sql("INSERT INTO T VALUES (2, 'b')")
+
+ checkAnswer(sql("SELECT snapshot_id FROM `T$snapshots` WHERE snapshot_id >
2"), Nil)
+ checkAnswer(sql("SELECT schema_id FROM `T$schemas` WHERE schema_id > 0"),
Nil)
+
+ sql("CALL paimon.sys.expire_snapshots(table => 'test.T', retain_max => 1,
retain_min => 1)")
+ checkAnswer(sql("SELECT snapshot_id FROM `T$snapshots` WHERE snapshot_id <
2"), Nil)
+ }
}
