[WIP][AMORO-4292][AMS] Guard IcebergTableMaintainer orphan cleanup against shared table locations

amoro-ams/src/main/java/org/apache/amoro/server/optimizing/maintainer/DefaultTableMaintainerContext.java

@@ -67,6 +67,16 @@ public void recordOrphanDataFilesCleaned(int expected, int cleaned) {
       public void recordOrphanMetadataFilesCleaned(int expected, int cleaned) {
         metrics.completeOrphanMetadataFiles(expected, cleaned);
       }
+
+      @Override
+      public void recordSuccess() {
+        metrics.recordSuccess();
+      }
+
+      @Override
+      public void recordFailure(MaintainerMetrics.CleanFailureReason reason) {
+        metrics.recordFailure(reason);
+      }
     };
   }
 

amoro-ams/src/main/java/org/apache/amoro/server/table/TableConfigurations.java

@@ -104,6 +104,11 @@ public static TableConfiguration parseTableConfig(Map<String, String> properties
                     properties,
                     TableProperties.ENABLE_DANGLING_DELETE_FILES_CLEAN,
                     TableProperties.ENABLE_DANGLING_DELETE_FILES_CLEAN_DEFAULT))
+        .setIgnoreLocationConflictWhenCleanOrphan(
+            CompatiblePropertyUtil.propertyAsBoolean(
+                properties,
+                TableProperties.IGNORE_LOCATION_CONFLICT_WHEN_CLEAN_ORPHAN,
+                TableProperties.IGNORE_LOCATION_CONFLICT_WHEN_CLEAN_ORPHAN_DEFAULT))
         .setOptimizingConfig(parseOptimizingConfig(properties))
         .setExpiringDataConfig(parseDataExpirationConfig(properties))
         .setTagConfiguration(parseTagConfiguration(properties));

amoro-ams/src/main/java/org/apache/amoro/server/table/TableOrphanFilesCleaningMetrics.java

@@ -19,10 +19,13 @@
 package org.apache.amoro.server.table;
 
 import static org.apache.amoro.metrics.MetricDefine.defineCounter;
+import static org.apache.amoro.metrics.MetricDefine.defineGauge;
 
 import org.apache.amoro.ServerTableIdentifier;
 import org.apache.amoro.maintainer.MaintainerMetrics;
+import org.apache.amoro.maintainer.MaintainerMetrics.CleanFailureReason;
 import org.apache.amoro.metrics.Counter;
+import org.apache.amoro.metrics.Gauge;
 import org.apache.amoro.metrics.MetricDefine;
 import org.apache.amoro.metrics.MetricRegistry;
 
@@ -35,6 +38,13 @@ public class TableOrphanFilesCleaningMetrics extends AbstractTableMetrics
   private final Counter orphanMetadataFilesCount = new Counter();
   private final Counter expectedOrphanMetadataFilesCount = new Counter();
 
+  // ---- last-value gauges ----
+  private volatile int lastStatus = STATUS_SUCCESS;
+  private volatile long lastFailureTimestampMs = 0L;
+
+  // --- status constants ---
+  public static final int STATUS_SUCCESS = 0;
+
   public TableOrphanFilesCleaningMetrics(ServerTableIdentifier identifier) {
     super(identifier);
   }
@@ -65,6 +75,24 @@ public TableOrphanFilesCleaningMetrics(ServerTableIdentifier identifier) {
           .withTags("catalog", "database", "table")
           .build();
 
+  // ---- new orphan-file-cleaning status metrics ----
+
+  public static final MetricDefine TABLE_ORPHAN_FILE_CLEANING_LAST_STATUS =
+      defineGauge("table_orphan_file_cleaning_last_status")
+          .withDescription(
+              "Status of the most recent orphan-file-cleaning attempt; "
+                  + "see MaintainerMetrics.CleanFailureReason: "
+                  + "0=SUCCESS, 1=LOCATION_CONFLICT, 2=LOCATION_CONFLICT_CHECK_FAILED, "
+                  + "3=EXECUTION_FAILED")
+          .withTags("catalog", "database", "table")
+          .build();
+
+  public static final MetricDefine TABLE_ORPHAN_FILE_CLEANING_LAST_FAILURE_TIMESTAMP_MS =
+      defineGauge("table_orphan_file_cleaning_last_failure_timestamp_ms")
+          .withDescription("Epoch millis of the last real orphan-file-cleaning failure")
+          .withTags("catalog", "database", "table")
+          .build();
+
   @Override
   public void registerMetrics(MetricRegistry registry) {
     if (globalRegistry == null) {
@@ -78,6 +106,15 @@ public void registerMetrics(MetricRegistry registry) {
           registry,
           TABLE_EXPECTED_ORPHAN_METADATA_FILE_CLEANING_COUNT,
           expectedOrphanMetadataFilesCount);
+
+      // new gauges
+      registerMetric(
+          registry, TABLE_ORPHAN_FILE_CLEANING_LAST_STATUS, (Gauge<Integer>) () -> lastStatus);
+      registerMetric(
+          registry,
+          TABLE_ORPHAN_FILE_CLEANING_LAST_FAILURE_TIMESTAMP_MS,
+          (Gauge<Long>) () -> lastFailureTimestampMs);
+
       globalRegistry = registry;
     }
   }
@@ -101,4 +138,40 @@ public void recordOrphanDataFilesCleaned(int expected, int cleaned) {
   public void recordOrphanMetadataFilesCleaned(int expected, int cleaned) {
     completeOrphanMetadataFiles(expected, cleaned);
   }
+
+  // ---- public mutation API ----
+
+  /**
+   * Record a successful orphan-file-cleaning run. Resets {@code last_status} to {@link
+   * #STATUS_SUCCESS}.
+   *
+   * <p>Note: {@code lastFailureTimestampMs} is intentionally preserved across success runs. It
+   * tracks the time of the most recent real failure and is needed by monitoring to alert on
+   * stale-failure windows. Only an actual failure (via {@link #recordFailure(CleanFailureReason)})
+   * refreshes it; a success run is independent.
+   */
+  @Override
+  public void recordSuccess() {
+    this.lastStatus = STATUS_SUCCESS;
+  }
+
+  /**
+   * Record a failure event. Updates {@code last_status} to the failure reason and refreshes {@code
+   * lastFailureTimestampMs}.
+   */
+  @Override
+  public void recordFailure(CleanFailureReason reason) {
+    this.lastStatus = reason.statusCode();
+    this.lastFailureTimestampMs = System.currentTimeMillis();
+  }
+
+  // ---- package-private / test-visible accessors ----
+
+  int getLastStatus() {
+    return lastStatus;
+  }
+
+  long getLastFailureTimestampMs() {
+    return lastFailureTimestampMs;
+  }
 }

amoro-ams/src/test/java/org/apache/amoro/server/optimizing/maintainer/TestIcebergOrphanFileLocationConflict.java

@@ -0,0 +1,185 @@
+/*
+ * 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.maintainer;
+
+import static org.apache.amoro.formats.iceberg.maintainer.IcebergTableMaintainer.DATA_FOLDER_NAME;
+
+import org.apache.amoro.BasicTableTestHelper;
+import org.apache.amoro.TableFormat;
+import org.apache.amoro.TableTestHelper;
+import org.apache.amoro.catalog.BasicCatalogTestHelper;
+import org.apache.amoro.catalog.CatalogTestHelper;
+import org.apache.amoro.formats.iceberg.maintainer.IcebergTableMaintainer;
+import org.apache.amoro.formats.iceberg.utils.IcebergTableUtil;
+import org.apache.amoro.server.scheduler.inline.ExecutorTestBase;
+import org.apache.amoro.table.TableProperties;
+import org.apache.amoro.table.UnkeyedTable;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.iceberg.io.FileIO;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.mockito.Mockito;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collections;
+
+@RunWith(Parameterized.class)
+public class TestIcebergOrphanFileLocationConflict extends ExecutorTestBase {
+
+  @Parameterized.Parameters(name = "{0}, {1}")
+  public static Object[] parameters() {
+    return new Object[][] {
+      {new BasicCatalogTestHelper(TableFormat.ICEBERG), new BasicTableTestHelper(false, true)},
+      {new BasicCatalogTestHelper(TableFormat.ICEBERG), new BasicTableTestHelper(false, false)}
+    };
+  }
+
+  public TestIcebergOrphanFileLocationConflict(
+      CatalogTestHelper catalogTestHelper, TableTestHelper tableTestHelper) {
+    super(catalogTestHelper, tableTestHelper);
+  }
+
+  private UnkeyedTable baseTable() {
+    return getMixedTable().asUnkeyedTable();
+  }
+
+  /** No conflict: a freshly committed Iceberg table is the only one in its metadata location. */
+  @Test
+  public void testHasOtherTableInLocationNoConflict() {
+    baseTable().newAppend().commit();
+    Assert.assertFalse(IcebergTableUtil.hasOtherTableInLocation(baseTable()));
+  }
+
+  /**
+   * Fail-safe conflict: a corrupt/legacy metadata json (unreadable uuid) in the metadata directory
+   * is treated as a conflict and returns {@code true}.
+   */
+  @Test
+  public void testHasOtherTableInLocationWithCorruptMetadata() throws IOException {
+    baseTable().newAppend().commit();
+    String corruptMeta =
+        baseTable().location() + File.separator + "metadata" + File.separator + "v0.metadata.json";
+    baseTable().io().newOutputFile(corruptMeta).createOrOverwrite().close();
+    Assert.assertTrue(baseTable().io().exists(corruptMeta));
+    Assert.assertTrue(IcebergTableUtil.hasOtherTableInLocation(baseTable()));
+  }
+
+  /**
+   * When the FileIO does not support prefix operations, the conflict detection throws a {@link
+   * ValidationException} instead of silently proceeding.
+   */
+  @Test
+  public void testHasOtherTableInLocationThrowsWhenFileIoLacksPrefixSupport() {
+    Table table =
+        Mockito.mock(Table.class, Mockito.withSettings().extraInterfaces(HasTableOperations.class));
+    TableOperations ops = Mockito.mock(TableOperations.class);
+    TableMetadata current = Mockito.mock(TableMetadata.class);
+    FileIO io = Mockito.mock(FileIO.class);
+
+    Mockito.when(((HasTableOperations) table).operations()).thenReturn(ops);
+    Mockito.when(ops.current()).thenReturn(current);
+    Mockito.when(current.uuid()).thenReturn("my-uuid");
+    Mockito.when(current.metadataFileLocation()).thenReturn("/tmp/meta/metadata/v1.metadata.json");
+    Mockito.when(current.previousFiles()).thenReturn(Collections.emptyList());
+    Mockito.when(table.io()).thenReturn(io);
+
+    try {
+      IcebergTableUtil.hasOtherTableInLocation(table);
+      Assert.fail("Expected ValidationException because the FileIO lacks prefix support");
+    } catch (ValidationException e) {
+      // expected
+    }
+  }
+
+  /**
+   * Default behavior: when a location conflict is detected, {@code cleanOrphanFiles} skips cleanup
+   * so it does not risk deleting another table's files.
+   */
+  @Test
+  public void testLocationConflictSkipsCleanupByDefault() throws IOException {
+    baseTable().newAppend().commit();
+    UnkeyedTable baseTable = baseTable();
+
+    String orphanDir =
+        baseTable.location() + File.separator + DATA_FOLDER_NAME + File.separator + "testLocation";
+    String orphanFile = orphanDir + File.separator + "orphan.parquet";
+    baseTable.io().newOutputFile(orphanFile).createOrOverwrite().close();
+    Assert.assertTrue(baseTable.io().exists(orphanFile));
+
+    // Simulate a location conflict: a corrupt metadata json in the metadata directory.
+    String corruptMeta =
+        baseTable.location() + File.separator + "metadata" + File.separator + "v0.metadata.json";
+    baseTable.io().newOutputFile(corruptMeta).createOrOverwrite().close();
+    Assert.assertTrue(IcebergTableUtil.hasOtherTableInLocation(baseTable));
+
+    baseTable
+        .updateProperties()
+        .set(TableProperties.ENABLE_ORPHAN_CLEAN, "true")
+        .set(TableProperties.MIN_ORPHAN_FILE_EXISTING_TIME, "0")
+        .commit();
+
+    new IcebergTableMaintainer(baseTable, baseTable.id(), TestTableMaintainerContext.of(baseTable))
+        .cleanOrphanFiles();
+
+    // Conflict detected -> cleanup skipped, orphan file must remain.
+    Assert.assertTrue(baseTable.io().exists(orphanFile));
+  }
+
+  /**
+   * When {@code clean-orphan-file.ignore-location-conflict=true}, the conflict is ignored and
+   * orphan files are cleaned up even though another table appears to share the location.
+   */
+  @Test
+  public void testLocationConflictIgnoredWhenPropertyEnabled() throws IOException {
+    baseTable().newAppend().commit();
+    UnkeyedTable baseTable = baseTable();
+
+    String orphanDir =
+        baseTable.location() + File.separator + DATA_FOLDER_NAME + File.separator + "testLocation";
+    String orphanFile = orphanDir + File.separator + "orphan.parquet";
+    baseTable.io().newOutputFile(orphanFile).createOrOverwrite().close();
+    Assert.assertTrue(baseTable.io().exists(orphanFile));
+
+    // Simulate a location conflict: a corrupt metadata json in the metadata directory.
+    String corruptMeta =
+        baseTable.location() + File.separator + "metadata" + File.separator + "v0.metadata.json";
+    baseTable.io().newOutputFile(corruptMeta).createOrOverwrite().close();
+    Assert.assertTrue(IcebergTableUtil.hasOtherTableInLocation(baseTable));
+
+    baseTable
+        .updateProperties()
+        .set(TableProperties.ENABLE_ORPHAN_CLEAN, "true")
+        .set(TableProperties.MIN_ORPHAN_FILE_EXISTING_TIME, "0")
+        .set(TableProperties.IGNORE_LOCATION_CONFLICT_WHEN_CLEAN_ORPHAN, "true")
+        .commit();
+
+    new IcebergTableMaintainer(baseTable, baseTable.id(), TestTableMaintainerContext.of(baseTable))
+        .cleanOrphanFiles();
+
+    // Conflict ignored -> cleanup proceeds, orphan file must be deleted.
+    Assert.assertFalse(baseTable.io().exists(orphanFile));
+  }
+}