[AMORO-4022] [Improvement]: AMS Iceberg maintainer moved to the amoro-iceberg module

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

@@ -0,0 +1,109 @@
+/*
+ * 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 org.apache.amoro.config.TableConfiguration;
+import org.apache.amoro.maintainer.MaintainerMetrics;
+import org.apache.amoro.maintainer.OptimizingInfo;
+import org.apache.amoro.maintainer.TableMaintainerContext;
+import org.apache.amoro.server.table.DefaultTableRuntime;
+import org.apache.amoro.server.table.TableOrphanFilesCleaningMetrics;
+import org.apache.amoro.server.utils.HiveLocationUtil;
+import org.apache.amoro.table.MixedTable;
+
+import java.util.Collections;
+import java.util.Set;
+
+/**
+ * Default implementation of TableMaintainerContext for AMS. Adapts DefaultTableRuntime to
+ * TableMaintainerContext interface.
+ */
+public class DefaultTableMaintainerContext implements TableMaintainerContext {
+
+  private final DefaultTableRuntime tableRuntime;
+  private final MixedTable mixedTable;
+
+  public DefaultTableMaintainerContext(DefaultTableRuntime tableRuntime) {
+    this.tableRuntime = tableRuntime;
+    this.mixedTable = null;
+  }
+
+  public DefaultTableMaintainerContext(DefaultTableRuntime tableRuntime, MixedTable mixedTable) {
+    this.tableRuntime = tableRuntime;
+    this.mixedTable = mixedTable;
+  }
+
+  @Override
+  public TableConfiguration getTableConfiguration() {
+    return tableRuntime.getTableConfiguration();
+  }
+
+  @Override
+  public MaintainerMetrics getMetrics() {
+    TableOrphanFilesCleaningMetrics metrics = tableRuntime.getOrphanFilesCleaningMetrics();
+    return new MaintainerMetrics() {
+      @Override
+      public void recordOrphanDataFilesCleaned(int expected, int cleaned) {
+        metrics.completeOrphanDataFiles(expected, cleaned);
+      }
+
+      @Override
+      public void recordOrphanMetadataFilesCleaned(int expected, int cleaned) {
+        metrics.completeOrphanMetadataFiles(expected, cleaned);
+      }
+    };
+  }
+
+  @Override
+  public OptimizingInfo getOptimizingInfo() {
+    return new DefaultOptimizingInfo(tableRuntime);
+  }
+
+  @Override
+  public Set<String> getHiveLocationPaths() {
+    // Use HiveLocationUtil to get Hive location paths
+    if (mixedTable == null) {
+      return Collections.emptySet();
+    }
+    return HiveLocationUtil.getHiveLocation(mixedTable);
+  }
+
+  /** OptimizingInfo implementation based on DefaultTableRuntime. */
+  private static class DefaultOptimizingInfo implements OptimizingInfo {
+
+    private final DefaultTableRuntime tableRuntime;
+
+    DefaultOptimizingInfo(DefaultTableRuntime tableRuntime) {
+      this.tableRuntime = tableRuntime;
+    }
+
+    @Override
+    public boolean isProcessing() {
+      return tableRuntime.getOptimizingStatus().isProcessing();
+    }
+
+    @Override
+    public long getTargetSnapshotId() {
+      if (!isProcessing()) {
+        return Long.MAX_VALUE;
+      }
+      return tableRuntime.getOptimizingProcess().getTargetSnapshotId();
+    }
+  }
+}

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

@@ -0,0 +1,75 @@
+/*
+ * 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 org.apache.amoro.AmoroTable;
+import org.apache.amoro.TableFormat;
+import org.apache.amoro.TableRuntime;
+import org.apache.amoro.formats.iceberg.maintainer.IcebergTableMaintainer;
+import org.apache.amoro.formats.iceberg.maintainer.MixedTableMaintainer;
+import org.apache.amoro.maintainer.TableMaintainer;
+import org.apache.amoro.server.table.DefaultTableRuntime;
+import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
+import org.apache.amoro.table.MixedTable;
+import org.apache.iceberg.Table;
+
+/** Factory for creating {@link TableMaintainer} instances. */
+public class TableMaintainerFactory {
+
+  /**
+   * Create an Iceberg table maintainer with AMS context.
+   *
+   * @param table the Iceberg table
+   * @param tableRuntime the AMS table runtime
+   * @return IcebergTableMaintainer instance
+   */
+  public static IcebergTableMaintainer createIcebergMaintainer(
+      Table table, DefaultTableRuntime tableRuntime) {
+    return new IcebergTableMaintainer(
+        table,
+        tableRuntime.getTableIdentifier().getIdentifier(),
+        new DefaultTableMaintainerContext(tableRuntime));
+  }
+
+  /**
+   * Create a {@link TableMaintainer} for the given table.
+   *
+   * @param amoroTable the Amoro table
+   * @param tableRuntime the table runtime
+   * @return TableMaintainer instance
+   */
+  public static TableMaintainer create(AmoroTable<?> amoroTable, TableRuntime tableRuntime) {
+    Preconditions.checkArgument(tableRuntime instanceof DefaultTableRuntime);
+    DefaultTableRuntime runtime = (DefaultTableRuntime) tableRuntime;
+    TableFormat format = amoroTable.format();
+
+    if (format.in(TableFormat.MIXED_HIVE, TableFormat.MIXED_ICEBERG)) {
+      MixedTable mixedTable = (MixedTable) amoroTable.originalTable();
+      return new MixedTableMaintainer(
+          mixedTable, new DefaultTableMaintainerContext(runtime, mixedTable));
+    } else if (TableFormat.ICEBERG.equals(format)) {
+      return new IcebergTableMaintainer(
+          (Table) amoroTable.originalTable(),
+          amoroTable.id(),
+          new DefaultTableMaintainerContext(runtime));
+    } else {
+      throw new RuntimeException("Unsupported table type" + amoroTable.originalTable().getClass());
+    }
+  }
+}

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

@@ -21,25 +21,39 @@
 import org.apache.amoro.AmoroTable;
 import org.apache.amoro.TableFormat;
 import org.apache.amoro.TableRuntime;
+import org.apache.amoro.formats.iceberg.maintainer.MixedTableMaintainer;
+import org.apache.amoro.maintainer.TableMaintainer;
 import org.apache.amoro.server.table.DefaultTableRuntime;
-import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
 import org.apache.amoro.table.MixedTable;
 import org.apache.iceberg.Table;
 
 /** Factory for creating {@link TableMaintainer}. */
 public class TableMaintainers {
 
-  /** Create a {@link TableMaintainer} for the given table. */
+  /**
+   * Create a {@link TableMaintainer} for the given table.
+   *
+   * @deprecated Use {@link TableMaintainerFactory#create(AmoroTable, TableRuntime)} instead.
+   */
   public static TableMaintainer create(AmoroTable<?> amoroTable, TableRuntime tableRuntime) {
+    return TableMaintainerFactory.create(amoroTable, tableRuntime);
+  }
+
+  /**
+   * Create a {@link TableMaintainer} for the given table with DefaultTableRuntime.
+   *
+   * @deprecated Use {@link TableMaintainerFactory#createIcebergMaintainer(Table,
+   *     DefaultTableRuntime)} instead.
+   */
+  public static TableMaintainer create(AmoroTable<?> amoroTable, DefaultTableRuntime tableRuntime) {
     TableFormat format = amoroTable.format();
     if (format.in(TableFormat.MIXED_HIVE, TableFormat.MIXED_ICEBERG)) {
-      Preconditions.checkArgument(tableRuntime instanceof DefaultTableRuntime);
+      MixedTable mixedTable = (MixedTable) amoroTable.originalTable();
       return new MixedTableMaintainer(
-          (MixedTable) amoroTable.originalTable(), (DefaultTableRuntime) tableRuntime);
+          mixedTable, new DefaultTableMaintainerContext(tableRuntime, mixedTable));
     } else if (TableFormat.ICEBERG.equals(format)) {
-      Preconditions.checkArgument(tableRuntime instanceof DefaultTableRuntime);
-      return new IcebergTableMaintainer(
-          (Table) amoroTable.originalTable(), amoroTable.id(), (DefaultTableRuntime) tableRuntime);
+      return TableMaintainerFactory.createIcebergMaintainer(
+          (Table) amoroTable.originalTable(), tableRuntime);
     } else {
       throw new RuntimeException("Unsupported table type" + amoroTable.originalTable().getClass());
     }

amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/DanglingDeleteFilesCleaningExecutor.java

@@ -21,7 +21,7 @@
 import org.apache.amoro.AmoroTable;
 import org.apache.amoro.TableRuntime;
 import org.apache.amoro.config.TableConfiguration;
-import org.apache.amoro.server.optimizing.maintainer.TableMaintainer;
+import org.apache.amoro.maintainer.TableMaintainer;
 import org.apache.amoro.server.optimizing.maintainer.TableMaintainers;
 import org.apache.amoro.server.scheduler.PeriodicTableScheduler;
 import org.apache.amoro.server.table.DefaultTableRuntime;

amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/DataExpiringExecutor.java

@@ -21,7 +21,7 @@
 import org.apache.amoro.AmoroTable;
 import org.apache.amoro.TableRuntime;
 import org.apache.amoro.config.TableConfiguration;
-import org.apache.amoro.server.optimizing.maintainer.TableMaintainer;
+import org.apache.amoro.maintainer.TableMaintainer;
 import org.apache.amoro.server.optimizing.maintainer.TableMaintainers;
 import org.apache.amoro.server.scheduler.PeriodicTableScheduler;
 import org.apache.amoro.server.table.TableService;

amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/OrphanFilesCleaningExecutor.java

@@ -21,7 +21,7 @@
 import org.apache.amoro.AmoroTable;
 import org.apache.amoro.TableRuntime;
 import org.apache.amoro.config.TableConfiguration;
-import org.apache.amoro.server.optimizing.maintainer.TableMaintainer;
+import org.apache.amoro.maintainer.TableMaintainer;
 import org.apache.amoro.server.optimizing.maintainer.TableMaintainers;
 import org.apache.amoro.server.scheduler.PeriodicTableScheduler;
 import org.apache.amoro.server.table.TableService;

amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/SnapshotsExpiringExecutor.java

@@ -21,7 +21,7 @@
 import org.apache.amoro.AmoroTable;
 import org.apache.amoro.TableRuntime;
 import org.apache.amoro.config.TableConfiguration;
-import org.apache.amoro.server.optimizing.maintainer.TableMaintainer;
+import org.apache.amoro.maintainer.TableMaintainer;
 import org.apache.amoro.server.optimizing.maintainer.TableMaintainers;
 import org.apache.amoro.server.scheduler.PeriodicTableScheduler;
 import org.apache.amoro.server.table.TableService;

amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/TagsAutoCreatingExecutor.java

@@ -22,7 +22,7 @@
 import org.apache.amoro.TableFormat;
 import org.apache.amoro.TableRuntime;
 import org.apache.amoro.config.TableConfiguration;
-import org.apache.amoro.server.optimizing.maintainer.TableMaintainer;
+import org.apache.amoro.maintainer.TableMaintainer;
 import org.apache.amoro.server.optimizing.maintainer.TableMaintainers;
 import org.apache.amoro.server.scheduler.PeriodicTableScheduler;
 import org.apache.amoro.server.table.TableService;

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

@@ -31,7 +31,6 @@
 import org.apache.amoro.optimizing.OptimizingType;
 import org.apache.amoro.server.AmoroServiceConstants;
 import org.apache.amoro.server.optimizing.OptimizingStatus;
-import org.apache.amoro.server.optimizing.maintainer.IcebergTableMaintainer;
 import org.apache.amoro.shade.guava32.com.google.common.collect.ImmutableMap;
 import org.apache.amoro.shade.guava32.com.google.common.primitives.Longs;
 import org.apache.iceberg.Snapshot;
@@ -350,7 +349,10 @@ public void nonMaintainedSnapshotTime(Snapshot snapshot) {
     }
     // ignore snapshot which is created by amoro maintain commits or no files added
     if (snapshot.summary().values().stream()
-            .anyMatch(IcebergTableMaintainer.AMORO_MAINTAIN_COMMITS::contains)
+            .anyMatch(
+                org.apache.amoro.formats.iceberg.maintainer.IcebergTableMaintainer
+                        .AMORO_MAINTAIN_COMMITS
+                    ::contains)
         || Long.parseLong(snapshot.summary().getOrDefault(SnapshotSummary.ADDED_FILES_PROP, "0"))
             == 0) {
       return;

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

@@ -21,12 +21,14 @@
 import static org.apache.amoro.metrics.MetricDefine.defineCounter;
 
 import org.apache.amoro.ServerTableIdentifier;
+import org.apache.amoro.maintainer.MaintainerMetrics;
 import org.apache.amoro.metrics.Counter;
 import org.apache.amoro.metrics.MetricDefine;
 import org.apache.amoro.metrics.MetricRegistry;
 
 /** Table Orphan Files Cleaning metrics. */
-public class TableOrphanFilesCleaningMetrics extends AbstractTableMetrics {
+public class TableOrphanFilesCleaningMetrics extends AbstractTableMetrics
+    implements MaintainerMetrics {
   private final Counter orphanDataFilesCount = new Counter();
   private final Counter expectedOrphanDataFilesCount = new Counter();
 
@@ -89,4 +91,14 @@ public void completeOrphanMetadataFiles(int expected, int cleaned) {
     expectedOrphanMetadataFilesCount.inc(expected);
     orphanMetadataFilesCount.inc(cleaned);
   }
+
+  @Override
+  public void recordOrphanDataFilesCleaned(int expected, int cleaned) {
+    completeOrphanDataFiles(expected, cleaned);
+  }
+
+  @Override
+  public void recordOrphanMetadataFilesCleaned(int expected, int cleaned) {
+    completeOrphanMetadataFiles(expected, cleaned);
+  }
 }