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 ad6e394a26fb..84cdecba481b 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.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.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.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 void rollbackSchema(Identifier identifier, long schemaId) { 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 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 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 4cef311061a5..d51a3119e1fb 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 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 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 4fe52b1d0b22..bfc9e3bf4e70 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 000000000000..1ac383a8b7c0 --- /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 000000000000..493386abde84 --- /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 { + + private static final String FIELD_SCHEMAS = "schemas"; + private static final String FIELD_NEXT_PAGE_TOKEN = "nextPageToken"; + + @JsonProperty(FIELD_SCHEMAS) + private final List schemas; + + @JsonProperty(FIELD_NEXT_PAGE_TOKEN) + private final String nextPageToken; + + public ListSchemasResponse(@JsonProperty(FIELD_SCHEMAS) List schemas) { + this(schemas, null); + } + + @JsonCreator + public ListSchemasResponse( + @JsonProperty(FIELD_SCHEMAS) List schemas, + @JsonProperty(FIELD_NEXT_PAGE_TOKEN) String nextPageToken) { + this.schemas = schemas; + this.nextPageToken = nextPageToken; + } + + @JsonGetter(FIELD_SCHEMAS) + public List getSchemas() { + return schemas; + } + + @Override + public List 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 7d26e03290a4..329abcf67b6c 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.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 @@ default void rollbackSchema(Identifier identifier, long schemaId) 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 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 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 2f20d38fdea9..bd342e1013de 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.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 PagedList listSnapshotsPaged( return wrapped.listSnapshotsPaged(identifier, maxResults, pageToken); } + @Override + public Optional loadSchema(Identifier identifier, String version) + throws TableNotExistException { + return wrapped.loadSchema(identifier, version); + } + + @Override + public PagedList 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 4fa683d7d4b1..0fb356966d61 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 boolean supportsVersionManagement() { return true; } + @Override + public Optional 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 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 aec8fb775e5d..210db841ebb9 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.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 @@ public void tearDown() throws Exception { } } + @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 59e9b870bc13..90a948395b11 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.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.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.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 void setPartitionOptionsCreateSupported(boolean partitionOptionsCreateSup this.partitionOptionsCreateSupported = partitionOptionsCreateSupported; } + public void setSchemaEndpointsSupported(boolean schemaEndpointsSupported) { + this.schemaEndpointsSupported = schemaEndpointsSupported; + } + public void clearReceivedListPartitionsByFilterRequests() { receivedListPartitionsByFilterRequests.clear(); } @@ -488,9 +490,7 @@ && isTableByIdRequest(request.getPath())) { 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 @@ && isTableByIdRequest(request.getPath())) { 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 @@ && isTableByIdRequest(request.getPath())) { 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 @@ private MockResponse snapshotHandle(Identifier identifier) throws Exception { private MockResponse listSnapshots(Identifier identifier) throws Exception { FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); - Iterator snapshots = table.snapshotManager().snapshots(); - List 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 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 @@ private MockResponse resetConsumer(Identifier identifier, String data) throws Ex } 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 = 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 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 000000000000..3fb47f146a18 --- /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 snapshots = table.snapshotManager().snapshots(); + List 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 = 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 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 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 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 61958cb7bd19..1cf891dbd3bd 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.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 void testRollbackSchemaFailedWithSnapshotReference() throws Exception { + " 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 expected = local.listAll(); + expected.sort(java.util.Comparator.comparingLong(TableSchema::id).reversed()); + PagedList firstPage = catalog.listSchemasPaged(identifier, 2, null); + assertThat(firstPage.getElements()).containsExactlyElementsOf(expected.subList(0, 2)); + assertThat(firstPage.getNextPageToken()).isNotNull(); + + PagedList 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();