[ARORO-3289] Avoid calling getMixedTablePartitionSpecById in the scan loop

amoro-ams/src/main/java/org/apache/amoro/server/optimizing/plan/OptimizingEvaluator.java

@@ -36,7 +36,6 @@
 import org.apache.amoro.shade.guava32.com.google.common.collect.Sets;
 import org.apache.amoro.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
 import org.apache.amoro.table.MixedTable;
-import org.apache.amoro.utils.MixedTableUtil;
 import org.apache.amoro.utils.TablePropertyUtil;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Snapshot;
@@ -138,9 +137,7 @@ private void initPartitionPlans(TableFileScanHelper tableFileScanHelper) {
     try (CloseableIterable<TableFileScanHelper.FileScanResult> results =
         tableFileScanHelper.scan()) {
       for (TableFileScanHelper.FileScanResult fileScanResult : results) {
-        PartitionSpec partitionSpec =
-            MixedTableUtil.getMixedTablePartitionSpecById(
-                mixedTable, fileScanResult.file().specId());
+        PartitionSpec partitionSpec = tableFileScanHelper.getSpec(fileScanResult.file().specId());
         StructLike partition = fileScanResult.file().partition();
         String partitionPath = partitionSpec.partitionToPath(partition);
         PartitionEvaluator evaluator =

amoro-ams/src/main/java/org/apache/amoro/server/optimizing/scan/IcebergTableFileScanHelper.java

@@ -21,19 +21,24 @@
 import org.apache.amoro.server.AmoroServiceConstants;
 import org.apache.amoro.shade.guava32.com.google.common.collect.Lists;
 import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.io.CloseableIterable;
 
+import java.util.Map;
+
 public class IcebergTableFileScanHelper implements TableFileScanHelper {
   private final Table table;
   private Expression partitionFilter = Expressions.alwaysTrue();
   private final long snapshotId;
+  private final Map<Integer, PartitionSpec> specs;
 
   public IcebergTableFileScanHelper(Table table, long snapshotId) {
     this.table = table;
     this.snapshotId = snapshotId;
+    this.specs = table.specs();
   }
 
   @Override
@@ -55,4 +60,9 @@ public TableFileScanHelper withPartitionFilter(Expression partitionFilter) {
     this.partitionFilter = partitionFilter;
     return this;
   }
+
+  @Override
+  public PartitionSpec getSpec(int specId) {
+    return specs.get(specId);
+  }
 }

amoro-ams/src/main/java/org/apache/amoro/server/optimizing/scan/KeyedTableFileScanHelper.java

@@ -67,11 +67,13 @@ public class KeyedTableFileScanHelper implements TableFileScanHelper {
   private final long changeSnapshotId;
   private final long baseSnapshotId;
   private Expression partitionFilter = Expressions.alwaysTrue();
+  private final PartitionSpec spec;
 
   public KeyedTableFileScanHelper(KeyedTable keyedTable, KeyedTableSnapshot snapshot) {
     this.keyedTable = keyedTable;
     this.baseSnapshotId = snapshot.baseSnapshotId();
     this.changeSnapshotId = snapshot.changeSnapshotId();
+    this.spec = keyedTable.spec();
   }
 
   /**
@@ -439,4 +441,13 @@ public void setMinTransactionIdAfter(long minTransactionIdAfter) {
       this.minTransactionIdAfter = minTransactionIdAfter;
     }
   }
+
+  @Override
+  public PartitionSpec getSpec(int specId) {
+    if (specId != spec.specId()) {
+      throw new IllegalArgumentException(
+          "Partition spec id " + specId + " not found in table " + keyedTable.name());
+    }
+    return spec;
+  }
 }

amoro-ams/src/main/java/org/apache/amoro/server/optimizing/scan/TableFileScanHelper.java

@@ -20,6 +20,7 @@
 
 import org.apache.iceberg.ContentFile;
 import org.apache.iceberg.DataFile;
+import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.io.CloseableIterable;
 
@@ -47,4 +48,6 @@ public List<ContentFile<?>> deleteFiles() {
   CloseableIterable<FileScanResult> scan();
 
   TableFileScanHelper withPartitionFilter(Expression partitionFilter);
+
+  PartitionSpec getSpec(int specId);
 }