From 7b77845a4bb1d96c9afbba300563b6a5ad3190fc Mon Sep 17 00:00:00 2001 From: "zhangyongxiang.alpha" Date: Mon, 7 Sep 2026 15:51:11 +0800 Subject: [PATCH 1/3] [core][rest] Support schema management over REST in RESTCatalog Add a Catalog-level API for listing schemas and expose it over the REST protocol so that RESTCatalog-backed tables can read historical schemas without direct filesystem access. * Introduce `Catalog#supportsSchemaManagement` and `Catalog#listSchemas(Identifier, SchemaFilter)` with a `SchemaFilter` value object (all / latest / earliest / by id / by range). * Add `GET /v1/{prefix}/databases/{db}/tables/{obj}/schemas` with a `ListSchemaResponse` payload; encode `SchemaFilter` as query parameters (`latest`, `earliest`, `schemaId`, `maxSchemaId`, `minSchemaId`). * Implement `RESTCatalog#supportsSchemaManagement`/`listSchemas` and map REST errors to catalog exceptions. * Add `CatalogSchemaManager`, a `SchemaManager` that delegates to the owning `Catalog` (analogous to `CatalogBranchManager`); reads go through `listSchemas`, writes reuse existing `Catalog#createTable`/`alterTable`/`rollbackSchema`. * Wire `AbstractFileStoreTable#schemaManager` to prefer `CatalogSchemaManager` whenever `supportsSchemaManagement()` is true. * Extend the REST mock server and add tests covering the new filter variants and the catalog-backed schema manager. Co-authored-by: TRAE CLI --- .../java/org/apache/paimon/rest/RESTApi.java | 39 +++ .../org/apache/paimon/rest/ResourcePaths.java | 12 + .../rest/responses/ListSchemaResponse.java | 95 ++++++ .../apache/paimon/schema/SchemaFilter.java | 160 ++++++++++ .../org/apache/paimon/catalog/Catalog.java | 49 +++ .../apache/paimon/catalog/CatalogUtils.java | 3 +- .../org/apache/paimon/rest/RESTCatalog.java | 27 ++ .../paimon/schema/CatalogSchemaManager.java | 296 ++++++++++++++++++ .../paimon/table/AbstractFileStoreTable.java | 5 + .../paimon/table/CatalogEnvironment.java | 33 +- .../apache/paimon/rest/RESTCatalogServer.java | 91 +++++- .../apache/paimon/rest/RESTCatalogTest.java | 121 +++++++ 12 files changed, 925 insertions(+), 6 deletions(-) create mode 100644 paimon-api/src/main/java/org/apache/paimon/rest/responses/ListSchemaResponse.java create mode 100644 paimon-api/src/main/java/org/apache/paimon/schema/SchemaFilter.java create mode 100644 paimon-core/src/main/java/org/apache/paimon/schema/CatalogSchemaManager.java 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..23037e4d3e20 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 @@ -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}. + * + *

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 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. * 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..c21024f232a3 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,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, diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/responses/ListSchemaResponse.java b/paimon-api/src/main/java/org/apache/paimon/rest/responses/ListSchemaResponse.java new file mode 100644 index 000000000000..ac4abe12a59b --- /dev/null +++ b/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 schemas; + + @JsonCreator + public ListSchemaResponse(@JsonProperty(FIELD_SCHEMAS) List schemas) { + this.schemas = schemas; + } + + @JsonGetter(FIELD_SCHEMAS) + public List 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; + } + } +} diff --git a/paimon-api/src/main/java/org/apache/paimon/schema/SchemaFilter.java b/paimon-api/src/main/java/org/apache/paimon/schema/SchemaFilter.java new file mode 100644 index 000000000000..9b9ae9085ee8 --- /dev/null +++ b/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. + * + *

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 + + '}'; + } +} 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..8bf125024d35 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,8 @@ import org.apache.paimon.rest.responses.GetTagResponse; import org.apache.paimon.schema.Schema; import org.apache.paimon.schema.SchemaChange; +import org.apache.paimon.schema.SchemaFilter; +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 +882,53 @@ default void rollbackSchema(Identifier identifier, long schemaId) throw new UnsupportedOperationException(); } + // ==================== Schema management methods ========================== + + /** + * Whether this catalog supports schema management for tables. If not, {@link + * #listSchemas(Identifier, SchemaFilter)} will throw an {@link UnsupportedOperationException}. + * + *

This is orthogonal to {@link #supportsVersionManagement()}: version management covers + * snapshot / tag / branch APIs, while schema management covers reading historical {@link + * TableSchema}s of a table. A catalog may reasonably support one without the other. Write-side + * operations on schemas ({@link #createTable(Identifier, Schema, boolean)}, {@link + * #alterTable(Identifier, List, boolean)} and {@link #rollbackSchema(Identifier, long)}) are + * already exposed by the corresponding methods on this interface. + */ + default boolean supportsSchemaManagement() { + return false; + } + + /** + * List schemas of a table, filtered by the given {@link SchemaFilter}. + * + *

All schema read patterns (latest / earliest / by id / by range / all) share this single + * method; callers select the desired subset by populating {@link SchemaFilter}. Implementations + * must interpret the filter fields consistently: + * + *

+ * + *

The returned list is not required to be sorted; callers that need a specific order should + * sort by {@link TableSchema#id()} themselves. + * + * @param identifier path of the table + * @param filter which schemas to return, must not be {@code null} + * @throws TableNotExistException if the table does not exist + * @throws UnsupportedOperationException if the catalog does not {@link + * #supportsSchemaManagement()} + */ + default List listSchemas(Identifier identifier, SchemaFilter filter) + 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/CatalogUtils.java b/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java index 916f63350c7b..c625e00c161f 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java @@ -394,7 +394,8 @@ public static Table loadTable( isRestCatalog ? null : lockContext, catalogContext, catalog.supportsVersionManagement(), - catalog.supportsPartitionModification()); + catalog.supportsPartitionModification(), + catalog.supportsSchemaManagement()); Path path = new Path(schema.options().get(PATH.key())); FileStoreTable table = FileStoreTableFactory.create(dataFileIO.apply(path), path, schema, catalogEnv); 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..4d6e22684bfd 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 @@ -59,9 +59,11 @@ import org.apache.paimon.rest.responses.GetTableResponse; import org.apache.paimon.rest.responses.GetTagResponse; import org.apache.paimon.rest.responses.GetViewResponse; +import org.apache.paimon.rest.responses.ListSchemaResponse; import org.apache.paimon.schema.FileSystemSchemaManager; import org.apache.paimon.schema.Schema; import org.apache.paimon.schema.SchemaChange; +import org.apache.paimon.schema.SchemaFilter; import org.apache.paimon.schema.SchemaManager; import org.apache.paimon.schema.TableSchema; import org.apache.paimon.table.FormatTable; @@ -478,6 +480,31 @@ public boolean supportsVersionManagement() { return true; } + @Override + public boolean supportsSchemaManagement() { + return true; + } + + @Override + public List listSchemas(Identifier identifier, SchemaFilter filter) + throws TableNotExistException { + try { + ListSchemaResponse response = api.listSchemas(identifier, filter); + if (response.getSchemas() == null || response.getSchemas().isEmpty()) { + return Collections.emptyList(); + } + List result = new ArrayList<>(response.getSchemas().size()); + for (ListSchemaResponse.SchemaItem item : response.getSchemas()) { + result.add(TableSchema.create(item.getSchemaId(), item.getSchema())); + } + return result; + } 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/main/java/org/apache/paimon/schema/CatalogSchemaManager.java b/paimon-core/src/main/java/org/apache/paimon/schema/CatalogSchemaManager.java new file mode 100644 index 000000000000..2a71cbc01d47 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/schema/CatalogSchemaManager.java @@ -0,0 +1,296 @@ +/* + * 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 org.apache.paimon.catalog.Catalog; +import org.apache.paimon.catalog.CatalogLoader; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.fs.Path; +import org.apache.paimon.table.SchemaModification; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.ChangelogManager; +import org.apache.paimon.utils.FunctionWithException; +import org.apache.paimon.utils.SnapshotManager; +import org.apache.paimon.utils.TagManager; +import org.apache.paimon.utils.ThrowingConsumer; + +import javax.annotation.Nullable; +import javax.annotation.concurrent.ThreadSafe; + +import java.io.FileNotFoundException; +import java.io.IOException; +import java.util.Comparator; +import java.util.List; +import java.util.Optional; +import java.util.function.Predicate; +import java.util.stream.Collectors; + +/** + * A {@link SchemaManager} implementation that delegates every schema query and mutation to the + * owning {@link Catalog}. This mirrors the {@link org.apache.paimon.utils.CatalogBranchManager} + * pattern for the {@code BranchManager} interface: on-filesystem operations (reading raw {@code + * schema-*} files, deleting schemas, resolving schema paths) are intentionally not supported. + * + *

Read side collapses onto the single {@link Catalog#listSchemas(Identifier, SchemaFilter)} + * endpoint; different callers populate the filter differently (latest / earliest / by id / by range + * / all). Write side is routed through {@link Catalog#createTable(Identifier, Schema, boolean)}, + * {@link Catalog#alterTable(Identifier, List, boolean)} and {@link + * Catalog#rollbackSchema(Identifier, long)}. + */ +@ThreadSafe +public class CatalogSchemaManager implements SchemaManager { + + private static final long serialVersionUID = 1L; + + private final CatalogLoader catalogLoader; + private final Identifier identifier; + + public CatalogSchemaManager(CatalogLoader catalogLoader, Identifier identifier) { + this.catalogLoader = catalogLoader; + this.identifier = identifier; + } + + @Override + public SchemaManager copyWithBranch(String branchName) { + Identifier branchIdentifier = + new Identifier(identifier.getDatabaseName(), identifier.getTableName(), branchName); + return new CatalogSchemaManager(catalogLoader, branchIdentifier); + } + + @Override + public Optional latest() { + return executeGet( + catalog -> { + List schemas = + catalog.listSchemas(identifier, SchemaFilter.latest()); + if (schemas.isEmpty()) { + return Optional.empty(); + } + return Optional.of(schemas.get(0)); + }); + } + + @Override + public TableSchema latestOrThrow(String message) { + return latest().orElseThrow(() -> new RuntimeException(message)); + } + + @Override + public long earliestCreationTime() { + return executeGet( + catalog -> { + List schemas = + catalog.listSchemas(identifier, SchemaFilter.earliest()); + if (schemas.isEmpty()) { + throw new IllegalStateException("Table " + identifier + " has no schema."); + } + return schemas.get(0).timeMillis(); + }); + } + + @Override + public List listAll() { + return executeGet( + catalog -> { + List schemas = catalog.listSchemas(identifier, SchemaFilter.all()); + schemas.sort(Comparator.comparingLong(TableSchema::id)); + return schemas; + }); + } + + @Override + public List listAllIds() { + return listAll().stream().map(TableSchema::id).collect(Collectors.toList()); + } + + @Override + public TableSchema createTable(Schema schema) throws Exception { + return createTable(schema, false); + } + + @Override + public TableSchema createTable(Schema schema, boolean externalTable) throws Exception { + executePost(catalog -> catalog.createTable(identifier, schema, false)); + return latestOrThrow( + "Failed to load the newly created schema for table " + identifier + "."); + } + + @Override + public TableSchema commitChanges(SchemaChange... changes) throws Exception { + return commitChanges(java.util.Arrays.asList(changes)); + } + + @Override + public TableSchema commitChanges(List changes) + throws Catalog.TableNotExistException, Catalog.ColumnAlreadyExistException, + Catalog.ColumnNotExistException { + try (Catalog catalog = catalogLoader.load()) { + catalog.alterTable(identifier, changes, false); + } catch (Catalog.TableNotExistException + | Catalog.ColumnAlreadyExistException + | Catalog.ColumnNotExistException e) { + throw e; + } catch (RuntimeException e) { + throw e; + } catch (Exception e) { + throw new RuntimeException(e); + } + return latestOrThrow( + "Failed to load the latest schema after altering table " + identifier + "."); + } + + @Override + public boolean mergeSchema( + RowType rowType, + boolean typeWidening, + boolean allowExplicitCast, + boolean caseSensitive, + @Nullable SchemaModification schemaModification) { + TableSchema current = + latest().orElseThrow( + () -> + new RuntimeException( + "It requires that the current schema to exist when calling 'mergeSchema'")); + TableSchema update = + SchemaMergingUtils.mergeSchemas( + current, rowType, typeWidening, allowExplicitCast, caseSensitive); + if (current.equals(update)) { + return false; + } + List changes = + SchemaMergingUtils.diffSchemaChanges(current, update, caseSensitive); + try { + if (schemaModification != null) { + schemaModification.alterSchema(changes); + } else { + commitChanges(changes); + } + return true; + } catch (RuntimeException e) { + throw e; + } catch (Exception e) { + throw new RuntimeException("Failed to commit the schema.", e); + } + } + + @Override + public boolean commit(TableSchema newSchema) { + throw new UnsupportedOperationException( + "CatalogSchemaManager does not support committing a TableSchema directly; " + + "use commitChanges or the catalog createTable / alterTable APIs instead."); + } + + @Override + public TableSchema schema(long id) { + return executeGet( + catalog -> { + List schemas = + catalog.listSchemas(identifier, SchemaFilter.withId(id)); + if (schemas.isEmpty()) { + throw new IllegalStateException( + "Schema " + id + " not found for table " + identifier + "."); + } + return schemas.get(0); + }); + } + + @Override + public TableSchema tryGetSchema(long id) throws FileNotFoundException { + List schemas = + executeGet(catalog -> catalog.listSchemas(identifier, SchemaFilter.withId(id))); + if (schemas.isEmpty()) { + throw new FileNotFoundException( + "Schema " + id + " not found for table " + identifier + "."); + } + return schemas.get(0); + } + + @Override + public boolean schemaExists(long id) { + List schemas = + executeGet(catalog -> catalog.listSchemas(identifier, SchemaFilter.withId(id))); + return !schemas.isEmpty(); + } + + @Override + public Path schemaDirectory() { + throw new UnsupportedOperationException( + "CatalogSchemaManager does not expose a filesystem schema directory."); + } + + @Override + public Path toSchemaPath(long schemaId) { + throw new UnsupportedOperationException( + "CatalogSchemaManager does not expose a filesystem schema path."); + } + + @Override + public List schemaPaths(Predicate predicate) throws IOException { + throw new UnsupportedOperationException( + "CatalogSchemaManager does not expose filesystem schema paths."); + } + + @Override + public void deleteSchema(long schemaId) { + throw new UnsupportedOperationException( + "CatalogSchemaManager does not support deleting a single schema; " + + "use catalog.rollbackSchema instead."); + } + + @Override + public void rollbackTo( + long targetSchemaId, + SnapshotManager snapshotManager, + TagManager tagManager, + ChangelogManager changelogManager) { + executePost(catalog -> catalog.rollbackSchema(identifier, targetSchemaId)); + } + + private void executePost(ThrowingConsumer func) { + executeGet( + catalog -> { + try { + func.accept(catalog); + return null; + } catch (Catalog.TableNotExistException e) { + throw new IllegalArgumentException( + String.format( + "Table '%s' doesn't exist.", e.identifier().getFullName())); + } catch (Catalog.DatabaseNotExistException e) { + throw new IllegalArgumentException( + String.format("Database '%s' doesn't exist.", e.database())); + } catch (Catalog.TableAlreadyExistException e) { + throw new IllegalArgumentException( + String.format( + "Table '%s' already exists.", + e.identifier().getFullName())); + } + }); + } + + private T executeGet(FunctionWithException func) { + try (Catalog catalog = catalogLoader.load()) { + return func.apply(catalog); + } catch (RuntimeException e) { + throw e; + } catch (Exception e) { + throw new RuntimeException(e); + } + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java index 2117f4c4dd8f..a0d1b1df7a9b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java @@ -33,6 +33,7 @@ import org.apache.paimon.options.ExpireConfig; import org.apache.paimon.options.Options; import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.schema.CatalogSchemaManager; import org.apache.paimon.schema.FileSystemSchemaManager; import org.apache.paimon.schema.SchemaManager; import org.apache.paimon.schema.SchemaValidation; @@ -423,6 +424,10 @@ public FileStoreTable copy(TableSchema newTableSchema) { @Override public SchemaManager schemaManager() { + if (catalogEnvironment.catalogLoader() != null + && catalogEnvironment.supportsSchemaManagement()) { + return new CatalogSchemaManager(catalogEnvironment.catalogLoader(), identifier()); + } return new FileSystemSchemaManager(fileIO(), path, currentBranch()); } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java b/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java index 66979ba811f2..b963336cf081 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java @@ -63,6 +63,7 @@ public class CatalogEnvironment implements Serializable { @Nullable private final CatalogContext catalogContext; private final boolean supportsVersionManagement; private final boolean supportsPartitionModification; + private final boolean supportsSchemaManagement; public CatalogEnvironment( @Nullable Identifier identifier, @@ -73,6 +74,28 @@ public CatalogEnvironment( @Nullable CatalogContext catalogContext, boolean supportsVersionManagement, boolean supportsPartitionModification) { + this( + identifier, + uuid, + catalogLoader, + lockFactory, + lockContext, + catalogContext, + supportsVersionManagement, + supportsPartitionModification, + false); + } + + public CatalogEnvironment( + @Nullable Identifier identifier, + @Nullable String uuid, + @Nullable CatalogLoader catalogLoader, + @Nullable CatalogLockFactory lockFactory, + @Nullable CatalogLockContext lockContext, + @Nullable CatalogContext catalogContext, + boolean supportsVersionManagement, + boolean supportsPartitionModification, + boolean supportsSchemaManagement) { this.identifier = identifier; this.uuid = uuid; this.catalogLoader = catalogLoader; @@ -81,10 +104,11 @@ public CatalogEnvironment( this.catalogContext = catalogContext; this.supportsVersionManagement = supportsVersionManagement; this.supportsPartitionModification = supportsPartitionModification; + this.supportsSchemaManagement = supportsSchemaManagement; } public static CatalogEnvironment empty() { - return new CatalogEnvironment(null, null, null, null, null, null, false, false); + return new CatalogEnvironment(null, null, null, null, null, null, false, false, false); } @Nullable @@ -122,6 +146,10 @@ public boolean supportsVersionManagement() { return supportsVersionManagement; } + public boolean supportsSchemaManagement() { + return supportsSchemaManagement; + } + @Nullable public SchemaModification schemaModification() { if (catalogLoader == null) { @@ -253,7 +281,8 @@ public CatalogEnvironment copy(Identifier identifier) { lockContext, catalogContext, supportsVersionManagement, - supportsPartitionModification); + supportsPartitionModification, + supportsSchemaManagement); } public TableQueryAuth tableQueryAuth(CoreOptions 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..b18ef0274bcd 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 @@ -107,6 +107,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; @@ -118,6 +119,7 @@ import org.apache.paimon.schema.FileSystemSchemaManager; import org.apache.paimon.schema.Schema; import org.apache.paimon.schema.SchemaChange; +import org.apache.paimon.schema.SchemaFilter; import org.apache.paimon.schema.SchemaManager; import org.apache.paimon.schema.TableSchema; import org.apache.paimon.table.CatalogEnvironment; @@ -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,10 @@ && 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 isListConsumers = resources.length == 4 && ResourcePaths.TABLES.equals(resources[1]) @@ -692,6 +696,8 @@ && isTableByIdRequest(request.getPath())) { return snapshotHandle(identifier); } else if (isListSnapshots) { return listSnapshots(identifier); + } else if (isListSchemas) { + return listSchemas(identifier, parameters); } else if (isListConsumers) { return listConsumers(identifier); } else if (isResetConsumer) { @@ -1001,6 +1007,85 @@ private MockResponse listSnapshots(Identifier identifier) throws Exception { return new MockResponse().setResponseCode(200).setBody(RESTApi.toJson(response)); } + private MockResponse listSchemas(Identifier identifier, Map parameters) + throws Exception { + if (noPermissionTables.contains(identifier.getFullName())) { + throw new Catalog.TableNoPermissionException(identifier); + } + if (!tableMetadataStore.containsKey(identifier.getFullName())) { + throw new Catalog.TableNotExistException(identifier); + } + FileStoreTable table = getFileTable(identifier); + SchemaManager schemaManager = new FileSystemSchemaManager(table.fileIO(), table.location()); + SchemaFilter filter = parseSchemaFilter(parameters); + List all = schemaManager.listAll(); + all.sort(Comparator.comparingLong(TableSchema::id).reversed()); + List items; + if (filter.isLatest()) { + items = + all.isEmpty() + ? Collections.emptyList() + : Collections.singletonList(toSchemaItem(all.get(0))); + } else if (filter.isEarliest()) { + items = + all.isEmpty() + ? Collections.emptyList() + : Collections.singletonList(toSchemaItem(all.get(all.size() - 1))); + } else if (filter.schemaId() != null) { + long target = filter.schemaId(); + items = + all.stream() + .filter(s -> s.id() == target) + .findFirst() + .map(s -> Collections.singletonList(toSchemaItem(s))) + .orElse(Collections.emptyList()); + } else { + items = + all.stream() + .filter( + s -> + filter.maxSchemaId() == null + || s.id() <= filter.maxSchemaId()) + .filter( + s -> + filter.minSchemaId() == null + || s.id() >= filter.minSchemaId()) + .map(RESTCatalogServer::toSchemaItem) + .collect(Collectors.toList()); + } + ListSchemaResponse response = new ListSchemaResponse(items); + return new MockResponse().setResponseCode(200).setBody(RESTApi.toJson(response)); + } + + private static SchemaFilter parseSchemaFilter(Map parameters) { + if (parameters == null || parameters.isEmpty()) { + return SchemaFilter.all(); + } + if ("true".equalsIgnoreCase(parameters.get("latest"))) { + return SchemaFilter.latest(); + } + if ("true".equalsIgnoreCase(parameters.get("earliest"))) { + return SchemaFilter.earliest(); + } + String schemaId = parameters.get("schemaId"); + if (schemaId != null) { + return SchemaFilter.withId(Long.parseLong(schemaId)); + } + String maxSchemaId = parameters.get("maxSchemaId"); + String minSchemaId = parameters.get("minSchemaId"); + Long max = maxSchemaId == null ? null : Long.parseLong(maxSchemaId); + Long min = minSchemaId == null ? null : Long.parseLong(minSchemaId); + if (max == null && min == null) { + return SchemaFilter.all(); + } + return SchemaFilter.range(max, min); + } + + private static ListSchemaResponse.SchemaItem toSchemaItem(TableSchema schema) { + return new ListSchemaResponse.SchemaItem( + schema.id(), schema.toSchema(), schema.timeMillis()); + } + private MockResponse listConsumers(Identifier identifier) throws Exception { FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); ConsumerManager consumerManager = 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..15c99babe5cf 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 @@ -80,7 +80,9 @@ import org.apache.paimon.schema.FileSystemSchemaManager; import org.apache.paimon.schema.Schema; import org.apache.paimon.schema.SchemaChange; +import org.apache.paimon.schema.SchemaFilter; 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 +2472,125 @@ public void testRollbackSchemaFailedWithSnapshotReference() throws Exception { + " is still referenced by snapshots/tags/changelogs"); } + @Test + public void testSupportsSchemaManagement() { + assertThat(catalog.supportsSchemaManagement()).isTrue(); + } + + @Test + public void testListSchemasAll() throws Exception { + Identifier identifier = Identifier.create("test_list_schemas", "table_all"); + createTable(identifier, Maps.newHashMap(), Lists.newArrayList("col1")); + + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); + SchemaManager local = new FileSystemSchemaManager(table.fileIO(), table.location()); + long firstSchemaId = local.latest().get().id(); + + catalog.alterTable(identifier, SchemaChange.setOption("aa", "bb"), false); + catalog.alterTable(identifier, SchemaChange.setOption("cc", "dd"), false); + + List all = catalog.listSchemas(identifier, SchemaFilter.all()); + assertThat(all).hasSize(3); + all.sort(java.util.Comparator.comparingLong(TableSchema::id)); + assertThat(all.get(0).id()).isEqualTo(firstSchemaId); + assertThat(all.get(2).id()).isEqualTo(firstSchemaId + 2); + } + + @Test + public void testListSchemasLatestAndEarliest() throws Exception { + Identifier identifier = Identifier.create("test_list_schemas", "table_latest_earliest"); + createTable(identifier, Maps.newHashMap(), Lists.newArrayList("col1")); + + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); + SchemaManager local = new FileSystemSchemaManager(table.fileIO(), table.location()); + long firstSchemaId = local.latest().get().id(); + + catalog.alterTable(identifier, SchemaChange.setOption("aa", "bb"), false); + long secondSchemaId = local.latest().get().id(); + + List latest = catalog.listSchemas(identifier, SchemaFilter.latest()); + assertThat(latest).hasSize(1); + assertThat(latest.get(0).id()).isEqualTo(secondSchemaId); + + List earliest = catalog.listSchemas(identifier, SchemaFilter.earliest()); + assertThat(earliest).hasSize(1); + assertThat(earliest.get(0).id()).isEqualTo(firstSchemaId); + } + + @Test + public void testListSchemasById() throws Exception { + Identifier identifier = Identifier.create("test_list_schemas", "table_by_id"); + createTable(identifier, Maps.newHashMap(), Lists.newArrayList("col1")); + + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); + SchemaManager local = new FileSystemSchemaManager(table.fileIO(), table.location()); + long firstSchemaId = local.latest().get().id(); + catalog.alterTable(identifier, SchemaChange.setOption("aa", "bb"), false); + + List byId = + catalog.listSchemas(identifier, SchemaFilter.withId(firstSchemaId)); + assertThat(byId).hasSize(1); + assertThat(byId.get(0).id()).isEqualTo(firstSchemaId); + + List missing = catalog.listSchemas(identifier, SchemaFilter.withId(9999L)); + assertThat(missing).isEmpty(); + } + + @Test + public void testListSchemasByRange() throws Exception { + Identifier identifier = Identifier.create("test_list_schemas", "table_by_range"); + createTable(identifier, Maps.newHashMap(), Lists.newArrayList("col1")); + + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); + SchemaManager local = new FileSystemSchemaManager(table.fileIO(), table.location()); + long firstSchemaId = local.latest().get().id(); + 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 range = + catalog.listSchemas( + identifier, SchemaFilter.range(firstSchemaId + 2, firstSchemaId + 1)); + assertThat(range).hasSize(2); + range.sort(java.util.Comparator.comparingLong(TableSchema::id)); + assertThat(range.get(0).id()).isEqualTo(firstSchemaId + 1); + assertThat(range.get(1).id()).isEqualTo(firstSchemaId + 2); + } + + @Test + public void testListSchemasTableNotExist() { + Identifier missing = Identifier.create("test_list_schemas", "missing_table"); + assertThatThrownBy(() -> catalog.listSchemas(missing, SchemaFilter.all())) + .isInstanceOf(Catalog.TableNotExistException.class); + } + + @Test + public void testCatalogSchemaManagerBackedTableUsesRest() throws Exception { + Identifier identifier = Identifier.create("test_list_schemas", "table_catalog_backed"); + createTable(identifier, Maps.newHashMap(), Lists.newArrayList("col1")); + + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); + // AbstractFileStoreTable should build a CatalogSchemaManager because + // RESTCatalog#supportsSchemaManagement is true. + assertThat(table.schemaManager().getClass().getSimpleName()) + .isEqualTo("CatalogSchemaManager"); + long firstSchemaId = table.schemaManager().latest().get().id(); + + catalog.alterTable(identifier, SchemaChange.setOption("aa", "bb"), false); + // The catalog-backed schema manager should observe the newest schema over REST. + assertThat(table.schemaManager().latest().get().id()).isEqualTo(firstSchemaId + 1); + + // A rollback via the catalog-backed manager must reach the REST server. + table.schemaManager() + .rollbackTo( + firstSchemaId, + table.snapshotManager(), + table.tagManager(), + new org.apache.paimon.utils.ChangelogManager( + table.fileIO(), table.location(), null)); + assertThat(table.schemaManager().latest().get().id()).isEqualTo(firstSchemaId); + } + @Test public void testDataTokenExpired() throws Exception { this.catalog = newRestCatalogWithDataToken(); From 5b271fa0ccf7eb7868d5b3503be28a9a5884706f Mon Sep 17 00:00:00 2001 From: "zhangyongxiang.alpha" Date: Mon, 7 Sep 2026 17:56:14 +0800 Subject: [PATCH 2/3] [test] Extract REST catalog metadata handlers --- .../apache/paimon/rest/RESTCatalogServer.java | 87 +----------- .../RESTCatalogServerMetadataHandler.java | 126 ++++++++++++++++++ 2 files changed, 128 insertions(+), 85 deletions(-) create mode 100644 paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServerMetadataHandler.java 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 b18ef0274bcd..616c97e64e2d 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 @@ -107,8 +107,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.ListSchemaResponse; -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; @@ -119,7 +117,6 @@ import org.apache.paimon.schema.FileSystemSchemaManager; import org.apache.paimon.schema.Schema; import org.apache.paimon.schema.SchemaChange; -import org.apache.paimon.schema.SchemaFilter; import org.apache.paimon.schema.SchemaManager; import org.apache.paimon.schema.TableSchema; import org.apache.paimon.table.CatalogEnvironment; @@ -169,7 +166,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; @@ -998,92 +994,13 @@ 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 { - if (noPermissionTables.contains(identifier.getFullName())) { - throw new Catalog.TableNoPermissionException(identifier); - } - if (!tableMetadataStore.containsKey(identifier.getFullName())) { - throw new Catalog.TableNotExistException(identifier); - } FileStoreTable table = getFileTable(identifier); - SchemaManager schemaManager = new FileSystemSchemaManager(table.fileIO(), table.location()); - SchemaFilter filter = parseSchemaFilter(parameters); - List all = schemaManager.listAll(); - all.sort(Comparator.comparingLong(TableSchema::id).reversed()); - List items; - if (filter.isLatest()) { - items = - all.isEmpty() - ? Collections.emptyList() - : Collections.singletonList(toSchemaItem(all.get(0))); - } else if (filter.isEarliest()) { - items = - all.isEmpty() - ? Collections.emptyList() - : Collections.singletonList(toSchemaItem(all.get(all.size() - 1))); - } else if (filter.schemaId() != null) { - long target = filter.schemaId(); - items = - all.stream() - .filter(s -> s.id() == target) - .findFirst() - .map(s -> Collections.singletonList(toSchemaItem(s))) - .orElse(Collections.emptyList()); - } else { - items = - all.stream() - .filter( - s -> - filter.maxSchemaId() == null - || s.id() <= filter.maxSchemaId()) - .filter( - s -> - filter.minSchemaId() == null - || s.id() >= filter.minSchemaId()) - .map(RESTCatalogServer::toSchemaItem) - .collect(Collectors.toList()); - } - ListSchemaResponse response = new ListSchemaResponse(items); - return new MockResponse().setResponseCode(200).setBody(RESTApi.toJson(response)); - } - - private static SchemaFilter parseSchemaFilter(Map parameters) { - if (parameters == null || parameters.isEmpty()) { - return SchemaFilter.all(); - } - if ("true".equalsIgnoreCase(parameters.get("latest"))) { - return SchemaFilter.latest(); - } - if ("true".equalsIgnoreCase(parameters.get("earliest"))) { - return SchemaFilter.earliest(); - } - String schemaId = parameters.get("schemaId"); - if (schemaId != null) { - return SchemaFilter.withId(Long.parseLong(schemaId)); - } - String maxSchemaId = parameters.get("maxSchemaId"); - String minSchemaId = parameters.get("minSchemaId"); - Long max = maxSchemaId == null ? null : Long.parseLong(maxSchemaId); - Long min = minSchemaId == null ? null : Long.parseLong(minSchemaId); - if (max == null && min == null) { - return SchemaFilter.all(); - } - return SchemaFilter.range(max, min); - } - - private static ListSchemaResponse.SchemaItem toSchemaItem(TableSchema schema) { - return new ListSchemaResponse.SchemaItem( - schema.id(), schema.toSchema(), schema.timeMillis()); + return RESTCatalogServerMetadataHandler.listSchemas(table, parameters); } private MockResponse listConsumers(Identifier identifier) throws Exception { 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..ee025c07bc8d --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServerMetadataHandler.java @@ -0,0 +1,126 @@ +/* + * 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.ListSchemaResponse; +import org.apache.paimon.rest.responses.ListSnapshotsResponse; +import org.apache.paimon.schema.FileSystemSchemaManager; +import org.apache.paimon.schema.SchemaFilter; +import org.apache.paimon.schema.SchemaManager; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.table.FileStoreTable; + +import okhttp3.mockwebserver.MockResponse; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +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 new MockResponse().setResponseCode(200).setBody(RESTApi.toJson(response)); + } + + static MockResponse listSchemas(FileStoreTable table, Map parameters) + throws Exception { + SchemaManager schemaManager = new FileSystemSchemaManager(table.fileIO(), table.location()); + SchemaFilter filter = parseSchemaFilter(parameters); + List all = schemaManager.listAll(); + all.sort(Comparator.comparingLong(TableSchema::id).reversed()); + List items; + if (filter.isLatest()) { + items = + all.isEmpty() + ? Collections.emptyList() + : Collections.singletonList(toSchemaItem(all.get(0))); + } else if (filter.isEarliest()) { + items = + all.isEmpty() + ? Collections.emptyList() + : Collections.singletonList(toSchemaItem(all.get(all.size() - 1))); + } else if (filter.schemaId() != null) { + long target = filter.schemaId(); + items = + all.stream() + .filter(s -> s.id() == target) + .findFirst() + .map(s -> Collections.singletonList(toSchemaItem(s))) + .orElse(Collections.emptyList()); + } else { + items = + all.stream() + .filter( + s -> + filter.maxSchemaId() == null + || s.id() <= filter.maxSchemaId()) + .filter( + s -> + filter.minSchemaId() == null + || s.id() >= filter.minSchemaId()) + .map(RESTCatalogServerMetadataHandler::toSchemaItem) + .collect(Collectors.toList()); + } + ListSchemaResponse response = new ListSchemaResponse(items); + return new MockResponse().setResponseCode(200).setBody(RESTApi.toJson(response)); + } + + private static SchemaFilter parseSchemaFilter(Map parameters) { + if (parameters == null || parameters.isEmpty()) { + return SchemaFilter.all(); + } + if ("true".equalsIgnoreCase(parameters.get("latest"))) { + return SchemaFilter.latest(); + } + if ("true".equalsIgnoreCase(parameters.get("earliest"))) { + return SchemaFilter.earliest(); + } + String schemaId = parameters.get("schemaId"); + if (schemaId != null) { + return SchemaFilter.withId(Long.parseLong(schemaId)); + } + String maxSchemaId = parameters.get("maxSchemaId"); + String minSchemaId = parameters.get("minSchemaId"); + Long max = maxSchemaId == null ? null : Long.parseLong(maxSchemaId); + Long min = minSchemaId == null ? null : Long.parseLong(minSchemaId); + if (max == null && min == null) { + return SchemaFilter.all(); + } + return SchemaFilter.range(max, min); + } + + private static ListSchemaResponse.SchemaItem toSchemaItem(TableSchema schema) { + return new ListSchemaResponse.SchemaItem( + schema.id(), schema.toSchema(), schema.timeMillis()); + } +} From 95d1b09617c17df90bf1b0406167d10016f4ec08 Mon Sep 17 00:00:00 2001 From: "zhangyongxiang.alpha" Date: Tue, 8 Sep 2026 21:01:24 +0800 Subject: [PATCH 3/3] [core][rest] Address schema API review comments --- .../java/org/apache/paimon/rest/RESTApi.java | 64 ++-- .../org/apache/paimon/rest/ResourcePaths.java | 4 + .../paimon/rest/responses/ErrorResponse.java | 2 + .../rest/responses/GetSchemaResponse.java | 47 +++ .../rest/responses/ListSchemaResponse.java | 95 ------ .../rest/responses/ListSchemasResponse.java | 69 ++++ .../apache/paimon/schema/SchemaFilter.java | 160 ---------- .../org/apache/paimon/catalog/Catalog.java | 51 +-- .../apache/paimon/catalog/CatalogUtils.java | 3 +- .../paimon/catalog/DelegateCatalog.java | 14 + .../org/apache/paimon/rest/RESTCatalog.java | 29 +- .../paimon/schema/CatalogSchemaManager.java | 296 ------------------ .../paimon/table/AbstractFileStoreTable.java | 5 - .../paimon/table/CatalogEnvironment.java | 33 +- .../paimon/rest/MockRESTCatalogTest.java | 39 +++ .../apache/paimon/rest/RESTCatalogServer.java | 55 ++-- .../RESTCatalogServerMetadataHandler.java | 156 +++++---- .../apache/paimon/rest/RESTCatalogTest.java | 119 ++----- 18 files changed, 372 insertions(+), 869 deletions(-) create mode 100644 paimon-api/src/main/java/org/apache/paimon/rest/responses/GetSchemaResponse.java delete mode 100644 paimon-api/src/main/java/org/apache/paimon/rest/responses/ListSchemaResponse.java create mode 100644 paimon-api/src/main/java/org/apache/paimon/rest/responses/ListSchemasResponse.java delete mode 100644 paimon-api/src/main/java/org/apache/paimon/schema/SchemaFilter.java delete mode 100644 paimon-core/src/main/java/org/apache/paimon/schema/CatalogSchemaManager.java 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 23037e4d3e20..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,7 +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.ListSchemaResponse; +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; @@ -104,7 +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.SchemaFilter; +import org.apache.paimon.schema.TableSchema; import org.apache.paimon.table.Instant; import org.apache.paimon.table.TableSnapshot; import org.apache.paimon.utils.JsonSerdeUtil; @@ -764,41 +765,32 @@ public void rollbackSchema(Identifier identifier, long schemaId) { restAuthFunction); } - /** - * List schemas of a table filtered by the given {@link SchemaFilter}. - * - *

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 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()); + /** 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 client.get( - resourcePaths.schemas(identifier.getDatabaseName(), identifier.getObjectName()), - queryParams, - ListSchemaResponse.class, - restAuthFunction); + return new PagedList<>(schemas, response.getNextPageToken()); } /** 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 c21024f232a3..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 @@ -235,6 +235,10 @@ public String schemas(String databaseName, String 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/ListSchemaResponse.java b/paimon-api/src/main/java/org/apache/paimon/rest/responses/ListSchemaResponse.java deleted file mode 100644 index ac4abe12a59b..000000000000 --- a/paimon-api/src/main/java/org/apache/paimon/rest/responses/ListSchemaResponse.java +++ /dev/null @@ -1,95 +0,0 @@ -/* - * 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 schemas; - - @JsonCreator - public ListSchemaResponse(@JsonProperty(FIELD_SCHEMAS) List schemas) { - this.schemas = schemas; - } - - @JsonGetter(FIELD_SCHEMAS) - public List 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; - } - } -} 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-api/src/main/java/org/apache/paimon/schema/SchemaFilter.java b/paimon-api/src/main/java/org/apache/paimon/schema/SchemaFilter.java deleted file mode 100644 index 9b9ae9085ee8..000000000000 --- a/paimon-api/src/main/java/org/apache/paimon/schema/SchemaFilter.java +++ /dev/null @@ -1,160 +0,0 @@ -/* - * 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. - * - *

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 - + '}'; - } -} 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 8bf125024d35..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,7 +31,6 @@ import org.apache.paimon.rest.responses.GetTagResponse; import org.apache.paimon.schema.Schema; import org.apache.paimon.schema.SchemaChange; -import org.apache.paimon.schema.SchemaFilter; import org.apache.paimon.schema.TableSchema; import org.apache.paimon.table.CatalogEnvironment; import org.apache.paimon.table.Instant; @@ -882,49 +881,33 @@ default void rollbackSchema(Identifier identifier, long schemaId) throw new UnsupportedOperationException(); } - // ==================== Schema management methods ========================== - /** - * Whether this catalog supports schema management for tables. If not, {@link - * #listSchemas(Identifier, SchemaFilter)} will throw an {@link UnsupportedOperationException}. + * Return the schema of a table for the given version. The version can be {@code EARLIEST}, + * {@code LATEST}, or a schema ID. * - *

This is orthogonal to {@link #supportsVersionManagement()}: version management covers - * snapshot / tag / branch APIs, while schema management covers reading historical {@link - * TableSchema}s of a table. A catalog may reasonably support one without the other. Write-side - * operations on schemas ({@link #createTable(Identifier, Schema, boolean)}, {@link - * #alterTable(Identifier, List, boolean)} and {@link #rollbackSchema(Identifier, long)}) are - * already exposed by the corresponding methods on this interface. + * @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 boolean supportsSchemaManagement() { - return false; + default Optional loadSchema(Identifier identifier, String version) + throws TableNotExistException { + throw new UnsupportedOperationException(); } /** - * List schemas of a table, filtered by the given {@link SchemaFilter}. - * - *

All schema read patterns (latest / earliest / by id / by range / all) share this single - * method; callers select the desired subset by populating {@link SchemaFilter}. Implementations - * must interpret the filter fields consistently: - * - *

- * - *

The returned list is not required to be sorted; callers that need a specific order should - * sort by {@link TableSchema#id()} themselves. + * Get a paged schema list of a table in descending schema ID order. * * @param identifier path of the table - * @param filter which schemas to return, must not be {@code null} + * @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 {@link - * #supportsSchemaManagement()} + * @throws UnsupportedOperationException if the catalog does not support listing schemas */ - default List listSchemas(Identifier identifier, SchemaFilter filter) + default PagedList listSchemasPaged( + Identifier identifier, @Nullable Integer maxResults, @Nullable String pageToken) throws TableNotExistException { throw new UnsupportedOperationException(); } diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java b/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java index c625e00c161f..916f63350c7b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java @@ -394,8 +394,7 @@ public static Table loadTable( isRestCatalog ? null : lockContext, catalogContext, catalog.supportsVersionManagement(), - catalog.supportsPartitionModification(), - catalog.supportsSchemaManagement()); + catalog.supportsPartitionModification()); Path path = new Path(schema.options().get(PATH.key())); FileStoreTable table = FileStoreTableFactory.create(dataFileIO.apply(path), path, schema, catalogEnv); 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 4d6e22684bfd..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 @@ -59,11 +59,9 @@ import org.apache.paimon.rest.responses.GetTableResponse; import org.apache.paimon.rest.responses.GetTagResponse; import org.apache.paimon.rest.responses.GetViewResponse; -import org.apache.paimon.rest.responses.ListSchemaResponse; import org.apache.paimon.schema.FileSystemSchemaManager; import org.apache.paimon.schema.Schema; import org.apache.paimon.schema.SchemaChange; -import org.apache.paimon.schema.SchemaFilter; import org.apache.paimon.schema.SchemaManager; import org.apache.paimon.schema.TableSchema; import org.apache.paimon.table.FormatTable; @@ -481,23 +479,26 @@ public boolean supportsVersionManagement() { } @Override - public boolean supportsSchemaManagement() { - return true; + 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 List listSchemas(Identifier identifier, SchemaFilter filter) + public PagedList listSchemasPaged( + Identifier identifier, @Nullable Integer maxResults, @Nullable String pageToken) throws TableNotExistException { try { - ListSchemaResponse response = api.listSchemas(identifier, filter); - if (response.getSchemas() == null || response.getSchemas().isEmpty()) { - return Collections.emptyList(); - } - List result = new ArrayList<>(response.getSchemas().size()); - for (ListSchemaResponse.SchemaItem item : response.getSchemas()) { - result.add(TableSchema.create(item.getSchemaId(), item.getSchema())); - } - return result; + return api.listSchemasPaged(identifier, maxResults, pageToken); } catch (NoSuchResourceException e) { throw new TableNotExistException(identifier); } catch (ForbiddenException e) { diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/CatalogSchemaManager.java b/paimon-core/src/main/java/org/apache/paimon/schema/CatalogSchemaManager.java deleted file mode 100644 index 2a71cbc01d47..000000000000 --- a/paimon-core/src/main/java/org/apache/paimon/schema/CatalogSchemaManager.java +++ /dev/null @@ -1,296 +0,0 @@ -/* - * 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 org.apache.paimon.catalog.Catalog; -import org.apache.paimon.catalog.CatalogLoader; -import org.apache.paimon.catalog.Identifier; -import org.apache.paimon.fs.Path; -import org.apache.paimon.table.SchemaModification; -import org.apache.paimon.types.RowType; -import org.apache.paimon.utils.ChangelogManager; -import org.apache.paimon.utils.FunctionWithException; -import org.apache.paimon.utils.SnapshotManager; -import org.apache.paimon.utils.TagManager; -import org.apache.paimon.utils.ThrowingConsumer; - -import javax.annotation.Nullable; -import javax.annotation.concurrent.ThreadSafe; - -import java.io.FileNotFoundException; -import java.io.IOException; -import java.util.Comparator; -import java.util.List; -import java.util.Optional; -import java.util.function.Predicate; -import java.util.stream.Collectors; - -/** - * A {@link SchemaManager} implementation that delegates every schema query and mutation to the - * owning {@link Catalog}. This mirrors the {@link org.apache.paimon.utils.CatalogBranchManager} - * pattern for the {@code BranchManager} interface: on-filesystem operations (reading raw {@code - * schema-*} files, deleting schemas, resolving schema paths) are intentionally not supported. - * - *

Read side collapses onto the single {@link Catalog#listSchemas(Identifier, SchemaFilter)} - * endpoint; different callers populate the filter differently (latest / earliest / by id / by range - * / all). Write side is routed through {@link Catalog#createTable(Identifier, Schema, boolean)}, - * {@link Catalog#alterTable(Identifier, List, boolean)} and {@link - * Catalog#rollbackSchema(Identifier, long)}. - */ -@ThreadSafe -public class CatalogSchemaManager implements SchemaManager { - - private static final long serialVersionUID = 1L; - - private final CatalogLoader catalogLoader; - private final Identifier identifier; - - public CatalogSchemaManager(CatalogLoader catalogLoader, Identifier identifier) { - this.catalogLoader = catalogLoader; - this.identifier = identifier; - } - - @Override - public SchemaManager copyWithBranch(String branchName) { - Identifier branchIdentifier = - new Identifier(identifier.getDatabaseName(), identifier.getTableName(), branchName); - return new CatalogSchemaManager(catalogLoader, branchIdentifier); - } - - @Override - public Optional latest() { - return executeGet( - catalog -> { - List schemas = - catalog.listSchemas(identifier, SchemaFilter.latest()); - if (schemas.isEmpty()) { - return Optional.empty(); - } - return Optional.of(schemas.get(0)); - }); - } - - @Override - public TableSchema latestOrThrow(String message) { - return latest().orElseThrow(() -> new RuntimeException(message)); - } - - @Override - public long earliestCreationTime() { - return executeGet( - catalog -> { - List schemas = - catalog.listSchemas(identifier, SchemaFilter.earliest()); - if (schemas.isEmpty()) { - throw new IllegalStateException("Table " + identifier + " has no schema."); - } - return schemas.get(0).timeMillis(); - }); - } - - @Override - public List listAll() { - return executeGet( - catalog -> { - List schemas = catalog.listSchemas(identifier, SchemaFilter.all()); - schemas.sort(Comparator.comparingLong(TableSchema::id)); - return schemas; - }); - } - - @Override - public List listAllIds() { - return listAll().stream().map(TableSchema::id).collect(Collectors.toList()); - } - - @Override - public TableSchema createTable(Schema schema) throws Exception { - return createTable(schema, false); - } - - @Override - public TableSchema createTable(Schema schema, boolean externalTable) throws Exception { - executePost(catalog -> catalog.createTable(identifier, schema, false)); - return latestOrThrow( - "Failed to load the newly created schema for table " + identifier + "."); - } - - @Override - public TableSchema commitChanges(SchemaChange... changes) throws Exception { - return commitChanges(java.util.Arrays.asList(changes)); - } - - @Override - public TableSchema commitChanges(List changes) - throws Catalog.TableNotExistException, Catalog.ColumnAlreadyExistException, - Catalog.ColumnNotExistException { - try (Catalog catalog = catalogLoader.load()) { - catalog.alterTable(identifier, changes, false); - } catch (Catalog.TableNotExistException - | Catalog.ColumnAlreadyExistException - | Catalog.ColumnNotExistException e) { - throw e; - } catch (RuntimeException e) { - throw e; - } catch (Exception e) { - throw new RuntimeException(e); - } - return latestOrThrow( - "Failed to load the latest schema after altering table " + identifier + "."); - } - - @Override - public boolean mergeSchema( - RowType rowType, - boolean typeWidening, - boolean allowExplicitCast, - boolean caseSensitive, - @Nullable SchemaModification schemaModification) { - TableSchema current = - latest().orElseThrow( - () -> - new RuntimeException( - "It requires that the current schema to exist when calling 'mergeSchema'")); - TableSchema update = - SchemaMergingUtils.mergeSchemas( - current, rowType, typeWidening, allowExplicitCast, caseSensitive); - if (current.equals(update)) { - return false; - } - List changes = - SchemaMergingUtils.diffSchemaChanges(current, update, caseSensitive); - try { - if (schemaModification != null) { - schemaModification.alterSchema(changes); - } else { - commitChanges(changes); - } - return true; - } catch (RuntimeException e) { - throw e; - } catch (Exception e) { - throw new RuntimeException("Failed to commit the schema.", e); - } - } - - @Override - public boolean commit(TableSchema newSchema) { - throw new UnsupportedOperationException( - "CatalogSchemaManager does not support committing a TableSchema directly; " - + "use commitChanges or the catalog createTable / alterTable APIs instead."); - } - - @Override - public TableSchema schema(long id) { - return executeGet( - catalog -> { - List schemas = - catalog.listSchemas(identifier, SchemaFilter.withId(id)); - if (schemas.isEmpty()) { - throw new IllegalStateException( - "Schema " + id + " not found for table " + identifier + "."); - } - return schemas.get(0); - }); - } - - @Override - public TableSchema tryGetSchema(long id) throws FileNotFoundException { - List schemas = - executeGet(catalog -> catalog.listSchemas(identifier, SchemaFilter.withId(id))); - if (schemas.isEmpty()) { - throw new FileNotFoundException( - "Schema " + id + " not found for table " + identifier + "."); - } - return schemas.get(0); - } - - @Override - public boolean schemaExists(long id) { - List schemas = - executeGet(catalog -> catalog.listSchemas(identifier, SchemaFilter.withId(id))); - return !schemas.isEmpty(); - } - - @Override - public Path schemaDirectory() { - throw new UnsupportedOperationException( - "CatalogSchemaManager does not expose a filesystem schema directory."); - } - - @Override - public Path toSchemaPath(long schemaId) { - throw new UnsupportedOperationException( - "CatalogSchemaManager does not expose a filesystem schema path."); - } - - @Override - public List schemaPaths(Predicate predicate) throws IOException { - throw new UnsupportedOperationException( - "CatalogSchemaManager does not expose filesystem schema paths."); - } - - @Override - public void deleteSchema(long schemaId) { - throw new UnsupportedOperationException( - "CatalogSchemaManager does not support deleting a single schema; " - + "use catalog.rollbackSchema instead."); - } - - @Override - public void rollbackTo( - long targetSchemaId, - SnapshotManager snapshotManager, - TagManager tagManager, - ChangelogManager changelogManager) { - executePost(catalog -> catalog.rollbackSchema(identifier, targetSchemaId)); - } - - private void executePost(ThrowingConsumer func) { - executeGet( - catalog -> { - try { - func.accept(catalog); - return null; - } catch (Catalog.TableNotExistException e) { - throw new IllegalArgumentException( - String.format( - "Table '%s' doesn't exist.", e.identifier().getFullName())); - } catch (Catalog.DatabaseNotExistException e) { - throw new IllegalArgumentException( - String.format("Database '%s' doesn't exist.", e.database())); - } catch (Catalog.TableAlreadyExistException e) { - throw new IllegalArgumentException( - String.format( - "Table '%s' already exists.", - e.identifier().getFullName())); - } - }); - } - - private T executeGet(FunctionWithException func) { - try (Catalog catalog = catalogLoader.load()) { - return func.apply(catalog); - } catch (RuntimeException e) { - throw e; - } catch (Exception e) { - throw new RuntimeException(e); - } - } -} diff --git a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java index a0d1b1df7a9b..2117f4c4dd8f 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java @@ -33,7 +33,6 @@ import org.apache.paimon.options.ExpireConfig; import org.apache.paimon.options.Options; import org.apache.paimon.predicate.Predicate; -import org.apache.paimon.schema.CatalogSchemaManager; import org.apache.paimon.schema.FileSystemSchemaManager; import org.apache.paimon.schema.SchemaManager; import org.apache.paimon.schema.SchemaValidation; @@ -424,10 +423,6 @@ public FileStoreTable copy(TableSchema newTableSchema) { @Override public SchemaManager schemaManager() { - if (catalogEnvironment.catalogLoader() != null - && catalogEnvironment.supportsSchemaManagement()) { - return new CatalogSchemaManager(catalogEnvironment.catalogLoader(), identifier()); - } return new FileSystemSchemaManager(fileIO(), path, currentBranch()); } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java b/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java index b963336cf081..66979ba811f2 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java @@ -63,7 +63,6 @@ public class CatalogEnvironment implements Serializable { @Nullable private final CatalogContext catalogContext; private final boolean supportsVersionManagement; private final boolean supportsPartitionModification; - private final boolean supportsSchemaManagement; public CatalogEnvironment( @Nullable Identifier identifier, @@ -74,28 +73,6 @@ public CatalogEnvironment( @Nullable CatalogContext catalogContext, boolean supportsVersionManagement, boolean supportsPartitionModification) { - this( - identifier, - uuid, - catalogLoader, - lockFactory, - lockContext, - catalogContext, - supportsVersionManagement, - supportsPartitionModification, - false); - } - - public CatalogEnvironment( - @Nullable Identifier identifier, - @Nullable String uuid, - @Nullable CatalogLoader catalogLoader, - @Nullable CatalogLockFactory lockFactory, - @Nullable CatalogLockContext lockContext, - @Nullable CatalogContext catalogContext, - boolean supportsVersionManagement, - boolean supportsPartitionModification, - boolean supportsSchemaManagement) { this.identifier = identifier; this.uuid = uuid; this.catalogLoader = catalogLoader; @@ -104,11 +81,10 @@ public CatalogEnvironment( this.catalogContext = catalogContext; this.supportsVersionManagement = supportsVersionManagement; this.supportsPartitionModification = supportsPartitionModification; - this.supportsSchemaManagement = supportsSchemaManagement; } public static CatalogEnvironment empty() { - return new CatalogEnvironment(null, null, null, null, null, null, false, false, false); + return new CatalogEnvironment(null, null, null, null, null, null, false, false); } @Nullable @@ -146,10 +122,6 @@ public boolean supportsVersionManagement() { return supportsVersionManagement; } - public boolean supportsSchemaManagement() { - return supportsSchemaManagement; - } - @Nullable public SchemaModification schemaModification() { if (catalogLoader == null) { @@ -281,8 +253,7 @@ public CatalogEnvironment copy(Identifier identifier) { lockContext, catalogContext, supportsVersionManagement, - supportsPartitionModification, - supportsSchemaManagement); + supportsPartitionModification); } public TableQueryAuth tableQueryAuth(CoreOptions options) { 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 616c97e64e2d..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; @@ -274,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) { @@ -346,6 +346,10 @@ public void setPartitionOptionsCreateSupported(boolean partitionOptionsCreateSup this.partitionOptionsCreateSupported = partitionOptionsCreateSupported; } + public void setSchemaEndpointsSupported(boolean schemaEndpointsSupported) { + this.schemaEndpointsSupported = schemaEndpointsSupported; + } + public void clearReceivedListPartitionsByFilterRequests() { receivedListPartitionsByFilterRequests.clear(); } @@ -533,6 +537,10 @@ && isTableByIdRequest(request.getPath())) { 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,8 +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) { @@ -1000,7 +1012,12 @@ private MockResponse listSnapshots(Identifier identifier) throws Exception { private MockResponse listSchemas(Identifier identifier, Map parameters) throws Exception { FileStoreTable table = getFileTable(identifier); - return RESTCatalogServerMetadataHandler.listSchemas(table, parameters); + 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 { @@ -1033,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 index ee025c07bc8d..3fb47f146a18 100644 --- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServerMetadataHandler.java +++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServerMetadataHandler.java @@ -19,22 +19,25 @@ package org.apache.paimon.rest; import org.apache.paimon.Snapshot; -import org.apache.paimon.rest.responses.ListSchemaResponse; +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.SchemaFilter; 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.Collections; import java.util.Comparator; import java.util.Iterator; import java.util.List; -import java.util.Map; +import java.util.Optional; import java.util.stream.Collectors; /** Metadata response handlers used by {@link RESTCatalogServer}. */ @@ -49,78 +52,97 @@ static MockResponse listSnapshots(FileStoreTable table) throws Exception { snapshotList.add(snapshots.next()); } ListSnapshotsResponse response = new ListSnapshotsResponse(snapshotList, null); - return new MockResponse().setResponseCode(200).setBody(RESTApi.toJson(response)); + return response(response); } - static MockResponse listSchemas(FileStoreTable table, Map parameters) - throws Exception { - SchemaManager schemaManager = new FileSystemSchemaManager(table.fileIO(), table.location()); - SchemaFilter filter = parseSchemaFilter(parameters); - List all = schemaManager.listAll(); - all.sort(Comparator.comparingLong(TableSchema::id).reversed()); - List items; - if (filter.isLatest()) { - items = - all.isEmpty() - ? Collections.emptyList() - : Collections.singletonList(toSchemaItem(all.get(0))); - } else if (filter.isEarliest()) { - items = - all.isEmpty() - ? Collections.emptyList() - : Collections.singletonList(toSchemaItem(all.get(all.size() - 1))); - } else if (filter.schemaId() != null) { - long target = filter.schemaId(); - items = - all.stream() - .filter(s -> s.id() == target) - .findFirst() - .map(s -> Collections.singletonList(toSchemaItem(s))) - .orElse(Collections.emptyList()); - } else { - items = - all.stream() - .filter( - s -> - filter.maxSchemaId() == null - || s.id() <= filter.maxSchemaId()) - .filter( - s -> - filter.minSchemaId() == null - || s.id() >= filter.minSchemaId()) - .map(RESTCatalogServerMetadataHandler::toSchemaItem) - .collect(Collectors.toList()); + 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) { } - ListSchemaResponse response = new ListSchemaResponse(items); - return new MockResponse().setResponseCode(200).setBody(RESTApi.toJson(response)); - } - private static SchemaFilter parseSchemaFilter(Map parameters) { - if (parameters == null || parameters.isEmpty()) { - return SchemaFilter.all(); + if (snapshot == null) { + return notFound(ErrorResponse.RESOURCE_TYPE_SNAPSHOT, "No Snapshot"); } - if ("true".equalsIgnoreCase(parameters.get("latest"))) { - return SchemaFilter.latest(); - } - if ("true".equalsIgnoreCase(parameters.get("earliest"))) { - return SchemaFilter.earliest(); + 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) { + } + } } - String schemaId = parameters.get("schemaId"); - if (schemaId != null) { - return SchemaFilter.withId(Long.parseLong(schemaId)); + + if (schema == null) { + return notFound(ErrorResponse.RESOURCE_TYPE_SCHEMA, "No Schema"); } - String maxSchemaId = parameters.get("maxSchemaId"); - String minSchemaId = parameters.get("minSchemaId"); - Long max = maxSchemaId == null ? null : Long.parseLong(maxSchemaId); - Long min = minSchemaId == null ? null : Long.parseLong(minSchemaId); - if (max == null && min == null) { - return SchemaFilter.all(); + 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()); } - return SchemaFilter.range(max, min); + + 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 ListSchemaResponse.SchemaItem toSchemaItem(TableSchema schema) { - return new ListSchemaResponse.SchemaItem( - schema.id(), schema.toSchema(), schema.timeMillis()); + 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 15c99babe5cf..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 @@ -80,7 +80,6 @@ import org.apache.paimon.schema.FileSystemSchemaManager; import org.apache.paimon.schema.Schema; import org.apache.paimon.schema.SchemaChange; -import org.apache.paimon.schema.SchemaFilter; import org.apache.paimon.schema.SchemaManager; import org.apache.paimon.schema.TableSchema; import org.apache.paimon.table.FileStoreTable; @@ -2473,122 +2472,54 @@ public void testRollbackSchemaFailedWithSnapshotReference() throws Exception { } @Test - public void testSupportsSchemaManagement() { - assertThat(catalog.supportsSchemaManagement()).isTrue(); - } - - @Test - public void testListSchemasAll() throws Exception { - Identifier identifier = Identifier.create("test_list_schemas", "table_all"); - createTable(identifier, Maps.newHashMap(), Lists.newArrayList("col1")); - - FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); - SchemaManager local = new FileSystemSchemaManager(table.fileIO(), table.location()); - long firstSchemaId = local.latest().get().id(); - - catalog.alterTable(identifier, SchemaChange.setOption("aa", "bb"), false); - catalog.alterTable(identifier, SchemaChange.setOption("cc", "dd"), false); - - List all = catalog.listSchemas(identifier, SchemaFilter.all()); - assertThat(all).hasSize(3); - all.sort(java.util.Comparator.comparingLong(TableSchema::id)); - assertThat(all.get(0).id()).isEqualTo(firstSchemaId); - assertThat(all.get(2).id()).isEqualTo(firstSchemaId + 2); - } - - @Test - public void testListSchemasLatestAndEarliest() throws Exception { - Identifier identifier = Identifier.create("test_list_schemas", "table_latest_earliest"); + 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()); - long firstSchemaId = local.latest().get().id(); - - catalog.alterTable(identifier, SchemaChange.setOption("aa", "bb"), false); - long secondSchemaId = local.latest().get().id(); - - List latest = catalog.listSchemas(identifier, SchemaFilter.latest()); - assertThat(latest).hasSize(1); - assertThat(latest.get(0).id()).isEqualTo(secondSchemaId); - - List earliest = catalog.listSchemas(identifier, SchemaFilter.earliest()); - assertThat(earliest).hasSize(1); - assertThat(earliest.get(0).id()).isEqualTo(firstSchemaId); - } + TableSchema first = local.latest().get(); - @Test - public void testListSchemasById() throws Exception { - Identifier identifier = Identifier.create("test_list_schemas", "table_by_id"); - createTable(identifier, Maps.newHashMap(), Lists.newArrayList("col1")); - - FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); - SchemaManager local = new FileSystemSchemaManager(table.fileIO(), table.location()); - long firstSchemaId = local.latest().get().id(); catalog.alterTable(identifier, SchemaChange.setOption("aa", "bb"), false); + TableSchema latest = local.latest().get(); - List byId = - catalog.listSchemas(identifier, SchemaFilter.withId(firstSchemaId)); - assertThat(byId).hasSize(1); - assertThat(byId.get(0).id()).isEqualTo(firstSchemaId); - - List missing = catalog.listSchemas(identifier, SchemaFilter.withId(9999L)); - assertThat(missing).isEmpty(); + 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 testListSchemasByRange() throws Exception { - Identifier identifier = Identifier.create("test_list_schemas", "table_by_range"); + 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()); - long firstSchemaId = local.latest().get().id(); 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 range = - catalog.listSchemas( - identifier, SchemaFilter.range(firstSchemaId + 2, firstSchemaId + 1)); - assertThat(range).hasSize(2); - range.sort(java.util.Comparator.comparingLong(TableSchema::id)); - assertThat(range.get(0).id()).isEqualTo(firstSchemaId + 1); - assertThat(range.get(1).id()).isEqualTo(firstSchemaId + 2); + 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 testListSchemasTableNotExist() { + public void testSchemaMethodsTableNotExist() { Identifier missing = Identifier.create("test_list_schemas", "missing_table"); - assertThatThrownBy(() -> catalog.listSchemas(missing, SchemaFilter.all())) + assertThatThrownBy(() -> catalog.loadSchema(missing, "LATEST")) + .isInstanceOf(Catalog.TableNotExistException.class); + assertThatThrownBy(() -> catalog.listSchemasPaged(missing, null, null)) .isInstanceOf(Catalog.TableNotExistException.class); - } - - @Test - public void testCatalogSchemaManagerBackedTableUsesRest() throws Exception { - Identifier identifier = Identifier.create("test_list_schemas", "table_catalog_backed"); - createTable(identifier, Maps.newHashMap(), Lists.newArrayList("col1")); - - FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); - // AbstractFileStoreTable should build a CatalogSchemaManager because - // RESTCatalog#supportsSchemaManagement is true. - assertThat(table.schemaManager().getClass().getSimpleName()) - .isEqualTo("CatalogSchemaManager"); - long firstSchemaId = table.schemaManager().latest().get().id(); - - catalog.alterTable(identifier, SchemaChange.setOption("aa", "bb"), false); - // The catalog-backed schema manager should observe the newest schema over REST. - assertThat(table.schemaManager().latest().get().id()).isEqualTo(firstSchemaId + 1); - - // A rollback via the catalog-backed manager must reach the REST server. - table.schemaManager() - .rollbackTo( - firstSchemaId, - table.snapshotManager(), - table.tagManager(), - new org.apache.paimon.utils.ChangelogManager( - table.fileIO(), table.location(), null)); - assertThat(table.schemaManager().latest().get().id()).isEqualTo(firstSchemaId); } @Test