Skip to content

[core] Decouple catalog alter and drop from runtime table loading #10830

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JingsongLi merged 1 commit into from
Oct 27, 2026
Merged

[core] Decouple catalog alter and drop from runtime table loading #10830

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 14 additions & 13 deletions paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@
import org.apache.paimon.table.system.SystemTableLoader;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.FileSystemBranchManager;
import org.apache.paimon.utils.SnapshotNotExistException;

import org.slf4j.Logger;
Expand Down Expand Up @@ -383,18 +384,17 @@ public void dropTable(Identifier identifier, boolean ignoreIfNotExists)
}

Set<Path> externalPaths = new HashSet<>();
if (tableExistsInFileSystem(getTableLocation(identifier), DEFAULT_MAIN_BRANCH)) {
Table table = getTable(identifier);
if (table instanceof FileStoreTable) {
FileStoreTable fileStoreTable = (FileStoreTable) table;
List<Path> schemaExternalPaths =
getSchemaExternalPaths(fileStoreTable.schemaManager().listAll());
externalPaths.addAll(schemaExternalPaths);
List<String> branches = fileStoreTable.branchManager().branches();
for (String branch : branches) {
SchemaManager schemaManager =
fileStoreTable.schemaManager().copyWithBranch(branch);
externalPaths.addAll(getSchemaExternalPaths(schemaManager.listAll()));
Path tablePath = getTableLocation(identifier);
if (tableExistsInFileSystem(tablePath, DEFAULT_MAIN_BRANCH)) {
TableType tableType = CoreOptions.fromMap(loadTableSchema(identifier).options()).type();
if (tableType == TableType.TABLE || tableType == TableType.MATERIALIZED_TABLE) {
// Cleanup only needs persisted metadata, even if runtime options are invalid.
FileIO tableFileIO = fileIO(tablePath);
SchemaManager schemaManager = new FileSystemSchemaManager(tableFileIO, tablePath);
externalPaths.addAll(getSchemaExternalPaths(schemaManager.listAll()));
for (String branch : FileSystemBranchManager.branches(tableFileIO, tablePath)) {
externalPaths.addAll(
getSchemaExternalPaths(schemaManager.copyWithBranch(branch).listAll()));
}
}
}
Expand Down Expand Up @@ -503,7 +503,8 @@ public void alterTable(
checkNotSystemTable(identifier, "alterTable");

try {
getTable(identifier);
// Validate the resulting schema, rather than the old runtime options.
loadTableSchema(identifier);
} catch (TableNotExistException e) {
if (ignoreIfNotExists) {
return;
Expand Down
12 changes: 6 additions & 6 deletions paimon-core/src/main/java/org/apache/paimon/utils/FileSystemBranchManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,11 +72,6 @@ public FileSystemBranchManager(
this.mergeHandler = mergeHandler;
}

/** Return the root Directory of branch. */
private Path branchDirectory() {
return new Path(tablePath + "/branch");
}

/** Return the path of a branch. */
public Path branchPath(String branchName) {
return new Path(BranchManager.branchPath(tablePath, branchName));
Expand Down Expand Up @@ -458,8 +453,13 @@ public void renameBranch(String fromBranch, String toBranch) {

@Override
public List<String> branches() {
return branches(fileIO, tablePath);
}

/** List branches without constructing the table's runtime managers. */
public static List<String> branches(FileIO fileIO, Path tablePath) {
try {
return listVersionedDirectories(fileIO, branchDirectory(), BRANCH_PREFIX)
return listVersionedDirectories(fileIO, new Path(tablePath, "branch"), BRANCH_PREFIX)
.map(status -> status.getPath().getName().substring(BRANCH_PREFIX.length()))
.collect(Collectors.toList());
} catch (IOException e) {
Expand Down
290 changes: 290 additions & 0 deletions paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,19 +20,39 @@

import org.apache.paimon.CoreOptions;
import org.apache.paimon.TableType;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.fs.Path;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.FileSystemSchemaManager;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.BatchTableCommit;
import org.apache.paimon.table.sink.BatchTableWrite;
import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.utils.SnapshotManager;

import org.apache.paimon.shade.guava30.com.google.common.collect.Lists;

import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;
import org.junit.jupiter.params.provider.ValueSource;

import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.assertj.core.api.Assertions.tuple;

/** Tests for {@link FileSystemCatalog}. */
public class FileSystemCatalogTest extends CatalogTestBase {
Expand Down Expand Up @@ -151,4 +171,274 @@ public void testPartitionsFromCatalogAreRejectedOutsideRestCatalog() throws Exce
.hasMessageContaining(CoreOptions.METASTORE_PARTITIONED_TABLE.key())
.hasMessageContaining("REST catalog");
}

@ParameterizedTest
@ValueSource(booleans = {false, true})
public void testAlterTableWithInvalidOptions(boolean cacheEnabled) throws Exception {
Identifier identifier = Identifier.create("test_db", "legacy_table");
FileSystemSchemaManager schemaManager = createLegacyTable(identifier, cacheEnabled);

assertThatThrownBy(() -> catalog.getTable(identifier))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("primary key");

// Fixing just one of the two invalid options must not persist an invalid schema.
assertThatThrownBy(
() ->
catalog.alterTable(
identifier,
SchemaChange.removeOption("full-compaction.delta-commits"),
false))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("primary key");
assertThat(schemaManager.listAllIds()).containsExactly(0L);

catalog.alterTable(
identifier,
Arrays.asList(
SchemaChange.removeOption("full-compaction.delta-commits"),
SchemaChange.setOption("fields.seq.sequence-group", "v")),
false);

assertThat(schemaManager.latest().get().id()).isEqualTo(1L);
assertThat(catalog.getTable(identifier).options())
.doesNotContainKey("full-compaction.delta-commits")
.containsEntry("fields.seq.sequence-group", "v");

assertThatThrownBy(
() ->
catalog.alterTable(
identifier,
SchemaChange.setOption("fields.seq.sequence-group", "id,v"),
false))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("primary key");
assertThat(schemaManager.latest().get().id()).isEqualTo(1L);
}

@ParameterizedTest
@ValueSource(booleans = {false, true})
public void testAlterTableWithInvalidOptionsAndSnapshots(boolean cacheEnabled)
throws Exception {
Identifier identifier = Identifier.create("test_db", "legacy_table");
FileSystemSchemaManager schemaManager = createLegacyTable(identifier, cacheEnabled, true);
Path tablePath =
((FileSystemCatalog) DelegateCatalog.rootCatalog(catalog))
.getTableLocation(identifier);
SnapshotManager snapshotManager =
new SnapshotManager(fileIO, tablePath, Identifier.DEFAULT_MAIN_BRANCH, null, null);
assertThat(snapshotManager.latestSnapshot()).isNotNull();
long snapshotId = snapshotManager.latestSnapshot().id();
assertThat(snapshotManager.latestSnapshot().schemaId()).isZero();
assertThatThrownBy(() -> catalog.getTable(identifier))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("primary key");

List<SchemaChange> repairChanges =
Arrays.asList(
SchemaChange.removeOption("full-compaction.delta-commits"),
SchemaChange.setOption("fields.seq.sequence-group", "v"));
List<SchemaChange> immutableChanges = new ArrayList<>(repairChanges);
immutableChanges.add(SchemaChange.setOption("merge-engine", "deduplicate"));
assertThatThrownBy(() -> catalog.alterTable(identifier, immutableChanges, false))
.isInstanceOf(UnsupportedOperationException.class)
.hasMessage("Change 'merge-engine' is not supported yet.");
assertThat(schemaManager.listAllIds()).containsExactly(0L);
assertThat(schemaManager.latest().get().options())
.containsEntry("merge-engine", "partial-update")
.containsEntry("full-compaction.delta-commits", "1")
.containsEntry("fields.seq.sequence-group", "id,v");

catalog.alterTable(identifier, repairChanges, false);

FileStoreTable repairedTable = (FileStoreTable) catalog.getTable(identifier);
assertThat(repairedTable.schema().id()).isEqualTo(1L);
assertThat(repairedTable.options())
.containsEntry("merge-engine", "partial-update")
.containsEntry("fields.seq.sequence-group", "v")
.doesNotContainKey("full-compaction.delta-commits");
assertThat(snapshotManager.latestSnapshotId()).isEqualTo(snapshotId);
assertThat(snapshotManager.latestSnapshot().schemaId()).isZero();
assertThat(read(repairedTable, null, null, null, null))
.extracting(row -> row.getInt(0), row -> row.getInt(1), row -> row.getInt(2))
.containsExactlyInAnyOrder(tuple(1, 10, 100), tuple(2, 20, 200));
}

@ParameterizedTest
@ValueSource(booleans = {false, true})
public void testAlterTableWithInvalidOptionsOnBranch(boolean cacheEnabled) throws Exception {
Identifier identifier = Identifier.create("test_db", "legacy_table");
FileSystemSchemaManager mainSchemaManager = createLegacyTable(identifier, cacheEnabled);
TableSchema legacySchema = mainSchemaManager.latest().get();
FileSystemSchemaManager branchSchemaManager = mainSchemaManager.copyWithBranch("dev");
writeLegacySchema(branchSchemaManager, legacySchema, 0, legacySchema.options());
Identifier branchIdentifier = new Identifier("test_db", "legacy_table", "dev");

catalog.alterTable(
branchIdentifier,
Arrays.asList(
SchemaChange.removeOption("full-compaction.delta-commits"),
SchemaChange.setOption("fields.seq.sequence-group", "v")),
false);

assertThat(branchSchemaManager.latest().get().id()).isEqualTo(1L);
assertThat(mainSchemaManager.latest().get().id()).isZero();
assertThat(catalog.getTable(branchIdentifier).options())
.containsEntry("fields.seq.sequence-group", "v");
assertThatThrownBy(() -> catalog.getTable(identifier))
.isInstanceOf(IllegalArgumentException.class);
}

@ParameterizedTest
@ValueSource(booleans = {false, true})
public void testDropTableWithInvalidOptions(boolean cacheEnabled) throws Exception {
Identifier identifier = Identifier.create("test_db", "legacy_table");
createLegacyTable(identifier, cacheEnabled);
Path tablePath =
((FileSystemCatalog) DelegateCatalog.rootCatalog(catalog))
.getTableLocation(identifier);

assertThatThrownBy(() -> catalog.getTable(identifier))
.isInstanceOf(IllegalArgumentException.class);
catalog.dropTable(identifier, false);

assertThat(fileIO.exists(tablePath)).isFalse();
assertThatThrownBy(() -> catalog.dropTable(identifier, false))
.isInstanceOf(Catalog.TableNotExistException.class);
catalog.dropTable(identifier, true);
}

@ParameterizedTest
@ValueSource(booleans = {false, true})
public void testDropTableWithInvalidOptionsAndExternalPaths(boolean cacheEnabled)
throws Exception {
Identifier identifier = Identifier.create("test_db", "legacy_table");
FileSystemSchemaManager schemaManager = createLegacyTable(identifier, cacheEnabled);
TableSchema legacySchema = schemaManager.latest().get();
java.nio.file.Path oldMainPath = Files.createDirectory(tempFile.resolve("old-main-data"));
java.nio.file.Path newMainPath = Files.createDirectory(tempFile.resolve("new-main-data"));
java.nio.file.Path oldBranchPath =
Files.createDirectory(tempFile.resolve("old-branch-index"));
java.nio.file.Path newBranchPath =
Files.createDirectory(tempFile.resolve("new-branch-index"));
java.nio.file.Path unrelatedPath = Files.createDirectory(tempFile.resolve("unrelated"));

Map<String, String> mainOptions = new HashMap<>(legacySchema.options());
mainOptions.put(CoreOptions.DATA_FILE_EXTERNAL_PATHS.key(), oldMainPath.toUri().toString());
writeLegacySchema(schemaManager, legacySchema, 0, mainOptions);
mainOptions.put(CoreOptions.DATA_FILE_EXTERNAL_PATHS.key(), newMainPath.toUri().toString());
writeLegacySchema(schemaManager, legacySchema, 1, mainOptions);

FileSystemSchemaManager branchSchemaManager = schemaManager.copyWithBranch("dev");
Map<String, String> branchOptions = new HashMap<>(legacySchema.options());
branchOptions.put(
CoreOptions.GLOBAL_INDEX_EXTERNAL_PATH.key(), oldBranchPath.toUri().toString());
writeLegacySchema(branchSchemaManager, legacySchema, 0, branchOptions);
branchOptions.put(
CoreOptions.GLOBAL_INDEX_EXTERNAL_PATH.key(), newBranchPath.toUri().toString());
writeLegacySchema(branchSchemaManager, legacySchema, 1, branchOptions);

assertThatThrownBy(() -> catalog.getTable(identifier))
.isInstanceOf(IllegalArgumentException.class);
catalog.dropTable(identifier, false);

for (java.nio.file.Path externalPath :
Arrays.asList(oldMainPath, newMainPath, oldBranchPath, newBranchPath)) {
assertThat(externalPath).doesNotExist();
}
assertThat(unrelatedPath).exists();
}

@ParameterizedTest
@EnumSource(
value = TableType.class,
names = {"FORMAT_TABLE", "OBJECT_TABLE"})
public void testDropNonDataTablePreservesExternalPaths(TableType tableType) throws Exception {
Identifier identifier = Identifier.create("test_db", "non_data_table");
java.nio.file.Path externalPath = Files.createDirectory(tempFile.resolve("external-data"));
catalog.createDatabase(identifier.getDatabaseName(), false);
// The filesystem catalog can load these table types but does not create them.
Path tablePath = ((FileSystemCatalog) catalog).getTableLocation(identifier);
new FileSystemSchemaManager(fileIO, tablePath)
.createTable(
Schema.newBuilder()
.column("id", DataTypes.INT())
.option(CoreOptions.TYPE.key(), tableType.toString())
.option(
CoreOptions.DATA_FILE_EXTERNAL_PATHS.key(),
externalPath.toUri().toString())
.build());

catalog.dropTable(identifier, false);

assertThat(externalPath).exists();
}

private FileSystemSchemaManager createLegacyTable(Identifier identifier, boolean cacheEnabled)
throws Exception {
return createLegacyTable(identifier, cacheEnabled, false);
}

private FileSystemSchemaManager createLegacyTable(
Identifier identifier, boolean cacheEnabled, boolean withData) throws Exception {
if (cacheEnabled) {
catalog = new CachingCatalog(catalog, new Options());
}
catalog.createDatabase(identifier.getDatabaseName(), false);
catalog.createTable(
identifier,
Schema.newBuilder()
.column("id", DataTypes.INT().notNull())
.column("seq", DataTypes.INT())
.column("v", DataTypes.INT())
.primaryKey("id")
.option("bucket", "1")
.option("merge-engine", "partial-update")
.option("changelog-producer", "lookup")
.option("fields.seq.sequence-group", "v")
.build(),
false);
Path tablePath =
((FileSystemCatalog) DelegateCatalog.rootCatalog(catalog))
.getTableLocation(identifier);
FileSystemSchemaManager schemaManager = new FileSystemSchemaManager(fileIO, tablePath);
if (withData) {
BatchWriteBuilder writeBuilder = catalog.getTable(identifier).newBatchWriteBuilder();
try (IOManager ioManager = IOManager.create(tempFile.resolve("io").toString());
BatchTableWrite write = writeBuilder.newWrite().withIOManager(ioManager);
BatchTableCommit commit = writeBuilder.newCommit()) {
write.write(GenericRow.of(1, 10, 100));
write.write(GenericRow.of(2, 20, 200));
commit.commit(write.prepareCommit());
}
catalog.invalidateTable(identifier);
}
TableSchema schema = schemaManager.latest().get();
Map<String, String> options = new HashMap<>(schema.options());
options.put("full-compaction.delta-commits", "1");
options.put("fields.seq.sequence-group", "id,v");
// Persist a legacy schema without passing it through the current CREATE validator.
writeLegacySchema(schemaManager, schema, schema.id(), options);
return schemaManager;
}

private void writeLegacySchema(
FileSystemSchemaManager schemaManager,
TableSchema schema,
long id,
Map<String, String> options)
throws Exception {
TableSchema legacySchema =
new TableSchema(
id,
schema.fields(),
schema.highestFieldId(),
schema.partitionKeys(),
schema.primaryKeys(),
new HashMap<>(options),
schema.comment());
Path schemaPath = schemaManager.toSchemaPath(id);
fileIO.mkdirs(schemaPath.getParent());
fileIO.overwriteFileUtf8(schemaPath, legacySchema.toString());
}
}
Loading
Loading