This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 6b2a7d87603 [fix](fe) Protect historical Iceberg columns in direct
catalogs (#69408)
6b2a7d87603 is described below
commit 6b2a7d87603246833ecd1e3f67a6f2c7a7a4a924
Author: Oliveira <[email protected]>
AuthorDate: Tue Oct 27 16:33:03 2026 +0800
[fix](fe) Protect historical Iceberg columns in direct catalogs (#69408)
### What problem does this PR solve?
Issue Number: None
Related PR: #66627
Problem Summary:
In direct Iceberg catalogs, dropping a source column that is still
referenced by a retained historical partition spec can leave metadata
that other engines cannot read. Checking only the current/default spec
misses references left after partition evolution.
Validate the target and all descendants against every retained spec,
including historical specs without live files and format-v1 void fields.
Pin the loaded table UUID and target subtree across refresh, stage the
standard SDK schema update against exactly that validated base, and
submit the original base once through `TableOperations.commit`. Catalogs
with full metadata CAS, including HMS/JDBC implementations, reject
intervening metadata changes rather than retrying an unvalidated schema
update. Unreferenced columns remain droppable.
Immediately before commit, reread the metadata-file UUID using the
existing FileIO, following the repository's transaction validation
pattern. This detects Hadoop's known cached-v1 metadata alias after
table recreation while retaining the original object for the SDK CAS. It
does not replace Hadoop's underlying numeric-version protocol with a
general atomic UUID fence.
Scope: REST retains branch-4.2's ordinary SDK table-update endpoint,
loaded-table authentication and table-handle/cache behavior. This PR
adds no REST transaction requirement or blanket partition-history
restriction. Complete atomic protection of REST retained specs is a
separate known gap tracked in #69544, including imported low partition
IDs and stale writers after metadata cleanup; it is not claimed fixed
here. ADD/RENAME, SDK dependencies and catalog configuration remain
unchanged.
The additional metadata read uses the existing catalog FileIO and
metadata location under ALTER TABLE authorization. For HTTP-backed
storage, it reuses the existing catalog/storage SSRF surface; no new
user-supplied endpoint or artifact-loading property is introduced.
### Release note
Reject direct Iceberg catalog column drops when any retained partition
spec references the target or its descendants. Preserve legal
unreferenced drops and the existing REST SDK behavior. Complete REST
retained-spec concurrency protection is tracked separately in #69544.
### Check List (For Author)
- Test
- [x] Unit Test: 370 focused FE tests passed, zero
failures/errors/skips. Includes low-ID restored-spec CAS races,
currently unpartitioned tables, UUID/subtree replacement, real Hadoop
directory recreation and three real-HTTP REST compatibility cases. FE
Checkstyle passed with zero violations.
- [x] Regression test: complete native `iceberg_schema_change_ddl`,
`test_iceberg_delete_unpartitioned_evolution` and
`test_iceberg_historical_filter_planning` suites passed with zero
failed/fatal/skipped suites. These verify REST baseline compatibility.
The task ASAN runtime used a 32GiB BE process limit; the first 8GiB
attempt hit the local process cap and its logs are retained.
- [x] Manual test: standard Clang21 ASAN BE+FE build and verified
source/artifact/serving identities before and after regression, using
fresh task state. Direct-path tests use real InMemory/Hadoop commits; a
standalone HMS server integration was not run locally.
- Behavior changed:
- [x] Yes. Direct-catalog unsafe source-column drops are rejected; REST
preserves branch baseline behavior.
- Does this need documentation?
- [x] No. Scope, catalog atomicity assumptions and the REST follow-up
are stated above.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
.../datasource/iceberg/IcebergMetadataOps.java | 219 ++++++-
.../datasource/iceberg/IcebergColumnDropTest.java | 688 +++++++++++++++++++++
.../iceberg/IcebergMetadataOpsValidationTest.java | 5 +-
.../IcebergRestColumnDropCompatibilityTest.java | 210 +++++++
4 files changed, 1097 insertions(+), 25 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataOps.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataOps.java
index e7cd9c13a0e..728f4a1c3cd 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataOps.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataOps.java
@@ -56,6 +56,8 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Splitter;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.HasTableOperations;
import org.apache.iceberg.ManageSnapshots;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.RowLevelOperationMode;
@@ -63,6 +65,9 @@ import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.SnapshotRef;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableMetadataParser;
+import org.apache.iceberg.TableOperations;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.UpdatePartitionSpec;
import org.apache.iceberg.UpdateProperties;
@@ -73,10 +78,13 @@ import org.apache.iceberg.catalog.SessionCatalog;
import org.apache.iceberg.catalog.SupportsNamespaces;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.catalog.ViewCatalog;
+import org.apache.iceberg.encryption.EncryptionManager;
import org.apache.iceberg.exceptions.NoSuchNamespaceException;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.expressions.Literal;
import org.apache.iceberg.expressions.Term;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.LocationProvider;
import org.apache.iceberg.jdbc.JdbcCatalog;
import org.apache.iceberg.rest.RESTSessionCatalog;
import org.apache.iceberg.types.Type;
@@ -117,9 +125,7 @@ public class IcebergMetadataOps implements
ExternalMetadataOps {
// catalog types it is the catalog itself when it implements ViewCatalog.
Empty when views are unsupported
// or disabled.
private final Optional<ViewCatalog> defaultViewCatalog;
- // Non-null only when the backing catalog supports per-user dynamic
session (an Iceberg REST catalog with
- // iceberg.rest.session=user). Captured once at construction; the
per-request decision is delegated to it so
- // this class never re-checks the catalog type or reads REST properties
directly.
+ // Session-aware catalog capability for ordinary metadata calls.
private final IcebergUserSessionCatalog userSessionCatalog;
private IcebergSessionCatalogAdapter sessionCatalogAdapter;
private ExecutionAuthenticator executionAuthenticator;
@@ -141,10 +147,8 @@ public class IcebergMetadataOps implements
ExternalMetadataOps {
public IcebergMetadataOps(ExternalCatalog dorisCatalog, Catalog catalog) {
this.dorisCatalog = dorisCatalog;
this.catalog = catalog;
- // Session-aware behavior exists only when the backing catalog
advertises the IcebergUserSessionCatalog
- // capability (an Iceberg REST catalog with dynamic identity). Capture
it once and read what we need
- // from the capability, so no call-time path has to know about the
concrete REST catalog or its
- // properties. For any other catalog type this is null (session
disabled).
+ // Capture the session catalog and delegated-token policy with this
metadata-ops generation.
+ // Other catalog types have no session-aware capability.
this.userSessionCatalog = dorisCatalog instanceof
IcebergUserSessionCatalog
? (IcebergUserSessionCatalog) dorisCatalog : null;
RESTSessionCatalog restSessionCatalog =
@@ -959,15 +963,9 @@ public class IcebergMetadataOps implements
ExternalMetadataOps {
@Override
public void dropColumn(ExternalTable dorisTable, String columnName, long
updateTime) throws UserException {
try (IcebergExternalMetaCache.WritableTableLease lease =
acquireWritableTable(dorisTable)) {
- Table icebergTable = lease.getTable();
- validateRowLineageColumnMutation(icebergTable, columnName, "drop");
- ResolvedColumnPath columnPath =
resolveColumnPath(icebergTable.schema(), ColumnPath.of(columnName), "drop");
- UpdateSchema updateSchema = icebergTable.updateSchema();
- updateSchema.deleteColumn(columnPath.getFullPath());
- commitTableUpdate(lease, updateSchema::commit,
- "Failed to drop column: " + columnName + " from table: " +
icebergTable.name());
+ dropColumnSafely(lease, ColumnPath.of(columnName));
}
- refreshTable(dorisTable, updateTime);
+ refreshAfterColumnDrop(dorisTable, updateTime);
}
@Override
@@ -977,16 +975,191 @@ public class IcebergMetadataOps implements
ExternalMetadataOps {
return;
}
try (IcebergExternalMetaCache.WritableTableLease lease =
acquireWritableTable(dorisTable)) {
- Table icebergTable = lease.getTable();
- ResolvedColumnPath resolvedPath =
validateNestedStructFieldPath(icebergTable.schema(), columnPath, "drop");
+ dropColumnSafely(lease, columnPath);
+ }
+ refreshAfterColumnDrop(dorisTable, updateTime);
+ }
- UpdateSchema updateSchema = icebergTable.updateSchema();
- updateSchema.deleteColumn(resolvedPath.getFullPath());
- commitTableUpdate(lease, updateSchema::commit,
- "Failed to drop nested column: " +
columnPath.getFullPath() + " from table: "
- + icebergTable.name());
+ private void dropColumnSafely(IcebergExternalMetaCache.WritableTableLease
lease, ColumnPath path)
+ throws UserException {
+ try {
+ // A refresh can perform authenticated catalog IO just like the
eventual commit. Keep target
+ // validation, refresh and commit inside the one catalog
generation retained by this lease.
+ lease.getAuthenticator().execute(() -> {
+ commitColumnDrop(lease, path);
+ return null;
+ });
+ } catch (UserException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new UserException("Failed to drop column: " +
path.getFullPath() + ", error message is: "
+ + ExceptionUtils.getRootCauseMessage(e), e);
+ }
+ }
+
+ private void refreshAfterColumnDrop(ExternalTable dorisTable, long
updateTime) throws UserException {
+ try {
+ refreshTable(dorisTable, updateTime);
+ } catch (RuntimeException e) {
+ throw new UserException("Iceberg column DROP committed, but Doris
metadata refresh failed for "
+ + dorisTable.getRemoteDbName() + "." +
dorisTable.getRemoteName()
+ + "; refresh the table before another operation: " +
e.getMessage(), e);
+ }
+ }
+
+ private void commitColumnDrop(IcebergExternalMetaCache.WritableTableLease
lease, ColumnPath path)
+ throws UserException {
+ Table table = lease.getTable();
+ if (!path.isNested()) {
+ validateRowLineageColumnMutation(table, path.getTopLevelName(),
"drop");
+ }
+ if (lease.getRuntimeContext().getMetastoreProperties() instanceof
IcebergRestProperties) {
+ // Preserve the ordinary SDK REST endpoint, table authentication
and cache handling.
+ // REST requirements cannot atomically fence the retained-spec
set; that fix is separate.
+ ResolvedColumnPath resolvedPath = resolveDropPath(table.schema(),
path);
+
table.updateSchema().deleteColumn(resolvedPath.getFullPath()).commit();
+ return;
+ }
+ TableOperations operations = ((HasTableOperations) table).operations();
+ TableMetadata loaded = operations.current();
+ if (loaded == null) {
+ throw new UserException("Cannot drop column from an unloaded
Iceberg table: " + table.name());
+ }
+ ResolvedColumnPath loadedPath = resolveDropPath(loaded.schema(), path);
+ Set<Integer> loadedSubtreeIds = fieldSubtreeIds(loadedPath);
+
+ TableMetadata base = operations.refresh();
+ ResolvedColumnPath resolvedPath = validatePinnedDropIdentity(
+ table.name(), path, loaded, loadedSubtreeIds, base);
+ validateNotUsedByRetainedPartitionSpec(base, resolvedPath);
+ // Stage the standard SDK update against exactly the validated base.
Recapturing operations.current()
+ // here could instead delete a same-name replacement published on a
shared handle after refresh.
+ CapturingTableOperations schemaCapture = new
CapturingTableOperations(operations, base);
+ Table stagedTable = new BaseTable(schemaCapture, table.name());
+ if (!path.isNested()) {
+ validateRowLineageColumnMutation(stagedTable,
path.getTopLevelName(), "drop");
+ }
+
stagedTable.updateSchema().deleteColumn(resolvedPath.getFullPath()).commit();
+ TableMetadata schemaUpdate = schemaCapture.committed();
+ // Like IcebergTransaction, reread the file because Hadoop can reuse a
metadata version
+ // after drop/recreate while refresh() retains the old cached object.
+ TableMetadata currentFile = TableMetadataParser.read(operations.io(),
base.metadataFileLocation());
+ if (!Objects.equals(base.uuid(), currentFile.uuid())) {
+ throw concurrentDropTargetChange(table.name(), path);
+ }
+ // Catalogs with full metadata CAS reject changes after the validated
refresh.
+ operations.commit(base, schemaUpdate);
+ }
+
+ private ResolvedColumnPath validatePinnedDropIdentity(
+ String tableName, ColumnPath path, TableMetadata loaded,
+ Set<Integer> loadedSubtreeIds, TableMetadata refreshed) throws
UserException {
+ 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);
+ // 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 (UserException e) {
+ throw concurrentDropTargetChange(tableName, path);
+ }
+ }
+
+ private ResolvedColumnPath resolveDropPath(
+ Schema schema, ColumnPath path) throws UserException {
+ return path.isNested()
+ ? 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 UserException concurrentDropTargetChange(
+ String tableName, ColumnPath path) {
+ return new UserException("Iceberg table or drop target changed
concurrently: "
+ + tableName + "." + path.getFullPath());
+ }
+
+ private static void validateNotUsedByRetainedPartitionSpec(
+ TableMetadata metadata, ResolvedColumnPath columnPath) throws
UserException {
+ Set<Integer> droppedFieldIds = fieldSubtreeIds(columnPath);
+ // Every retained spec resolves partition types by source field ID
against the current schema,
+ // including the default spec and historical specs without live files.
+ boolean usedByRetainedSpec = metadata.specs().stream()
+ .anyMatch(spec -> spec.fields().stream().anyMatch(field ->
+ droppedFieldIds.contains(field.sourceId())));
+ if (usedByRetainedSpec) {
+ throw new UserException(
+ "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();
}
- refreshTable(dorisTable, updateTime);
}
@Override
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergColumnDropTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergColumnDropTest.java
new file mode 100644
index 00000000000..8710dc6be9d
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergColumnDropTest.java
@@ -0,0 +1,688 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.datasource.iceberg;
+
+import org.apache.doris.analysis.ColumnPath;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.security.authentication.ExecutionAuthenticator;
+import org.apache.doris.datasource.ExternalCatalog;
+import org.apache.doris.datasource.ExternalTable;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.MetadataUpdate;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.PartitionSpecParser;
+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.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.encryption.EncryptionManager;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.inmemory.InMemoryCatalog;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.LocationProvider;
+import org.apache.iceberg.mapping.MappingUtil;
+import org.apache.iceberg.mapping.NameMappingParser;
+import org.apache.iceberg.types.Types;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.util.Collections;
+import java.util.Optional;
+import java.util.concurrent.Callable;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+/** Exercises the public DROP entry points against real committed Iceberg
metadata. */
+public class IcebergColumnDropTest {
+ private InMemoryCatalog catalog;
+ private IcebergMetadataOps ops;
+ private ExternalCatalog dorisCatalog;
+ private ExecutionAuthenticator authenticator = new
ExecutionAuthenticator() { };
+
+ @Rule
+ public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ @Before
+ public void setUp() {
+ catalog = new InMemoryCatalog();
+ catalog.initialize("test", Collections.emptyMap());
+ catalog.createNamespace(Namespace.of("db"));
+ dorisCatalog = Mockito.mock(ExternalCatalog.class);
+
Mockito.when(dorisCatalog.getExecutionAuthenticator()).thenReturn(authenticator);
+
Mockito.when(dorisCatalog.getProperties()).thenReturn(Collections.emptyMap());
+
Mockito.doReturn(Optional.empty()).when(dorisCatalog).getDbForReplay(Mockito.anyString());
+ ops = new IcebergMetadataOps(dorisCatalog, catalog);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ catalog.close();
+ }
+
+ private Table createTable(String name, int format, boolean nested) {
+ Schema schema = nested ? new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "info", Types.StructType.of(
+ Types.NestedField.optional(3, "metric",
Types.IntegerType.get()),
+ Types.NestedField.optional(4, "other",
Types.IntegerType.get()))))
+ : new Schema(Types.NestedField.optional(1, "id",
Types.LongType.get()),
+ Types.NestedField.optional(2, "metric",
Types.IntegerType.get()),
+ Types.NestedField.optional(3, "other",
Types.IntegerType.get()));
+ return catalog.buildTable(TableIdentifier.of("db", name), schema)
+
.withPartitionSpec(PartitionSpec.builderFor(schema).identity(nested ?
"info.metric" : "metric").build())
+ .withProperty(TableProperties.FORMAT_VERSION,
Integer.toString(format)).create();
+ }
+
+ private Table createUnpartitionedTable(String name, int format) {
+ Schema schema = new Schema(Types.NestedField.optional(1, "id",
Types.LongType.get()),
+ Types.NestedField.optional(2, "metric",
Types.IntegerType.get()),
+ Types.NestedField.optional(3, "other",
Types.IntegerType.get()));
+ return catalog.buildTable(TableIdentifier.of("db", name), schema)
+ .withProperty(TableProperties.FORMAT_VERSION,
Integer.toString(format)).create();
+ }
+
+ private TableMetadata metadata(String name) {
+ return ((HasTableOperations)
catalog.loadTable(TableIdentifier.of("db", name))).operations().current();
+ }
+
+ private void drop(Table table, ColumnPath path) throws UserException {
+ ExternalTable dorisTable = Mockito.mock(ExternalTable.class);
+ Mockito.when(dorisTable.getRemoteDbName()).thenReturn("db");
+
Mockito.when(dorisTable.getRemoteName()).thenReturn(table.name().substring(table.name().lastIndexOf('.')
+ 1));
+ IcebergExternalMetaCache.WritableTableLease lease =
+
Mockito.mock(IcebergExternalMetaCache.WritableTableLease.class);
+
Mockito.when(lease.getRuntimeContext()).thenReturn(Mockito.mock(IcebergRuntimeContext.class));
+ Mockito.when(lease.getTable()).thenReturn(table);
+ Mockito.when(lease.getAuthenticator()).thenReturn(authenticator);
+ try (MockedStatic<IcebergUtils> utils =
Mockito.mockStatic(IcebergUtils.class, Mockito.CALLS_REAL_METHODS)) {
+ utils.when(() ->
IcebergUtils.acquireWritableIcebergTable(dorisTable, ops)).thenReturn(lease);
+ try {
+ if (path.isNested()) {
+ ops.dropColumn(dorisTable, path, 1L);
+ } else {
+ ops.dropColumn(dorisTable, path.getTopLevelName(), 1L);
+ }
+ } finally {
+ Mockito.verify(lease).close();
+ }
+ }
+ }
+
+ private void assertProtected(String name, Table table, ColumnPath path) {
+ String before = metadata(name).metadataFileLocation();
+ Assert.assertNotNull(before);
+ UserException error = Assert.assertThrows(UserException.class, () ->
drop(table, path));
+ Assert.assertTrue(error.getMessage(),
error.getMessage().contains("partition spec"));
+ Assert.assertEquals(before, metadata(name).metadataFileLocation());
+ }
+
+ @Test
+ public void testCurrentPartitionSource() {
+ Table table = createTable("current_source", 2, false);
+ assertProtected("current_source", table, ColumnPath.of("metric"));
+ }
+
+ @Test
+ public void testHistoricalSourceWithoutSnapshotsV2() {
+ assertFileFreeHistoricalSource(2);
+ }
+
+ @Test
+ public void testHistoricalSourceWithoutSnapshotsV3() {
+ assertFileFreeHistoricalSource(3);
+ }
+
+ private void assertFileFreeHistoricalSource(int format) {
+ Table table = createTable("empty", format, false);
+ table.updateSpec().removeField("metric").commit();
+ Assert.assertNull(table.currentSnapshot());
+ assertProtected("empty", table, ColumnPath.of("metric"));
+ }
+
+ @Test
+ public void testHistoricalSourceWithFile() {
+ Table table = createTable("file", 2, false);
+
table.newAppend().appendFile(DataFiles.builder(table.spec()).withPath("memory:/file/data.parquet")
+
.withFileSizeInBytes(1).withRecordCount(1).withPartitionPath("metric=5").build()).commit();
+ table.updateSpec().removeField("metric").commit();
+ assertProtected("file", table, ColumnPath.of("metric"));
+ }
+
+ @Test
+ public void testRenamedHistoricalSource() {
+ Table table = createTable("renamed", 2, false);
+ int sourceId = table.schema().findField("metric").fieldId();
+ table.updateSpec().removeField("metric").commit();
+ table.updateSchema().renameColumn("metric", "renamed_metric").commit();
+ Assert.assertEquals(sourceId,
table.schema().findField("renamed_metric").fieldId());
+ assertProtected("renamed", table, ColumnPath.of("RENAMED_METRIC"));
+ }
+
+ @Test
+ public void testNestedHistoricalSource() {
+ Table table = createTable("nested", 2, true);
+
table.updateSpec().removeField(table.spec().fields().get(0).name()).commit();
+ assertProtected("nested", table,
ColumnPath.fromDotName("INFO.METRIC"));
+ }
+
+ @Test
+ public void testParentOfHistoricalSource() {
+ Table table = createTable("parent", 2, true);
+
table.updateSpec().removeField(table.spec().fields().get(0).name()).commit();
+ assertProtected("parent", table, ColumnPath.of("info"));
+ }
+
+ @Test
+ public void testFormatV1VoidSource() {
+ Table table = createTable("void_source", 1, false);
+ table.updateSpec().removeField("metric").commit();
+ Assert.assertTrue(table.spec().fields().get(0).transform().isVoid());
+ assertProtected("void_source", table, ColumnPath.of("metric"));
+ }
+
+ @Test
+ public void testUnreferencedColumnRemainsDroppable() throws Exception {
+ Table table = createTable("ordinary", 2, false);
+ table.updateSpec().removeField("metric").commit();
+ drop(table, ColumnPath.of("other"));
+ Assert.assertNull(metadata("ordinary").schema().findField("other"));
+
Assert.assertNotNull(metadata("ordinary").schema().findField("metric"));
+ Assert.assertEquals(2, metadata("ordinary").specs().size());
+ }
+
+ @Test
+ public void testDirectDropRejectsConcurrentPartitionSource() {
+ Table table = createTable("direct_race", 2, false);
+ DirectRacingTableOperations racing = new
DirectRacingTableOperations(((HasTableOperations) table).operations());
+ Assert.assertThrows(UserException.class,
+ () -> drop(new BaseTable(racing, table.name()),
ColumnPath.of("other")));
+ Assert.assertEquals(1, racing.commitAttempts);
+
Assert.assertNotNull(metadata("direct_race").schema().findField("other"));
+ Assert.assertEquals(2, metadata("direct_race").specs().size());
+ }
+
+ @Test
+ public void testDirectDropAllowsExpiredPartitionSource() throws Exception {
+ Table table = createTable("direct_expired_source", 2, false);
+ table.updateSpec().removeField("metric").commit();
+ table.expireSnapshots().cleanExpiredMetadata(true).commit();
+ Assert.assertEquals(1,
metadata("direct_expired_source").specs().size());
+ drop(table, ColumnPath.of("metric"));
+
Assert.assertNull(metadata("direct_expired_source").schema().findField("metric"));
+ }
+
+ @Test
+ public void testDropRejectsRecreatedTableDuringRefresh() {
+ Table original = createTable("original", 2, true);
+ createTable("replacement", 2, true);
+ RefreshingTableOperations refreshing = new RefreshingTableOperations(
+ ((HasTableOperations) original).operations(),
+ ((HasTableOperations) original).operations().current(),
metadata("replacement"));
+ assertConcurrentTargetChange(original, refreshing,
ColumnPath.fromDotName("info.other"));
+ }
+
+ @Test
+ public void testDropRejectsSamePathFieldReplacement() {
+ Table table = createTable("field_replaced", 2, false);
+ TableOperations delegate = ((HasTableOperations) table).operations();
+ TableMetadata loaded = delegate.current();
+ table.updateSchema().deleteColumn("other").commit();
+ table.updateSchema().addColumn("other",
Types.IntegerType.get()).commit();
+ RefreshingTableOperations refreshing = new
RefreshingTableOperations(delegate, loaded, delegate.refresh());
+ assertConcurrentTargetChange(table, refreshing,
ColumnPath.of("other"));
+ }
+
+ @Test
+ public void testDropParentRejectsDescendantReplacement() {
+ Table table = createTable("subtree_replaced", 2, true);
+ TableOperations delegate = ((HasTableOperations) table).operations();
+ TableMetadata loaded = delegate.current();
+ table.updateSchema().deleteColumn("info.other").commit();
+ table.updateSchema().addColumn("info", "other",
Types.IntegerType.get()).commit();
+ RefreshingTableOperations refreshing = new
RefreshingTableOperations(delegate, loaded, delegate.refresh());
+ assertConcurrentTargetChange(table, refreshing, ColumnPath.of("info"));
+ }
+
+ private void assertConcurrentTargetChange(Table original,
RefreshingTableOperations refreshing, ColumnPath path) {
+ UserException error = Assert.assertThrows(UserException.class,
+ () -> drop(new BaseTable(refreshing, original.name()), path));
+ Assert.assertTrue(error.getMessage(),
error.getMessage().contains("changed concurrently"));
+ Assert.assertEquals(0, refreshing.commitAttempts);
+ }
+
+ @Test
+ public void testDropNestedRowLineageNameAndWriterProperties() throws
Exception {
+ Table table = createTable("nested_lineage", 3, true);
+ table.updateSchema().renameColumn("info.other", "_row_id").commit();
+ String metrics = TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX +
"info._row_id";
+ String bloom =
TableProperties.PARQUET_BLOOM_FILTER_COLUMN_ENABLED_PREFIX + "info._row_id";
+ String stats = TableProperties.PARQUET_COLUMN_STATS_ENABLED_PREFIX +
"info._row_id";
+ table.updateProperties().set(metrics, "full").set(bloom,
"true").set(stats, "true").commit();
+ drop(table, ColumnPath.fromDotName("info._row_id"));
+ TableMetadata after = metadata("nested_lineage");
+ Assert.assertNull(after.schema().findField("info._row_id"));
+ Assert.assertNotNull(after.schema().findField("info.metric"));
+ Assert.assertFalse(after.properties().containsKey(metrics));
+ Assert.assertFalse(after.properties().containsKey(bloom));
+ Assert.assertFalse(after.properties().containsKey(stats));
+ }
+
+ @Test
+ public void testRefreshAndCommitUseLeaseAuthenticator() throws Exception {
+ Table table = createTable("authenticated", 2, false);
+ AtomicBoolean authenticated = new AtomicBoolean();
+ authenticator = new ExecutionAuthenticator() {
+ @Override
+ public <T> T execute(Callable<T> task) throws Exception {
+ Assert.assertFalse(authenticated.getAndSet(true));
+ try {
+ return task.call();
+ } finally {
+ authenticated.set(false);
+ }
+ }
+ };
+ DelegatingTableOperations scoped = new
DelegatingTableOperations(((HasTableOperations) table).operations()) {
+ @Override
+ public TableMetadata current() {
+ Assert.assertTrue(authenticated.get());
+ return super.current();
+ }
+
+ @Override
+ public TableMetadata refresh() {
+ Assert.assertTrue(authenticated.get());
+ return super.refresh();
+ }
+
+ @Override
+ public void commit(TableMetadata base, TableMetadata next) {
+ Assert.assertTrue(authenticated.get());
+ super.commit(base, next);
+ }
+ };
+ drop(new BaseTable(scoped, table.name()), ColumnPath.of("other"));
+ Assert.assertFalse(authenticated.get());
+
Assert.assertNull(metadata("authenticated").schema().findField("other"));
+ }
+
+ @Test
+ public void testDropPinsValidatedBaseAfterRefreshReturns() {
+ Table table = createTable("post_refresh_replacement", 2, false);
+ int originalId = table.schema().findField("other").fieldId();
+ DelegatingTableOperations racing = new
DelegatingTableOperations(((HasTableOperations) table).operations()) {
+ @Override
+ public TableMetadata refresh() {
+ TableMetadata checked = delegate.refresh();
+ // A concurrent writer changes the shared handle after refresh
fetched its immutable result.
+ Table writer = new BaseTable(delegate, "replacement_writer");
+ writer.updateSchema().deleteColumn("other").commit();
+ writer.updateSchema().addColumn("other",
Types.IntegerType.get()).commit();
+ return checked;
+ }
+ };
+ Assert.assertThrows(UserException.class,
+ () -> drop(new BaseTable(racing, table.name()),
ColumnPath.of("other")));
+
Assert.assertNotNull(metadata("post_refresh_replacement").schema().findField("other"));
+ Assert.assertNotEquals(originalId,
+
metadata("post_refresh_replacement").schema().findField("other").fieldId());
+ }
+
+ @Test
+ public void testOnlyCurrentV1VoidSpecProtectsSource() {
+ Table table = createTable("current_void", 1, false);
+ int oldSpecId = table.spec().specId();
+ table.updateSpec().removeField("metric").commit();
+ TableOperations delegate = ((HasTableOperations) table).operations();
+ TableMetadata before = delegate.current();
+ TableMetadata.Builder cleaned = TableMetadata.buildFrom(before);
+ new
MetadataUpdate.RemovePartitionSpecs(Collections.singleton(oldSpecId)).applyTo(cleaned);
+ delegate.commit(before, cleaned.build());
+ Assert.assertEquals(1, metadata("current_void").specs().size());
+ assertProtected("current_void", table, ColumnPath.of("metric"));
+ }
+
+ @Test
+ public void testDirectV1DropDoesNotAllocatePartitionFieldId() throws
Exception {
+ Schema schema = new Schema(Types.NestedField.optional(1, "id",
Types.LongType.get()),
+ Types.NestedField.optional(2, "other",
Types.IntegerType.get()));
+ Table table = catalog.buildTable(TableIdentifier.of("db",
"v1_first_partition"), schema)
+ .withProperty(TableProperties.FORMAT_VERSION, "1").create();
+ drop(table, ColumnPath.of("other"));
+ table.updateSpec().addField("id").commit();
+ Assert.assertEquals(1000, table.spec().fields().get(0).fieldId());
+ }
+
+ @Test
+ public void testDirectDropWithOnlyCollectionRemaining() throws Exception {
+ Schema schema = new Schema(Types.NestedField.optional(1, "id",
Types.LongType.get()),
+ Types.NestedField.optional(2, "values",
Types.ListType.ofOptional(3, Types.IntegerType.get())));
+ Table table = catalog.buildTable(TableIdentifier.of("db",
"collection_remaining"), schema)
+ .withProperty(TableProperties.FORMAT_VERSION, "2").create();
+ drop(table, ColumnPath.of("id"));
+
Assert.assertNull(metadata("collection_remaining").schema().findField("id"));
+
Assert.assertNotNull(metadata("collection_remaining").schema().findField("values"));
+
Assert.assertTrue(metadata("collection_remaining").specs().stream().allMatch(spec
-> spec.fields().isEmpty()));
+ }
+
+ @Test
+ public void testDirectConsecutiveOrdinaryDrops() throws Exception {
+ Schema schema = new Schema(Types.NestedField.optional(1, "id",
Types.LongType.get()),
+ Types.NestedField.optional(2, "a", Types.LongType.get()),
+ Types.NestedField.optional(3, "b", Types.LongType.get()));
+ Table table = catalog.buildTable(TableIdentifier.of("db",
"consecutive_drops"), schema)
+ .withProperty(TableProperties.FORMAT_VERSION, "2").create();
+ drop(table, ColumnPath.of("b"));
+ table = catalog.loadTable(TableIdentifier.of("db",
"consecutive_drops"));
+ drop(table, ColumnPath.of("id"));
+ TableMetadata after = metadata("consecutive_drops");
+ Assert.assertNull(after.schema().findField("id"));
+ Assert.assertNull(after.schema().findField("b"));
+ Assert.assertNotNull(after.schema().findField("a"));
+ Assert.assertEquals(0, after.defaultSpecId());
+ }
+
+ @Test
+ public void testDropRevalidatesRowLineageAfterFormatUpgrade() {
+ Schema schema = new Schema(Types.NestedField.optional(1, "id",
Types.LongType.get()),
+ Types.NestedField.optional(2, "_row_id",
Types.LongType.get()));
+ Table table = catalog.buildTable(TableIdentifier.of("db",
"format_upgrade"), schema)
+ .withProperty(TableProperties.FORMAT_VERSION, "2").create();
+ AtomicInteger commits = new AtomicInteger();
+ DelegatingTableOperations upgrading = new
DelegatingTableOperations(((HasTableOperations) table).operations()) {
+ @Override
+ public TableMetadata refresh() {
+ new BaseTable(delegate, "upgrading_writer").updateProperties()
+ .set(TableProperties.FORMAT_VERSION, "3").commit();
+ return delegate.refresh();
+ }
+
+ @Override
+ public void commit(TableMetadata base, TableMetadata next) {
+ commits.incrementAndGet();
+ super.commit(base, next);
+ }
+ };
+ UserException error = Assert.assertThrows(UserException.class,
+ () -> drop(new BaseTable(upgrading, table.name()),
ColumnPath.of("_row_id")));
+ Assert.assertTrue(error.getMessage(), error.getMessage().contains("row
lineage"));
+ Assert.assertEquals(0, commits.get());
+ Assert.assertEquals(3, metadata("format_upgrade").formatVersion());
+
Assert.assertNotNull(metadata("format_upgrade").schema().findField("_row_id"));
+ }
+
+ @Test
+ public void testDirectDropCleansWriterProperties() throws Exception {
+ Table table = createTable("direct_properties", 2, false);
+ String property = TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX +
"other";
+ int historicalOtherId = table.schema().findField("other").fieldId();
+ table.updateProperties().set(property, "full")
+ .set(TableProperties.DEFAULT_NAME_MAPPING,
NameMappingParser.toJson(MappingUtil.create(table.schema())))
+ .commit();
+ drop(table, ColumnPath.of("other"));
+ TableMetadata after = metadata("direct_properties");
+ Assert.assertFalse(after.properties().containsKey(property));
+ Assert.assertEquals(Integer.valueOf(historicalOtherId),
NameMappingParser.fromJson(
+
after.properties().get(TableProperties.DEFAULT_NAME_MAPPING)).find("other").id());
+ }
+
+ private Table createLowIdPartitionHistory(String name) {
+ Table table = createUnpartitionedTable(name, 2);
+ TableOperations operations = ((HasTableOperations) table).operations();
+ TableMetadata base = operations.current();
+ PartitionSpec historical = PartitionSpecParser.fromJson(base.schema(),
"{\"spec-id\":1,\"fields\":["
+ +
"{\"name\":\"metric\",\"transform\":\"identity\",\"source-id\":2,\"field-id\":10}]}");
+ PartitionSpec current = PartitionSpecParser.fromJson(base.schema(),
"{\"spec-id\":2,\"fields\":["
+ +
"{\"name\":\"other\",\"transform\":\"identity\",\"source-id\":3,\"field-id\":11}]}");
+ operations.commit(base,
TableMetadata.buildFrom(base).setDefaultPartitionSpec(historical)
+ .setDefaultPartitionSpec(current).build());
+ table.refresh();
+ Assert.assertEquals(999, metadata(name).lastAssignedPartitionId());
+ return table;
+ }
+
+ @Test
+ public void testLowIdHistoricalSourceIsProtected() {
+ Table table = createLowIdPartitionHistory("low_id_history");
+ assertProtected("low_id_history", table, ColumnPath.of("metric"));
+ }
+
+ @Test
+ public void testLowIdUnreferencedColumnRemainsDroppable() throws Exception
{
+ Table table = createLowIdPartitionHistory("low_id_ordinary");
+ drop(table, ColumnPath.of("id"));
+
Assert.assertNull(metadata("low_id_ordinary").schema().findField("id"));
+ Assert.assertEquals(999,
metadata("low_id_ordinary").lastAssignedPartitionId());
+ }
+
+ @Test
+ public void testDirectCasRejectsLowIdSourceRestoredAfterExpiry() {
+ assertLowIdRestoreRejected(false);
+ }
+
+ @Test
+ public void testDirectCasRejectsLowIdRestoreOnNowUnpartitionedTable() {
+ assertLowIdRestoreRejected(true);
+ }
+
+ private void assertLowIdRestoreRejected(boolean unpartitioned) {
+ String name = "low_id_restore_" + unpartitioned;
+ Table table = createLowIdPartitionHistory(name);
+ PartitionSpec staleSpec =
table.updateSpec().addNonDefaultSpec().addField("metric").apply();
+ if (unpartitioned) {
+ table.updateSpec().removeField("other").commit();
+ }
+ table.expireSnapshots().cleanExpiredMetadata(true).commit();
+ Assert.assertEquals(1, metadata(name).specs().size());
+ Assert.assertEquals(999, metadata(name).lastAssignedPartitionId());
+ DelegatingTableOperations racing = new
DelegatingTableOperations(((HasTableOperations) table).operations()) {
+ @Override
+ public void commit(TableMetadata base, TableMetadata next) {
+ // Publish the spec prepared before expiry without increasing
the partition counter.
+ TableMetadata concurrent = delegate.refresh();
+ delegate.commit(concurrent,
TableMetadata.buildFrom(concurrent).addPartitionSpec(staleSpec).build());
+ Assert.assertEquals(999,
delegate.refresh().lastAssignedPartitionId());
+ super.commit(base, next);
+ }
+ };
+ Assert.assertThrows(UserException.class,
+ () -> drop(new BaseTable(racing, table.name()),
ColumnPath.of("metric")));
+ Assert.assertNotNull(metadata(name).schema().findField("metric"));
+ Assert.assertTrue(metadata(name).specs().stream().flatMap(spec ->
spec.fields().stream())
+ .anyMatch(field -> field.sourceId() == 2));
+ }
+
+ @Test
+ public void testHadoopCatalogHistoricalSourceAndOrdinaryDrop() throws
Exception {
+ try (HadoopCatalog hadoop = new HadoopCatalog(new Configuration(),
+
temporaryFolder.newFolder("hadoop-history").toURI().toString())) {
+ hadoop.createNamespace(Namespace.of("db"));
+ Schema schema = new Schema(Types.NestedField.optional(1, "id",
Types.LongType.get()),
+ Types.NestedField.optional(2, "metric",
Types.IntegerType.get()),
+ Types.NestedField.optional(3, "other",
Types.IntegerType.get()));
+ Table table = hadoop.createTable(TableIdentifier.of("db",
"history"), schema,
+
PartitionSpec.builderFor(schema).identity("metric").build());
+ table.updateSpec().removeField("metric").commit();
+ ops = new IcebergMetadataOps(dorisCatalog, hadoop);
+ String before = ((HasTableOperations)
table).operations().refresh().metadataFileLocation();
+ UserException error = Assert.assertThrows(UserException.class,
+ () -> drop(table, ColumnPath.of("metric")));
+ Assert.assertTrue(error.getMessage(),
error.getMessage().contains("partition spec"));
+ Assert.assertEquals(before, ((HasTableOperations)
table).operations().refresh().metadataFileLocation());
+ drop(table, ColumnPath.of("other"));
+ table.refresh();
+ Assert.assertNull(table.schema().findField("other"));
+ Assert.assertNotNull(table.schema().findField("metric"));
+ }
+ }
+
+ @Test
+ public void testHadoopRecreatedTableWithReusedMetadataVersion() throws
Exception {
+ assertHadoopRecreatedTableRejected(false);
+ }
+
+ @Test
+ public void testHadoopTableRecreatedDuringDropRefresh() throws Exception {
+ assertHadoopRecreatedTableRejected(true);
+ }
+
+ private void assertHadoopRecreatedTableRejected(boolean duringRefresh)
throws Exception {
+ try (HadoopCatalog hadoop = new HadoopCatalog(new Configuration(),
+
temporaryFolder.newFolder("hadoop-recreated").toURI().toString())) {
+ hadoop.createNamespace(Namespace.of("db"));
+ TableIdentifier identifier = TableIdentifier.of("db", "recreated");
+ Schema schema = new Schema(Types.NestedField.optional(1, "id",
Types.LongType.get()),
+ Types.NestedField.optional(2, "other",
Types.IntegerType.get()));
+ Table original = hadoop.createTable(identifier, schema);
+ TableOperations originalOperations = ((HasTableOperations)
original).operations();
+ TableMetadata loaded = originalOperations.current();
+ AtomicReference<TableMetadata> replacement = new
AtomicReference<>();
+ Runnable recreate = () -> {
+ Assert.assertTrue(hadoop.dropTable(identifier, true));
+ TableMetadata current = ((HasTableOperations)
hadoop.createTable(identifier, schema))
+ .operations().current();
+ Assert.assertNotEquals(loaded.uuid(), current.uuid());
+ Assert.assertEquals(loaded.metadataFileLocation(),
current.metadataFileLocation());
+ replacement.set(current);
+ };
+ Table retained = original;
+ if (duringRefresh) {
+ retained = new BaseTable(new
DelegatingTableOperations(originalOperations) {
+ @Override
+ public TableMetadata refresh() {
+ TableMetadata cached = delegate.refresh();
+ recreate.run();
+ return cached;
+ }
+ }, original.name());
+ } else {
+ recreate.run();
+ }
+ ops = new IcebergMetadataOps(dorisCatalog, hadoop);
+ Table target = retained;
+ UserException error = Assert.assertThrows(UserException.class,
+ () -> drop(target, ColumnPath.of("other")));
+ Assert.assertTrue(error.getMessage(),
error.getMessage().contains("changed concurrently"));
+ Table current = hadoop.loadTable(identifier);
+ Assert.assertEquals(replacement.get().uuid(),
((HasTableOperations) current).operations().current().uuid());
+ Assert.assertNotNull(current.schema().findField("other"));
+ }
+ }
+
+ private static class DelegatingTableOperations implements TableOperations {
+ protected final TableOperations delegate;
+
+ private DelegatingTableOperations(TableOperations delegate) {
+ this.delegate = delegate;
+ }
+
+ @Override
+ public TableMetadata current() {
+ return delegate.current();
+ }
+
+ @Override
+ public TableMetadata refresh() {
+ return delegate.refresh();
+ }
+
+ @Override
+ public void commit(TableMetadata base, TableMetadata next) {
+ delegate.commit(base, next);
+ }
+
+ @Override
+ public FileIO io() {
+ return delegate.io();
+ }
+
+ @Override
+ public EncryptionManager encryption() {
+ return delegate.encryption();
+ }
+
+ @Override
+ public String metadataFileLocation(String name) {
+ return delegate.metadataFileLocation(name);
+ }
+
+ @Override
+ public LocationProvider locationProvider() {
+ return delegate.locationProvider();
+ }
+ }
+
+ private static final class DirectRacingTableOperations extends
DelegatingTableOperations {
+ private int commitAttempts;
+
+ private DirectRacingTableOperations(TableOperations delegate) {
+ super(delegate);
+ }
+
+ @Override
+ public void commit(TableMetadata base, TableMetadata next) {
+ commitAttempts++;
+ new BaseTable(delegate,
"concurrent").updateSpec().addField("other").commit();
+ delegate.commit(base, next);
+ }
+ }
+
+ private static final class RefreshingTableOperations extends
DelegatingTableOperations {
+ private final TableMetadata loaded;
+ private final TableMetadata refreshed;
+ private boolean didRefresh;
+ private int commitAttempts;
+
+ private RefreshingTableOperations(TableOperations delegate,
TableMetadata loaded, TableMetadata refreshed) {
+ super(delegate);
+ this.loaded = loaded;
+ this.refreshed = refreshed;
+ }
+
+ @Override
+ public TableMetadata current() {
+ return didRefresh ? refreshed : loaded;
+ }
+
+ @Override
+ public TableMetadata refresh() {
+ didRefresh = true;
+ return refreshed;
+ }
+
+ @Override
+ public void commit(TableMetadata base, TableMetadata next) {
+ commitAttempts++;
+ Assert.fail("Changed drop target must never commit");
+ }
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergMetadataOpsValidationTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergMetadataOpsValidationTest.java
index 6d1b0b06a40..7431d89471a 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergMetadataOpsValidationTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergMetadataOpsValidationTest.java
@@ -1503,13 +1503,13 @@ public class IcebergMetadataOpsValidationTest {
ops.modifyColumn(v3DorisTable, ColumnPath.fromDotName("s._row_id"),
new Column("_row_id", Type.BIGINT, true), null, 1L);
ops.modifyColumnComment(v3DorisTable,
ColumnPath.fromDotName("s._row_id"), "comment", 1L);
- ops.dropColumn(v3DorisTable, ColumnPath.fromDotName("s._row_id"),
1L);
+ // DROP is covered with real TableOperations in
IcebergColumnDropTest.
ops.addColumn(v2DorisTable, new Column("_row_id", Type.BIGINT,
true), null, 1L);
ops.renameColumn(v2DorisTable, "id",
"_last_updated_sequence_number", 1L);
}
- Mockito.verify(v3UpdateSchema, Mockito.times(5)).commit();
+ Mockito.verify(v3UpdateSchema, Mockito.times(4)).commit();
Mockito.verify(v2UpdateSchema, Mockito.times(2)).commit();
}
@@ -1599,6 +1599,7 @@ public class IcebergMetadataOpsValidationTest {
Mockito.mock(IcebergExternalMetaCache.WritableTableLease.class);
Mockito.when(lease.getTable()).thenReturn(icebergTable);
Mockito.when(lease.getAuthenticator()).thenReturn(authenticator);
+
Mockito.when(lease.getRuntimeContext()).thenReturn(Mockito.mock(IcebergRuntimeContext.class));
mockedIcebergUtils.when(() ->
IcebergUtils.acquireWritableIcebergTable(dorisTable, expectedOps))
.thenReturn(lease);
return lease;
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergRestColumnDropCompatibilityTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergRestColumnDropCompatibilityTest.java
new file mode 100644
index 00000000000..e92fd2db234
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergRestColumnDropCompatibilityTest.java
@@ -0,0 +1,210 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.datasource.iceberg;
+
+import org.apache.doris.analysis.ColumnPath;
+import org.apache.doris.common.security.authentication.ExecutionAuthenticator;
+import org.apache.doris.datasource.ExternalCatalog;
+import org.apache.doris.datasource.ExternalTable;
+import org.apache.doris.datasource.property.metastore.IcebergRestProperties;
+import
org.apache.doris.datasource.property.metastore.IcebergRestProperties.DelegatedTokenMode;
+
+import com.google.common.io.ByteStreams;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+import org.apache.iceberg.CatalogProperties;
+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.catalog.TableIdentifier;
+import org.apache.iceberg.inmemory.InMemoryCatalog;
+import org.apache.iceberg.inmemory.InMemoryFileIO;
+import org.apache.iceberg.rest.Endpoint;
+import org.apache.iceberg.rest.RESTSessionCatalog;
+import org.apache.iceberg.rest.auth.OAuth2Properties;
+import org.apache.iceberg.rest.requests.UpdateTableRequest;
+import org.apache.iceberg.rest.requests.UpdateTableRequestParser;
+import org.apache.iceberg.rest.responses.ConfigResponse;
+import org.apache.iceberg.rest.responses.ConfigResponseParser;
+import org.apache.iceberg.rest.responses.LoadTableResponse;
+import org.apache.iceberg.rest.responses.LoadTableResponseParser;
+import org.apache.iceberg.types.Types;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+
+/** Keeps REST DROP on the baseline SDK table endpoint and authentication
path. */
+public class IcebergRestColumnDropCompatibilityTest {
+ private static final TableIdentifier ID = TableIdentifier.of("db",
"rest_drop_compatibility");
+ private final AtomicReference<Throwable> serverFailure = new
AtomicReference<>();
+ private InMemoryCatalog backend;
+ private RESTSessionCatalog rest;
+ private HttpServer server;
+ private boolean tableToken;
+ private int tablePosts;
+ private String tableAuth;
+
+ @Before
+ public void setUp() throws Exception {
+ backend = new InMemoryCatalog();
+ backend.initialize("compatibility", Collections.emptyMap());
+ backend.createNamespace(ID.namespace());
+ backend.createTable(ID, new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "metric",
Types.IntegerType.get()),
+ Types.NestedField.optional(3, "other",
Types.IntegerType.get())));
+ server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext("/", this::handle);
+ server.start();
+ Map<String, String> properties = new HashMap<>();
+ properties.put(CatalogProperties.URI, "http://127.0.0.1:"; +
server.getAddress().getPort());
+ properties.put(CatalogProperties.FILE_IO_IMPL,
InMemoryFileIO.class.getName());
+ properties.put("rest.auth.type", "oauth2");
+ properties.put(OAuth2Properties.TOKEN, "catalog-token");
+ properties.put(OAuth2Properties.TOKEN_REFRESH_ENABLED, "false");
+ rest = new RESTSessionCatalog();
+ rest.initialize("compatibility", properties);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ rest.close();
+ server.stop(0);
+ backend.close();
+ Assert.assertNull(serverFailure.get());
+ }
+
+ private TableOperations backendOps() {
+ return ((HasTableOperations) backend.loadTable(ID)).operations();
+ }
+
+ private Table drop(ColumnPath path) throws Exception {
+ org.apache.iceberg.catalog.Catalog catalog = rest.asCatalog(
+
org.apache.iceberg.catalog.SessionCatalog.SessionContext.createEmpty());
+ Table table = catalog.loadTable(ID);
+ ExternalCatalog dorisCatalog = Mockito.mock(ExternalCatalog.class,
+
Mockito.withSettings().extraInterfaces(IcebergUserSessionCatalog.class));
+ IcebergUserSessionCatalog capability = (IcebergUserSessionCatalog)
dorisCatalog;
+ Mockito.when(capability.getRestSessionCatalog()).thenReturn(rest);
+
Mockito.when(capability.getDelegatedTokenMode()).thenReturn(DelegatedTokenMode.ACCESS_TOKEN);
+ ExecutionAuthenticator authenticator = new ExecutionAuthenticator() {
};
+
Mockito.when(dorisCatalog.getExecutionAuthenticator()).thenReturn(authenticator);
+
Mockito.when(dorisCatalog.getProperties()).thenReturn(Collections.emptyMap());
+
Mockito.doReturn(Optional.empty()).when(dorisCatalog).getDbForReplay(Mockito.anyString());
+ IcebergMetadataOps ops = new IcebergMetadataOps(dorisCatalog, catalog);
+ ExternalTable dorisTable = Mockito.mock(ExternalTable.class);
+ Mockito.when(dorisTable.getRemoteDbName()).thenReturn("db");
+ Mockito.when(dorisTable.getRemoteName()).thenReturn(ID.name());
+ IcebergExternalMetaCache.WritableTableLease lease =
+
Mockito.mock(IcebergExternalMetaCache.WritableTableLease.class);
+ Mockito.when(lease.getTable()).thenReturn(table);
+ IcebergRuntimeContext runtime =
Mockito.mock(IcebergRuntimeContext.class);
+
Mockito.when(runtime.getMetastoreProperties()).thenReturn(Mockito.mock(IcebergRestProperties.class));
+ Mockito.when(lease.getRuntimeContext()).thenReturn(runtime);
+ Mockito.when(lease.getAuthenticator()).thenReturn(authenticator);
+ try (MockedStatic<IcebergUtils> utils =
Mockito.mockStatic(IcebergUtils.class, Mockito.CALLS_REAL_METHODS)) {
+ utils.when(() ->
IcebergUtils.acquireWritableIcebergTable(dorisTable, ops)).thenReturn(lease);
+ ops.dropColumn(dorisTable, path, 1L);
+ }
+ Mockito.verify(lease).close();
+ return table;
+ }
+
+ @Test
+ public void testUnpartitionedDropDoesNotRequireTransactionEndpoint()
throws Exception {
+ Table table = drop(ColumnPath.of("other"));
+ Assert.assertNull(table.schema().findField("other"));
+ Assert.assertNull(backendOps().refresh().schema().findField("other"));
+ Assert.assertEquals(1, tablePosts);
+ }
+
+ @Test
+ public void testPartitionedTableAllowsUnreferencedColumn() throws
Exception {
+ backend.loadTable(ID).updateSpec().addField("metric").commit();
+ drop(ColumnPath.of("other"));
+ Assert.assertNull(backendOps().refresh().schema().findField("other"));
+
Assert.assertNotNull(backendOps().current().schema().findField("metric"));
+ Assert.assertEquals(1, tablePosts);
+ }
+
+ @Test
+ public void testLoadedTableCredentialsArePreserved() throws Exception {
+ tableToken = true;
+ drop(ColumnPath.of("other"));
+ Assert.assertEquals("Bearer table-token", tableAuth);
+ Assert.assertNull(backendOps().refresh().schema().findField("other"));
+ }
+
+ private void handle(HttpExchange exchange) throws IOException {
+ try {
+ String path = exchange.getRequestURI().getPath();
+ if (path.equals("/v1/config")) {
+ send(exchange,
ConfigResponseParser.toJson(ConfigResponse.builder().withEndpoints(
+ Arrays.asList(Endpoint.V1_LOAD_TABLE,
Endpoint.V1_UPDATE_TABLE)).build()));
+ return;
+ }
+ Assert.assertTrue(path, path.endsWith("/namespaces/db/tables/" +
ID.name()));
+ if (exchange.getRequestMethod().equals("POST")) {
+ tablePosts++;
+ tableAuth =
exchange.getRequestHeaders().getFirst("Authorization");
+ UpdateTableRequest request = UpdateTableRequestParser.fromJson(
+ new
String(ByteStreams.toByteArray(exchange.getRequestBody()),
StandardCharsets.UTF_8));
+ TableOperations operations = backendOps();
+ TableMetadata base = operations.refresh();
+ request.requirements().forEach(requirement ->
requirement.validate(base));
+ TableMetadata.Builder updated = TableMetadata.buildFrom(base);
+ request.updates().forEach(update -> update.applyTo(updated));
+ operations.commit(base, updated.build());
+ }
+ LoadTableResponse.Builder response = LoadTableResponse.builder()
+ .withTableMetadata(backendOps().refresh());
+ if (tableToken) {
+ response.addConfig(OAuth2Properties.TOKEN, "table-token");
+ response.addConfig(OAuth2Properties.TOKEN_REFRESH_ENABLED,
"false");
+ }
+ send(exchange, LoadTableResponseParser.toJson(response.build()));
+ } catch (Throwable failure) {
+ serverFailure.compareAndSet(null, failure);
+ exchange.sendResponseHeaders(500, -1);
+ } finally {
+ exchange.close();
+ }
+ }
+
+ private void send(HttpExchange exchange, String body) throws IOException {
+ byte[] bytes = body.getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().set("Content-Type", "application/json");
+ exchange.sendResponseHeaders(200, bytes.length);
+ exchange.getResponseBody().write(bytes);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
