From e8261d57cc8ea7f882274eb2d2d0f108c16783b8 Mon Sep 17 00:00:00 2001
From: herefree <841043203@qq.com>
Date: Tue, 3 Sep 2024 14:05:02 +0800
Subject: [PATCH] fix query schema when specifying schema_id does not exist
 throw exception

---
 .../org/apache/paimon/schema/SchemaManager.java     | 13 +++++++++++++
 .../apache/paimon/table/system/SchemasTable.java    | 10 ++++++----
 .../org/apache/paimon/flink/CatalogTableITCase.java |  6 ++++++
 3 files changed, 25 insertions(+), 4 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 5d4adeb40f18..e164213e2bf8 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
@@ -554,6 +554,19 @@ public TableSchema schema(long id) {
         }
     }
 
+    /** Check if a schema exists. */
+    public boolean schemaExists(long id) {
+        Path path = toSchemaPath(id);
+        try {
+            return fileIO.exists(path);
+        } catch (IOException e) {
+            throw new RuntimeException(
+                    String.format(
+                            "Failed to determine if schema '%s' exists in path %s.", id, path),
+                    e);
+        }
+    }
+
     public static TableSchema fromPath(FileIO fileIO, Path path) {
         try {
             return JsonSerdeUtil.fromJson(fileIO.readFileUtf8(path), TableSchema.class);
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 b79b75edd338..06a4b726d104 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
@@ -225,10 +225,12 @@ public RecordReader<InternalRow> createReader(Split split) {
             SchemaManager manager = new SchemaManager(fileIO, location, branch);
 
             Collection<TableSchema> tableSchemas = Collections.emptyList();
-            if (predicate != null && predicate.function() instanceof Equal) {
-                Object equalValue = predicate.literals().get(0);
-                if (equalValue instanceof Long) {
-                    tableSchemas = Collections.singletonList(manager.schema((Long) equalValue));
+            if (predicate != null
+                    && predicate.function() instanceof Equal
+                    && predicate.literals().get(0) instanceof Long) {
+                Long equalValue = (Long) predicate.literals().get(0);
+                if (manager.schemaExists(equalValue)) {
+                    tableSchemas = Collections.singletonList(manager.schema(equalValue));
                 }
             } else {
                 tableSchemas = manager.listAll();
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 e1bbcbcf2382..fd0aa79c1fb0 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
@@ -257,6 +257,12 @@ public void testSchemasTable() {
                                 + "{\"id\":1,\"name\":\"b\",\"type\":\"INT\"},"
                                 + "{\"id\":2,\"name\":\"c\",\"type\":\"STRING\"}], [], [\"a\"], "
                                 + "{\"a.aa.aaa\":\"val1\",\"b.bb.bbb\":\"val2\"}, ]]");
+
+        result =
+                sql(
+                        "SELECT schema_id, fields, partition_keys, "
+                                + "primary_keys, options, `comment` FROM T$schemas where schema_id = 5");
+        assertThat(result.toString()).isEqualTo("[]");
     }
 
     @Test