This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 259c332b282 [fix](fe) Reject unsafe Iceberg column drops (#66627)
259c332b282 is described below
commit 259c332b28228e8f53fbf54bb6aca05ca7557da5
Author: Gabriel <[email protected]>
AuthorDate: Thu Aug 13 11:50:46 2026 +0800
[fix](fe) Reject unsafe Iceberg column drops (#66627)
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary: Dropping an Iceberg column can leave historical
partition specs referencing a field ID that no longer exists in the
current schema, which makes historical metadata unreadable. The same
failure occurs when dropping a struct ancestor of a referenced nested
field. Resolve top-level and nested drop targets before committing the
schema update, then reject the drop when any historical partition spec
references the target field or a field in its subtree.
### Release note
Iceberg column drops now fail explicitly when the column, or a nested
field beneath it, is referenced by a historical partition spec.
### Check List (For Author)
- Test
- [ ] Regression test
- [x] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason
- Behavior changed:
- [ ] No.
- [x] Yes. Invalid Iceberg column drops are rejected before corrupting
historical partition metadata.
- Does this need documentation?
- [x] No.
- [ ] Yes.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
.../doris/connector/iceberg/IcebergCatalogOps.java | 4 +-
.../iceberg/IcebergNestedColumnEvolution.java | 248 ++++++++++++-
...BackedIcebergCatalogOpsColumnEvolutionTest.java | 23 ++
.../iceberg/IcebergNestedColumnEvolutionTest.java | 394 +++++++++++++++++++++
4 files changed, 660 insertions(+), 9 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogOps.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogOps.java
index 80fa39ae8b8..4a7d516702f 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogOps.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogOps.java
@@ -458,9 +458,7 @@ public interface IcebergCatalogOps {
@Override
public void dropColumn(String dbName, String tableName, String
columnName) {
- UpdateSchema updateSchema = loadTable(dbName,
tableName).updateSchema();
- updateSchema.deleteColumn(columnName);
- updateSchema.commit();
+ IcebergNestedColumnEvolution.dropTopLevelColumn(loadTable(dbName,
tableName), columnName);
}
@Override
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergNestedColumnEvolution.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergNestedColumnEvolution.java
index 06a850e8a94..b4ca1c02c7d 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergNestedColumnEvolution.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergNestedColumnEvolution.java
@@ -21,15 +21,25 @@ import
org.apache.doris.connector.spi.DorisConnectorException;
import org.apache.doris.connector.spi.ddl.ConnectorColumnPath;
import org.apache.doris.connector.spi.ddl.ConnectorColumnPosition;
+import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.HasTableOperations;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableOperations;
import org.apache.iceberg.UpdateSchema;
+import org.apache.iceberg.encryption.EncryptionManager;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.LocationProvider;
+import org.apache.iceberg.transforms.Transforms;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.TypeUtil;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.types.Types.NestedField;
import java.util.ArrayList;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Objects;
import java.util.Set;
@@ -47,8 +57,8 @@ import java.util.TreeSet;
* {@link IcebergColumnChange}), so this class only ever touches iceberg +
neutral SPI types — never a Doris type.
* Every failure is raised as a {@link DorisConnectorException} (the caller
maps it to a {@code DdlException}).</p>
*
- * <p><b>No partial commit:</b> every {@code UpdateSchema.commit()} is the
final statement of each entry point, so
- * any validation/resolution throw aborts the whole change before anything is
committed (legacy parity).</p>
+ * <p><b>No partial commit:</b> every entry point validates before its single
metadata commit, so any
+ * validation/resolution throw aborts the whole change before anything is
committed (legacy parity).</p>
*
* <p>The TOP-LEVEL (flat) column ops stay on {@link IcebergCatalogOps}, but
they share this class's name
* resolution and validators — iceberg matches field names case-SENSITIVELY
while Doris column names are
@@ -57,6 +67,8 @@ import java.util.TreeSet;
*/
public final class IcebergNestedColumnEvolution {
+ private static final String REST_DROP_FENCE_PREFIX =
"_doris_schema_drop_fence_";
+
private IcebergNestedColumnEvolution() {
}
@@ -87,10 +99,234 @@ public final class IcebergNestedColumnEvolution {
/** Drops the nested field at {@code path}; its parent must resolve to a
struct that contains the leaf. */
public static void dropColumn(Table table, ConnectorColumnPath path) {
- ResolvedColumnPath resolvedPath =
validateNestedStructFieldPath(table.schema(), path, "drop");
- UpdateSchema updateSchema = table.updateSchema();
- updateSchema.deleteColumn(resolvedPath.getFullPath());
- updateSchema.commit();
+ dropColumnSafely(table, path, true);
+ }
+
+ static void dropTopLevelColumn(Table table, String columnName) {
+ dropColumnSafely(table, ConnectorColumnPath.of(columnName), false);
+ }
+
+ private static void dropColumnSafely(
+ Table table, ConnectorColumnPath path, boolean nested) {
+ TableOperations operations = ((HasTableOperations) table).operations();
+ TableMetadata loaded = operations.current();
+ if (loaded == null) {
+ throw new DorisConnectorException("Cannot drop column from an
unloaded Iceberg table: " + table.name());
+ }
+ ResolvedColumnPath loadedPath = resolveDropPath(loaded.schema(), path,
nested);
+ Set<Integer> loadedSubtreeIds = fieldSubtreeIds(loadedPath);
+
+ TableMetadata base = operations.refresh();
+ ResolvedColumnPath resolvedPath = validatePinnedDropIdentity(
+ table.name(), path, nested, loaded, loadedSubtreeIds, base);
+ validateNotUsedByRetainedPartitionSpec(base, resolvedPath);
+
+ if (isRestTableOperations(operations)) {
+ commitRestDropWithSpecFence(operations, table.name(), base,
resolvedPath);
+ return;
+ }
+
+ // Use Iceberg's standard commit so column-scoped writer properties
and name mappings are transformed.
+ // Direct catalogs atomically reject any metadata change after the
refresh through their metadata CAS.
+ new BaseTable(operations, table.name()).updateSchema()
+ .deleteColumn(resolvedPath.getFullPath()).commit();
+ }
+
+ private static void commitRestDropWithSpecFence(
+ TableOperations operations, String tableName, TableMetadata base,
ResolvedColumnPath resolvedPath) {
+ if (base.formatVersion() < 2) {
+ throw new DorisConnectorException(
+ "Cannot safely drop a column from a format-v1 Iceberg REST
table: " + tableName);
+ }
+
+ Set<Integer> droppedFieldIds = fieldSubtreeIds(resolvedPath);
+ String fenceSource = findRestFenceSource(base, droppedFieldIds);
+ if (fenceSource == null) {
+ throw new DorisConnectorException(
+ "Cannot safely fence a column drop in Iceberg REST table:
" + tableName);
+ }
+ String fenceName = newRestFenceName(base);
+
+ // Validation proved that no retained spec references the drop target,
so any concurrent new reference
+ // must allocate a partition field ID. Adding a distinct void field
makes REST assert that ID counter.
+ CapturingTableOperations schemaCapture = new
CapturingTableOperations(operations, base);
+ new BaseTable(schemaCapture, tableName).updateSchema()
+ .deleteColumn(resolvedPath.getFullPath()).commit();
+
+ CapturingTableOperations fenceCapture = new
CapturingTableOperations(operations, base);
+ new BaseTable(fenceCapture, tableName).updateSpec()
+ .addField(fenceName, Expressions.transform(fenceSource,
Transforms.alwaysNull()))
+ .addNonDefaultSpec().commit();
+
+ TableMetadata.Builder combined = TableMetadata.buildFrom(base);
+ schemaCapture.committed().changes().forEach(update ->
update.applyTo(combined));
+ fenceCapture.committed().changes().forEach(update ->
update.applyTo(combined));
+ // The durable fence must stay in metadata: removing its
client-predicted spec ID could delete a
+ // concurrently added spec after REST renumbers the fence during
server-side rebase.
+ operations.commit(base, combined.build());
+ }
+
+ private static String findRestFenceSource(TableMetadata metadata,
Set<Integer> droppedFieldIds) {
+ LinkedHashSet<Integer> candidates = new LinkedHashSet<>();
+ metadata.specs().forEach(spec -> spec.fields().forEach(field ->
candidates.add(field.sourceId())));
+ candidates.addAll(metadata.schema().identifierFieldIds());
+ collectStructPrimitiveFieldIds(metadata.schema().asStruct(),
droppedFieldIds, candidates);
+
+ return candidates.stream()
+ .filter(id -> !droppedFieldIds.contains(id))
+ .filter(id -> metadata.schema().findField(id) != null)
+ .filter(id -> metadata.spec().fields().stream()
+ .noneMatch(field -> field.sourceId() == id &&
field.transform().isVoid()))
+ .map(metadata.schema()::findColumnName)
+ .filter(Objects::nonNull)
+ .findFirst()
+ .orElse(null);
+ }
+
+ private static void collectStructPrimitiveFieldIds(
+ Types.StructType struct, Set<Integer> droppedFieldIds,
Set<Integer> candidates) {
+ for (NestedField field : struct.fields()) {
+ if (droppedFieldIds.contains(field.fieldId())) {
+ continue;
+ }
+ if (field.type().isPrimitiveType()) {
+ candidates.add(field.fieldId());
+ } else if (field.type().isStructType()) {
+ collectStructPrimitiveFieldIds(field.type().asStructType(),
droppedFieldIds, candidates);
+ }
+ }
+ }
+
+ private static String newRestFenceName(TableMetadata metadata) {
+ int suffix = metadata.lastAssignedPartitionId() + 1;
+ String candidate = REST_DROP_FENCE_PREFIX + suffix;
+ while (hasFieldOrPartitionName(metadata, candidate)) {
+ candidate = REST_DROP_FENCE_PREFIX + ++suffix;
+ }
+ return candidate;
+ }
+
+ private static boolean hasFieldOrPartitionName(TableMetadata metadata,
String candidate) {
+ boolean fieldNameExists =
TypeUtil.indexById(metadata.schema().asStruct()).values().stream()
+ .anyMatch(field -> field.name().equalsIgnoreCase(candidate));
+ return fieldNameExists || metadata.specs().stream().flatMap(spec ->
spec.fields().stream())
+ .anyMatch(field -> field.name().equalsIgnoreCase(candidate));
+ }
+
+ private static ResolvedColumnPath validatePinnedDropIdentity(
+ String tableName, ConnectorColumnPath path, boolean nested,
TableMetadata loaded,
+ Set<Integer> loadedSubtreeIds, TableMetadata refreshed) {
+ if (!Objects.equals(loaded.uuid(), refreshed.uuid())
+ || (loaded.uuid() == null && loaded != refreshed)) {
+ // UUID-less legacy metadata has no generation identity, so only
the exact refreshed object can be
+ // trusted; accepting a different object could retarget the DROP
to a recreated table.
+ throw concurrentDropTargetChange(tableName, path);
+ }
+ try {
+ ResolvedColumnPath refreshedPath =
resolveDropPath(refreshed.schema(), path, nested);
+ // A parent DROP pins every descendant ID as well as the parent
ID, preventing a same-path subtree
+ // replacement from turning an old request into deletion of a
newly created object.
+ if (!loadedSubtreeIds.equals(fieldSubtreeIds(refreshedPath))) {
+ throw concurrentDropTargetChange(tableName, path);
+ }
+ return refreshedPath;
+ } catch (DorisConnectorException e) {
+ throw concurrentDropTargetChange(tableName, path);
+ }
+ }
+
+ private static ResolvedColumnPath resolveDropPath(
+ Schema schema, ConnectorColumnPath path, boolean nested) {
+ return nested
+ ? validateNestedStructFieldPath(schema, path, "drop")
+ : resolveColumnPath(schema, path, "drop");
+ }
+
+ private static Set<Integer> fieldSubtreeIds(ResolvedColumnPath path) {
+ return new
TreeSet<>(TypeUtil.indexById(Types.StructType.of(path.getField())).keySet());
+ }
+
+ private static DorisConnectorException concurrentDropTargetChange(
+ String tableName, ConnectorColumnPath path) {
+ return new DorisConnectorException("Iceberg table or drop target
changed concurrently: "
+ + tableName + "." + path.getFullPath());
+ }
+
+ private static boolean isRestTableOperations(TableOperations operations) {
+ return
"RESTTableOperations".equalsIgnoreCase(operations.getClass().getSimpleName());
+ }
+
+ private static void validateNotUsedByRetainedPartitionSpec(
+ TableMetadata metadata, ResolvedColumnPath columnPath) {
+ Set<Integer> droppedFieldIds = TypeUtil.indexById(
+ Types.StructType.of(columnPath.getField())).keySet();
+ // Every retained spec resolves partition types by source field ID
against the current schema. Checking
+ // the current spec too establishes that a later REST fence can detect
any newly introduced reference.
+ boolean usedByRetainedSpec = metadata.specs().stream()
+ .anyMatch(spec -> spec.fields().stream().anyMatch(field ->
+ droppedFieldIds.contains(field.sourceId())));
+ if (usedByRetainedSpec) {
+ throw new DorisConnectorException(
+ "Cannot drop column which is used by an old partition spec
or current retained partition spec: "
+ + columnPath.getFullPath());
+ }
+ }
+
+ private static final class CapturingTableOperations implements
TableOperations {
+ private final TableOperations delegate;
+ private TableMetadata current;
+ private TableMetadata committed;
+
+ private CapturingTableOperations(TableOperations delegate,
TableMetadata base) {
+ this.delegate = delegate;
+ this.current = base;
+ }
+
+ private TableMetadata committed() {
+ if (committed == null) {
+ throw new IllegalStateException("Iceberg metadata update did
not commit to the capture");
+ }
+ return committed;
+ }
+
+ @Override
+ public TableMetadata current() {
+ return current;
+ }
+
+ @Override
+ public TableMetadata refresh() {
+ return current;
+ }
+
+ @Override
+ public void commit(TableMetadata base, TableMetadata metadata) {
+ if (base != current) {
+ throw new IllegalStateException("Iceberg metadata capture
committed from an unexpected base");
+ }
+ current = metadata;
+ committed = metadata;
+ }
+
+ @Override
+ public FileIO io() {
+ return delegate.io();
+ }
+
+ @Override
+ public EncryptionManager encryption() {
+ return delegate.encryption();
+ }
+
+ @Override
+ public String metadataFileLocation(String fileName) {
+ return delegate.metadataFileLocation(fileName);
+ }
+
+ @Override
+ public LocationProvider locationProvider() {
+ return delegate.locationProvider();
+ }
}
/**
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsColumnEvolutionTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsColumnEvolutionTest.java
index b80f0f6df38..c0b9089b6fd 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsColumnEvolutionTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsColumnEvolutionTest.java
@@ -23,8 +23,11 @@ import org.apache.doris.connector.spi.ConnectorType;
import org.apache.doris.connector.spi.DorisConnectorException;
import org.apache.doris.connector.spi.ddl.ConnectorColumnPosition;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
import org.apache.iceberg.inmemory.InMemoryCatalog;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
@@ -124,6 +127,26 @@ public class
CatalogBackedIcebergCatalogOpsColumnEvolutionTest {
Assertions.assertNull(reload().findField("val"));
}
+ @Test
+ public void testDropColumnUsedByHistoricalPartitionSpecFailsLoud() {
+ Table table = ops.loadTable("db1", "t1");
+ table.updateSpec().addField("val").commit();
+ String partitionName = table.spec().fields().get(0).name();
+ DataFile dataFile = DataFiles.builder(table.spec())
+ .withPath("file:/warehouse/t1/data.parquet")
+ .withFileSizeInBytes(1)
+ .withRecordCount(1)
+ .withPartitionPath(partitionName + "=1")
+ .build();
+ table.newAppend().appendFile(dataFile).commit();
+ table.updateSpec().removeField(partitionName).commit();
+
+ DorisConnectorException ex =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> ops.dropColumn("db1", "t1", "val"));
+ Assertions.assertTrue(ex.getMessage().contains("used by an old
partition spec"), ex.getMessage());
+ Assertions.assertNotNull(reload().findField("val"));
+ }
+
@Test
public void testRenameColumn() {
ops.renameColumn("db1", "t1", "name", "full_name");
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergNestedColumnEvolutionTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergNestedColumnEvolutionTest.java
index 154c85f286b..0f74d82f735 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergNestedColumnEvolutionTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergNestedColumnEvolutionTest.java
@@ -23,9 +23,26 @@ import
org.apache.doris.connector.spi.DorisConnectorException;
import org.apache.doris.connector.spi.ddl.ConnectorColumnPath;
import org.apache.doris.connector.spi.ddl.ConnectorColumnPosition;
+import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.MetadataTableType;
+import org.apache.iceberg.MetadataTableUtils;
+import org.apache.iceberg.MetadataUpdate;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.UpdateRequirement;
+import org.apache.iceberg.UpdateRequirements;
+import org.apache.iceberg.encryption.EncryptionManager;
+import org.apache.iceberg.exceptions.CommitFailedException;
import org.apache.iceberg.inmemory.InMemoryCatalog;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.LocationProvider;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.AfterEach;
@@ -36,6 +53,8 @@ import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import java.util.Map;
+import java.util.Set;
import java.util.stream.Collectors;
/**
@@ -79,6 +98,11 @@ public class IcebergNestedColumnEvolutionTest {
IcebergSchemaBuilder.buildTableProperties(Collections.emptyMap()));
}
+ private void createTable(String table, Schema schema, PartitionSpec spec,
Map<String, String> properties) {
+ ops.createTable("db1", table, schema, spec, null,
+ IcebergSchemaBuilder.buildTableProperties(properties));
+ }
+
private Schema reload(String table) {
return ops.loadTable("db1", table).schema();
}
@@ -267,6 +291,376 @@ public class IcebergNestedColumnEvolutionTest {
Assertions.assertNotNull(s.field("x"));
}
+ @Test
+ public void testDropNestedFieldUsedByHistoricalPartitionSpecFailsLoud() {
+ createTable("d_old_spec", flatNestedSchema());
+ Table table = ops.loadTable("db1", "d_old_spec");
+ table.updateSpec().addField("s.a").commit();
+ String partitionName = table.spec().fields().get(0).name();
+ DataFile dataFile = DataFiles.builder(table.spec())
+ .withPath("file:/warehouse/d_old_spec/data.parquet")
+ .withFileSizeInBytes(1)
+ .withRecordCount(1)
+ .withPartitionPath(partitionName + "=1")
+ .build();
+ table.newAppend().appendFile(dataFile).commit();
+ table.updateSpec().removeField(partitionName).commit();
+
+ DorisConnectorException ex =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> ops.dropNestedColumn("db1", "d_old_spec", path("s",
"a")));
+ Assertions.assertTrue(ex.getMessage().contains("used by an old
partition spec"), ex.getMessage());
+ Assertions.assertNotNull(reload("d_old_spec").findField("s.a"));
+ }
+
+ @Test
+ public void testDropNestedFieldUsedByCurrentFormatV1VoidSpecFailsLoud() {
+ Schema schema = flatNestedSchema();
+ PartitionSpec initialSpec =
PartitionSpec.builderFor(schema).identity("s.a").build();
+ createTable("d_v1_void", schema, initialSpec,
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "1"));
+ Table table = ops.loadTable("db1", "d_v1_void");
+ int oldSpecId = table.spec().specId();
+
table.updateSpec().removeField(table.spec().fields().get(0).name()).commit();
+
+ // Simulate metadata cleanup so only the v1 current void field retains
the source ID. Ignoring the
+ // current spec would let schema deletion corrupt metadata-table
schema construction.
+ table.refresh();
+ TableOperations tableOps = ((HasTableOperations) table).operations();
+ TableMetadata base = tableOps.current();
+ TableMetadata.Builder builder = TableMetadata.buildFrom(base);
+ new
MetadataUpdate.RemovePartitionSpecs(Set.of(oldSpecId)).applyTo(builder);
+ tableOps.commit(base, builder.build());
+ table.refresh();
+
+
Assertions.assertTrue(table.spec().fields().get(0).transform().isVoid());
+ Assertions.assertEquals(1, table.specs().size());
+ assertFailsLoud(() -> ops.dropNestedColumn("db1", "d_v1_void",
path("s", "a")),
+ "partition spec");
+ Assertions.assertNotNull(reload("d_v1_void").findField("s.a"));
+ Assertions.assertDoesNotThrow(() ->
MetadataTableUtils.createMetadataTableInstance(
+ ops.loadTable("db1", "d_v1_void"),
MetadataTableType.POSITION_DELETES).schema());
+ }
+
+ @Test
+ public void
testRestDropSurvivesSpecIdRebaseWithoutRemovingConcurrentSpec() {
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "drop_me",
Types.IntegerType.get()));
+ createTable("d_concurrent_spec", schema,
+ PartitionSpec.builderFor(schema).identity("id").build(),
Collections.emptyMap());
+ Table original = ops.loadTable("db1", "d_concurrent_spec");
+ String metricsKey = TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX +
"drop_me";
+ original.updateProperties().set(metricsKey, "full").commit();
+ RestTableOperations concurrentOps = new RestTableOperations(
+ ((HasTableOperations) original).operations(), null);
+ Table racedTable = new BaseTable(concurrentOps, original.name());
+
+ IcebergNestedColumnEvolution.dropTopLevelColumn(racedTable, "drop_me");
+
+ Assertions.assertEquals(1, concurrentOps.commitAttempts);
+ Assertions.assertTrue(concurrentOps.requirements.stream()
+
.anyMatch(UpdateRequirement.AssertLastAssignedPartitionId.class::isInstance));
+ Assertions.assertFalse(concurrentOps.clientChanges.stream()
+
.anyMatch(MetadataUpdate.RemovePartitionSpecs.class::isInstance));
+ original.refresh();
+ Assertions.assertNull(original.schema().findField("drop_me"));
+ Assertions.assertFalse(original.properties().containsKey(metricsKey),
+ "the REST path must preserve SchemaUpdate's column-property
cleanup");
+ Assertions.assertEquals(3, original.specs().size(),
+ "both the rebased concurrent spec and the durable fence must
be retained");
+
Assertions.assertTrue(original.specs().values().stream().anyMatch(PartitionSpec::isUnpartitioned));
+ Assertions.assertTrue(original.specs().values().stream().flatMap(spec
-> spec.fields().stream())
+ .anyMatch(field ->
field.name().startsWith("_doris_schema_drop_fence_")
+ && field.sourceId() == schema.findField("id").fieldId()
+ && field.transform().isVoid()));
+ }
+
+ @Test
+ public void testRestDropRejectsConcurrentSpecUsingDropTarget() {
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "drop_me",
Types.IntegerType.get()));
+ createTable("d_target_spec_race", schema);
+ Table original = ops.loadTable("db1", "d_target_spec_race");
+ RestTableOperations concurrentOps = new RestTableOperations(
+ ((HasTableOperations) original).operations(), "drop_me");
+ Table racedTable = new BaseTable(concurrentOps, original.name());
+
+ Assertions.assertThrows(CommitFailedException.class,
+ () ->
IcebergNestedColumnEvolution.dropTopLevelColumn(racedTable, "drop_me"));
+
+ Assertions.assertEquals(1, concurrentOps.commitAttempts);
+ original.refresh();
+ Assertions.assertNotNull(original.schema().findField("drop_me"));
+ Assertions.assertTrue(original.specs().values().stream().flatMap(spec
-> spec.fields().stream())
+ .anyMatch(field -> field.sourceId() ==
schema.findField("drop_me").fieldId()));
+ }
+
+ @Test
+ public void testRestFormatV1DropFailsBeforeCommit() {
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "drop_me",
Types.IntegerType.get()));
+ createTable("d_rest_v1", schema, PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "1"));
+ Table original = ops.loadTable("db1", "d_rest_v1");
+ RestTableOperations concurrentOps = new RestTableOperations(
+ ((HasTableOperations) original).operations(), null);
+
+ assertFailsLoud(() -> IcebergNestedColumnEvolution.dropTopLevelColumn(
+ new BaseTable(concurrentOps, original.name()), "drop_me"),
"format-v1");
+
+ Assertions.assertEquals(0, concurrentOps.commitAttempts);
+ Assertions.assertNotNull(original.schema().findField("drop_me"));
+ }
+
+ @Test
+ public void testDropAllowsSchemaContainingOldSyntheticFenceName() {
+ Schema schema = new Schema(
+ Types.NestedField.optional(1, "_doris_schema_drop_fence_1000",
Types.StringType.get()),
+ Types.NestedField.optional(2, "drop_me",
Types.IntegerType.get()));
+ createTable("d_fence_name", schema);
+ Table original = ops.loadTable("db1", "d_fence_name");
+ RestTableOperations restOps = new RestTableOperations(
+ ((HasTableOperations) original).operations(), null);
+
+ IcebergNestedColumnEvolution.dropTopLevelColumn(
+ new BaseTable(restOps, original.name()), "drop_me");
+
+ original.refresh();
+
Assertions.assertNotNull(original.schema().findField("_doris_schema_drop_fence_1000"));
+ Assertions.assertNull(original.schema().findField("drop_me"));
+ Assertions.assertTrue(original.specs().values().stream().flatMap(spec
-> spec.fields().stream())
+ .anyMatch(field ->
field.name().equals("_doris_schema_drop_fence_1001")));
+ }
+
+ @Test
+ public void testFormatV1DropPreservesFirstPartitionFieldId() {
+ createTable("d_v1_first_partition", flatNestedSchema(),
PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "1"));
+
+ ops.dropNestedColumn("db1", "d_v1_first_partition", path("s", "a"));
+ Table table = ops.loadTable("db1", "d_v1_first_partition");
+ Assertions.assertDoesNotThrow(() ->
table.updateSpec().addField("s.x").commit());
+ Assertions.assertEquals(1000, table.spec().fields().get(0).fieldId());
+ }
+
+ @Test
+ public void testDropRemovesColumnScopedWriterProperties() {
+ createTable("d_column_props", flatNestedSchema());
+ Table table = ops.loadTable("db1", "d_column_props");
+ String metricsKey = TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX +
"s.a";
+ String bloomKey =
TableProperties.PARQUET_BLOOM_FILTER_COLUMN_ENABLED_PREFIX + "s.a";
+ String statsKey = TableProperties.PARQUET_COLUMN_STATS_ENABLED_PREFIX
+ "s.a";
+ table.updateProperties()
+ .set(metricsKey, "full")
+ .set(bloomKey, "true")
+ .set(statsKey, "true")
+ .commit();
+
+ ops.dropNestedColumn("db1", "d_column_props", path("s", "a"));
+
+ Map<String, String> properties = ops.loadTable("db1",
"d_column_props").properties();
+ Assertions.assertFalse(properties.containsKey(metricsKey));
+ Assertions.assertFalse(properties.containsKey(bloomKey));
+ Assertions.assertFalse(properties.containsKey(statsKey));
+ }
+
+ @Test
+ public void testDropRejectsTableReplacementDuringInitialRefresh() {
+ createTable("d_original_identity", flatNestedSchema());
+ createTable("d_replacement_identity", flatNestedSchema());
+ Table original = ops.loadTable("db1", "d_original_identity");
+ Table replacement = ops.loadTable("db1", "d_replacement_identity");
+ RefreshingTableOperations refreshing = new RefreshingTableOperations(
+ ((HasTableOperations) original).operations().current(),
+ ((HasTableOperations) replacement).operations().current(),
+ ((HasTableOperations) original).operations());
+
+ assertFailsLoud(() -> IcebergNestedColumnEvolution.dropColumn(
+ new BaseTable(refreshing, original.name()), path("s", "a")),
"changed concurrently");
+ Assertions.assertEquals(0, refreshing.commitAttempts);
+ }
+
+ @Test
+ public void testDropRejectsSamePathFieldReplacementDuringRefresh() {
+ createTable("d_field_identity", flatNestedSchema());
+ Table table = ops.loadTable("db1", "d_field_identity");
+ TableOperations delegate = ((HasTableOperations) table).operations();
+ TableMetadata loaded = delegate.current();
+ int originalFieldId = loaded.schema().findField("s.a").fieldId();
+ table.updateSchema().deleteColumn("s.a").commit();
+ table.updateSchema().addColumn("s", "a",
Types.IntegerType.get()).commit();
+ TableMetadata replacement = delegate.refresh();
+ Assertions.assertNotEquals(originalFieldId,
replacement.schema().findField("s.a").fieldId());
+ RefreshingTableOperations refreshing = new
RefreshingTableOperations(loaded, replacement, delegate);
+
+ assertFailsLoud(() -> IcebergNestedColumnEvolution.dropColumn(
+ new BaseTable(refreshing, table.name()), path("s", "a")),
"changed concurrently");
+ Assertions.assertEquals(0, refreshing.commitAttempts);
+ }
+
+ @Test
+ public void testDropParentRejectsSubtreeReplacementDuringRefresh() {
+ createTable("d_subtree_identity", flatNestedSchema());
+ Table table = ops.loadTable("db1", "d_subtree_identity");
+ TableOperations delegate = ((HasTableOperations) table).operations();
+ TableMetadata loaded = delegate.current();
+ table.updateSchema().deleteColumn("s.a").commit();
+ table.updateSchema().addColumn("s", "a",
Types.IntegerType.get()).commit();
+ RefreshingTableOperations refreshing = new RefreshingTableOperations(
+ loaded, delegate.refresh(), delegate);
+
+ assertFailsLoud(() -> IcebergNestedColumnEvolution.dropTopLevelColumn(
+ new BaseTable(refreshing, table.name()), "s"), "changed
concurrently");
+ Assertions.assertEquals(0, refreshing.commitAttempts);
+ }
+
+ /** Simulates REST applying the client updates after a concurrent
partition-spec commit. */
+ private static final class RestTableOperations implements TableOperations {
+ private final TableOperations delegate;
+ private final String sourceColumn;
+ private boolean injectConcurrentSpec = true;
+ private int commitAttempts;
+ private List<UpdateRequirement> requirements = Collections.emptyList();
+ private List<MetadataUpdate> clientChanges = Collections.emptyList();
+
+ private RestTableOperations(TableOperations delegate, String
sourceColumn) {
+ this.delegate = delegate;
+ this.sourceColumn = sourceColumn;
+ }
+
+ @Override
+ public TableMetadata current() {
+ return delegate.current();
+ }
+
+ @Override
+ public TableMetadata refresh() {
+ return delegate.refresh();
+ }
+
+ @Override
+ public void commit(TableMetadata base, TableMetadata metadata) {
+ commitAttempts++;
+ if (!injectConcurrentSpec) {
+ delegate.commit(base, metadata);
+ return;
+ }
+ injectConcurrentSpec = false;
+ clientChanges = metadata.changes();
+ requirements = UpdateRequirements.forUpdateTable(base,
clientChanges);
+
+ int newSpecId =
base.specs().stream().mapToInt(PartitionSpec::specId).max().orElse(-1) + 1;
+ PartitionSpec concurrentSpec = sourceColumn == null
+ ?
PartitionSpec.builderFor(base.schema()).withSpecId(newSpecId).build()
+ :
PartitionSpec.builderFor(base.schema()).withSpecId(newSpecId).identity(sourceColumn).build();
+ TableMetadata.Builder concurrent =
TableMetadata.buildFrom(base).addPartitionSpec(concurrentSpec);
+ delegate.commit(base, concurrent.build());
+ TableMetadata rebased = delegate.refresh();
+ for (UpdateRequirement requirement : requirements) {
+ requirement.validate(rebased);
+ }
+ TableMetadata.Builder rebasedUpdate =
TableMetadata.buildFrom(rebased);
+ for (MetadataUpdate update : clientChanges) {
+ update.applyTo(rebasedUpdate);
+ }
+ delegate.commit(rebased, rebasedUpdate.build());
+ }
+
+ @Override
+ public FileIO io() {
+ return delegate.io();
+ }
+
+ @Override
+ public EncryptionManager encryption() {
+ return delegate.encryption();
+ }
+
+ @Override
+ public String metadataFileLocation(String fileName) {
+ return delegate.metadataFileLocation(fileName);
+ }
+
+ @Override
+ public LocationProvider locationProvider() {
+ return delegate.locationProvider();
+ }
+ }
+
+ private static final class RefreshingTableOperations implements
TableOperations {
+ private final TableMetadata loaded;
+ private final TableMetadata refreshed;
+ private final TableOperations delegate;
+ private boolean didRefresh;
+ private int commitAttempts;
+
+ private RefreshingTableOperations(
+ TableMetadata loaded, TableMetadata refreshed, TableOperations
delegate) {
+ this.loaded = loaded;
+ this.refreshed = refreshed;
+ this.delegate = delegate;
+ }
+
+ @Override
+ public TableMetadata current() {
+ return didRefresh ? refreshed : loaded;
+ }
+
+ @Override
+ public TableMetadata refresh() {
+ didRefresh = true;
+ return refreshed;
+ }
+
+ @Override
+ public void commit(TableMetadata base, TableMetadata metadata) {
+ commitAttempts++;
+ }
+
+ @Override
+ public FileIO io() {
+ return delegate.io();
+ }
+
+ @Override
+ public EncryptionManager encryption() {
+ return delegate.encryption();
+ }
+
+ @Override
+ public String metadataFileLocation(String fileName) {
+ return delegate.metadataFileLocation(fileName);
+ }
+
+ @Override
+ public LocationProvider locationProvider() {
+ return delegate.locationProvider();
+ }
+ }
+
+ @Test
+ public void testDropParentOfHistoricalPartitionFieldFailsLoud() {
+ createTable("d_old_spec_parent", flatNestedSchema());
+ Table table = ops.loadTable("db1", "d_old_spec_parent");
+ table.updateSpec().addField("s.a").commit();
+ String partitionName = table.spec().fields().get(0).name();
+ DataFile dataFile = DataFiles.builder(table.spec())
+ .withPath("file:/warehouse/d_old_spec_parent/data.parquet")
+ .withFileSizeInBytes(1)
+ .withRecordCount(1)
+ .withPartitionPath(partitionName + "=1")
+ .build();
+ table.newAppend().appendFile(dataFile).commit();
+ table.updateSpec().removeField(partitionName).commit();
+
+ DorisConnectorException ex =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> ops.dropColumn("db1", "d_old_spec_parent", "s"));
+ Assertions.assertTrue(ex.getMessage().contains("used by an old
partition spec"), ex.getMessage());
+ Assertions.assertNotNull(reload("d_old_spec_parent").findField("s"));
+ }
+
@Test
public void testRenameNestedStructField() {
createTable("d_rename", flatNestedSchema());
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
