This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 030b38f994 [core] Decouple catalog alter and drop from runtime table
loading (#10830)
030b38f994 is described below
commit 030b38f9948217d0851d29b3583f4f2ee37fd7ed
Author: zhoulii <[email protected]>
AuthorDate: Tue Oct 27 19:26:42 2026 +0800
[core] Decouple catalog alter and drop from runtime table loading (#10830)
---
.../org/apache/paimon/catalog/AbstractCatalog.java | 27 +-
.../paimon/utils/FileSystemBranchManager.java | 12 +-
.../paimon/catalog/FileSystemCatalogTest.java | 290 +++++++++++++++++++++
.../org/apache/paimon/hive/HiveCatalogTest.java | 54 ++++
4 files changed, 364 insertions(+), 19 deletions(-)
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 48587b1770..d8a43f019b 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.sink.TableCommitImpl;
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 abstract class AbstractCatalog implements Catalog {
}
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()));
}
}
}
@@ -503,7 +503,8 @@ public abstract class AbstractCatalog implements Catalog {
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 872d864eeb..c84b2559ca 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 class FileSystemBranchManager implements
BranchManager {
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 class FileSystemBranchManager implements
BranchManager {
@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) {
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 d6706b4bda..c50bebd439 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 @@ package org.apache.paimon.catalog;
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 class FileSystemCatalogTest extends
CatalogTestBase {
.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());
+ }
}
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 31bb910b97..9e22b024e4 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.Partition;
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.apache.thrift.TException;
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 class HiveCatalogTest extends CatalogTestBase {
}
}
+ @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<String, String> 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";
