This is an automated email from the ASF dual-hosted git repository.
zhoujinsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/amoro.git
The following commit(s) were added to refs/heads/master by this push:
new e697eb3a3 [AMORO-4375][Iceberg] Prevent automatic changes to
unsupported V3 tables (#4376)
e697eb3a3 is described below
commit e697eb3a32f777b0435230fb405a3861fe383e8e
Author: Jiwon Park <[email protected]>
AuthorDate: Mon Sep 14 20:04:32 2026 +0900
[AMORO-4375][Iceberg] Prevent automatic changes to unsupported V3 tables
(#4376)
* [AMORO-4375][Iceberg] Prevent automatic changes to unsupported V3 tables
Signed-off-by: Jiwon Park <[email protected]>
* [AMORO-4375] Move optimizing close overloads to interface defaults
Signed-off-by: Jiwon Park <[email protected]>
---------
Signed-off-by: Jiwon Park <[email protected]>
---
.../amoro/server/optimizing/OptimizingProcess.java | 11 +-
.../amoro/server/optimizing/OptimizingQueue.java | 39 +-
.../server/optimizing/UnKeyedTableCommit.java | 4 +-
.../inline/TableRuntimeRefreshExecutor.java | 21 +
.../amoro/server/table/DefaultTableRuntime.java | 9 +
.../TestIcebergV3OptimizingPlanning.java | 445 +++++++++++++++++++++
.../server/optimizing/TestOptimizingProcess.java} | 43 +-
.../commit/TestIcebergV3OptimizingCommit.java | 356 +++++++++++++++++
.../iceberg/TestIcebergV3MaintenanceRecovery.java | 125 ++++++
.../iceberg/IcebergMaintenanceCompatibility.java | 159 ++++++++
.../iceberg/maintainer/IcebergTableMaintainer.java | 10 +
.../iceberg/IcebergV3MaintenanceFixture.java | 245 ++++++++++++
.../amoro/formats/iceberg/IcebergV3TestTables.java | 170 ++++++++
.../maintainer/TestIcebergV3MaintenanceGuard.java | 188 +++++++++
docs/formats/iceberg.md | 21 +-
15 files changed, 1818 insertions(+), 28 deletions(-)
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingProcess.java
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingProcess.java
index b088f6f27..eb3eec04a 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingProcess.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingProcess.java
@@ -28,7 +28,16 @@ public interface OptimizingProcess {
long getProcessId();
- void close(boolean needCommit);
+ void close(boolean needCommit, String reason);
+
+ default void close(boolean needCommit) {
+ close(needCommit, null);
+ }
+
+ /** Close without committing and persist the reason in the process history.
*/
+ default void close(String reason) {
+ close(false, reason);
+ }
boolean isClosed();
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
index c090979a2..2f7ba59c1 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
@@ -21,12 +21,15 @@ package org.apache.amoro.server.optimizing;
import org.apache.amoro.AmoroTable;
import org.apache.amoro.OptimizerProperties;
import org.apache.amoro.ServerTableIdentifier;
+import org.apache.amoro.TableFormat;
import org.apache.amoro.api.BlockableOperation;
import org.apache.amoro.api.OptimizingTaskId;
import org.apache.amoro.api.OptimizingTaskResult;
import org.apache.amoro.exception.OptimizingClosedException;
import org.apache.amoro.exception.PersistenceException;
import org.apache.amoro.exception.TaskNotFoundException;
+import org.apache.amoro.formats.iceberg.IcebergMaintenanceCompatibility;
+import
org.apache.amoro.formats.iceberg.IcebergMaintenanceCompatibility.UnsupportedTableException;
import org.apache.amoro.optimizing.MetricsSummary;
import org.apache.amoro.optimizing.OptimizingType;
import org.apache.amoro.optimizing.RewriteFilesInput;
@@ -64,6 +67,7 @@ import org.apache.amoro.utils.MixedDataFiles;
import org.apache.amoro.utils.TablePropertyUtil;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
import org.apache.iceberg.util.StructLikeMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -138,6 +142,29 @@ public class OptimizingQueue extends PersistentBase {
return;
}
+ if (process != null && tableRuntime.getFormat() == TableFormat.ICEBERG) {
+ AmoroTable<?> table = null;
+ try {
+ table =
catalogManager.loadTable(tableRuntime.getTableIdentifier().getIdentifier());
+ } catch (Exception e) {
+ // Catalog availability must not prevent recovery. The writer checks
again at commit.
+ LOG.warn(
+ "Cannot check table version during recovery of {}; continuing
recovery",
+ tableRuntime.getTableIdentifier(),
+ e);
+ }
+ if (table != null) {
+ try {
+ IcebergMaintenanceCompatibility.checkSupported((Table)
table.originalTable());
+ } catch (UnsupportedTableException e) {
+ process.close(e.getMessage());
+ tableRuntime.suspendUnsupportedOptimizing();
+ LOG.warn("Not resuming automatic optimizing: {}", e.getMessage());
+ return;
+ }
+ }
+ }
+
tableRuntime.resetTaskQuotas(
System.currentTimeMillis() -
AmoroServiceConstants.QUOTA_LOOK_BACK_TIME);
@@ -382,6 +409,9 @@ public class OptimizingQueue extends PersistentBase {
try {
ServerTableIdentifier identifier = tableRuntime.getTableIdentifier();
AmoroTable<?> table =
catalogManager.loadTable(identifier.getIdentifier());
+ if (table.format() == TableFormat.ICEBERG) {
+ IcebergMaintenanceCompatibility.checkSupported((Table)
table.originalTable());
+ }
AbstractOptimizingPlanner planner =
IcebergTableUtil.createOptimizingPlanner(
tableRuntime.refresh(table),
@@ -394,6 +424,10 @@ public class OptimizingQueue extends PersistentBase {
tableRuntime.completeEmptyProcess();
return null;
}
+ } catch (UnsupportedTableException e) {
+ tableRuntime.suspendUnsupportedOptimizing();
+ LOG.warn("Skipping automatic optimizing: {}", e.getMessage());
+ return null;
} catch (Throwable throwable) {
tableRuntime.planFailed();
LOG.error("Planning table {} failed", tableRuntime.getTableIdentifier(),
throwable);
@@ -654,7 +688,7 @@ public class OptimizingQueue extends PersistentBase {
}
@Override
- public void close(boolean needCommit) {
+ public void close(boolean needCommit, String reason) {
lock.lock();
try {
if (this.status != ProcessStatus.RUNNING) {
@@ -663,6 +697,9 @@ public class OptimizingQueue extends PersistentBase {
if (tableRuntime.isAllowPartialCommit() && needCommit) {
tableRuntime.beginCommitting();
} else {
+ if (reason != null) {
+ this.failedReason = reason;
+ }
this.status = ProcessStatus.CLOSED;
this.endTime = System.currentTimeMillis();
persistAndSetCompleted(false);
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/UnKeyedTableCommit.java
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/UnKeyedTableCommit.java
index d71ae484b..878fd2000 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/UnKeyedTableCommit.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/UnKeyedTableCommit.java
@@ -24,6 +24,7 @@ import static
org.apache.amoro.hive.op.UpdateHiveFiles.SYNC_DATA_TO_HIVE;
import org.apache.amoro.api.CommitMetaProducer;
import org.apache.amoro.data.FileNameRules;
import org.apache.amoro.exception.OptimizingCommitException;
+import org.apache.amoro.formats.iceberg.IcebergMaintenanceCompatibility;
import org.apache.amoro.hive.HMSClientPool;
import org.apache.amoro.hive.table.SupportHive;
import org.apache.amoro.hive.utils.HivePartitionUtil;
@@ -250,7 +251,8 @@ public class UnKeyedTableCommit {
}
});
try {
- Transaction transaction = table.asUnkeyedTable().newTransaction();
+ Transaction transaction =
+
IcebergMaintenanceCompatibility.forUpdate(table.asUnkeyedTable()).newTransaction();
if (removedDeleteFiles.isEmpty() && !addedDeleteFiles.isEmpty()) {
/* In order to avoid the validation in
{@link
org.apache.iceberg.BaseRewriteFiles#validateReplacedAndAddedFiles} which will
throw
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/TableRuntimeRefreshExecutor.java
b/amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/TableRuntimeRefreshExecutor.java
index ae9a1ea1d..86401e414 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/TableRuntimeRefreshExecutor.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/TableRuntimeRefreshExecutor.java
@@ -19,9 +19,12 @@
package org.apache.amoro.server.scheduler.inline;
import org.apache.amoro.AmoroTable;
+import org.apache.amoro.TableFormat;
import org.apache.amoro.TableRuntime;
import org.apache.amoro.config.OptimizingConfig;
import org.apache.amoro.config.TableConfiguration;
+import org.apache.amoro.formats.iceberg.IcebergMaintenanceCompatibility;
+import
org.apache.amoro.formats.iceberg.IcebergMaintenanceCompatibility.UnsupportedTableException;
import org.apache.amoro.optimizing.evaluation.MetadataBasedEvaluationEvent;
import org.apache.amoro.optimizing.plan.AbstractOptimizingEvaluator;
import org.apache.amoro.process.ProcessStatus;
@@ -156,6 +159,24 @@ public class TableRuntimeRefreshExecutor extends
PeriodicTableScheduler {
AmoroTable<?> table = loadTable(tableRuntime);
defaultTableRuntime.refresh(table);
MixedTable mixedTable = (MixedTable) table.originalTable();
+ if (table.format() == TableFormat.ICEBERG) {
+ try {
+
IcebergMaintenanceCompatibility.checkSupported(mixedTable.asUnkeyedTable());
+ } catch (UnsupportedTableException e) {
+ OptimizingProcess process =
defaultTableRuntime.getOptimizingProcess();
+ if (process != null && process.getStatus() == ProcessStatus.RUNNING)
{
+ process.close(e.getMessage());
+ }
+ defaultTableRuntime.suspendUnsupportedOptimizing();
+ defaultTableRuntime.setLatestEvaluatedNeedOptimizing(false);
+ logger.warn("Skipping automatic optimizing: {}", e.getMessage());
+ OptimizingConfig config = defaultTableRuntime.getOptimizingConfig();
+ if (!config.isEnabled() && config.isTableSummaryEnabled()) {
+ tryEvaluatingPendingInput(defaultTableRuntime, mixedTable);
+ }
+ return;
+ }
+ }
boolean snapshotChanged =
(mixedTable.isKeyedTable()
&& (lastOptimizedSnapshotId !=
defaultTableRuntime.getCurrentSnapshotId()
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java
b/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java
index fe8813709..db5576b8b 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java
@@ -353,6 +353,15 @@ public class DefaultTableRuntime extends
AbstractTableRuntime {
store().begin().updateStatusCode(code ->
OptimizingStatus.PENDING.getCode()).commit();
}
+ /** Stop scheduling unsupported metadata without marking its snapshot as
optimized. */
+ public void suspendUnsupportedOptimizing() {
+ store()
+ .begin()
+ .updateStatusCode(code -> OptimizingStatus.IDLE.getCode())
+ .updateState(PENDING_INPUT_KEY, any -> new
AbstractOptimizingEvaluator.PendingInput())
+ .commit();
+ }
+
public void beginProcess(OptimizingProcess optimizingProcess) {
this.optimizingProcess = optimizingProcess;
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestIcebergV3OptimizingPlanning.java
b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestIcebergV3OptimizingPlanning.java
new file mode 100644
index 000000000..6bad7f4db
--- /dev/null
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestIcebergV3OptimizingPlanning.java
@@ -0,0 +1,445 @@
+/*
+ * 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.amoro.server.optimizing;
+
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+import org.apache.amoro.AmoroTable;
+import org.apache.amoro.ServerTableIdentifier;
+import org.apache.amoro.TableFormat;
+import org.apache.amoro.api.OptimizingTaskResult;
+import org.apache.amoro.formats.iceberg.IcebergTable;
+import org.apache.amoro.formats.iceberg.IcebergV3TestTables;
+import org.apache.amoro.io.IcebergDataTestHelpers;
+import org.apache.amoro.optimizing.RewriteFilesOutput;
+import org.apache.amoro.optimizing.RewriteStageTask;
+import org.apache.amoro.process.ProcessStatus;
+import org.apache.amoro.resource.ResourceGroup;
+import org.apache.amoro.server.catalog.CatalogManager;
+import org.apache.amoro.server.persistence.PersistentBase;
+import org.apache.amoro.server.persistence.TableRuntimeMeta;
+import org.apache.amoro.server.persistence.mapper.TableMetaMapper;
+import org.apache.amoro.server.persistence.mapper.TableProcessMapper;
+import org.apache.amoro.server.persistence.mapper.TableRuntimeMapper;
+import org.apache.amoro.server.process.TableProcessMeta;
+import org.apache.amoro.server.resource.OptimizerThread;
+import org.apache.amoro.server.scheduler.inline.TableRuntimeRefreshExecutor;
+import org.apache.amoro.server.table.DefaultTableRuntime;
+import org.apache.amoro.server.table.DefaultTableRuntimeStore;
+import org.apache.amoro.server.table.DerbyPersistence;
+import org.apache.amoro.server.table.TableService;
+import org.apache.amoro.table.TableMetaStore;
+import org.apache.amoro.table.TableProperties;
+import org.apache.amoro.table.UnkeyedTable;
+import org.apache.amoro.utils.SerializationUtil;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.DeleteFile;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.stream.Collectors;
+
+class TestIcebergV3OptimizingPlanning extends PersistentBase {
+ @TempDir private Path temp;
+
+ private final TestPersistence persistence = new TestPersistence();
+ private final CatalogManager catalogManager = mock(CatalogManager.class);
+ private OptimizingQueue queue;
+
+ @AfterEach
+ void cleanUp() {
+ if (queue != null) {
+ queue.dispose();
+ }
+ persistence.cleanUp();
+ }
+
+ @Test
+ void testV3IsNotScheduledAndV2StillGetsTask() throws Exception {
+ DefaultTableRuntime v3 = createRuntime("v3", 3);
+ UnkeyedTable table = (UnkeyedTable) v3.loadTable().originalTable();
+ String metadataLocation = IcebergV3TestTables.metadataLocation(table);
+ byte[] metadataBytes = IcebergV3TestTables.metadataBytes(table);
+ queue = createQueue(v3);
+
+ TaskRuntime<?> rejectedTask = queue.pollTask(thread(1), 0);
+
+ // Add the supported table after the V3 planning attempt, without removing
V3.
+ DefaultTableRuntime v2 = createRuntime("v2", 2);
+ queue.refreshTable(v2);
+ TaskRuntime<?> supportedTask = queue.pollTask(thread(2), 0);
+
+ assertAll(
+ () -> assertNull(rejectedTask, "An initially V3 table received an
optimizing task"),
+ () -> assertEquals(0L, v3.getProcessId(), "V3 planning created a
process"),
+ () ->
+ assertTrue(
+ queue
+ .collectTasks(task -> task.getTableId() ==
v3.getTableIdentifier().getId())
+ .isEmpty()),
+ () -> assertFalse(v3.getOptimizingStatus().isProcessing()),
+ () ->
+ assertTrue(
+ v3.getOptimizingConfig().isEnabled(), "Do not rewrite the
user's enabled setting"),
+ () -> assertTrue(v3.getLastPlanTime() > 0, "The V3 planning attempt
did not run"),
+ () ->
verify(catalogManager).loadTable(v3.getTableIdentifier().getIdentifier()),
+ () -> assertScheduledTask(v2, supportedTask),
+ () -> IcebergV3TestTables.assertV3MetadataUnchanged(table,
metadataLocation, metadataBytes),
+ () -> assertEquals(Arrays.asList(1, 2),
IcebergV3TestTables.readIds(table)));
+ }
+
+ @Test
+ void testV2PlanningSchedulesRealInputFiles() throws Exception {
+ DefaultTableRuntime runtime = createRuntime("v2", 2);
+ UnkeyedTable table = (UnkeyedTable) runtime.loadTable().originalTable();
+ String metadataLocation = IcebergV3TestTables.metadataLocation(table);
+ byte[] metadataBytes = IcebergV3TestTables.metadataBytes(table);
+ queue = createQueue(runtime);
+
+ TaskRuntime<?> task = queue.pollTask(thread(1), 0);
+
+ assertScheduledTask(runtime, task);
+ table.refresh();
+ assertAll(
+ () -> assertEquals(metadataLocation,
IcebergV3TestTables.metadataLocation(table)),
+ () -> assertArrayEquals(metadataBytes,
IcebergV3TestTables.metadataBytes(table)),
+ () -> assertEquals(Arrays.asList(1, 2),
IcebergV3TestTables.readIds(table)));
+ }
+
+ @ParameterizedTest(name = "refresh initially V{0} from IDLE")
+ @ValueSource(ints = {3, 2})
+ void testRefreshOnlyAdmitsSupportedTables(int formatVersion) throws
Exception {
+ DefaultTableRuntime runtime = createRuntime("refresh", formatVersion,
OptimizingStatus.IDLE);
+ AmoroTable<?> amoroTable = runtime.loadTable();
+ UnkeyedTable table = (UnkeyedTable) amoroTable.originalTable();
+ String metadataLocation = IcebergV3TestTables.metadataLocation(table);
+ byte[] metadataBytes = IcebergV3TestTables.metadataBytes(table);
+ long lastOptimizedSnapshotId = runtime.getLastOptimizedSnapshotId();
+ assertEquals(OptimizingStatus.IDLE, runtime.getOptimizingStatus());
+ assertEquals(0, runtime.getPendingInput().getDataFileCount());
+ assertNotEquals(runtime.getCurrentSnapshotId(), lastOptimizedSnapshotId);
+
+ TableService tableService = mock(TableService.class);
+
doReturn(amoroTable).when(tableService).loadTable(runtime.getTableIdentifier());
+ TableRuntimeRefreshExecutor executor =
+ new TableRuntimeRefreshExecutor(tableService, 1, 60000L, 1);
+ try {
+ // Call the real refresh/evaluation path without starting periodic
scheduling.
+ executor.execute(runtime);
+ } finally {
+ executor.gracefulShutdown();
+ }
+
+ verify(tableService).loadTable(runtime.getTableIdentifier());
+ table.refresh();
+ assertAll(
+ () -> {
+ if (formatVersion == 3) {
+ assertNotEquals(
+ OptimizingStatus.PENDING,
+ runtime.getOptimizingStatus(),
+ "Refresh admitted a V3 table as an optimizing candidate");
+ } else {
+ assertEquals(OptimizingStatus.PENDING,
runtime.getOptimizingStatus());
+ }
+ },
+ () ->
+ assertEquals(formatVersion == 3 ? 0 : 2,
runtime.getPendingInput().getDataFileCount()),
+ () -> assertFalse(runtime.getOptimizingStatus().isProcessing()),
+ () -> assertEquals(0L, runtime.getProcessId()),
+ () -> assertTrue(runtime.getOptimizingConfig().isEnabled()),
+ () -> assertEquals(lastOptimizedSnapshotId,
runtime.getLastOptimizedSnapshotId()),
+ () -> assertEquals(metadataLocation,
IcebergV3TestTables.metadataLocation(table)),
+ () -> assertArrayEquals(metadataBytes,
IcebergV3TestTables.metadataBytes(table)),
+ () -> assertEquals(Arrays.asList(1, 2),
IcebergV3TestTables.readIds(table)));
+ }
+
+ private DefaultTableRuntime createRuntime(String name, int formatVersion)
throws Exception {
+ return createRuntime(name, formatVersion, OptimizingStatus.PENDING);
+ }
+
+ @Test
+ void testV3ReadOnlySummaryStillCollected() throws Exception {
+ DefaultTableRuntime runtime = createRuntime("summary", 2,
OptimizingStatus.IDLE);
+ UnkeyedTable table = (UnkeyedTable) runtime.loadTable().originalTable();
+ table
+ .updateProperties()
+ .set("self-optimizing.enabled", "false")
+ .set("table-summary.enabled", "true")
+ .commit();
+ AmoroTable<?> loaded = upgradedTable(runtime, table);
+ UnkeyedTable current = (UnkeyedTable) loaded.originalTable();
+ String location = IcebergV3TestTables.metadataLocation(current);
+ byte[] bytes = IcebergV3TestTables.metadataBytes(current);
+ long lastOptimized = runtime.getLastOptimizedSnapshotId();
+
+ executeRefresh(runtime, loaded);
+
+ TableRuntimeMeta meta =
+ getAs(
+ TableRuntimeMapper.class,
+ mapper ->
mapper.selectRuntime(runtime.getTableIdentifier().getId()));
+ assertAll(
+ () -> assertEquals(2, meta.getTableSummary().getTotalFileCount()),
+ () -> assertFalse(runtime.getOptimizingConfig().isEnabled()),
+ () -> assertEquals(OptimizingStatus.IDLE,
runtime.getOptimizingStatus()),
+ () -> assertEquals(0, runtime.getPendingInput().getDataFileCount()),
+ () -> assertEquals(lastOptimized,
runtime.getLastOptimizedSnapshotId()),
+ () -> IcebergV3TestTables.assertV3MetadataUnchanged(current, location,
bytes));
+ }
+
+ @Test
+ void testRefreshPersistsUnsupportedProcessReason() throws Exception {
+ DefaultTableRuntime runtime = createRuntime("running", 2);
+ UnkeyedTable table = (UnkeyedTable) runtime.loadTable().originalTable();
+ queue = createQueue(runtime);
+ assertNotNull(queue.pollTask(thread(1), 0));
+ long processId = runtime.getProcessId();
+ long lastOptimized = runtime.getLastOptimizedSnapshotId();
+ AmoroTable<?> loaded = upgradedTable(runtime, table);
+ UnkeyedTable current = (UnkeyedTable) loaded.originalTable();
+ String location = IcebergV3TestTables.metadataLocation(current);
+ byte[] bytes = IcebergV3TestTables.metadataBytes(current);
+
+ executeRefresh(runtime, loaded);
+
+ TableProcessMeta meta =
+ getAs(TableProcessMapper.class, mapper ->
mapper.getProcessMeta(processId));
+ assertAll(
+ () -> assertEquals(ProcessStatus.CLOSED, meta.getStatus()),
+ () ->
+ assertTrue(
+ meta.getFailMessage() != null
+ && meta.getFailMessage().contains("Unsupported Iceberg
format version: 3")),
+ () -> assertNull(runtime.getOptimizingProcess()),
+ () -> assertEquals(OptimizingStatus.IDLE,
runtime.getOptimizingStatus()),
+ () -> assertEquals(lastOptimized,
runtime.getLastOptimizedSnapshotId()),
+ () -> IcebergV3TestTables.assertV3MetadataUnchanged(current, location,
bytes));
+ }
+
+ private AmoroTable<?> upgradedTable(DefaultTableRuntime runtime,
UnkeyedTable table)
+ throws Exception {
+ return IcebergTable.newIcebergTable(
+ runtime.getTableIdentifier().getIdentifier(),
+ IcebergV3TestTables.upgradeToV3(table),
+ TableMetaStore.builder().withConfiguration(new
Configuration()).build(),
+ Collections.emptyMap());
+ }
+
+ private void executeRefresh(DefaultTableRuntime runtime, AmoroTable<?>
loaded) {
+ TableService service = mock(TableService.class);
+ doReturn(loaded).when(service).loadTable(runtime.getTableIdentifier());
+ TableRuntimeRefreshExecutor executor = new
TableRuntimeRefreshExecutor(service, 1, 60000L, 1);
+ try {
+ executor.execute(runtime);
+ } finally {
+ executor.gracefulShutdown();
+ }
+ }
+
+ @ParameterizedTest(name = "recover {0} on V{1}, catalog unavailable: {2}")
+ @CsvSource({
+ "tasks,3,false",
+ "commit,3,false",
+ "tasks,2,false",
+ "commit,2,false",
+ "tasks,2,true",
+ "commit,2,true"
+ })
+ void testOptimizingRecoveryChecksCurrentVersion(
+ String stage, int version, boolean catalogUnavailable) throws Exception {
+ DefaultTableRuntime runtime = createRuntime("recovery", 2);
+ UnkeyedTable table = (UnkeyedTable) runtime.loadTable().originalTable();
+ queue = createQueue(runtime);
+ OptimizerThread worker = thread(1);
+ TaskRuntime<?> task = queue.pollTask(worker, 0);
+ assertNotNull(task);
+ if (stage.equals("commit")) {
+ queue.ackTask(task.getTaskId(), worker);
+ RewriteFilesOutput output =
+ new RewriteFilesOutput(
+ IcebergDataTestHelpers.insert(table,
IcebergV3TestTables.records(table, 1, 2))
+ .dataFiles(),
+ new DeleteFile[0],
+ null);
+ OptimizingTaskResult result =
+ new OptimizingTaskResult(task.getTaskId(), worker.getThreadId());
+ result.setTaskOutput(SerializationUtil.simpleSerialize(output));
+ queue.completeTask(worker, result);
+ // Reproduce a restart after persisting the result but before persisting
COMMITTING.
+ runtime
+ .store()
+ .begin()
+ .updateStatusCode(ignored ->
OptimizingStatus.MINOR_OPTIMIZING.getCode())
+ .commit();
+ } else {
+ queue.retryTask(task);
+ }
+ queue.dispose();
+ UnkeyedTable current = version == 3 ?
IcebergV3TestTables.upgradeToV3(table) : table;
+ IcebergTable loaded =
+ IcebergTable.newIcebergTable(
+ runtime.getTableIdentifier().getIdentifier(),
+ current,
+ TableMetaStore.builder().withConfiguration(new
Configuration()).build(),
+ Collections.emptyMap());
+
doReturn(loaded).when(catalogManager).loadTable(runtime.getTableIdentifier().getIdentifier());
+ if (catalogUnavailable) {
+ // The recovery lookup fails once; the catalog is available again at
commit time.
+ doThrow(new IllegalStateException("Catalog temporarily unavailable"))
+ .doReturn(loaded)
+ .when(catalogManager)
+ .loadTable(runtime.getTableIdentifier().getIdentifier());
+ }
+ String location = IcebergV3TestTables.metadataLocation(current);
+ byte[] bytes = IcebergV3TestTables.metadataBytes(current);
+ ServerTableIdentifier identifier = runtime.getTableIdentifier();
+ DefaultTableRuntime restored =
+ new DefaultTableRuntime(
+ new DefaultTableRuntimeStore(
+ identifier,
+ getAs(TableRuntimeMapper.class, mapper ->
mapper.selectRuntime(identifier.getId())),
+ DefaultTableRuntime.REQUIRED_STATES,
+ getAs(TableRuntimeMapper.class,
TableRuntimeMapper::selectAllStates).stream()
+ .filter(state -> state.getTableId() == identifier.getId())
+ .collect(Collectors.toList())),
+ () -> loaded);
+ queue = createQueue(restored);
+ if (stage.equals("commit")) {
+ if (version != 3) {
+ assertEquals(OptimizingStatus.COMMITTING,
restored.getOptimizingStatus());
+ }
+ if (restored.getOptimizingStatus() == OptimizingStatus.COMMITTING) {
+ restored.getOptimizingProcess().commit();
+ }
+ if (version != 3) {
+ current.refresh();
+ assertEquals(1, IcebergV3TestTables.dataFiles(current).size());
+ }
+ } else {
+ TaskRuntime<?> resumed = queue.pollTask(thread(2), 0);
+ if (version == 3) {
+ assertNull(resumed, "Recovery rescheduled work on V3");
+ } else {
+ assertNotNull(resumed);
+ assertEquals(task.getTaskId(), resumed.getTaskId());
+ }
+ }
+ if (version == 3) {
+ assertAll(
+ () -> assertFalse(restored.getOptimizingStatus().isProcessing()),
+ () -> assertTrue(queue.collectTasks().isEmpty()),
+ () -> IcebergV3TestTables.assertV3MetadataUnchanged(current,
location, bytes));
+ }
+ assertEquals(Arrays.asList(1, 2), IcebergV3TestTables.readIds(current));
+ }
+
+ private DefaultTableRuntime createRuntime(
+ String name, int formatVersion, OptimizingStatus initialStatus) throws
Exception {
+ UnkeyedTable table = IcebergV3TestTables.create(temp.resolve(name), 2);
+ // Configure while still V2 so setup does not corrupt the V3 fixture.
+ table
+ .updateProperties()
+ .set("self-optimizing.enabled", "true")
+ .set(TableProperties.SELF_OPTIMIZING_MINOR_TRIGGER_FILE_CNT, "2")
+ .commit();
+ if (formatVersion == 3) {
+ table = IcebergV3TestTables.upgradeToV3(table);
+ }
+ ServerTableIdentifier identifier =
+ ServerTableIdentifier.of("test_catalog", "test_database", name,
TableFormat.ICEBERG);
+ IcebergTable amoroTable =
+ IcebergTable.newIcebergTable(
+ identifier.getIdentifier(),
+ table,
+ TableMetaStore.builder().withConfiguration(new
Configuration()).build(),
+ Collections.emptyMap());
+
doReturn(amoroTable).when(catalogManager).loadTable(identifier.getIdentifier());
+
+ TableRuntimeMeta meta = new TableRuntimeMeta();
+ meta.setGroupName("default");
+ meta.setStatusCode(initialStatus.getCode());
+ doAs(TableMetaMapper.class, mapper -> mapper.insertTable(identifier));
+ meta.setTableId(identifier.getId());
+ doAs(TableRuntimeMapper.class, mapper -> mapper.insertRuntime(meta));
+ DefaultTableRuntimeStore store =
+ new DefaultTableRuntimeStore(
+ identifier, meta, DefaultTableRuntime.REQUIRED_STATES,
Collections.emptyList());
+ DefaultTableRuntime runtime = new DefaultTableRuntime(store, () ->
amoroTable);
+ runtime.refresh(amoroTable);
+ assertTrue(runtime.getOptimizingConfig().isEnabled());
+ assertEquals(2,
IcebergV3TestTables.dataFiles(amoroTable.originalTable()).size());
+ return runtime;
+ }
+
+ private OptimizingQueue createQueue(DefaultTableRuntime runtime) {
+ // Run planning synchronously. pollTask(..., 0) still drives planning and
then
+ // fetches the produced task, without sleeps or a background executor.
+ return new OptimizingQueue(
+ catalogManager,
+ new ResourceGroup.Builder("default", "local").build(),
+ group -> 1,
+ Runnable::run,
+ Collections.singletonList(runtime),
+ 1);
+ }
+
+ private void assertScheduledTask(DefaultTableRuntime runtime, TaskRuntime<?>
task) {
+
verify(catalogManager).loadTable(runtime.getTableIdentifier().getIdentifier());
+ assertNotNull(task, "A supported V2 table must still receive work");
+ assertEquals(runtime.getTableIdentifier().getId(), task.getTableId());
+ assertEquals(TaskRuntime.Status.SCHEDULED, task.getStatus());
+ assertTrue(runtime.getProcessId() > 0);
+ RewriteStageTask rewrite = (RewriteStageTask) task.getTaskDescriptor();
+ assertEquals(2, rewrite.getInput().rewrittenDataFiles().length);
+ }
+
+ private OptimizerThread thread(int id) {
+ return new OptimizerThread(id, null) {
+ @Override
+ public String getToken() {
+ return "planning-test";
+ }
+ };
+ }
+
+ private static class TestPersistence extends DerbyPersistence {
+ void cleanUp() {
+ super.after();
+ }
+ }
+}
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingProcess.java
b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingProcess.java
similarity index 53%
copy from
amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingProcess.java
copy to
amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingProcess.java
index b088f6f27..16926976c 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingProcess.java
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingProcess.java
@@ -18,35 +18,30 @@
package org.apache.amoro.server.optimizing;
-import org.apache.amoro.optimizing.MetricsSummary;
-import org.apache.amoro.optimizing.OptimizingType;
-import org.apache.amoro.process.ProcessStatus;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
-public interface OptimizingProcess {
+import org.junit.jupiter.api.Test;
- long getTableId();
+class TestOptimizingProcess {
+ @Test
+ void testCloseWithCommitPreference() {
+ OptimizingProcess process = mock(OptimizingProcess.class,
CALLS_REAL_METHODS);
- long getProcessId();
+ process.close(true);
+ verify(process).close(true, null);
- void close(boolean needCommit);
+ process.close(false);
+ verify(process).close(false, null);
+ }
- boolean isClosed();
+ @Test
+ void testCloseWithReasonDoesNotCommit() {
+ OptimizingProcess process = mock(OptimizingProcess.class,
CALLS_REAL_METHODS);
- long getTargetSnapshotId();
+ process.close("Unsupported Iceberg format version: 3");
- long getTargetChangeSnapshotId();
-
- long getPlanTime();
-
- long getDuration();
-
- OptimizingType getOptimizingType();
-
- ProcessStatus getStatus();
-
- long getRunningQuotaTime(long calculatingStartTime, long calculatingEndTime);
-
- void commit();
-
- MetricsSummary getSummary();
+ verify(process).close(false, "Unsupported Iceberg format version: 3");
+ }
}
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/commit/TestIcebergV3OptimizingCommit.java
b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/commit/TestIcebergV3OptimizingCommit.java
new file mode 100644
index 000000000..90967acf7
--- /dev/null
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/commit/TestIcebergV3OptimizingCommit.java
@@ -0,0 +1,356 @@
+/*
+ * 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.amoro.server.optimizing.commit;
+
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+import org.apache.amoro.exception.OptimizingCommitException;
+import org.apache.amoro.formats.iceberg.IcebergV3TestTables;
+import org.apache.amoro.io.IcebergDataTestHelpers;
+import org.apache.amoro.optimizing.RewriteFilesInput;
+import org.apache.amoro.optimizing.RewriteFilesOutput;
+import org.apache.amoro.optimizing.RewriteStageTask;
+import org.apache.amoro.server.optimizing.TaskRuntime;
+import org.apache.amoro.server.optimizing.UnKeyedTableCommit;
+import org.apache.amoro.table.UnkeyedTable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.data.FileHelpers;
+import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+
+class TestIcebergV3OptimizingCommit {
+ @TempDir private Path temp;
+
+ @Test
+ void testV3OptimizingCommitDoesNotChangeMetadata() throws Exception {
+ UnkeyedTable table = IcebergV3TestTables.create(temp.resolve("v3"), 3);
+ assertCommitRejected(table, prepareCommit(table));
+ }
+
+ @ParameterizedTest(name = "external V{1} update at {0}")
+ @CsvSource({
+ "transaction-commit, 3",
+ "first-storage-commit, 3",
+ "transaction-commit, 2",
+ "first-storage-commit, 2"
+ })
+ void testOptimizingCommitAcrossConcurrentMetadataChange(String upgradePoint,
int externalVersion)
+ throws Exception {
+ UnkeyedTable original = IcebergV3TestTables.create(temp.resolve("race"),
2);
+ TableOperations operations = spy(((HasTableOperations)
original).operations());
+ UnkeyedTable table = spy(IcebergV3TestTables.wrap(new
BaseTable(operations, original.name())));
+ UnKeyedTableCommit commit = prepareCommit(table);
+ List<Integer> attemptedVersions = new ArrayList<>();
+ List<Integer> committedVersions = new ArrayList<>();
+ AtomicInteger conflicts = new AtomicInteger();
+ AtomicReference<UnkeyedTable> external = new AtomicReference<>();
+ AtomicReference<String> externalLocation = new AtomicReference<>();
+ AtomicReference<byte[]> externalBytes = new AtomicReference<>();
+
+ // This hook observes real storage commits. It does not manufacture a
conflict:
+ // the independent upgrade occupies the next Hadoop metadata version,
causing the
+ // stale commit to fail and Iceberg's real transaction retry to refresh
its base.
+ doAnswer(
+ invocation -> {
+ TableMetadata base = invocation.getArgument(0);
+ TableMetadata updated = invocation.getArgument(1);
+ attemptedVersions.add(updated.formatVersion());
+ if (external.get() == null) {
+ assertEquals("first-storage-commit", upgradePoint);
+ assertEquals(2, base.formatVersion());
+ external.set(commitExternalChange(table, externalVersion));
+
externalLocation.set(IcebergV3TestTables.metadataLocation(external.get()));
+
externalBytes.set(IcebergV3TestTables.metadataBytes(external.get()));
+ }
+ try {
+ Object result = invocation.callRealMethod();
+ committedVersions.add(updated.formatVersion());
+ return result;
+ } catch (CommitFailedException conflict) {
+ conflicts.incrementAndGet();
+ throw conflict;
+ }
+ })
+ .when(operations)
+ .commit(any(TableMetadata.class), any(TableMetadata.class));
+
+ if (upgradePoint.equals("transaction-commit")) {
+ long originalSnapshotId = table.currentSnapshot().snapshotId();
+ doAnswer(
+ invocation -> {
+ TableMetadata staged = invocation.getArgument(0);
+ // Observe the staged rewrite through the underlying
operations, independent
+ // of which Table facade the production code uses for its
transaction.
+ if (external.get() == null
+ && staged.currentSnapshot().snapshotId() !=
originalSnapshotId) {
+ assertEquals(2, staged.formatVersion());
+ external.set(commitExternalChange(table, externalVersion));
+
externalLocation.set(IcebergV3TestTables.metadataLocation(external.get()));
+
externalBytes.set(IcebergV3TestTables.metadataBytes(external.get()));
+ }
+ return invocation.callRealMethod();
+ })
+ .when(operations)
+ .temp(any(TableMetadata.class));
+ }
+
+ long snapshotId = table.currentSnapshot().snapshotId();
+ Set<String> files = filePaths(table);
+ if (externalVersion == 2) {
+ // Exercise exactly the same hooks and retry while the table remains
supported.
+ commit.commit();
+ assertNotNull(external.get(), "The external update hook was not
reached");
+ external.get().refresh();
+ assertAll(
+ () -> assertEquals(upgradePoint.equals("first-storage-commit") ? 1 :
0, conflicts.get()),
+ () ->
+ assertEquals(
+ upgradePoint.equals("first-storage-commit")
+ ? Arrays.asList(2, 2)
+ : Collections.singletonList(2),
+ attemptedVersions),
+ () -> assertEquals(Collections.singletonList(2), committedVersions),
+ () -> assertEquals("true",
external.get().properties().get("test.concurrent-update")),
+ () -> assertNotEquals(snapshotId,
external.get().currentSnapshot().snapshotId()),
+ () -> assertEquals(1, filePaths(external.get()).size()),
+ () -> assertEquals(Arrays.asList(1, 2),
IcebergV3TestTables.readIds(external.get())));
+ return;
+ }
+ assertAll(
+ () -> assertUnsupportedCommit(commit),
+ () -> assertNotNull(external.get(), "The upgrade hook was not
reached"),
+ () ->
+ assertEquals(
+ upgradePoint.equals("first-storage-commit") ? 1 : 0,
+ conflicts.get(),
+ "The storage conflict must exercise the real Iceberg retry"),
+ () ->
+ assertEquals(
+ Collections.emptyList(),
+ committedVersions,
+ "V3 metadata was committed; attempted versions: " +
attemptedVersions),
+ () ->
+ IcebergV3TestTables.assertV3MetadataUnchanged(
+ external.get(), externalLocation.get(), externalBytes.get()),
+ () -> assertEquals(snapshotId,
external.get().currentSnapshot().snapshotId()),
+ () -> assertEquals(files, filePaths(external.get())),
+ () -> assertEquals(Arrays.asList(1, 2),
IcebergV3TestTables.readIds(external.get())));
+ }
+
+ private UnkeyedTable commitExternalChange(UnkeyedTable table, int
formatVersion)
+ throws Exception {
+ if (formatVersion == 3) {
+ return IcebergV3TestTables.upgradeToV3(table);
+ }
+ Table external = new HadoopTables(new
Configuration()).load(table.location());
+ external.updateProperties().set("test.concurrent-update", "true").commit();
+ return IcebergV3TestTables.wrap(external);
+ }
+
+ private void assertCommitRejected(UnkeyedTable table, UnKeyedTableCommit
commit)
+ throws Exception {
+ String metadataLocation = IcebergV3TestTables.metadataLocation(table);
+ byte[] metadataBytes = IcebergV3TestTables.metadataBytes(table);
+ Set<String> files = filePaths(table);
+ long snapshotId = table.currentSnapshot().snapshotId();
+
+ assertAll(
+ () -> assertUnsupportedCommit(commit),
+ () -> IcebergV3TestTables.assertV3MetadataUnchanged(table,
metadataLocation, metadataBytes),
+ () -> assertEquals(snapshotId, table.currentSnapshot().snapshotId()),
+ () -> assertEquals(files, filePaths(table), "Optimizing replaced
committed data files"),
+ () -> assertEquals(Arrays.asList(1, 2),
IcebergV3TestTables.readIds(table)));
+ }
+
+ private void assertUnsupportedCommit(UnKeyedTableCommit commit) {
+ OptimizingCommitException failure =
+ assertThrows(OptimizingCommitException.class, commit::commit);
+ assertTrue(
+ failure.getCause() instanceof IllegalArgumentException,
+ "Reject unsupported metadata explicitly, not because the fixture
failed");
+ assertTrue(failure.getCause().getMessage().contains("Unsupported Iceberg
format version: 3"));
+ }
+
+ @Test
+ void testV2OptimizingCommitReplacesFiles() throws Exception {
+ UnkeyedTable table = IcebergV3TestTables.create(temp.resolve("v2"), 2);
+ String metadataLocation = IcebergV3TestTables.metadataLocation(table);
+ long snapshotId = table.currentSnapshot().snapshotId();
+ assertEquals(2, filePaths(table).size());
+
+ prepareCommit(table).commit();
+ table.refresh();
+
+ assertAll(
+ () -> assertNotEquals(metadataLocation,
IcebergV3TestTables.metadataLocation(table)),
+ () -> assertNotEquals(snapshotId,
table.currentSnapshot().snapshotId()),
+ () -> assertEquals(1, filePaths(table).size()),
+ () -> assertEquals(Arrays.asList(1, 2),
IcebergV3TestTables.readIds(table)));
+ }
+
+ @ParameterizedTest(name = "{0} on V{1}")
+ @CsvSource({
+ "delete-rewrite,3",
+ "rewrite-and-add-delete,3",
+ "delete-rewrite,2",
+ "rewrite-and-add-delete,2"
+ })
+ void testDeleteCommitBranches(String mode, int version) throws Exception {
+ UnkeyedTable initial = IcebergV3TestTables.create(temp.resolve("deletes"),
2);
+ List<DataFile> inputs = IcebergV3TestTables.dataFiles(initial);
+ DeleteFile oldDelete = null;
+ if (mode.equals("delete-rewrite")) {
+ oldDelete = equalityDelete(initial, inputs.get(0), 1);
+ initial.newRowDelta().addDeletes(oldDelete).commit();
+ }
+ UnkeyedTable table = version == 3 ?
IcebergV3TestTables.upgradeToV3(initial) : initial;
+ List<Integer> expectedIds = IcebergV3TestTables.readIds(table);
+ DataFile[] outputData;
+ DeleteFile outputDelete;
+ RewriteFilesInput input;
+ if (oldDelete != null) {
+ outputData = new DataFile[0];
+ outputDelete = equalityDelete(table, inputs.get(0), 1);
+ input =
+ new RewriteFilesInput(
+ new DataFile[0],
+ new DataFile[0],
+ new DeleteFile[0],
+ new DeleteFile[] {oldDelete},
+ table);
+ } else {
+ // A valid staged output with one redundant row removed by the new
equality delete.
+ outputData =
+ IcebergDataTestHelpers.insert(table,
IcebergV3TestTables.records(table, 1, 2, 3))
+ .dataFiles();
+ outputDelete = equalityDelete(table, outputData[0], 3);
+ input =
+ new RewriteFilesInput(
+ inputs.toArray(new DataFile[0]),
+ new DataFile[0],
+ new DeleteFile[0],
+ new DeleteFile[0],
+ table);
+ }
+ UnKeyedTableCommit commit =
+ commitFor(
+ table,
+ input,
+ new RewriteFilesOutput(outputData, new DeleteFile[]
{outputDelete}, null));
+ String location = IcebergV3TestTables.metadataLocation(table);
+ byte[] bytes = IcebergV3TestTables.metadataBytes(table);
+ long snapshotId = table.currentSnapshot().snapshotId();
+ if (version == 3) {
+ assertAll(
+ () -> assertUnsupportedCommit(commit),
+ () -> IcebergV3TestTables.assertV3MetadataUnchanged(table, location,
bytes),
+ () -> assertEquals(snapshotId, table.currentSnapshot().snapshotId()),
+ () -> assertEquals(expectedIds, IcebergV3TestTables.readIds(table)));
+ } else {
+ commit.commit();
+ table.refresh();
+ assertNotEquals(snapshotId, table.currentSnapshot().snapshotId());
+ assertEquals(
+ mode.equals("delete-rewrite") ? 2 : 1,
IcebergV3TestTables.dataFiles(table).size());
+ assertEquals(expectedIds, IcebergV3TestTables.readIds(table));
+ }
+ }
+
+ private DeleteFile equalityDelete(UnkeyedTable table, DataFile data, int id)
throws Exception {
+ return FileHelpers.writeDeleteFile(
+ table,
+ OutputFileFactory.builderFor(table, 1, 1)
+ .build()
+ .newOutputFile(data.partition())
+ .encryptingOutputFile(),
+ data.partition(),
+ IcebergV3TestTables.records(table, id),
+ table.schema());
+ }
+
+ private UnKeyedTableCommit prepareCommit(UnkeyedTable table) throws
Exception {
+ List<DataFile> inputs = IcebergV3TestTables.dataFiles(table);
+ assertEquals(2, inputs.size());
+ // Prepare real output without committing it; this test targets the AMS
commit boundary.
+ DataFile[] outputs =
+ IcebergDataTestHelpers.insert(table,
IcebergV3TestTables.records(table, 1, 2)).dataFiles();
+ assertEquals(1, outputs.length);
+ RewriteFilesInput input =
+ new RewriteFilesInput(
+ inputs.toArray(new DataFile[0]),
+ new DataFile[0],
+ new DeleteFile[0],
+ new DeleteFile[0],
+ table);
+ return commitFor(table, input, new RewriteFilesOutput(outputs, new
DeleteFile[0], null));
+ }
+
+ @SuppressWarnings("unchecked")
+ private UnKeyedTableCommit commitFor(
+ UnkeyedTable table, RewriteFilesInput input, RewriteFilesOutput output) {
+ RewriteStageTask task = mock(RewriteStageTask.class);
+ when(task.getInput()).thenReturn(input);
+ when(task.getOutput()).thenReturn(output);
+ when(task.getPartition()).thenReturn("");
+ TaskRuntime<RewriteStageTask> runtime = mock(TaskRuntime.class);
+ when(runtime.getStatus()).thenReturn(TaskRuntime.Status.SUCCESS);
+ when(runtime.getTaskDescriptor()).thenReturn(task);
+ when(runtime.getProperties()).thenReturn(Collections.emptyMap());
+ return new UnKeyedTableCommit(
+ table.currentSnapshot().snapshotId(), table,
Collections.singletonList(runtime));
+ }
+
+ private Set<String> filePaths(UnkeyedTable table) throws Exception {
+ return IcebergV3TestTables.dataFiles(table).stream()
+ .map(file -> file.path().toString())
+ .collect(Collectors.toSet());
+ }
+}
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/process/iceberg/TestIcebergV3MaintenanceRecovery.java
b/amoro-ams/src/test/java/org/apache/amoro/server/process/iceberg/TestIcebergV3MaintenanceRecovery.java
new file mode 100644
index 000000000..79ef3631e
--- /dev/null
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/process/iceberg/TestIcebergV3MaintenanceRecovery.java
@@ -0,0 +1,125 @@
+/*
+ * 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.amoro.server.process.iceberg;
+
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+
+import org.apache.amoro.Action;
+import org.apache.amoro.IcebergActions;
+import org.apache.amoro.ServerTableIdentifier;
+import org.apache.amoro.TableFormat;
+import org.apache.amoro.formats.iceberg.IcebergTable;
+import org.apache.amoro.formats.iceberg.IcebergV3MaintenanceFixture;
+import org.apache.amoro.process.LocalExecutionEngine;
+import org.apache.amoro.process.LocalProcess;
+import org.apache.amoro.process.TableProcessStore;
+import org.apache.amoro.server.optimizing.OptimizingStatus;
+import org.apache.amoro.server.table.DefaultTableRuntime;
+import org.apache.amoro.server.table.TableOrphanFilesCleaningMetrics;
+import org.apache.amoro.table.TableMetaStore;
+import org.apache.hadoop.conf.Configuration;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+import java.nio.file.Path;
+import java.util.Collections;
+
+class TestIcebergV3MaintenanceRecovery {
+ @TempDir private Path temp;
+
+ @ParameterizedTest(name = "recover {0} on V{1}")
+ @CsvSource({
+ "snapshot,3",
+ "dangling,3",
+ "data,3",
+ "orphan,3",
+ "tag,3",
+ "snapshot,2",
+ "dangling,2",
+ "data,2",
+ "orphan,2",
+ "tag,2"
+ })
+ void testRecoveredMaintenanceChecksVersion(String action, int version)
throws Exception {
+ IcebergV3MaintenanceFixture fixture =
+ new IcebergV3MaintenanceFixture(temp.resolve(action), version, action);
+ DefaultTableRuntime runtime = mock(DefaultTableRuntime.class);
+ doReturn(ServerTableIdentifier.of("test_catalog", "test_database", action,
TableFormat.ICEBERG))
+ .when(runtime)
+ .getTableIdentifier();
+ doReturn(fixture.configuration).when(runtime).getTableConfiguration();
+ doReturn(OptimizingStatus.IDLE).when(runtime).getOptimizingStatus();
+ doReturn(mock(TableOrphanFilesCleaningMetrics.class))
+ .when(runtime)
+ .getOrphanFilesCleaningMetrics();
+ doReturn(
+ IcebergTable.newIcebergTable(
+ fixture.table.id(),
+ fixture.table,
+ TableMetaStore.builder().withConfiguration(new
Configuration()).build(),
+ Collections.emptyMap()))
+ .when(runtime)
+ .loadTable();
+ LocalExecutionEngine engine = mock(LocalExecutionEngine.class);
+ doReturn(LocalExecutionEngine.ENGINE_NAME).when(engine).name();
+ IcebergProcessFactory factory = new IcebergProcessFactory();
+ factory.availableExecuteEngines(Collections.singletonList(engine));
+ TableProcessStore store = mock(TableProcessStore.class);
+ doReturn(action(action)).when(store).getAction();
+ LocalProcess recovered = (LocalProcess) factory.recover(runtime, store);
+ if (version == 3) {
+ assertAll(
+ () -> {
+ RuntimeException failure = assertThrows(RuntimeException.class,
recovered::run);
+ Throwable cause = failure;
+ while (cause.getCause() != null) {
+ cause = cause.getCause();
+ }
+ assertTrue(cause instanceof IllegalArgumentException);
+ assertTrue(cause.getMessage().contains("Unsupported Iceberg format
version: 3"));
+ },
+ fixture::assertPreserved);
+ } else {
+ recovered.run();
+ fixture.assertChanged();
+ }
+ }
+
+ private Action action(String name) {
+ switch (name) {
+ case "snapshot":
+ return IcebergActions.EXPIRE_SNAPSHOTS;
+ case "dangling":
+ return IcebergActions.CLEAN_DANGLING_DELETE;
+ case "data":
+ return IcebergActions.EXPIRE_DATA;
+ case "orphan":
+ return IcebergActions.CLEAN_ORPHAN;
+ case "tag":
+ return IcebergActions.AUTO_CREATE_TAGS;
+ default:
+ throw new IllegalArgumentException(name);
+ }
+ }
+}
diff --git
a/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/IcebergMaintenanceCompatibility.java
b/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/IcebergMaintenanceCompatibility.java
new file mode 100644
index 000000000..a9c997812
--- /dev/null
+++
b/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/IcebergMaintenanceCompatibility.java
@@ -0,0 +1,159 @@
+/*
+ * 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.amoro.formats.iceberg;
+
+import org.apache.amoro.TableFormat;
+import org.apache.amoro.table.MixedTable;
+import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.encryption.EncryptionManager;
+import org.apache.iceberg.exceptions.CleanableFailure;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.LocationProvider;
+
+/** Compatibility boundary for automatic changes made by Amoro's bundled
Iceberg writer. */
+public final class IcebergMaintenanceCompatibility {
+ private IcebergMaintenanceCompatibility() {}
+
+ /**
+ * Check already loaded metadata without an extra refresh. Iceberg 1.8 can
parse V3 but cannot
+ * safely write metadata produced by newer V3 writers. Automatic writers
must use forUpdate.
+ */
+ public static void checkSupported(Table table) {
+ if (isMixedTable(table)) {
+ return;
+ }
+ if (!(table instanceof HasTableOperations)) {
+ throw new UnsupportedTableException(
+ "Cannot determine Iceberg format version for table " + table.name());
+ }
+ checkMetadata(((HasTableOperations) table).operations().current(),
table.name());
+ }
+
+ /**
+ * Return a facade used only by automatic writers. Catalog reads and user
table settings remain
+ * unchanged. Every storage commit, including transaction retries, checks
both base and result.
+ */
+ public static Table forUpdate(Table table) {
+ if (isMixedTable(table)) {
+ return table;
+ }
+ table.refresh();
+ checkSupported(table);
+ TableOperations operations = ((HasTableOperations) table).operations();
+ if (operations instanceof GuardedOperations) {
+ return table;
+ }
+ return new BaseTable(new GuardedOperations(operations, table.name()),
table.name());
+ }
+
+ private static boolean isMixedTable(Table table) {
+ return table instanceof MixedTable && ((MixedTable) table).format() !=
TableFormat.ICEBERG;
+ }
+
+ private static void checkMetadata(TableMetadata metadata, String tableName) {
+ if (metadata == null) {
+ throw new UnsupportedTableException(
+ "Cannot determine Iceberg format version for table " + tableName);
+ }
+ if (metadata.formatVersion() > 2) {
+ throw new UnsupportedTableException(
+ "Unsupported Iceberg format version: "
+ + metadata.formatVersion()
+ + " for table "
+ + tableName
+ + "; Amoro automatic maintenance supports V1 and V2");
+ }
+ }
+
+ /** A definite rejection, safe for Iceberg to clean up its uncommitted
metadata files. */
+ public static class UnsupportedTableException extends
IllegalArgumentException
+ implements CleanableFailure {
+ private UnsupportedTableException(String message) {
+ super(message);
+ }
+ }
+
+ private static class GuardedOperations implements TableOperations {
+ private final TableOperations delegate;
+ private final String tableName;
+
+ private GuardedOperations(TableOperations delegate, String tableName) {
+ this.delegate = delegate;
+ this.tableName = tableName;
+ }
+
+ @Override
+ public TableMetadata current() {
+ return delegate.current();
+ }
+
+ @Override
+ public TableMetadata refresh() {
+ return delegate.refresh();
+ }
+
+ @Override
+ public void commit(TableMetadata base, TableMetadata metadata) {
+ checkMetadata(base, tableName);
+ checkMetadata(metadata, tableName);
+ // Preserve the catalog's atomic conflict detection. A stale V2 commit
fails there;
+ // if Iceberg retries against V3, the checks above reject it before
serialization.
+ delegate.commit(base, 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();
+ }
+
+ @Override
+ public TableOperations temp(TableMetadata metadata) {
+ return new GuardedOperations(delegate.temp(metadata), tableName);
+ }
+
+ @Override
+ public long newSnapshotId() {
+ return delegate.newSnapshotId();
+ }
+
+ @Override
+ public boolean requireStrictCleanup() {
+ return delegate.requireStrictCleanup();
+ }
+ }
+}
diff --git
a/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/maintainer/IcebergTableMaintainer.java
b/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/maintainer/IcebergTableMaintainer.java
index 5333856ce..dad6b7daa 100644
---
a/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/maintainer/IcebergTableMaintainer.java
+++
b/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/maintainer/IcebergTableMaintainer.java
@@ -24,6 +24,8 @@ import org.apache.amoro.api.CommitMetaProducer;
import org.apache.amoro.config.DataExpirationConfig;
import org.apache.amoro.config.TableConfiguration;
import org.apache.amoro.config.TagConfiguration;
+import org.apache.amoro.formats.iceberg.IcebergMaintenanceCompatibility;
+import
org.apache.amoro.formats.iceberg.IcebergMaintenanceCompatibility.UnsupportedTableException;
import org.apache.amoro.formats.iceberg.utils.IcebergTableUtil;
import org.apache.amoro.formats.iceberg.utils.RollingFileCleaner;
import org.apache.amoro.iceberg.Constants;
@@ -143,6 +145,8 @@ public class IcebergTableMaintainer implements
TableMaintainer {
if (!tableConfiguration.isCleanOrphanEnabled()) {
return Maps.newHashMap();
}
+ table.refresh();
+ IcebergMaintenanceCompatibility.checkSupported(table);
long keepTime = tableConfiguration.getOrphanExistingMinutes() * 60 * 1000;
@@ -166,6 +170,7 @@ public class IcebergTableMaintainer implements
TableMaintainer {
if (!tableConfiguration.isDeleteDanglingDeleteFilesEnabled()) {
return Maps.newHashMap();
}
+ table = IcebergMaintenanceCompatibility.forUpdate(table);
Snapshot currentSnapshot = table.currentSnapshot();
if (currentSnapshot == null) {
@@ -191,6 +196,7 @@ public class IcebergTableMaintainer implements
TableMaintainer {
if (!expireSnapshotEnabled()) {
return Maps.newHashMap();
}
+ table = IcebergMaintenanceCompatibility.forUpdate(table);
int cleaned =
expireSnapshots(
mustOlderThan(System.currentTimeMillis()),
@@ -281,6 +287,7 @@ public class IcebergTableMaintainer implements
TableMaintainer {
@Override
public Map<String, String> expireData() {
+ table = IcebergMaintenanceCompatibility.forUpdate(table);
DataExpirationConfig expirationConfig =
context.getTableConfiguration().getExpiringDataConfig();
try {
Types.NestedField field =
table.schema().findField(expirationConfig.getExpirationField());
@@ -289,6 +296,8 @@ public class IcebergTableMaintainer implements
TableMaintainer {
}
return expireDataFrom(expirationConfig,
expireBaseOnRule(expirationConfig, field));
+ } catch (UnsupportedTableException e) {
+ throw e;
} catch (Throwable t) {
LOG.error("Unexpected purge error for table {} ", tableIdentifier, t);
return Maps.newHashMap();
@@ -357,6 +366,7 @@ public class IcebergTableMaintainer implements
TableMaintainer {
@Override
public void autoCreateTags() {
+ table = IcebergMaintenanceCompatibility.forUpdate(table);
TagConfiguration tagConfiguration =
context.getTableConfiguration().getTagConfiguration();
new AutoCreateIcebergTagAction(table, tagConfiguration,
LocalDateTime.now()).execute();
}
diff --git
a/amoro-format-iceberg/src/test/java/org/apache/amoro/formats/iceberg/IcebergV3MaintenanceFixture.java
b/amoro-format-iceberg/src/test/java/org/apache/amoro/formats/iceberg/IcebergV3MaintenanceFixture.java
new file mode 100644
index 000000000..10d04da55
--- /dev/null
+++
b/amoro-format-iceberg/src/test/java/org/apache/amoro/formats/iceberg/IcebergV3MaintenanceFixture.java
@@ -0,0 +1,245 @@
+/*
+ * 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.amoro.formats.iceberg;
+
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import org.apache.amoro.config.DataExpirationConfig;
+import org.apache.amoro.config.TableConfiguration;
+import org.apache.amoro.config.TagConfiguration;
+import org.apache.amoro.formats.iceberg.maintainer.IcebergTableMaintainer;
+import org.apache.amoro.io.IcebergDataTestHelpers;
+import org.apache.amoro.maintainer.MaintainerMetrics;
+import org.apache.amoro.maintainer.OptimizingInfo;
+import org.apache.amoro.maintainer.TableMaintainerContext;
+import org.apache.amoro.table.UnkeyedTable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.FileHelpers;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.Pair;
+
+import java.io.OutputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.attribute.FileTime;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.StreamSupport;
+
+/** Real maintenance work, shared by direct action and recovered-process
tests. */
+public class IcebergV3MaintenanceFixture {
+ public final UnkeyedTable table;
+ public final TableConfiguration configuration;
+ private final String action;
+ private final String metadataLocation;
+ private final byte[] metadataBytes;
+ private final long snapshotId;
+ private final Map<?, ?> refs;
+ private final long snapshots;
+ private final Path orphan;
+
+ public IcebergV3MaintenanceFixture(Path location, int version, String
action) throws Exception {
+ this.action = action;
+ Schema schema =
+ new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.required(2, "event_time", Types.LongType.get()));
+ Table raw =
+ new HadoopTables(new Configuration())
+ .create(
+ schema,
+ PartitionSpec.unpartitioned(),
+ Collections.singletonMap("format-version", version == 1 ? "1"
: "2"),
+ location.toString());
+ GenericRecord record = GenericRecord.create(schema);
+ IcebergDataTestHelpers.append(
+ raw, Collections.singletonList(record.copy("id", 1, "event_time",
1L)));
+ IcebergDataTestHelpers.append(
+ raw, Collections.singletonList(record.copy("id", 2, "event_time",
2L)));
+
+ if (action.equals("dangling")) {
+ DataFile oldFile = IcebergV3TestTables.dataFiles(raw).get(0);
+ DeleteFile delete =
+ FileHelpers.writeDeleteFile(
+ raw,
+ OutputFileFactory.builderFor(raw, 1, 1)
+ .build()
+ .newOutputFile(oldFile.partition())
+ .encryptingOutputFile(),
+ oldFile.partition(),
+ Collections.singletonList(Pair.of(oldFile.path(), 0L)))
+ .first();
+ raw.newRowDelta().addDeletes(delete).commit();
+ int id;
+ try
(org.apache.iceberg.io.CloseableIterable<org.apache.iceberg.data.Record> rows =
+ org.apache.iceberg.data.IcebergGenerics.read(raw).build()) {
+ id = rows.iterator().next().getField("id").equals(1) ? 2 : 1;
+ }
+ DataFile replacement =
+ IcebergDataTestHelpers.insert(
+ raw, Collections.singletonList(record.copy("id", id,
"event_time", (long) id)))
+ .dataFiles()[0];
+ raw.newRewrite()
+ .rewriteFiles(Collections.singleton(oldFile),
Collections.singleton(replacement))
+ .validateFromSnapshot(raw.currentSnapshot().snapshotId())
+ .commit();
+ assertEquals(
+ 1,
+
org.apache.amoro.formats.iceberg.utils.IcebergTableUtil.getDanglingDeleteFiles(raw)
+ .size());
+ }
+ table = version == 3 ? IcebergV3TestTables.upgradeToV3(raw) :
IcebergV3TestTables.wrap(raw);
+ configuration =
+ new TableConfiguration()
+ .setExpireSnapshotEnabled(true)
+ .setSnapshotTTLMinutes(0)
+ .setSnapshotMinCount(1)
+ .setDeleteDanglingDeleteFilesEnabled(true)
+ .setCleanOrphanEnabled(true)
+ .setOrphanExistingMinutes(1);
+ configuration.setExpiringDataConfig(
+ new DataExpirationConfig()
+ .setEnabled(true)
+ .setExpirationField("event_time")
+ .setExpirationLevel(DataExpirationConfig.ExpireLevel.FILE)
+ .setRetentionTime(1)
+ .setDateTimePattern("yyyy-MM-dd HH:mm:ss")
+ .setNumberDateFormat(IcebergTableMaintainer.EXPIRE_TIMESTAMP_MS)
+ .setBaseOnRule(DataExpirationConfig.BaseOnRule.CURRENT_TIME));
+ TagConfiguration tags = new TagConfiguration();
+ tags.setAutoCreateTag(true);
+ tags.setTriggerPeriod(TagConfiguration.Period.MONTHLY);
+ tags.setTagFormat("'tag-'yyyy-MM");
+ tags.setMaxDelayMinutes(0);
+ configuration.setTagConfiguration(tags);
+ orphan = location.resolve("data").resolve("orphan.parquet");
+ if (action.equals("orphan")) {
+ try (OutputStream output =
table.io().newOutputFile(orphan.toString()).create()) {
+ output.write(1);
+ }
+ Files.setLastModifiedTime(orphan, FileTime.fromMillis(1));
+ assertTrue(Files.exists(orphan));
+ }
+ metadataLocation = IcebergV3TestTables.metadataLocation(table);
+ metadataBytes = IcebergV3TestTables.metadataBytes(table);
+ snapshotId = table.currentSnapshot().snapshotId();
+ refs = new HashMap<>(table.refs());
+ snapshots = snapshotCount();
+ assertEquals(Arrays.asList(1, 2), IcebergV3TestTables.readIds(table));
+ }
+
+ public IcebergTableMaintainer maintainer() {
+ TableMaintainerContext context = mock(TableMaintainerContext.class);
+ when(context.getTableConfiguration()).thenReturn(configuration);
+ when(context.getMetrics()).thenReturn(mock(MaintainerMetrics.class));
+ when(context.getOptimizingInfo()).thenReturn(mock(OptimizingInfo.class));
+ return new IcebergTableMaintainer(table, table.id(), context);
+ }
+
+ public void run(IcebergTableMaintainer maintainer) {
+ switch (action) {
+ case "snapshot":
+ maintainer.expireSnapshots();
+ break;
+ case "dangling":
+ maintainer.cleanDanglingDeleteFiles();
+ break;
+ case "data":
+ maintainer.expireData();
+ break;
+ case "orphan":
+ maintainer.cleanOrphanFiles();
+ break;
+ case "tag":
+ maintainer.autoCreateTags();
+ break;
+ default:
+ throw new IllegalArgumentException(action);
+ }
+ }
+
+ public void assertPreserved() throws Exception {
+ assertAll(
+ () -> IcebergV3TestTables.assertV3MetadataUnchanged(table,
metadataLocation, metadataBytes),
+ () -> assertEquals(snapshotId, table.currentSnapshot().snapshotId()),
+ () -> assertEquals(snapshots, snapshotCount()),
+ () -> assertEquals(refs, table.refs()),
+ () -> assertEquals(Arrays.asList(1, 2),
IcebergV3TestTables.readIds(table)),
+ () -> {
+ if (action.equals("orphan")) {
+ assertTrue(Files.exists(orphan));
+ }
+ });
+ }
+
+ public void assertChanged() throws Exception {
+ table.refresh();
+ if (action.equals("orphan")) {
+ assertFalse(Files.exists(orphan));
+ assertEquals(metadataLocation,
IcebergV3TestTables.metadataLocation(table));
+ } else {
+ assertNotEquals(metadataLocation,
IcebergV3TestTables.metadataLocation(table));
+ }
+ switch (action) {
+ case "snapshot":
+ assertEquals(1, snapshotCount());
+ break;
+ case "dangling":
+ assertEquals(
+ 0,
+
org.apache.amoro.formats.iceberg.utils.IcebergTableUtil.getDanglingDeleteFiles(table)
+ .size());
+ assertEquals("0",
table.currentSnapshot().summary().get("total-delete-files"));
+ break;
+ case "data":
+ assertTrue(IcebergV3TestTables.dataFiles(table).isEmpty());
+ break;
+ case "tag":
+ assertEquals(
+ 1,
+
table.refs().values().stream().filter(org.apache.iceberg.SnapshotRef::isTag).count());
+ break;
+ default:
+ break;
+ }
+ assertEquals(
+ action.equals("data") ? Collections.emptyList() : Arrays.asList(1, 2),
+ IcebergV3TestTables.readIds(table));
+ }
+
+ private long snapshotCount() {
+ return StreamSupport.stream(table.snapshots().spliterator(),
false).count();
+ }
+}
diff --git
a/amoro-format-iceberg/src/test/java/org/apache/amoro/formats/iceberg/IcebergV3TestTables.java
b/amoro-format-iceberg/src/test/java/org/apache/amoro/formats/iceberg/IcebergV3TestTables.java
new file mode 100644
index 000000000..fc616c966
--- /dev/null
+++
b/amoro-format-iceberg/src/test/java/org/apache/amoro/formats/iceberg/IcebergV3TestTables.java
@@ -0,0 +1,170 @@
+/*
+ * 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.amoro.formats.iceberg;
+
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.apache.amoro.io.IcebergDataTestHelpers;
+import org.apache.amoro.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import
org.apache.amoro.shade.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+import
org.apache.amoro.shade.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.amoro.table.TableIdentifier;
+import org.apache.amoro.table.TableMetaStore;
+import org.apache.amoro.table.UnkeyedTable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.IcebergGenerics;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.types.Types;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/** Local tables for testing metadata preservation by unsupported-version
guards. */
+public final class IcebergV3TestTables {
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private IcebergV3TestTables() {}
+
+ public static UnkeyedTable create(Path location, int formatVersion) throws
IOException {
+ Configuration conf = new Configuration();
+ HadoopTables tables = new HadoopTables(conf);
+ Schema schema = new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()));
+ Table table =
+ tables.create(
+ schema,
+ PartitionSpec.unpartitioned(),
+ Collections.singletonMap("format-version", "2"),
+ location.toString());
+ IcebergDataTestHelpers.append(table, records(table, 1));
+ IcebergDataTestHelpers.append(table, records(table, 2));
+
+ if (formatVersion == 3) {
+ return upgradeToV3(table);
+ }
+ assertEquals(2, formatVersion);
+ return wrap(table);
+ }
+
+ public static UnkeyedTable wrap(Table table) {
+ return IcebergTable.newIcebergTable(
+ TableIdentifier.of("test_catalog", "test_database", "test_table"),
+ table,
+ TableMetaStore.builder().withConfiguration(new
Configuration()).build(),
+ Collections.emptyMap())
+ .originalTable();
+ }
+
+ public static UnkeyedTable upgradeToV3(Table table) throws IOException {
+ // Commit through an independent handle so existing readers retain their
cached V2
+ // metadata and Hadoop's normal version-conflict detection remains
effective.
+ Table external = new HadoopTables(new
Configuration()).load(table.location());
+ String previousLocation = metadataLocation(external);
+ external.updateProperties().set("format-version", "3").commit();
+ external.refresh();
+ assertFalse(previousLocation.equals(metadataLocation(external)));
+
+ // Complete the synthetic external upgrade with the modern V3 JSON fields.
Only
+ // this freshly committed temporary fixture is edited, before control
returns to
+ // the old writer. Existing V2 snapshots remain valid and no row IDs are
allocated.
+ ObjectNode metadata = (ObjectNode)
MAPPER.readTree(metadataBytes(external));
+ metadata.put("next-row-id", 0L);
+ metadata.remove("row-lineage");
+ try (OutputStream output =
+
external.io().newOutputFile(metadataLocation(external)).createOrOverwrite()) {
+ MAPPER.writeValue(output, metadata);
+ }
+ UnkeyedTable upgraded = wrap(new HadoopTables(new
Configuration()).load(external.location()));
+ JsonNode raw = MAPPER.readTree(metadataBytes(upgraded));
+ assertEquals(3, raw.get("format-version").asInt());
+ assertTrue(raw.has("next-row-id"));
+ assertFalse(raw.has("row-lineage"));
+ return upgraded;
+ }
+
+ public static String metadataLocation(Table table) {
+ return ((HasTableOperations)
table).operations().current().metadataFileLocation();
+ }
+
+ public static byte[] metadataBytes(Table table) throws IOException {
+ try (InputStream input =
table.io().newInputFile(metadataLocation(table)).newStream()) {
+ return input.readAllBytes();
+ }
+ }
+
+ public static void assertV3MetadataUnchanged(
+ Table table, String originalLocation, byte[] originalBytes) throws
IOException {
+ table.refresh();
+ byte[] currentBytes = metadataBytes(table);
+ JsonNode currentJson = MAPPER.readTree(currentBytes);
+ assertAll(
+ () -> assertEquals(originalLocation, metadataLocation(table),
"Metadata pointer changed"),
+ () -> assertArrayEquals(originalBytes, currentBytes, "Current metadata
JSON changed"),
+ () -> assertTrue(currentJson.has("next-row-id"), "Commit dropped
next-row-id"),
+ () -> assertEquals(3, currentJson.get("format-version").asInt()));
+ }
+
+ public static List<DataFile> dataFiles(Table table) throws IOException {
+ List<DataFile> files = new ArrayList<>();
+ try (CloseableIterable<FileScanTask> tasks = table.newScan().planFiles()) {
+ for (FileScanTask task : tasks) {
+ files.add(task.file().copy());
+ }
+ }
+ return files;
+ }
+
+ public static List<Record> records(Table table, int... ids) {
+ List<Record> records = new ArrayList<>();
+ for (int id : ids) {
+ GenericRecord record = GenericRecord.create(table.schema());
+ record.setField("id", id);
+ records.add(record);
+ }
+ return records;
+ }
+
+ public static List<Integer> readIds(Table table) throws IOException {
+ List<Integer> ids = new ArrayList<>();
+ try (CloseableIterable<Record> records =
IcebergGenerics.read(table).build()) {
+ for (Record record : records) {
+ ids.add((Integer) record.getField("id"));
+ }
+ }
+ Collections.sort(ids);
+ return ids;
+ }
+}
diff --git
a/amoro-format-iceberg/src/test/java/org/apache/amoro/formats/iceberg/maintainer/TestIcebergV3MaintenanceGuard.java
b/amoro-format-iceberg/src/test/java/org/apache/amoro/formats/iceberg/maintainer/TestIcebergV3MaintenanceGuard.java
new file mode 100644
index 000000000..2f08e6517
--- /dev/null
+++
b/amoro-format-iceberg/src/test/java/org/apache/amoro/formats/iceberg/maintainer/TestIcebergV3MaintenanceGuard.java
@@ -0,0 +1,188 @@
+/*
+ * 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.amoro.formats.iceberg.maintainer;
+
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.apache.amoro.config.TableConfiguration;
+import org.apache.amoro.config.TagConfiguration;
+import org.apache.amoro.formats.iceberg.IcebergMaintenanceCompatibility;
+import org.apache.amoro.formats.iceberg.IcebergV3MaintenanceFixture;
+import org.apache.amoro.formats.iceberg.IcebergV3TestTables;
+import org.apache.amoro.maintainer.TableMaintainerContext;
+import org.apache.amoro.table.UnkeyedTable;
+import org.apache.iceberg.SnapshotRef;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+class TestIcebergV3MaintenanceGuard {
+ @TempDir private Path temp;
+
+ @ParameterizedTest
+ @ValueSource(strings = {"snapshot", "dangling", "orphan"})
+ void testDisabledMaintenanceRemainsNoOp(String action) throws Exception {
+ IcebergV3MaintenanceFixture fixture =
+ new IcebergV3MaintenanceFixture(temp.resolve(action), 3, action);
+ fixture.configuration.setExpireSnapshotEnabled(false);
+ fixture.configuration.setDeleteDanglingDeleteFilesEnabled(false);
+ fixture.configuration.setCleanOrphanEnabled(false);
+
+ IcebergTableMaintainer maintainer = fixture.maintainer();
+ assertAll(
+ () -> {
+ Map<String, String> result;
+ switch (action) {
+ case "snapshot":
+ result = maintainer.expireSnapshots();
+ break;
+ case "dangling":
+ result = maintainer.cleanDanglingDeleteFiles();
+ break;
+ case "orphan":
+ result = maintainer.cleanOrphanFiles();
+ break;
+ default:
+ throw new IllegalArgumentException(action);
+ }
+ assertTrue(result.isEmpty());
+ },
+ fixture::assertPreserved);
+ }
+
+ @Test
+ void testLoadedMetadataCheckDoesNotRefreshButWriterDoes() throws Exception {
+ UnkeyedTable table =
spy(IcebergV3TestTables.create(temp.resolve("refresh"), 2));
+
+ IcebergMaintenanceCompatibility.checkSupported(table);
+ verify(table, never()).refresh();
+
+ IcebergMaintenanceCompatibility.forUpdate(table);
+ verify(table).refresh();
+ }
+
+ @ParameterizedTest(name = "{0} on V{1}")
+ @CsvSource({
+ "snapshot,3",
+ "dangling,3",
+ "data,3",
+ "orphan,3",
+ "snapshot,2",
+ "dangling,2",
+ "data,2",
+ "orphan,2",
+ "snapshot,1",
+ "data,1",
+ "orphan,1",
+ "tag,1"
+ })
+ void testMaintenancePreservesUnsupportedTables(String action, int version)
throws Exception {
+ IcebergV3MaintenanceFixture fixture =
+ new IcebergV3MaintenanceFixture(temp.resolve(action), version, action);
+ IcebergTableMaintainer maintainer = fixture.maintainer();
+ if (version == 3) {
+ assertAll(
+ () -> {
+ IllegalArgumentException failure =
+ assertThrows(IllegalArgumentException.class, () ->
fixture.run(maintainer));
+ assertTrue(failure.getMessage().contains("Unsupported Iceberg
format version: 3"));
+ },
+ fixture::assertPreserved);
+ } else {
+ fixture.run(maintainer);
+ fixture.assertChanged();
+ if (action.equals("orphan")) {
+ assertSame(fixture.table, maintainer.table, "File cleanup does not
need a commit wrapper");
+ }
+ }
+ }
+
+ @Test
+ void testV3AutoTagDoesNotChangeMetadata() throws Exception {
+ UnkeyedTable table = IcebergV3TestTables.create(temp.resolve("v3"), 3);
+ assertAutoTagRejected(table, maintainer(table));
+ }
+
+ private void assertAutoTagRejected(UnkeyedTable table,
IcebergTableMaintainer maintainer)
+ throws Exception {
+ String metadataLocation = IcebergV3TestTables.metadataLocation(table);
+ byte[] metadataBytes = IcebergV3TestTables.metadataBytes(table);
+ Map<String, SnapshotRef> refs = new HashMap<>(table.refs());
+ long snapshotId = table.currentSnapshot().snapshotId();
+
+ // assertAll also checks persisted metadata when the missing rejection
assertion fails.
+ assertAll(
+ () -> {
+ IllegalArgumentException failure =
+ assertThrows(IllegalArgumentException.class,
maintainer::autoCreateTags);
+ assertTrue(failure.getMessage().contains("Unsupported Iceberg format
version: 3"));
+ },
+ () -> IcebergV3TestTables.assertV3MetadataUnchanged(table,
metadataLocation, metadataBytes),
+ () -> assertEquals(refs, table.refs(), "Auto tag changed snapshot
references"),
+ () -> assertEquals(snapshotId, table.currentSnapshot().snapshotId()),
+ () -> assertEquals(Arrays.asList(1, 2),
IcebergV3TestTables.readIds(table)));
+ }
+
+ @Test
+ void testV2AutoTagCommits() throws Exception {
+ UnkeyedTable table = IcebergV3TestTables.create(temp.resolve("v2"), 2);
+ String metadataLocation = IcebergV3TestTables.metadataLocation(table);
+ long snapshotId = table.currentSnapshot().snapshotId();
+ assertEquals(0,
table.refs().values().stream().filter(SnapshotRef::isTag).count());
+
+ maintainer(table).autoCreateTags();
+ table.refresh();
+
+ assertAll(
+ () -> assertNotEquals(metadataLocation,
IcebergV3TestTables.metadataLocation(table)),
+ () -> assertEquals(1,
table.refs().values().stream().filter(SnapshotRef::isTag).count()),
+ () -> assertEquals(snapshotId, table.currentSnapshot().snapshotId()),
+ () -> assertEquals(Arrays.asList(1, 2),
IcebergV3TestTables.readIds(table)));
+ }
+
+ private IcebergTableMaintainer maintainer(UnkeyedTable table) {
+ TagConfiguration tags = new TagConfiguration();
+ tags.setAutoCreateTag(true);
+ // The snapshot was just created, so it is eligible for the current
monthly boundary.
+ tags.setTriggerPeriod(TagConfiguration.Period.MONTHLY);
+ tags.setTagFormat("'tag-'yyyy-MM");
+ tags.setMaxDelayMinutes(0);
+ TableConfiguration configuration = new
TableConfiguration().setTagConfiguration(tags);
+ TableMaintainerContext context = mock(TableMaintainerContext.class);
+ when(context.getTableConfiguration()).thenReturn(configuration);
+ return new IcebergTableMaintainer(table, table.id(), context);
+ }
+}
diff --git a/docs/formats/iceberg.md b/docs/formats/iceberg.md
index 922bb9bd6..981ce57e1 100644
--- a/docs/formats/iceberg.md
+++ b/docs/formats/iceberg.md
@@ -37,7 +37,26 @@ However, as data and delete files are written, the read
performance and availabi
Starting from Amoro v0.4, Iceberg format including v1 and v2 is supported.
Users only need to register Iceberg's catalog in Amoro to host the table for
Amoro maintenance. For detailed operation steps, please refer to [Managing
Catalogs](../managing-catalogs/).
Amoro maintains the performance and economic availability of Iceberg tables
with minimal read/write costs through means such as small file merging,
eq-delete file conversion to pos-delete files, duplicate data elimination, and
file cleaning, and Amoro has no intrusive impact on the functionality of
Iceberg.
-Iceberg format has full upward and downward compatibility features, and in
general, users do not have to worry about the compatibility of the Iceberg
version used by the engine client with the Iceberg version on which Amoro
depends.
+### Format version compatibility
+
+Amoro automatic maintenance supports Iceberg format versions 1 and 2. Format
version 3
+is not supported by the bundled Iceberg writer: a metadata commit can omit
fields required
+by newer V3 readers and make the table inaccessible to those readers.
+
+Amoro skips optimizing evaluation and planning for V3 tables and rejects their
maintenance
+and optimizing commits. This also applies to a table upgraded while a
transaction is committing.
+Recovery checks the table version before resuming optimizing jobs. If the
catalog lookup fails,
+recovery continues with a warning; the writer still checks compatibility
before committing.
+Snapshot expiration, dangling delete cleanup, data expiration, automatic tags,
and orphan file
+cleanup are all blocked for unsupported tables.
+
+The compatibility guard does not block catalog discovery or metadata reads and
does not
+change the table's maintenance settings. Read-only table summary collection
remains available
+when optimizing is disabled and table summary collection is enabled. Loading
and reading still
+depend on the features supported by the bundled Iceberg reader. The table
name, unsupported
+version, and supported versions are reported in AMS logs; rejected maintenance
and optimizing
+processes closed by the guard also use the existing failure or closure
reporting. This protection
+does not repair metadata already written without required V3 fields.
Amoro supports all catalog types supported by Iceberg, including but not
limited to: Hadoop, Hive, Glue, JDBC, Nessie, Snowflake, and so on.
