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..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,7 +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.ListSnapshotsResponse; import org.apache.paimon.rest.responses.ListTableDetailsResponse; import org.apache.paimon.rest.responses.ListTablesGloballyResponse; import org.apache.paimon.rest.responses.ListTablesResponse; @@ -167,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; @@ -488,9 +486,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 +529,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 +692,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) { @@ -992,13 +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 { + FileStoreTable table = getFileTable(identifier); + 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()); + } +} 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();