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 7d7a672811 feat(core): Support schema management over REST in
RESTCatalog (#9673)
7d7a672811 is described below
commit 7d7a672811b98b543c44d42b7d949059c5cad48b
Author: baiyangtx <[email protected]>
AuthorDate: Tue Sep 8 22:09:20 2026 +0800
feat(core): Support schema management over REST in RESTCatalog (#9673)
---
.../main/java/org/apache/paimon/rest/RESTApi.java | 31 +++++
.../java/org/apache/paimon/rest/ResourcePaths.java | 16 +++
.../paimon/rest/responses/ErrorResponse.java | 2 +
.../paimon/rest/responses/GetSchemaResponse.java | 47 +++++++
.../paimon/rest/responses/ListSchemasResponse.java | 69 ++++++++++
.../java/org/apache/paimon/catalog/Catalog.java | 32 +++++
.../org/apache/paimon/catalog/DelegateCatalog.java | 14 ++
.../java/org/apache/paimon/rest/RESTCatalog.java | 28 ++++
.../apache/paimon/rest/MockRESTCatalogTest.java | 39 ++++++
.../org/apache/paimon/rest/RESTCatalogServer.java | 79 +++++------
.../rest/RESTCatalogServerMetadataHandler.java | 148 +++++++++++++++++++++
.../org/apache/paimon/rest/RESTCatalogTest.java | 52 ++++++++
12 files changed, 511 insertions(+), 46 deletions(-)
diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
b/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
index ad6e394a26..84cdecba48 100644
--- a/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
+++ b/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
@@ -77,6 +77,7 @@ import
org.apache.paimon.rest.responses.DropPartitionsResponse;
import org.apache.paimon.rest.responses.ErrorResponse;
import org.apache.paimon.rest.responses.GetDatabaseResponse;
import org.apache.paimon.rest.responses.GetFunctionResponse;
+import org.apache.paimon.rest.responses.GetSchemaResponse;
import org.apache.paimon.rest.responses.GetTableResponse;
import org.apache.paimon.rest.responses.GetTableSnapshotResponse;
import org.apache.paimon.rest.responses.GetTableTokenResponse;
@@ -92,6 +93,7 @@ import org.apache.paimon.rest.responses.ListFunctionsResponse;
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.ListSchemasResponse;
import org.apache.paimon.rest.responses.ListSnapshotsResponse;
import org.apache.paimon.rest.responses.ListTableDetailsResponse;
import org.apache.paimon.rest.responses.ListTablesGloballyResponse;
@@ -103,6 +105,7 @@ import org.apache.paimon.rest.responses.ListViewsResponse;
import org.apache.paimon.rest.responses.PagedResponse;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
+import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.Instant;
import org.apache.paimon.table.TableSnapshot;
import org.apache.paimon.utils.JsonSerdeUtil;
@@ -762,6 +765,34 @@ public class RESTApi {
restAuthFunction);
}
+ /** Load the schema of a table for the given version. */
+ public TableSchema loadSchema(Identifier identifier, String version) {
+ GetSchemaResponse response =
+ client.get(
+ resourcePaths.schemas(
+ identifier.getDatabaseName(),
identifier.getObjectName(), version),
+ GetSchemaResponse.class,
+ restAuthFunction);
+ return response.getSchema();
+ }
+
+ /** Get a paged schema list of a table in descending schema ID order. */
+ public PagedList<TableSchema> listSchemasPaged(
+ Identifier identifier, @Nullable Integer maxResults, @Nullable
String pageToken) {
+ ListSchemasResponse response =
+ client.get(
+ resourcePaths.schemas(
+ identifier.getDatabaseName(),
identifier.getObjectName()),
+ buildPagedQueryParams(maxResults, pageToken),
+ ListSchemasResponse.class,
+ restAuthFunction);
+ List<TableSchema> schemas = response.getSchemas();
+ if (schemas == null) {
+ return new PagedList<>(emptyList(), null);
+ }
+ return new PagedList<>(schemas, response.getNextPageToken());
+ }
+
/**
* Create table.
*
diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/ResourcePaths.java
b/paimon-api/src/main/java/org/apache/paimon/rest/ResourcePaths.java
index 4cef311061..d51a3119e1 100644
--- a/paimon-api/src/main/java/org/apache/paimon/rest/ResourcePaths.java
+++ b/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,21 @@ public class ResourcePaths {
SNAPSHOTS);
}
+ public String schemas(String databaseName, String objectName) {
+ return SLASH.join(
+ V1,
+ prefix,
+ DATABASES,
+ encodeString(databaseName),
+ TABLES,
+ encodeString(objectName),
+ SCHEMAS);
+ }
+
+ public String schemas(String databaseName, String objectName, String
version) {
+ return SLASH.join(schemas(databaseName, objectName),
encodeString(version));
+ }
+
public String authTable(String databaseName, String objectName) {
return SLASH.join(
V1,
diff --git
a/paimon-api/src/main/java/org/apache/paimon/rest/responses/ErrorResponse.java
b/paimon-api/src/main/java/org/apache/paimon/rest/responses/ErrorResponse.java
index 4fe52b1d0b..bfc9e3bf4e 100644
---
a/paimon-api/src/main/java/org/apache/paimon/rest/responses/ErrorResponse.java
+++
b/paimon-api/src/main/java/org/apache/paimon/rest/responses/ErrorResponse.java
@@ -41,6 +41,8 @@ public class ErrorResponse implements RESTResponse {
public static final String RESOURCE_TYPE_SNAPSHOT = "SNAPSHOT";
+ public static final String RESOURCE_TYPE_SCHEMA = "SCHEMA";
+
public static final String RESOURCE_TYPE_BRANCH = "BRANCH";
public static final String RESOURCE_TYPE_TAG = "TAG";
diff --git
a/paimon-api/src/main/java/org/apache/paimon/rest/responses/GetSchemaResponse.java
b/paimon-api/src/main/java/org/apache/paimon/rest/responses/GetSchemaResponse.java
new file mode 100644
index 0000000000..1ac383a8b7
--- /dev/null
+++
b/paimon-api/src/main/java/org/apache/paimon/rest/responses/GetSchemaResponse.java
@@ -0,0 +1,47 @@
+/*
+ * 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.TableSchema;
+
+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;
+
+/** Response for table schema by a version. */
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class GetSchemaResponse implements RESTResponse {
+
+ private static final String FIELD_SCHEMA = "schema";
+
+ @JsonProperty(FIELD_SCHEMA)
+ private final TableSchema schema;
+
+ @JsonCreator
+ public GetSchemaResponse(@JsonProperty(FIELD_SCHEMA) TableSchema schema) {
+ this.schema = schema;
+ }
+
+ @JsonGetter(FIELD_SCHEMA)
+ public TableSchema getSchema() {
+ return schema;
+ }
+}
diff --git
a/paimon-api/src/main/java/org/apache/paimon/rest/responses/ListSchemasResponse.java
b/paimon-api/src/main/java/org/apache/paimon/rest/responses/ListSchemasResponse.java
new file mode 100644
index 0000000000..493386abde
--- /dev/null
+++
b/paimon-api/src/main/java/org/apache/paimon/rest/responses/ListSchemasResponse.java
@@ -0,0 +1,69 @@
+/*
+ * 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.schema.TableSchema;
+
+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 list schemas. */
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class ListSchemasResponse implements PagedResponse<TableSchema> {
+
+ private static final String FIELD_SCHEMAS = "schemas";
+ private static final String FIELD_NEXT_PAGE_TOKEN = "nextPageToken";
+
+ @JsonProperty(FIELD_SCHEMAS)
+ private final List<TableSchema> schemas;
+
+ @JsonProperty(FIELD_NEXT_PAGE_TOKEN)
+ private final String nextPageToken;
+
+ public ListSchemasResponse(@JsonProperty(FIELD_SCHEMAS) List<TableSchema>
schemas) {
+ this(schemas, null);
+ }
+
+ @JsonCreator
+ public ListSchemasResponse(
+ @JsonProperty(FIELD_SCHEMAS) List<TableSchema> schemas,
+ @JsonProperty(FIELD_NEXT_PAGE_TOKEN) String nextPageToken) {
+ this.schemas = schemas;
+ this.nextPageToken = nextPageToken;
+ }
+
+ @JsonGetter(FIELD_SCHEMAS)
+ public List<TableSchema> getSchemas() {
+ return schemas;
+ }
+
+ @Override
+ public List<TableSchema> data() {
+ return schemas;
+ }
+
+ @JsonGetter(FIELD_NEXT_PAGE_TOKEN)
+ public String getNextPageToken() {
+ return nextPageToken;
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
index 7d26e03290..329abcf67b 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
@@ -31,6 +31,7 @@ import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.rest.responses.GetTagResponse;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
+import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.CatalogEnvironment;
import org.apache.paimon.table.Instant;
import org.apache.paimon.table.Table;
@@ -880,6 +881,37 @@ public interface Catalog extends AutoCloseable {
throw new UnsupportedOperationException();
}
+ /**
+ * Return the schema of a table for the given version. The version can be
{@code EARLIEST},
+ * {@code LATEST}, or a schema ID.
+ *
+ * @param identifier path of the table
+ * @param version version of the schema
+ * @return the requested schema
+ * @throws TableNotExistException if the table does not exist
+ * @throws UnsupportedOperationException if the catalog does not support
loading schemas
+ */
+ default Optional<TableSchema> loadSchema(Identifier identifier, String
version)
+ throws TableNotExistException {
+ throw new UnsupportedOperationException();
+ }
+
+ /**
+ * Get a paged schema list of a table in descending schema ID order.
+ *
+ * @param identifier path of the table
+ * @param maxResults maximum number of results, or {@code null} for the
server default
+ * @param pageToken token from the previous response, or {@code null} for
the first page
+ * @return schemas and the token for the next page
+ * @throws TableNotExistException if the table does not exist
+ * @throws UnsupportedOperationException if the catalog does not support
listing schemas
+ */
+ default PagedList<TableSchema> listSchemasPaged(
+ Identifier identifier, @Nullable Integer maxResults, @Nullable
String pageToken)
+ throws TableNotExistException {
+ throw new UnsupportedOperationException();
+ }
+
/**
* Create a new branch for this table. By default, an empty branch will be
created using the
* latest schema. If you provide {@code #fromTag}, a branch will be
created from the tag and the
diff --git
a/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
b/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
index 2f20d38fde..bd342e1013 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
@@ -28,6 +28,7 @@ import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.rest.responses.GetTagResponse;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
+import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.Instant;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.TableSnapshot;
@@ -223,6 +224,19 @@ public abstract class DelegateCatalog implements Catalog {
return wrapped.listSnapshotsPaged(identifier, maxResults, pageToken);
}
+ @Override
+ public Optional<TableSchema> loadSchema(Identifier identifier, String
version)
+ throws TableNotExistException {
+ return wrapped.loadSchema(identifier, version);
+ }
+
+ @Override
+ public PagedList<TableSchema> listSchemasPaged(
+ Identifier identifier, @Nullable Integer maxResults, @Nullable
String pageToken)
+ throws TableNotExistException {
+ return wrapped.listSchemasPaged(identifier, maxResults, pageToken);
+ }
+
@Override
public void rollbackTo(Identifier identifier, Instant instant, @Nullable
Long fromSnapshot)
throws Catalog.TableNotExistException {
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
index 4fa683d7d4..0fb356966d 100644
--- a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
@@ -478,6 +478,34 @@ public class RESTCatalog implements Catalog {
return true;
}
+ @Override
+ public Optional<TableSchema> loadSchema(Identifier identifier, String
version)
+ throws TableNotExistException {
+ try {
+ return Optional.ofNullable(api.loadSchema(identifier, version));
+ } catch (NoSuchResourceException e) {
+ if (StringUtils.equals(e.resourceType(),
ErrorResponse.RESOURCE_TYPE_SCHEMA)) {
+ return Optional.empty();
+ }
+ throw new TableNotExistException(identifier);
+ } catch (ForbiddenException e) {
+ throw new TableNoPermissionException(identifier, e);
+ }
+ }
+
+ @Override
+ public PagedList<TableSchema> listSchemasPaged(
+ Identifier identifier, @Nullable Integer maxResults, @Nullable
String pageToken)
+ throws TableNotExistException {
+ try {
+ return api.listSchemasPaged(identifier, maxResults, pageToken);
+ } catch (NoSuchResourceException e) {
+ throw new TableNotExistException(identifier);
+ } catch (ForbiddenException e) {
+ throw new TableNoPermissionException(identifier, e);
+ }
+ }
+
@Override
public boolean commitSnapshot(
Identifier identifier,
diff --git
a/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java
b/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java
index aec8fb775e..210db841eb 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java
@@ -57,6 +57,7 @@ import
org.apache.paimon.rest.exceptions.NotAuthorizedException;
import org.apache.paimon.rest.exceptions.NotImplementedException;
import org.apache.paimon.rest.requests.CreatePartitionsRequest;
import org.apache.paimon.rest.responses.ConfigResponse;
+import org.apache.paimon.schema.FileSystemSchemaManager;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.table.BlobDescriptorReaderFactory;
@@ -142,6 +143,44 @@ class MockRESTCatalogTest extends RESTCatalogTest {
}
}
+ @Test
+ void testCompatibilityWithServerWithoutSchemaEndpoints() throws Exception {
+ restCatalogServer.setSchemaEndpointsSupported(false);
+ Identifier identifier = Identifier.create("schema_compatibility",
"table");
+ createTable(identifier, Collections.emptyMap(),
Collections.singletonList("col1"));
+ restCatalogServer.clearReceivedHeaders();
+
+ FileStoreTable table = (FileStoreTable) catalog.getTable(identifier);
+
assertThat(table.schemaManager()).isInstanceOf(FileSystemSchemaManager.class);
+ long firstSchemaId = table.schemaManager().latest().get().id();
+ catalog.alterTable(identifier, SchemaChange.setOption("key", "value"),
false);
+ assertThat(((FileStoreTable)
catalog.getTable(identifier)).schemaManager().latest().get())
+ .extracting(schema -> schema.options().get("key"))
+ .isEqualTo("value");
+ catalog.rollbackSchema(identifier, firstSchemaId);
+ assertThat(
+ ((FileStoreTable) catalog.getTable(identifier))
+ .schemaManager()
+ .latest()
+ .get()
+ .id())
+ .isEqualTo(firstSchemaId);
+
+ ResourcePaths paths = new ResourcePaths("paimon");
+ assertThat(
+ restCatalogServer.getReceivedHeaders(
+ paths.schemas(
+ identifier.getDatabaseName(),
identifier.getObjectName())))
+ .isEmpty();
+ assertThat(
+ restCatalogServer.getReceivedHeaders(
+ paths.schemas(
+ identifier.getDatabaseName(),
+ identifier.getObjectName(),
+ "LATEST")))
+ .isEmpty();
+ }
+
@Test
void testAuthFail() {
Options options = new Options();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
index 59e9b870bc..90a948395b 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
@@ -96,7 +96,6 @@ import org.apache.paimon.rest.responses.GetTableResponse;
import org.apache.paimon.rest.responses.GetTableSnapshotResponse;
import org.apache.paimon.rest.responses.GetTableTokenResponse;
import org.apache.paimon.rest.responses.GetTagResponse;
-import org.apache.paimon.rest.responses.GetVersionSnapshotResponse;
import org.apache.paimon.rest.responses.GetViewResponse;
import org.apache.paimon.rest.responses.ListBranchesResponse;
import org.apache.paimon.rest.responses.ListConsumersResponse;
@@ -107,7 +106,6 @@ import
org.apache.paimon.rest.responses.ListFunctionsResponse;
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.ListSnapshotsResponse;
import org.apache.paimon.rest.responses.ListTableDetailsResponse;
import org.apache.paimon.rest.responses.ListTablesGloballyResponse;
import org.apache.paimon.rest.responses.ListTablesResponse;
@@ -167,7 +165,6 @@ import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
-import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -276,6 +273,7 @@ public class RESTCatalogServer {
private volatile boolean partitionListingSupported = true;
private volatile boolean partitionOptionsCreateSupported = true;
+ private volatile boolean schemaEndpointsSupported = true;
public RESTCatalogServer(
String dataPath, AuthProvider authProvider, ConfigResponse config,
String warehouse) {
@@ -348,6 +346,10 @@ public class RESTCatalogServer {
this.partitionOptionsCreateSupported = partitionOptionsCreateSupported;
}
+ public void setSchemaEndpointsSupported(boolean schemaEndpointsSupported) {
+ this.schemaEndpointsSupported = schemaEndpointsSupported;
+ }
+
public void clearReceivedListPartitionsByFilterRequests() {
receivedListPartitionsByFilterRequests.clear();
}
@@ -488,9 +490,7 @@ public class RESTCatalogServer {
return functionsHandle(parameters);
} else if (request.getPath().startsWith(databaseUri)) {
String[] resources =
- request.getPath()
- .substring((databaseUri +
"/").length())
- .split("/");
+ resourcePath.substring((databaseUri +
"/").length()).split("/");
String databaseName =
RESTUtil.decodeString(resources[0]);
if (noPermissionDatabases.contains(databaseName)) {
throw new
Catalog.DatabaseNoPermissionException(databaseName);
@@ -533,6 +533,14 @@ public class RESTCatalogServer {
resources.length == 4
&&
ResourcePaths.TABLES.equals(resources[1])
&&
ResourcePaths.SNAPSHOTS.equals(resources[3]);
+ boolean isListSchemas =
+ resources.length == 4
+ &&
ResourcePaths.TABLES.equals(resources[1])
+ &&
ResourcePaths.SCHEMAS.equals(resources[3]);
+ boolean isLoadSchema =
+ resources.length == 5
+ &&
ResourcePaths.TABLES.equals(resources[1])
+ &&
ResourcePaths.SCHEMAS.equals(resources[3]);
boolean isListConsumers =
resources.length == 4
&&
ResourcePaths.TABLES.equals(resources[1])
@@ -692,6 +700,12 @@ public class RESTCatalogServer {
return snapshotHandle(identifier);
} else if (isListSnapshots) {
return listSnapshots(identifier);
+ } else if ((isListSchemas || isLoadSchema) &&
!schemaEndpointsSupported) {
+ return new MockResponse().setResponseCode(404);
+ } else if (isListSchemas) {
+ return listSchemas(identifier, parameters);
+ } else if (isLoadSchema) {
+ return loadSchema(identifier, resources[4]);
} else if (isListConsumers) {
return listConsumers(identifier);
} else if (isResetConsumer) {
@@ -992,13 +1006,18 @@ public class RESTCatalogServer {
private MockResponse listSnapshots(Identifier identifier) throws Exception
{
FileStoreTable table = (FileStoreTable) catalog.getTable(identifier);
- Iterator<Snapshot> snapshots = table.snapshotManager().snapshots();
- List<Snapshot> snapshotList = new ArrayList<>();
- while (snapshots.hasNext()) {
- snapshotList.add(snapshots.next());
- }
- ListSnapshotsResponse response = new
ListSnapshotsResponse(snapshotList, null);
- return new
MockResponse().setResponseCode(200).setBody(RESTApi.toJson(response));
+ return RESTCatalogServerMetadataHandler.listSnapshots(table);
+ }
+
+ private MockResponse listSchemas(Identifier identifier, Map<String,
String> parameters)
+ throws Exception {
+ FileStoreTable table = getFileTable(identifier);
+ return RESTCatalogServerMetadataHandler.listSchemas(
+ table, getMaxResults(parameters), parameters.get(PAGE_TOKEN));
+ }
+
+ private MockResponse loadSchema(Identifier identifier, String version)
throws Exception {
+ return
RESTCatalogServerMetadataHandler.loadSchema(getFileTable(identifier), version);
}
private MockResponse listConsumers(Identifier identifier) throws Exception
{
@@ -1031,40 +1050,8 @@ public class RESTCatalogServer {
}
private MockResponse loadSnapshot(Identifier identifier, String version)
throws Exception {
-
FileStoreTable table = (FileStoreTable) catalog.getTable(identifier);
- SnapshotManager snapshotManager = table.snapshotManager();
- Snapshot snapshot = null;
- try {
- if (version.equals("EARLIEST")) {
- snapshot = snapshotManager.earliestSnapshot();
- } else if (version.equals("LATEST")) {
- snapshot = snapshotManager.latestSnapshot();
- } else {
- try {
- long snapshotId = Long.parseLong(version);
- snapshot = snapshotManager.tryGetSnapshot(snapshotId);
- } catch (NumberFormatException e) {
- Optional<Tag> tag = table.tagManager().get(version);
- if (tag.isPresent()) {
- snapshot = tag.get().trimToSnapshot();
- }
- }
- }
- } catch (Exception ignored) {
- }
-
- if (snapshot == null) {
- RESTResponse response =
- new ErrorResponse(
- ErrorResponse.RESOURCE_TYPE_SNAPSHOT,
- identifier.getDatabaseName(),
- "No Snapshot",
- 404);
- return mockResponse(response, 404);
- }
- GetVersionSnapshotResponse response = new
GetVersionSnapshotResponse(snapshot);
- return new
MockResponse().setResponseCode(200).setBody(RESTApi.toJson(response));
+ return RESTCatalogServerMetadataHandler.loadSnapshot(table, version);
}
private Optional<MockResponse> checkTablePartitioned(Identifier
identifier) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServerMetadataHandler.java
b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServerMetadataHandler.java
new file mode 100644
index 0000000000..3fb47f146a
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServerMetadataHandler.java
@@ -0,0 +1,148 @@
+/*
+ * 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;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.rest.responses.ErrorResponse;
+import org.apache.paimon.rest.responses.GetSchemaResponse;
+import org.apache.paimon.rest.responses.GetVersionSnapshotResponse;
+import org.apache.paimon.rest.responses.ListSchemasResponse;
+import org.apache.paimon.rest.responses.ListSnapshotsResponse;
+import org.apache.paimon.schema.FileSystemSchemaManager;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.tag.Tag;
+import org.apache.paimon.utils.SnapshotManager;
+
+import okhttp3.mockwebserver.MockResponse;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/** Metadata response handlers used by {@link RESTCatalogServer}. */
+final class RESTCatalogServerMetadataHandler {
+
+ private RESTCatalogServerMetadataHandler() {}
+
+ static MockResponse listSnapshots(FileStoreTable table) throws Exception {
+ Iterator<Snapshot> snapshots = table.snapshotManager().snapshots();
+ List<Snapshot> snapshotList = new ArrayList<>();
+ while (snapshots.hasNext()) {
+ snapshotList.add(snapshots.next());
+ }
+ ListSnapshotsResponse response = new
ListSnapshotsResponse(snapshotList, null);
+ return response(response);
+ }
+
+ static MockResponse loadSnapshot(FileStoreTable table, String version)
throws Exception {
+ SnapshotManager snapshotManager = table.snapshotManager();
+ Snapshot snapshot = null;
+ try {
+ if (version.equals("EARLIEST")) {
+ snapshot = snapshotManager.earliestSnapshot();
+ } else if (version.equals("LATEST")) {
+ snapshot = snapshotManager.latestSnapshot();
+ } else {
+ try {
+ snapshot =
snapshotManager.tryGetSnapshot(Long.parseLong(version));
+ } catch (NumberFormatException e) {
+ Optional<Tag> tag = table.tagManager().get(version);
+ if (tag.isPresent()) {
+ snapshot = tag.get().trimToSnapshot();
+ }
+ }
+ }
+ } catch (Exception ignored) {
+ }
+
+ if (snapshot == null) {
+ return notFound(ErrorResponse.RESOURCE_TYPE_SNAPSHOT, "No
Snapshot");
+ }
+ return response(new GetVersionSnapshotResponse(snapshot));
+ }
+
+ static MockResponse loadSchema(FileStoreTable table, String version)
throws Exception {
+ SchemaManager schemaManager = schemaManager(table);
+ TableSchema schema = null;
+ if ("LATEST".equals(version)) {
+ schema = schemaManager.latest().orElse(null);
+ } else {
+ List<TableSchema> schemas = schemaManager.listAll();
+ if ("EARLIEST".equals(version)) {
+ schema =
+ schemas.stream()
+ .min(Comparator.comparingLong(TableSchema::id))
+ .orElse(null);
+ } else {
+ try {
+ long schemaId = Long.parseLong(version);
+ if (schemaManager.schemaExists(schemaId)) {
+ schema = schemaManager.schema(schemaId);
+ }
+ } catch (NumberFormatException ignored) {
+ }
+ }
+ }
+
+ if (schema == null) {
+ return notFound(ErrorResponse.RESOURCE_TYPE_SCHEMA, "No Schema");
+ }
+ return response(new GetSchemaResponse(schema));
+ }
+
+ static MockResponse listSchemas(FileStoreTable table, int maxResults,
String pageToken)
+ throws Exception {
+ List<TableSchema> schemas = schemaManager(table).listAll();
+ schemas.sort(Comparator.comparingLong(TableSchema::id).reversed());
+ if (pageToken != null) {
+ long previousSchemaId = Long.parseLong(pageToken);
+ schemas =
+ schemas.stream()
+ .filter(schema -> schema.id() < previousSchemaId)
+ .collect(Collectors.toList());
+ }
+
+ int resultSize = Math.min(maxResults, schemas.size());
+ List<TableSchema> result = new ArrayList<>(schemas.subList(0,
resultSize));
+ String nextPageToken =
+ resultSize < schemas.size()
+ ? Long.toString(result.get(result.size() - 1).id())
+ : null;
+ return response(new ListSchemasResponse(result, nextPageToken));
+ }
+
+ private static SchemaManager schemaManager(FileStoreTable table) {
+ return new FileSystemSchemaManager(table.fileIO(), table.location());
+ }
+
+ private static MockResponse notFound(String resourceType, String message)
throws Exception {
+ return new MockResponse()
+ .setResponseCode(404)
+ .setBody(RESTApi.toJson(new ErrorResponse(resourceType, null,
message, 404)));
+ }
+
+ private static MockResponse response(RESTResponse response) throws
Exception {
+ return new
MockResponse().setResponseCode(200).setBody(RESTApi.toJson(response));
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
index 61958cb7bd..1cf891dbd3 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
@@ -81,6 +81,7 @@ 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.FileStoreTable;
import org.apache.paimon.table.FormatTable;
import org.apache.paimon.table.Instant;
@@ -2470,6 +2471,57 @@ public abstract class RESTCatalogTest extends
CatalogTestBase {
+ " is still referenced by
snapshots/tags/changelogs");
}
+ @Test
+ public void testLoadSchema() throws Exception {
+ Identifier identifier = Identifier.create("test_list_schemas",
"table_load");
+ createTable(identifier, Maps.newHashMap(), Lists.newArrayList("col1"));
+
+ FileStoreTable table = (FileStoreTable) catalog.getTable(identifier);
+ SchemaManager local = new FileSystemSchemaManager(table.fileIO(),
table.location());
+ TableSchema first = local.latest().get();
+
+ catalog.alterTable(identifier, SchemaChange.setOption("aa", "bb"),
false);
+ TableSchema latest = local.latest().get();
+
+ assertThat(catalog.loadSchema(identifier, "EARLIEST")).contains(first);
+ assertThat(catalog.loadSchema(identifier, "LATEST")).contains(latest);
+ assertThat(catalog.loadSchema(identifier,
Long.toString(first.id()))).contains(first);
+ assertThat(catalog.loadSchema(identifier, "9999")).isEmpty();
+ assertThat(catalog.loadSchema(identifier,
"invalid-version")).isEmpty();
+ }
+
+ @Test
+ public void testListSchemasPaged() throws Exception {
+ Identifier identifier = Identifier.create("test_list_schemas",
"table_paged");
+ createTable(identifier, Maps.newHashMap(), Lists.newArrayList("col1"));
+
+ FileStoreTable table = (FileStoreTable) catalog.getTable(identifier);
+ SchemaManager local = new FileSystemSchemaManager(table.fileIO(),
table.location());
+ catalog.alterTable(identifier, SchemaChange.setOption("aa", "bb"),
false);
+ catalog.alterTable(identifier, SchemaChange.setOption("cc", "dd"),
false);
+ catalog.alterTable(identifier, SchemaChange.setOption("ee", "ff"),
false);
+
+ List<TableSchema> expected = local.listAll();
+
expected.sort(java.util.Comparator.comparingLong(TableSchema::id).reversed());
+ PagedList<TableSchema> firstPage =
catalog.listSchemasPaged(identifier, 2, null);
+
assertThat(firstPage.getElements()).containsExactlyElementsOf(expected.subList(0,
2));
+ assertThat(firstPage.getNextPageToken()).isNotNull();
+
+ PagedList<TableSchema> secondPage =
+ catalog.listSchemasPaged(identifier, 2,
firstPage.getNextPageToken());
+
assertThat(secondPage.getElements()).containsExactlyElementsOf(expected.subList(2,
4));
+ assertThat(secondPage.getNextPageToken()).isNull();
+ }
+
+ @Test
+ public void testSchemaMethodsTableNotExist() {
+ Identifier missing = Identifier.create("test_list_schemas",
"missing_table");
+ assertThatThrownBy(() -> catalog.loadSchema(missing, "LATEST"))
+ .isInstanceOf(Catalog.TableNotExistException.class);
+ assertThatThrownBy(() -> catalog.listSchemasPaged(missing, null, null))
+ .isInstanceOf(Catalog.TableNotExistException.class);
+ }
+
@Test
public void testDataTokenExpired() throws Exception {
this.catalog = newRestCatalogWithDataToken();
