You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "baiyangtx (via GitHub)" <gi...@apache.org> on 2026/08/11 12:23:24 UTC
[PR] refactor(core): make SchemaManager an interface [paimon]
baiyangtx opened a new pull request, #9174:
URL: https://github.com/apache/paimon/pull/9174
Move the existing file-system implementation to FileSystemSchemaManager and update construction sites to use it. Keep consumers typed against SchemaManager and preserve its static utility API.
### Purpose
### Tests
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(core): make SchemaManager an interface [paimon]
Posted by "baiyangtx (via GitHub)" <gi...@apache.org>.
baiyangtx commented on PR #9174:
URL: https://github.com/apache/paimon/pull/9174#issuecomment-5356064106
@JingsongLi Thanks for raising this concern.
Our concrete use case is that some tables need to access schemas from a remote service through REST APIs. We implemented an `HTTPSchemaManager` internally so that related tables can reuse remotely managed schemas.
`Catalog` and `RESTCatalog` already support pluggable snapshot and branch management, but there is no equivalent extension point for schema management. Making `SchemaManager` an interface enables this capability while retaining `FileSystemSchemaManager` as the default implementation.
I plan to add schema-related REST APIs to `RESTCatalog` in a separate PR to keep this change focused and reduce the review scope.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(core): make SchemaManager an interface [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #9174:
URL: https://github.com/apache/paimon/pull/9174#issuecomment-5524471078
+1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(core): make SchemaManager an interface [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #9174:
URL: https://github.com/apache/paimon/pull/9174
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(core): make SchemaManager an interface [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #9174:
URL: https://github.com/apache/paimon/pull/9174#issuecomment-5466121745
Please rebase master.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(core): make SchemaManager an interface [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #9174:
URL: https://github.com/apache/paimon/pull/9174#discussion_r3850458700
##########
paimon-core/src/main/java/org/apache/paimon/schema/FileSystemSchemaManager.java:
##########
@@ -0,0 +1,1554 @@
+/*
+ * 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.CoreOptions;
+import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.casting.CastExecutors;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.iceberg.IcebergOptions;
+import org.apache.paimon.schema.ColumnDirectiveUtils.ConvertedColumn;
+import org.apache.paimon.schema.SchemaChange.AddColumn;
+import org.apache.paimon.schema.SchemaChange.DropColumn;
+import org.apache.paimon.schema.SchemaChange.RemoveOption;
+import org.apache.paimon.schema.SchemaChange.RenameColumn;
+import org.apache.paimon.schema.SchemaChange.SetOption;
+import org.apache.paimon.schema.SchemaChange.UpdateColumnComment;
+import org.apache.paimon.schema.SchemaChange.UpdateColumnDefaultValue;
+import org.apache.paimon.schema.SchemaChange.UpdateColumnNullability;
+import org.apache.paimon.schema.SchemaChange.UpdateColumnPosition;
+import org.apache.paimon.schema.SchemaChange.UpdateColumnType;
+import org.apache.paimon.schema.SchemaChange.UpdateComment;
+import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.table.SchemaModification;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypeCasts;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.ReassignFieldId;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.BranchManager;
+import org.apache.paimon.utils.ChangelogManager;
+import org.apache.paimon.utils.LazyField;
+import org.apache.paimon.utils.Preconditions;
+import org.apache.paimon.utils.SnapshotManager;
+import org.apache.paimon.utils.StringUtils;
+import org.apache.paimon.utils.TagManager;
+
+import org.apache.paimon.shade.guava30.com.google.common.collect.FluentIterable;
+import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableList;
+import org.apache.paimon.shade.guava30.com.google.common.collect.Iterables;
+import org.apache.paimon.shade.guava30.com.google.common.collect.Maps;
+import org.apache.paimon.shade.guava30.com.google.common.collect.Streams;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.ThreadSafe;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.CoreOptions.AGG_FUNCTION;
+import static org.apache.paimon.CoreOptions.BUCKET_KEY;
+import static org.apache.paimon.CoreOptions.CLUSTERING_COLUMNS;
+import static org.apache.paimon.CoreOptions.DELETION_VECTORS_ENABLED;
+import static org.apache.paimon.CoreOptions.DELETION_VECTORS_MODIFIABLE;
+import static org.apache.paimon.CoreOptions.DISTINCT;
+import static org.apache.paimon.CoreOptions.FIELDS_PREFIX;
+import static org.apache.paimon.CoreOptions.IGNORE_DELETE;
+import static org.apache.paimon.CoreOptions.IGNORE_RETRACT;
+import static org.apache.paimon.CoreOptions.IGNORE_UPDATE_BEFORE;
+import static org.apache.paimon.CoreOptions.LIST_AGG_DELIMITER;
+import static org.apache.paimon.CoreOptions.MAP_SHARED_SHREDDING_COLUMN_PLACEMENT_POLICY;
+import static org.apache.paimon.CoreOptions.MAP_SHARED_SHREDDING_MAX_COLUMNS;
+import static org.apache.paimon.CoreOptions.MAP_STORAGE_LAYOUT;
+import static org.apache.paimon.CoreOptions.NESTED_KEY;
+import static org.apache.paimon.CoreOptions.PK_CLUSTERING_OVERRIDE;
+import static org.apache.paimon.CoreOptions.SEQUENCE_FIELD;
+import static org.apache.paimon.catalog.AbstractCatalog.DB_SUFFIX;
+import static org.apache.paimon.catalog.Identifier.DEFAULT_MAIN_BRANCH;
+import static org.apache.paimon.catalog.Identifier.UNKNOWN_DATABASE;
+import static org.apache.paimon.mergetree.compact.PartialUpdateMergeFunction.SEQUENCE_GROUP;
+import static org.apache.paimon.schema.ColumnDirectiveUtils.applyAddColumnDirective;
+import static org.apache.paimon.schema.ColumnDirectiveUtils.applyDirectives;
+import static org.apache.paimon.schema.ColumnDirectiveUtils.parseAddColumnComment;
+import static org.apache.paimon.types.BlobType.isBlobFileField;
+import static org.apache.paimon.utils.DefaultValueUtils.validateDefaultValue;
+import static org.apache.paimon.utils.FileUtils.listVersionedFiles;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.paimon.utils.Preconditions.checkState;
+
+/** File system implementation of {@link SchemaManager}. */
+@ThreadSafe
+public class FileSystemSchemaManager implements SchemaManager {
+
+ public static final String SCHEMA_PREFIX = "schema-";
+
+ private final FileIO fileIO;
+ private final Path tableRoot;
+
+ private final String branch;
+
+ public FileSystemSchemaManager(FileIO fileIO, Path tableRoot) {
+ this(fileIO, tableRoot, DEFAULT_MAIN_BRANCH);
+ }
+
+ /** Specify the default branch for data writing. */
+ public FileSystemSchemaManager(FileIO fileIO, Path tableRoot, String branch) {
+ this.fileIO = fileIO;
+ this.tableRoot = tableRoot;
+ this.branch = BranchManager.normalizeBranch(branch);
+ }
+
+ public FileSystemSchemaManager copyWithBranch(String branchName) {
+ return new FileSystemSchemaManager(fileIO, tableRoot, branchName);
+ }
+
+ public Optional<TableSchema> latest() {
+ try {
+ return listVersionedFiles(fileIO, schemaDirectory(), SCHEMA_PREFIX)
+ .reduce(Math::max)
+ .map(this::schema);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ public TableSchema latestOrThrow(String message) {
+ return latest().orElseThrow(() -> new RuntimeException(message));
+ }
+
+ public long earliestCreationTime() {
+ try {
+ long earliest = 0;
+ if (!schemaExists(0)) {
+ Optional<Long> min =
+ listVersionedFiles(fileIO, schemaDirectory(), SCHEMA_PREFIX)
+ .reduce(Math::min);
+ checkArgument(min.isPresent());
+ earliest = min.get();
+ }
+
+ Path schemaPath = toSchemaPath(earliest);
+ return fileIO.getFileStatus(schemaPath).getModificationTime();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ public List<TableSchema> listAll() {
+ return listAllIds().stream().map(this::schema).collect(Collectors.toList());
+ }
+
+ /** List all schema IDs. */
+ public List<Long> listAllIds() {
+ try {
+ return listVersionedFiles(fileIO, schemaDirectory(), SCHEMA_PREFIX)
+ .collect(Collectors.toList());
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ public TableSchema createTable(Schema schema) throws Exception {
+ return createTable(schema, false);
+ }
+
+ public TableSchema createTable(Schema schema, boolean externalTable) throws Exception {
+ while (true) {
+ Optional<TableSchema> latest = latest();
+ if (latest.isPresent()) {
+ TableSchema latestSchema = latest.get();
+ if (externalTable) {
+ checkSchemaForExternalTable(latestSchema.toSchema(), schema);
+ return latestSchema;
+ } else {
+ throw new IllegalStateException(
+ "Schema in filesystem exists, creation is not allowed.");
+ }
+ }
+
+ schema = applyDirectives(schema);
+ TableSchema newSchema = TableSchema.create(0, schema);
+
+ // validate table from creating table
+ FileStoreTableFactory.create(fileIO, tableRoot, newSchema).store();
+
+ boolean success = commit(newSchema);
+ if (success) {
+ return newSchema;
+ }
+ }
+ }
+
+ private void checkSchemaForExternalTable(Schema existsSchema, Schema newSchema) {
+ // When creating an external table, if the table already exists in the location, we can
+ // choose not to specify the fields.
+ if ((newSchema.fields().isEmpty()
+ || newSchema.rowType().equalsIgnoreFieldId(existsSchema.rowType()))
+ && (newSchema.partitionKeys().isEmpty()
+ || Objects.equals(newSchema.partitionKeys(), existsSchema.partitionKeys()))
+ && (newSchema.primaryKeys().isEmpty()
+ || Objects.equals(newSchema.primaryKeys(), existsSchema.primaryKeys()))) {
+ // check for options
+ Map<String, String> existsOptions = existsSchema.options();
+ Map<String, String> newOptions = newSchema.options();
+ newOptions.forEach(
+ (key, value) -> {
+ // ignore `owner` and `path`
+ if (!key.equals(Catalog.OWNER_PROP)
+ && !key.equals(CoreOptions.PATH.key())
+ && (!existsOptions.containsKey(key)
+ || !existsOptions.get(key).equals(value))) {
+ throw new RuntimeException(
+ "New schema's options are not equal to the exists schema's, new schema: "
+ + newOptions
+ + ", exists schema: "
+ + existsOptions);
+ }
+ });
+ } else {
+ throw new RuntimeException(
+ "New schema is not equal to exists schema, new schema: "
+ + newSchema
+ + ", exists schema: "
+ + existsSchema);
+ }
+ }
+
+ /** Update {@link SchemaChange}s. */
+ public TableSchema commitChanges(SchemaChange... changes) throws Exception {
+ return commitChanges(Arrays.asList(changes));
+ }
+
+ /** Update {@link SchemaChange}s. */
+ public TableSchema commitChanges(List<SchemaChange> changes)
+ throws Catalog.TableNotExistException, Catalog.ColumnAlreadyExistException,
+ Catalog.ColumnNotExistException {
+ SnapshotManager snapshotManager =
+ new SnapshotManager(fileIO, tableRoot, branch, null, null);
+ LazyField<Boolean> hasSnapshots =
+ new LazyField<>(() -> snapshotManager.latestSnapshot() != null);
+
+ while (true) {
+ TableSchema oldTableSchema =
+ latest().orElseThrow(
+ () ->
+ new Catalog.TableNotExistException(
+ identifierFromPath(
+ tableRoot.toString(), true, branch)));
+ LazyField<Identifier> lazyIdentifier =
+ new LazyField<>(() -> identifierFromPath(tableRoot.toString(), true, branch));
+ TableSchema newTableSchema =
+ generateTableSchema(oldTableSchema, changes, hasSnapshots, lazyIdentifier);
+ try {
+ boolean success = commit(newTableSchema);
+ if (success) {
+ return newTableSchema;
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ public static TableSchema generateTableSchema(
Review Comment:
This long list doesn't seem related to the FileSystem, does it? It feels like it could be moved into the SchemaManager.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(core): make SchemaManager an interface [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #9174:
URL: https://github.com/apache/paimon/pull/9174#issuecomment-5350584744
Are there any major benefits? Since this change would affect numerous upstream and downstream API implementations—entailing a significant cost—it wasn't made previously.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(core): make SchemaManager an interface [paimon]
Posted by "baiyangtx (via GitHub)" <gi...@apache.org>.
baiyangtx commented on code in PR #9174:
URL: https://github.com/apache/paimon/pull/9174#discussion_r3871526876
##########
paimon-core/src/main/java/org/apache/paimon/schema/FileSystemSchemaManager.java:
##########
@@ -0,0 +1,1554 @@
+/*
+ * 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.CoreOptions;
+import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.casting.CastExecutors;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.iceberg.IcebergOptions;
+import org.apache.paimon.schema.ColumnDirectiveUtils.ConvertedColumn;
+import org.apache.paimon.schema.SchemaChange.AddColumn;
+import org.apache.paimon.schema.SchemaChange.DropColumn;
+import org.apache.paimon.schema.SchemaChange.RemoveOption;
+import org.apache.paimon.schema.SchemaChange.RenameColumn;
+import org.apache.paimon.schema.SchemaChange.SetOption;
+import org.apache.paimon.schema.SchemaChange.UpdateColumnComment;
+import org.apache.paimon.schema.SchemaChange.UpdateColumnDefaultValue;
+import org.apache.paimon.schema.SchemaChange.UpdateColumnNullability;
+import org.apache.paimon.schema.SchemaChange.UpdateColumnPosition;
+import org.apache.paimon.schema.SchemaChange.UpdateColumnType;
+import org.apache.paimon.schema.SchemaChange.UpdateComment;
+import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.table.SchemaModification;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypeCasts;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.ReassignFieldId;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.BranchManager;
+import org.apache.paimon.utils.ChangelogManager;
+import org.apache.paimon.utils.LazyField;
+import org.apache.paimon.utils.Preconditions;
+import org.apache.paimon.utils.SnapshotManager;
+import org.apache.paimon.utils.StringUtils;
+import org.apache.paimon.utils.TagManager;
+
+import org.apache.paimon.shade.guava30.com.google.common.collect.FluentIterable;
+import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableList;
+import org.apache.paimon.shade.guava30.com.google.common.collect.Iterables;
+import org.apache.paimon.shade.guava30.com.google.common.collect.Maps;
+import org.apache.paimon.shade.guava30.com.google.common.collect.Streams;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.ThreadSafe;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.CoreOptions.AGG_FUNCTION;
+import static org.apache.paimon.CoreOptions.BUCKET_KEY;
+import static org.apache.paimon.CoreOptions.CLUSTERING_COLUMNS;
+import static org.apache.paimon.CoreOptions.DELETION_VECTORS_ENABLED;
+import static org.apache.paimon.CoreOptions.DELETION_VECTORS_MODIFIABLE;
+import static org.apache.paimon.CoreOptions.DISTINCT;
+import static org.apache.paimon.CoreOptions.FIELDS_PREFIX;
+import static org.apache.paimon.CoreOptions.IGNORE_DELETE;
+import static org.apache.paimon.CoreOptions.IGNORE_RETRACT;
+import static org.apache.paimon.CoreOptions.IGNORE_UPDATE_BEFORE;
+import static org.apache.paimon.CoreOptions.LIST_AGG_DELIMITER;
+import static org.apache.paimon.CoreOptions.MAP_SHARED_SHREDDING_COLUMN_PLACEMENT_POLICY;
+import static org.apache.paimon.CoreOptions.MAP_SHARED_SHREDDING_MAX_COLUMNS;
+import static org.apache.paimon.CoreOptions.MAP_STORAGE_LAYOUT;
+import static org.apache.paimon.CoreOptions.NESTED_KEY;
+import static org.apache.paimon.CoreOptions.PK_CLUSTERING_OVERRIDE;
+import static org.apache.paimon.CoreOptions.SEQUENCE_FIELD;
+import static org.apache.paimon.catalog.AbstractCatalog.DB_SUFFIX;
+import static org.apache.paimon.catalog.Identifier.DEFAULT_MAIN_BRANCH;
+import static org.apache.paimon.catalog.Identifier.UNKNOWN_DATABASE;
+import static org.apache.paimon.mergetree.compact.PartialUpdateMergeFunction.SEQUENCE_GROUP;
+import static org.apache.paimon.schema.ColumnDirectiveUtils.applyAddColumnDirective;
+import static org.apache.paimon.schema.ColumnDirectiveUtils.applyDirectives;
+import static org.apache.paimon.schema.ColumnDirectiveUtils.parseAddColumnComment;
+import static org.apache.paimon.types.BlobType.isBlobFileField;
+import static org.apache.paimon.utils.DefaultValueUtils.validateDefaultValue;
+import static org.apache.paimon.utils.FileUtils.listVersionedFiles;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.paimon.utils.Preconditions.checkState;
+
+/** File system implementation of {@link SchemaManager}. */
+@ThreadSafe
+public class FileSystemSchemaManager implements SchemaManager {
+
+ public static final String SCHEMA_PREFIX = "schema-";
+
+ private final FileIO fileIO;
+ private final Path tableRoot;
+
+ private final String branch;
+
+ public FileSystemSchemaManager(FileIO fileIO, Path tableRoot) {
+ this(fileIO, tableRoot, DEFAULT_MAIN_BRANCH);
+ }
+
+ /** Specify the default branch for data writing. */
+ public FileSystemSchemaManager(FileIO fileIO, Path tableRoot, String branch) {
+ this.fileIO = fileIO;
+ this.tableRoot = tableRoot;
+ this.branch = BranchManager.normalizeBranch(branch);
+ }
+
+ public FileSystemSchemaManager copyWithBranch(String branchName) {
+ return new FileSystemSchemaManager(fileIO, tableRoot, branchName);
+ }
+
+ public Optional<TableSchema> latest() {
+ try {
+ return listVersionedFiles(fileIO, schemaDirectory(), SCHEMA_PREFIX)
+ .reduce(Math::max)
+ .map(this::schema);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ public TableSchema latestOrThrow(String message) {
+ return latest().orElseThrow(() -> new RuntimeException(message));
+ }
+
+ public long earliestCreationTime() {
+ try {
+ long earliest = 0;
+ if (!schemaExists(0)) {
+ Optional<Long> min =
+ listVersionedFiles(fileIO, schemaDirectory(), SCHEMA_PREFIX)
+ .reduce(Math::min);
+ checkArgument(min.isPresent());
+ earliest = min.get();
+ }
+
+ Path schemaPath = toSchemaPath(earliest);
+ return fileIO.getFileStatus(schemaPath).getModificationTime();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ public List<TableSchema> listAll() {
+ return listAllIds().stream().map(this::schema).collect(Collectors.toList());
+ }
+
+ /** List all schema IDs. */
+ public List<Long> listAllIds() {
+ try {
+ return listVersionedFiles(fileIO, schemaDirectory(), SCHEMA_PREFIX)
+ .collect(Collectors.toList());
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ public TableSchema createTable(Schema schema) throws Exception {
+ return createTable(schema, false);
+ }
+
+ public TableSchema createTable(Schema schema, boolean externalTable) throws Exception {
+ while (true) {
+ Optional<TableSchema> latest = latest();
+ if (latest.isPresent()) {
+ TableSchema latestSchema = latest.get();
+ if (externalTable) {
+ checkSchemaForExternalTable(latestSchema.toSchema(), schema);
+ return latestSchema;
+ } else {
+ throw new IllegalStateException(
+ "Schema in filesystem exists, creation is not allowed.");
+ }
+ }
+
+ schema = applyDirectives(schema);
+ TableSchema newSchema = TableSchema.create(0, schema);
+
+ // validate table from creating table
+ FileStoreTableFactory.create(fileIO, tableRoot, newSchema).store();
+
+ boolean success = commit(newSchema);
+ if (success) {
+ return newSchema;
+ }
+ }
+ }
+
+ private void checkSchemaForExternalTable(Schema existsSchema, Schema newSchema) {
+ // When creating an external table, if the table already exists in the location, we can
+ // choose not to specify the fields.
+ if ((newSchema.fields().isEmpty()
+ || newSchema.rowType().equalsIgnoreFieldId(existsSchema.rowType()))
+ && (newSchema.partitionKeys().isEmpty()
+ || Objects.equals(newSchema.partitionKeys(), existsSchema.partitionKeys()))
+ && (newSchema.primaryKeys().isEmpty()
+ || Objects.equals(newSchema.primaryKeys(), existsSchema.primaryKeys()))) {
+ // check for options
+ Map<String, String> existsOptions = existsSchema.options();
+ Map<String, String> newOptions = newSchema.options();
+ newOptions.forEach(
+ (key, value) -> {
+ // ignore `owner` and `path`
+ if (!key.equals(Catalog.OWNER_PROP)
+ && !key.equals(CoreOptions.PATH.key())
+ && (!existsOptions.containsKey(key)
+ || !existsOptions.get(key).equals(value))) {
+ throw new RuntimeException(
+ "New schema's options are not equal to the exists schema's, new schema: "
+ + newOptions
+ + ", exists schema: "
+ + existsOptions);
+ }
+ });
+ } else {
+ throw new RuntimeException(
+ "New schema is not equal to exists schema, new schema: "
+ + newSchema
+ + ", exists schema: "
+ + existsSchema);
+ }
+ }
+
+ /** Update {@link SchemaChange}s. */
+ public TableSchema commitChanges(SchemaChange... changes) throws Exception {
+ return commitChanges(Arrays.asList(changes));
+ }
+
+ /** Update {@link SchemaChange}s. */
+ public TableSchema commitChanges(List<SchemaChange> changes)
+ throws Catalog.TableNotExistException, Catalog.ColumnAlreadyExistException,
+ Catalog.ColumnNotExistException {
+ SnapshotManager snapshotManager =
+ new SnapshotManager(fileIO, tableRoot, branch, null, null);
+ LazyField<Boolean> hasSnapshots =
+ new LazyField<>(() -> snapshotManager.latestSnapshot() != null);
+
+ while (true) {
+ TableSchema oldTableSchema =
+ latest().orElseThrow(
+ () ->
+ new Catalog.TableNotExistException(
+ identifierFromPath(
+ tableRoot.toString(), true, branch)));
+ LazyField<Identifier> lazyIdentifier =
+ new LazyField<>(() -> identifierFromPath(tableRoot.toString(), true, branch));
+ TableSchema newTableSchema =
+ generateTableSchema(oldTableSchema, changes, hasSnapshots, lazyIdentifier);
+ try {
+ boolean success = commit(newTableSchema);
+ if (success) {
+ return newTableSchema;
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ public static TableSchema generateTableSchema(
Review Comment:
Fixed. I moved the static methods to the SchemaManager interface while keeping the existing call sites compatible. Thanks for pointing this out!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(core): make SchemaManager an interface [paimon]
Posted by "baiyangtx (via GitHub)" <gi...@apache.org>.
baiyangtx commented on PR #9174:
URL: https://github.com/apache/paimon/pull/9174#issuecomment-5290494611
@Aitozi @JingsongLi PTAL
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org