diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java index 48587b1770fe..d8a43f019b68 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java @@ -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; @@ -383,18 +384,17 @@ public void dropTable(Identifier identifier, boolean ignoreIfNotExists) } Set externalPaths = new HashSet<>(); - if (tableExistsInFileSystem(getTableLocation(identifier), DEFAULT_MAIN_BRANCH)) { - Table table = getTable(identifier); - if (table instanceof FileStoreTable) { - FileStoreTable fileStoreTable = (FileStoreTable) table; - List schemaExternalPaths = - getSchemaExternalPaths(fileStoreTable.schemaManager().listAll()); - externalPaths.addAll(schemaExternalPaths); - List 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())); } } } @@ -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; diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/FileSystemBranchManager.java b/paimon-core/src/main/java/org/apache/paimon/utils/FileSystemBranchManager.java index 872d864eeb0f..c84b2559cabe 100644 --- a/paimon-core/src/main/java/org/apache/paimon/utils/FileSystemBranchManager.java +++ b/paimon-core/src/main/java/org/apache/paimon/utils/FileSystemBranchManager.java @@ -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)); @@ -458,8 +453,13 @@ public void renameBranch(String fromBranch, String toBranch) { @Override public List branches() { + return branches(fileIO, tablePath); + } + + /** List branches without constructing the table's runtime managers. */ + public static List 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) { diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java index d6706b4bdaf6..c50bebd4399f 100644 --- a/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java @@ -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 { @@ -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 repairChanges = + Arrays.asList( + SchemaChange.removeOption("full-compaction.delta-commits"), + SchemaChange.setOption("fields.seq.sequence-group", "v")); + List 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 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 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 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 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()); + } } diff --git a/paimon-hive/paimon-hive-catalog/src/test/java/org/apache/paimon/hive/HiveCatalogTest.java b/paimon-hive/paimon-hive-catalog/src/test/java/org/apache/paimon/hive/HiveCatalogTest.java index 31bb910b979e..9e22b024e483 100644 --- a/paimon-hive/paimon-hive-catalog/src/test/java/org/apache/paimon/hive/HiveCatalogTest.java +++ b/paimon-hive/paimon-hive-catalog/src/test/java/org/apache/paimon/hive/HiveCatalogTest.java @@ -34,6 +34,8 @@ import org.apache.paimon.partition.PartitionStatistics; import org.apache.paimon.schema.Schema; import org.apache.paimon.schema.SchemaChange; +import org.apache.paimon.schema.SchemaManager; +import org.apache.paimon.schema.TableSchema; import org.apache.paimon.table.FallbackReadFileStoreTable; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.object.ObjectTable; @@ -56,6 +58,8 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import java.lang.reflect.Field; import java.util.ArrayList; @@ -305,6 +309,56 @@ public void testAlterHiveTableParameters() { } } + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testAlterAndDropTableWithInvalidOptions(boolean external) throws Exception { + String database = "legacy_options_db"; + Identifier identifier = Identifier.create(database, "legacy_table"); + catalog.createDatabase(database, false); + Path externalPath = new Path(warehouse, "external-data"); + fileIO.mkdirs(externalPath); + fileIO.overwriteFileUtf8(new Path(externalPath, "data"), "data"); + catalog.createTable( + identifier, + Schema.newBuilder() + .column("id", DataTypes.INT()) + .option("EXTERNAL", Boolean.toString(external)) + .option(CoreOptions.DATA_FILE_EXTERNAL_PATHS.key(), externalPath.toString()) + .build(), + false); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); + SchemaManager schemaManager = table.schemaManager(); + TableSchema schema = schemaManager.latest().get(); + Map invalidOptions = new HashMap<>(schema.options()); + invalidOptions.put(CoreOptions.RECORD_LEVEL_TIME_FIELD.key(), "missing"); + // Simulate options persisted before the corresponding runtime validation was added. + fileIO.overwriteFileUtf8( + schemaManager.toSchemaPath(schema.id()), schema.copy(invalidOptions).toString()); + + assertThatThrownBy(() -> catalog.getTable(identifier)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("missing"); + catalog.alterTable( + identifier, + SchemaChange.removeOption(CoreOptions.RECORD_LEVEL_TIME_FIELD.key()), + false); + assertThat(catalog.getTable(identifier).options()) + .doesNotContainKey(CoreOptions.RECORD_LEVEL_TIME_FIELD.key()); + TableSchema repairedSchema = schemaManager.latest().get(); + assertThat(repairedSchema.id()).isEqualTo(schema.id() + 1); + + fileIO.overwriteFileUtf8( + schemaManager.toSchemaPath(repairedSchema.id()), + repairedSchema.copy(invalidOptions).toString()); + assertThatThrownBy(() -> catalog.getTable(identifier)) + .isInstanceOf(IllegalArgumentException.class); + catalog.dropTable(identifier, false); + + assertThat(catalog.listTables(database)).doesNotContain(identifier.getTableName()); + assertThat(fileIO.exists(table.location())).isEqualTo(external); + assertThat(fileIO.exists(externalPath)).isEqualTo(external); + } + @Test public void testDropTableWhenTablePathMissing() throws Exception { String databaseName = "test_db";