diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/maintainer/DefaultTableMaintainerContext.java b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/maintainer/DefaultTableMaintainerContext.java index ca47e22522..cbc9527fbe 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/maintainer/DefaultTableMaintainerContext.java +++ b/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); + } }; } diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableConfigurations.java b/amoro-ams/src/main/java/org/apache/amoro/server/table/TableConfigurations.java index 2c403ebf32..e89668b426 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableConfigurations.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/table/TableConfigurations.java @@ -104,6 +104,11 @@ public static TableConfiguration parseTableConfig(Map 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)); diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableOrphanFilesCleaningMetrics.java b/amoro-ams/src/main/java/org/apache/amoro/server/table/TableOrphanFilesCleaningMetrics.java index 481eb175b3..95595855b0 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableOrphanFilesCleaningMetrics.java +++ b/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) () -> lastStatus); + registerMetric( + registry, + TABLE_ORPHAN_FILE_CLEANING_LAST_FAILURE_TIMESTAMP_MS, + (Gauge) () -> 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}. + * + *

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; + } } diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/maintainer/TestIcebergOrphanFileLocationConflict.java b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/maintainer/TestIcebergOrphanFileLocationConflict.java new file mode 100644 index 0000000000..6bc5b2d250 --- /dev/null +++ b/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)); + } +} diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/table/TestTableOrphanFilesCleaningMetrics.java b/amoro-ams/src/test/java/org/apache/amoro/server/table/TestTableOrphanFilesCleaningMetrics.java new file mode 100644 index 0000000000..53d4397569 --- /dev/null +++ b/amoro-ams/src/test/java/org/apache/amoro/server/table/TestTableOrphanFilesCleaningMetrics.java @@ -0,0 +1,110 @@ +/* + * 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.table; + +import org.apache.amoro.ServerTableIdentifier; +import org.apache.amoro.TableFormat; +import org.apache.amoro.metrics.MetricRegistry; +import org.apache.amoro.maintainer.MaintainerMetrics.CleanFailureReason; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +/** + * Unit tests for {@link TableOrphanFilesCleaningMetrics} — the 2 last-value gauges introduced for + * orphan-file-cleaning monitoring. These tests exercise the {@code recordSuccess()} and {@code + * recordFailure(reason)} methods directly against the metric objects registered in a {@link + * MetricRegistry}, verifying the Gauge last-value semantics. + */ +public class TestTableOrphanFilesCleaningMetrics { + + private MetricRegistry registry; + private TableOrphanFilesCleaningMetrics metrics; + + @Before + public void setUp() { + registry = new MetricRegistry(); + metrics = + new TableOrphanFilesCleaningMetrics( + ServerTableIdentifier.of("test_catalog", "test_db", "test_table", TableFormat.ICEBERG)); + metrics.register(registry); + } + + @After + public void tearDown() { + if (metrics != null) { + metrics.unregister(); + } + } + + // ---- baseline initialization ---- + + @Test + public void testInitialStateIsSuccess() { + assertEquals(TableOrphanFilesCleaningMetrics.STATUS_SUCCESS, metrics.getLastStatus()); + assertEquals(0L, metrics.getLastFailureTimestampMs()); + } + + // ---- recordSuccess resets last_status but preserves lastFailureTimestampMs ---- + + @Test + public void testRecordSuccessAfterFailureResetsStatus() { + metrics.recordFailure(CleanFailureReason.LOCATION_CONFLICT); + assertEquals(CleanFailureReason.LOCATION_CONFLICT.statusCode(), metrics.getLastStatus()); + long failureTsBefore = metrics.getLastFailureTimestampMs(); + assertTrue("lastFailureTimestampMs should be > 0 after a real failure", failureTsBefore > 0); + + metrics.recordSuccess(); + assertEquals(TableOrphanFilesCleaningMetrics.STATUS_SUCCESS, metrics.getLastStatus()); + assertEquals( + "recordSuccess must NOT reset lastFailureTimestampMs — timestamp persists so monitoring" + + " can alert on stale-failure windows", + failureTsBefore, + metrics.getLastFailureTimestampMs()); + } + + // ---- recordFailure for each "real failure" reason updates last_status / ts ---- + + @Test + public void testLocationConflictUpdatesLastStatusAndTs() { + metrics.recordFailure(CleanFailureReason.LOCATION_CONFLICT); + assertEquals(CleanFailureReason.LOCATION_CONFLICT.statusCode(), metrics.getLastStatus()); + assertTrue( + "lastFailureTimestampMs should be > 0 on a real failure", + metrics.getLastFailureTimestampMs() > 0); + } + + @Test + public void testLocationCheckUnavailableUpdatesLastStatusAndTs() { + metrics.recordFailure(CleanFailureReason.LOCATION_CONFLICT_CHECK_FAILED); + assertEquals( + CleanFailureReason.LOCATION_CONFLICT_CHECK_FAILED.statusCode(), metrics.getLastStatus()); + assertTrue(metrics.getLastFailureTimestampMs() > 0); + } + + @Test + public void testExecutionFailedUpdatesLastStatusAndTs() { + metrics.recordFailure(CleanFailureReason.EXECUTION_FAILED); + assertEquals(CleanFailureReason.EXECUTION_FAILED.statusCode(), metrics.getLastStatus()); + assertTrue(metrics.getLastFailureTimestampMs() > 0); + } +} diff --git a/amoro-common/src/main/java/org/apache/amoro/config/TableConfiguration.java b/amoro-common/src/main/java/org/apache/amoro/config/TableConfiguration.java index 08d700d1ec..879efccbd7 100644 --- a/amoro-common/src/main/java/org/apache/amoro/config/TableConfiguration.java +++ b/amoro-common/src/main/java/org/apache/amoro/config/TableConfiguration.java @@ -33,6 +33,7 @@ public class TableConfiguration { private boolean cleanOrphanEnabled; private long orphanExistingMinutes; private boolean deleteDanglingDeleteFilesEnabled; + private boolean ignoreLocationConflictWhenCleanOrphan; private OptimizingConfig optimizingConfig; private DataExpirationConfig expiringDataConfig; private TagConfiguration tagConfiguration; @@ -121,6 +122,16 @@ public TableConfiguration setDeleteDanglingDeleteFilesEnabled( return this; } + public boolean isIgnoreLocationConflictWhenCleanOrphan() { + return ignoreLocationConflictWhenCleanOrphan; + } + + public TableConfiguration setIgnoreLocationConflictWhenCleanOrphan( + boolean ignoreLocationConflictWhenCleanOrphan) { + this.ignoreLocationConflictWhenCleanOrphan = ignoreLocationConflictWhenCleanOrphan; + return this; + } + public DataExpirationConfig getExpiringDataConfig() { return Optional.ofNullable(expiringDataConfig).orElse(new DataExpirationConfig()); } @@ -156,6 +167,7 @@ public boolean equals(Object o) { && cleanOrphanEnabled == that.cleanOrphanEnabled && orphanExistingMinutes == that.orphanExistingMinutes && deleteDanglingDeleteFilesEnabled == that.deleteDanglingDeleteFilesEnabled + && ignoreLocationConflictWhenCleanOrphan == that.ignoreLocationConflictWhenCleanOrphan && Objects.equal(optimizingConfig, that.optimizingConfig) && Objects.equal(expiringDataConfig, that.expiringDataConfig) && Objects.equal(tagConfiguration, that.tagConfiguration); @@ -172,6 +184,7 @@ public int hashCode() { cleanOrphanEnabled, orphanExistingMinutes, deleteDanglingDeleteFilesEnabled, + ignoreLocationConflictWhenCleanOrphan, optimizingConfig, expiringDataConfig, tagConfiguration); diff --git a/amoro-common/src/main/java/org/apache/amoro/maintainer/MaintainerMetrics.java b/amoro-common/src/main/java/org/apache/amoro/maintainer/MaintainerMetrics.java index 420c61e3b0..e49df6b25c 100644 --- a/amoro-common/src/main/java/org/apache/amoro/maintainer/MaintainerMetrics.java +++ b/amoro-common/src/main/java/org/apache/amoro/maintainer/MaintainerMetrics.java @@ -40,6 +40,36 @@ public interface MaintainerMetrics { */ void recordOrphanMetadataFilesCleaned(int expected, int cleaned); + /** Record a successful orphan-file-cleaning run. */ + void recordSuccess(); + + /** + * Record an orphan-file-cleaning failure event. + * + * @param reason the failure reason + */ + void recordFailure(CleanFailureReason reason); + + /** Atomic failure reasons for orphan-file-cleaning observability. */ + enum CleanFailureReason { + /** Another table uuid detected in the same location; cleanup skipped. */ + LOCATION_CONFLICT(1), + /** The location-conflict check itself failed (e.g. FileIO doesn't support prefix ops). */ + LOCATION_CONFLICT_CHECK_FAILED(2), + /** The actual cleaning execution threw an exception. */ + EXECUTION_FAILED(3); + + private final int statusCode; + + CleanFailureReason(int statusCode) { + this.statusCode = statusCode; + } + + public int statusCode() { + return statusCode; + } + } + /** No-op implementation that does nothing. */ MaintainerMetrics NOOP = new MaintainerMetrics() { @@ -48,5 +78,11 @@ public void recordOrphanDataFilesCleaned(int expected, int cleaned) {} @Override public void recordOrphanMetadataFilesCleaned(int expected, int cleaned) {} + + @Override + public void recordSuccess() {} + + @Override + public void recordFailure(CleanFailureReason reason) {} }; } 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 68d90e3a17..388886c0ef 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 @@ -31,6 +31,7 @@ import org.apache.amoro.io.PathInfo; import org.apache.amoro.io.SupportsFileSystemOperations; import org.apache.amoro.maintainer.MaintainerMetrics; +import org.apache.amoro.maintainer.MaintainerMetrics.CleanFailureReason; import org.apache.amoro.maintainer.OptimizingInfo; import org.apache.amoro.maintainer.TableMaintainer; import org.apache.amoro.maintainer.TableMaintainerContext; @@ -41,6 +42,7 @@ import org.apache.amoro.shade.guava32.com.google.common.collect.Maps; import org.apache.amoro.shade.guava32.com.google.common.collect.Sets; import org.apache.amoro.table.TableIdentifier; +import org.apache.amoro.table.TableProperties; import org.apache.amoro.utils.TableFileUtil; import org.apache.iceberg.ContentFile; import org.apache.iceberg.ContentScanTask; @@ -134,6 +136,18 @@ public IcebergTableMaintainer( this.context = context; } + /** Outcome of the location-conflict check before orphan-file cleanup. */ + private enum LocationConflictCheckResult { + /** The check was skipped (configured to ignore) — proceed normally. */ + CONFLICT_CHECK_SKIPPED, + /** No conflict detected — proceed normally. */ + NO_CONFLICT, + /** Another table uuid detected in the same location — skip cleanup. */ + CONFLICT_DETECTED, + /** The check itself failed (e.g. FileIO doesn't support prefix ops) — skip cleanup. */ + CONFLICT_CHECK_FAILED + } + @Override public Map cleanOrphanFiles() { TableConfiguration tableConfiguration = context.getTableConfiguration(); @@ -143,15 +157,39 @@ public Map cleanOrphanFiles() { return Maps.newHashMap(); } + LocationConflictCheckResult checkResult = checkLocationConflict(tableConfiguration); + + switch (checkResult) { + case CONFLICT_DETECTED: + metrics.recordFailure(CleanFailureReason.LOCATION_CONFLICT); + return Maps.newHashMap(); + case CONFLICT_CHECK_FAILED: + metrics.recordFailure(CleanFailureReason.LOCATION_CONFLICT_CHECK_FAILED); + return Maps.newHashMap(); + default: + break; + } + long keepTime = tableConfiguration.getOrphanExistingMinutes() * 60 * 1000; - int dataDeleted = cleanContentFiles(System.currentTimeMillis() - keepTime, metrics); + int dataDeleted; + int metadataDeleted; + try { + dataDeleted = cleanContentFiles(System.currentTimeMillis() - keepTime, metrics); + + // refresh + table.refresh(); - // refresh - table.refresh(); + // clear metadata files + metadataDeleted = cleanMetadata(System.currentTimeMillis() - keepTime, metrics); - // clear metadata files - int metadataDeleted = cleanMetadata(System.currentTimeMillis() - keepTime, metrics); + // cleanup completed — reset status to SUCCESS + metrics.recordSuccess(); + } catch (Exception e) { + metrics.recordFailure(CleanFailureReason.EXECUTION_FAILED); + LOG.error("Failed to clean orphan files for table {}", table.name(), e); + throw e; + } Map summary = Maps.newLinkedHashMap(); summary.put("orphan-data-files-cleaned", String.valueOf(dataDeleted)); @@ -159,6 +197,44 @@ public Map cleanOrphanFiles() { return summary; } + /** + * Checks whether the table's location is shared with another table before cleaning orphan files. + * All exceptions are handled internally — this method always returns a valid result. + * + * @return the location-conflict check result. + */ + private LocationConflictCheckResult checkLocationConflict(TableConfiguration tableConfiguration) { + if (tableConfiguration.isIgnoreLocationConflictWhenCleanOrphan()) { + LOG.warn( + "Table property {} is enabled for table {} at '{}'; skipping location-conflict check and proceeding with cleanup.", + TableProperties.IGNORE_LOCATION_CONFLICT_WHEN_CLEAN_ORPHAN, + table.name(), + table.location()); + return LocationConflictCheckResult.CONFLICT_CHECK_SKIPPED; + } + + try { + if (IcebergTableUtil.hasOtherTableInLocation(table)) { + LOG.warn( + "Table {} has other table in location {}, skip clean orphan files", + table.name(), + table.location()); + return LocationConflictCheckResult.CONFLICT_DETECTED; + } else { + return LocationConflictCheckResult.NO_CONFLICT; + } + } catch (Exception e) { + // Real IO failure (FileIO network timeout, S3/HDFS unreachable, …) — the system + // is unable to determine location-conflict safety, so skip the cleanup cycle. + LOG.warn( + "Location-conflict check failed for table {} at {}. Skipping orphan cleanup, exception is as follows: {}", + table.name(), + table.location(), + e); + return LocationConflictCheckResult.CONFLICT_CHECK_FAILED; + } + } + @Override public Map cleanDanglingDeleteFiles() { TableConfiguration tableConfiguration = context.getTableConfiguration(); diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/utils/IcebergTableUtil.java b/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/utils/IcebergTableUtil.java index 23a5b48a28..8be83a5af2 100644 --- a/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/utils/IcebergTableUtil.java +++ b/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/utils/IcebergTableUtil.java @@ -24,7 +24,9 @@ import org.apache.amoro.shade.guava32.com.google.common.base.Predicate; import org.apache.amoro.shade.guava32.com.google.common.collect.Iterables; import org.apache.amoro.shade.guava32.com.google.common.collect.Lists; +import org.apache.amoro.shade.guava32.com.google.common.collect.Sets; import org.apache.amoro.utils.TableFileUtil; +import org.apache.hadoop.fs.Path; import org.apache.iceberg.ContentFile; import org.apache.iceberg.DataOperations; import org.apache.iceberg.DeleteFile; @@ -37,9 +39,15 @@ import org.apache.iceberg.Snapshot; import org.apache.iceberg.StructLike; import org.apache.iceberg.Table; +import org.apache.iceberg.TableMetadata; +import org.apache.iceberg.TableMetadataParser; import org.apache.iceberg.TableOperations; import org.apache.iceberg.TableScan; +import org.apache.iceberg.exceptions.ValidationException; import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.io.FileInfo; +import org.apache.iceberg.io.SupportsPrefixOperations; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -193,4 +201,95 @@ public static Set getAllManifestFiles(Table table) { return allManifestFiles; } + + private static boolean isMetadataJson(String name) { + return name.endsWith(".metadata.json") || name.endsWith(".metadata.json.gz"); + } + + /** + * Returns {@code true} if another Iceberg table appears to share the same metadata location as + * the given table. + * + *

Lists the table's metadata directory via its own (storage-agnostic) {@link FileIO} and, for + * every {@code metadata.json} not in this table's own history (current + {@code + * previousFiles()}), reads its {@code table-uuid}: + * + *

    + *
  • same uuid → older version of this table, ignored; + *
  • different uuid → another table shares the location, returns {@code true}; + *
  • uuid missing/unreadable (legacy, corrupt, or compressed) → treated as a conflict, + * returns {@code true} (fail-safe). + *
+ * + *

Requires {@link SupportsPrefixOperations} for FileIO; otherwise a {@link + * ValidationException} is thrown. The caller may choose to skip the check when the storage + * backend is known to be used by a single table. On the no-conflict path only a single listing is + * done and no metadata file is read. + * + * @param table the table whose location is about to be cleaned + * @throws ValidationException if the table's {@code FileIO} does not support prefix operations + */ + public static boolean hasOtherTableInLocation(Table table) { + TableOperations ops = ((HasTableOperations) table).operations(); + TableMetadata current = ops.current(); + + String myUuid = current.uuid(); + Set myMetadataFiles = Sets.newHashSet(); + myMetadataFiles.add(new Path(current.metadataFileLocation()).getName()); + for (TableMetadata.MetadataLogEntry entry : current.previousFiles()) { + myMetadataFiles.add(new Path(entry.file()).getName()); + } + + Path metadataDir = new Path(current.metadataFileLocation()).getParent(); + + FileIO io = table.io(); + if (!(io instanceof SupportsPrefixOperations)) { + String msg = + String.format( + "Cannot detect location conflicts: the table's FileIO (%s) does not support prefix " + + "operations, which are required to inspect the metadata directory '%s'.", + io.getClass().getName(), metadataDir); + throw new ValidationException(msg); + } + + String prefix = metadataDir.toString(); + if (!prefix.endsWith("/")) { + prefix = prefix + "/"; + } + + SupportsPrefixOperations prefixIo = (SupportsPrefixOperations) io; + for (FileInfo info : prefixIo.listPrefix(prefix)) { + String name = new Path(info.location()).getName(); + if (!isMetadataJson(name) || myMetadataFiles.contains(name)) { + continue; + } + + String otherUuid = readTableUuid(table, info.location()); + if (otherUuid == null || !otherUuid.equals(myUuid)) { + LOG.warn( + "Another table (uuid={}) belonging to metadata file {} shares the metadata location with this table (uuid={}); " + + "treating the location as conflicting.", + otherUuid, + info.location(), + myUuid); + return true; + } + } + + return false; + } + + /** + * Reads the {@code table-uuid} from a metadata file, transparently handling gzip-compressed + * metadata ({@code .metadata.json.gz}). Returns {@code null} if the uuid cannot be determined + * (legacy file without a uuid, corrupt file, or any read failure). + */ + private static String readTableUuid(Table table, String metadataLocation) { + try { + return TableMetadataParser.read(table.io(), metadataLocation).uuid(); + } catch (Exception e) { + LOG.warn("Failed to read table-uuid from {}; treating as unreadable", metadataLocation, e); + return null; + } + } } diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/table/TableProperties.java b/amoro-format-iceberg/src/main/java/org/apache/amoro/table/TableProperties.java index 6c17f6ad0d..b615808f24 100644 --- a/amoro-format-iceberg/src/main/java/org/apache/amoro/table/TableProperties.java +++ b/amoro-format-iceberg/src/main/java/org/apache/amoro/table/TableProperties.java @@ -206,6 +206,17 @@ private TableProperties() {} "clean-orphan-file.min-existing-time-minutes"; public static final long MIN_ORPHAN_FILE_EXISTING_TIME_DEFAULT = 2880; // 2 Days + /** + * When true, ignore the shared-location conflict check and proceed cleaning orphan files even if + * the table location appears shared with another table. Warning: if the location is actually + * shared with another table, this may delete its files and corrupt that table. Default false + * skips cleanup. + */ + public static final String IGNORE_LOCATION_CONFLICT_WHEN_CLEAN_ORPHAN = + "clean-orphan-file.ignore-location-conflict"; + + public static final boolean IGNORE_LOCATION_CONFLICT_WHEN_CLEAN_ORPHAN_DEFAULT = false; + public static final String ENABLE_DANGLING_DELETE_FILES_CLEAN = "clean-dangling-delete-files.enabled"; public static final boolean ENABLE_DANGLING_DELETE_FILES_CLEAN_DEFAULT = true; diff --git a/docs/user-guides/configurations.md b/docs/user-guides/configurations.md index beb0920221..4adc4a9dbf 100644 --- a/docs/user-guides/configurations.md +++ b/docs/user-guides/configurations.md @@ -79,6 +79,7 @@ Data-cleaning configurations are applicable to both Iceberg Format and Mixed str | snapshot.keep.flink.checkpoint-retention | 7d(7 days) | The retention period for snapshots created by Flink checkpoints. Snapshots older than this duration may be cleaned up. The value should be specified as a duration string (e.g., "7d", "168h", "10080min") | | clean-orphan-file.enabled | false | Enables periodically clean orphan files | | clean-orphan-file.min-existing-time-minutes | 2880(2 days) | Cleaning orphan files keeps the files modified within a specified time in minutes | +| clean-orphan-file.ignore-location-conflict | false | Set to `true` to ignore the shared-location conflict check and proceed cleaning even if the table location appears shared. Warning: if the location is actually shared with another table, this may delete its files and corrupt that table. Default `false` skips cleanup | | clean-dangling-delete-files.enabled | true | Whether to enable cleaning of dangling delete files | | data-expire.enabled | false | Whether to enable data expiration | | data-expire.level | partition | Level of data expiration. Including partition and file |