feat(core): Support schema management over REST in RESTCatalog

paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java

@@ -92,6 +92,7 @@
 import org.apache.paimon.rest.responses.ListPartitionsResponse;
 import org.apache.paimon.rest.responses.ListPermissionsResponse;
 import org.apache.paimon.rest.responses.ListPoliciesResponse;
+import org.apache.paimon.rest.responses.ListSchemaResponse;
 import org.apache.paimon.rest.responses.ListSnapshotsResponse;
 import org.apache.paimon.rest.responses.ListTableDetailsResponse;
 import org.apache.paimon.rest.responses.ListTablesGloballyResponse;
@@ -103,6 +104,7 @@
 import org.apache.paimon.rest.responses.PagedResponse;
 import org.apache.paimon.schema.Schema;
 import org.apache.paimon.schema.SchemaChange;
+import org.apache.paimon.schema.SchemaFilter;
 import org.apache.paimon.table.Instant;
 import org.apache.paimon.table.TableSnapshot;
 import org.apache.paimon.utils.JsonSerdeUtil;
@@ -762,6 +764,43 @@ public void rollbackSchema(Identifier identifier, long schemaId) {
                 restAuthFunction);
     }
 
+    /**
+     * List schemas of a table filtered by the given {@link SchemaFilter}.
+     *
+     * <p>All schema read patterns (latest / earliest / by id / by range / all) share this single
+     * endpoint. The server is responsible for interpreting the filter and returning the matching
+     * schemas.
+     *
+     * @param identifier database name and table name.
+     * @param filter which schemas to return; see {@link SchemaFilter} for the allowed combinations.
+     * @throws NoSuchResourceException Exception thrown on HTTP 404 means the table not exists
+     * @throws ForbiddenException Exception thrown on HTTP 403 means don't have the permission for
+     *     this table
+     */
+    public ListSchemaResponse listSchemas(Identifier identifier, SchemaFilter filter) {
+        Map<String, String> queryParams = Maps.newHashMap();
+        if (filter.isLatest()) {
+            queryParams.put("latest", "true");
+        }
+        if (filter.isEarliest()) {
+            queryParams.put("earliest", "true");
+        }
+        if (filter.schemaId() != null) {
+            queryParams.put("schemaId", filter.schemaId().toString());
+        }
+        if (filter.maxSchemaId() != null) {
+            queryParams.put("maxSchemaId", filter.maxSchemaId().toString());
+        }
+        if (filter.minSchemaId() != null) {
+            queryParams.put("minSchemaId", filter.minSchemaId().toString());
+        }
+        return client.get(
+                resourcePaths.schemas(identifier.getDatabaseName(), identifier.getObjectName()),
+                queryParams,
+                ListSchemaResponse.class,
+                restAuthFunction);
+    }
+
     /**
      * Create table.
      *

paimon-api/src/main/java/org/apache/paimon/rest/ResourcePaths.java

@@ -37,6 +37,7 @@ public class ResourcePaths {
     protected static final String TAGS = "tags";
     protected static final String SNAPSHOTS = "snapshots";
     protected static final String CONSUMERS = "consumers";
+    protected static final String SCHEMAS = "schemas";
     protected static final String VIEWS = "views";
     protected static final String TABLE_DETAILS = "table-details";
     protected static final String VIEW_DETAILS = "view-details";
@@ -223,6 +224,17 @@ public String snapshots(String databaseName, String objectName) {
                 SNAPSHOTS);
     }
 
+    public String schemas(String databaseName, String objectName) {
+        return SLASH.join(
+                V1,
+                prefix,
+                DATABASES,
+                encodeString(databaseName),
+                TABLES,
+                encodeString(objectName),
+                SCHEMAS);
+    }
+
     public String authTable(String databaseName, String objectName) {
         return SLASH.join(
                 V1,

paimon-api/src/main/java/org/apache/paimon/rest/responses/ListSchemaResponse.java

@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.rest.responses;
+
+import org.apache.paimon.rest.RESTResponse;
+import org.apache.paimon.schema.Schema;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.List;
+
+/**
+ * Response for listing or getting table schemas. All schema queries (latest / earliest / by id / by
+ * range / list all) return this shape; the server is responsible for filtering.
+ */
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class ListSchemaResponse implements RESTResponse {
+
+    private static final String FIELD_SCHEMAS = "schemas";
+
+    @JsonProperty(FIELD_SCHEMAS)
+    private final List<SchemaItem> schemas;
+
+    @JsonCreator
+    public ListSchemaResponse(@JsonProperty(FIELD_SCHEMAS) List<SchemaItem> schemas) {
+        this.schemas = schemas;
+    }
+
+    @JsonGetter(FIELD_SCHEMAS)
+    public List<SchemaItem> getSchemas() {
+        return schemas;
+    }
+
+    /** One schema entry in a {@link ListSchemaResponse}. */
+    @JsonIgnoreProperties(ignoreUnknown = true)
+    public static class SchemaItem {
+
+        private static final String FIELD_SCHEMA_ID = "schemaId";
+        private static final String FIELD_SCHEMA = "schema";
+        private static final String FIELD_CREATED_AT = "createdAt";
+
+        @JsonProperty(FIELD_SCHEMA_ID)
+        private final long schemaId;
+
+        @JsonProperty(FIELD_SCHEMA)
+        private final Schema schema;
+
+        @JsonProperty(FIELD_CREATED_AT)
+        private final long createdAt;
+
+        @JsonCreator
+        public SchemaItem(
+                @JsonProperty(FIELD_SCHEMA_ID) long schemaId,
+                @JsonProperty(FIELD_SCHEMA) Schema schema,
+                @JsonProperty(FIELD_CREATED_AT) long createdAt) {
+            this.schemaId = schemaId;
+            this.schema = schema;
+            this.createdAt = createdAt;
+        }
+
+        @JsonGetter(FIELD_SCHEMA_ID)
+        public long getSchemaId() {
+            return schemaId;
+        }
+
+        @JsonGetter(FIELD_SCHEMA)
+        public Schema getSchema() {
+            return schema;
+        }
+
+        @JsonGetter(FIELD_CREATED_AT)
+        public long getCreatedAt() {
+            return createdAt;
+        }
+    }
+}

paimon-api/src/main/java/org/apache/paimon/schema/SchemaFilter.java

@@ -0,0 +1,160 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.schema;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/**
+ * Filter used by {@code Catalog#listSchemas} to express single-endpoint schema queries.
+ *
+ * <p>All schema read patterns (latest / earliest / by-id / by-range / all) share the same catalog
+ * method and are distinguished by which fields of this filter are populated. At most one of {@link
+ * #isLatest()}, {@link #isEarliest()}, {@link #schemaId()} may be set; when none of them is set,
+ * {@link #maxSchemaId()} / {@link #minSchemaId()} may optionally restrict the returned range.
+ */
+public class SchemaFilter implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    private static final SchemaFilter ALL = new SchemaFilter(false, false, null, null, null);
+    private static final SchemaFilter LATEST = new SchemaFilter(true, false, null, null, null);
+    private static final SchemaFilter EARLIEST = new SchemaFilter(false, true, null, null, null);
+
+    private final boolean latest;
+    private final boolean earliest;
+    @Nullable private final Long schemaId;
+    @Nullable private final Long maxSchemaId;
+    @Nullable private final Long minSchemaId;
+
+    private SchemaFilter(
+            boolean latest,
+            boolean earliest,
+            @Nullable Long schemaId,
+            @Nullable Long maxSchemaId,
+            @Nullable Long minSchemaId) {
+        int exclusive = 0;
+        if (latest) {
+            exclusive++;
+        }
+        if (earliest) {
+            exclusive++;
+        }
+        if (schemaId != null) {
+            exclusive++;
+        }
+        checkArgument(
+                exclusive <= 1,
+                "SchemaFilter is over-constrained: latest / earliest / schemaId are mutually exclusive.");
+        if (exclusive == 1) {
+            checkArgument(
+                    maxSchemaId == null && minSchemaId == null,
+                    "SchemaFilter is over-constrained: range cannot be combined with latest / earliest / schemaId.");
+        }
+        this.latest = latest;
+        this.earliest = earliest;
+        this.schemaId = schemaId;
+        this.maxSchemaId = maxSchemaId;
+        this.minSchemaId = minSchemaId;
+    }
+
+    public static SchemaFilter all() {
+        return ALL;
+    }
+
+    public static SchemaFilter latest() {
+        return LATEST;
+    }
+
+    public static SchemaFilter earliest() {
+        return EARLIEST;
+    }
+
+    public static SchemaFilter withId(long schemaId) {
+        return new SchemaFilter(false, false, schemaId, null, null);
+    }
+
+    public static SchemaFilter range(@Nullable Long maxSchemaId, @Nullable Long minSchemaId) {
+        return new SchemaFilter(false, false, null, maxSchemaId, minSchemaId);
+    }
+
+    public boolean isLatest() {
+        return latest;
+    }
+
+    public boolean isEarliest() {
+        return earliest;
+    }
+
+    @Nullable
+    public Long schemaId() {
+        return schemaId;
+    }
+
+    @Nullable
+    public Long maxSchemaId() {
+        return maxSchemaId;
+    }
+
+    @Nullable
+    public Long minSchemaId() {
+        return minSchemaId;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (!(o instanceof SchemaFilter)) {
+            return false;
+        }
+        SchemaFilter that = (SchemaFilter) o;
+        return latest == that.latest
+                && earliest == that.earliest
+                && Objects.equals(schemaId, that.schemaId)
+                && Objects.equals(maxSchemaId, that.maxSchemaId)
+                && Objects.equals(minSchemaId, that.minSchemaId);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(latest, earliest, schemaId, maxSchemaId, minSchemaId);
+    }
+
+    @Override
+    public String toString() {
+        return "SchemaFilter{"
+                + "latest="
+                + latest
+                + ", earliest="
+                + earliest
+                + ", schemaId="
+                + schemaId
+                + ", maxSchemaId="
+                + maxSchemaId
+                + ", minSchemaId="
+                + minSchemaId
+                + '}';
+    }
+}