diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/plan/OptimizingEvaluator.java b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/plan/OptimizingEvaluator.java index 5914d57bb9..b575d67fc0 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/plan/OptimizingEvaluator.java +++ b/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 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 = diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/scan/IcebergTableFileScanHelper.java b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/scan/IcebergTableFileScanHelper.java index 3a65bc3936..7fa9647994 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/scan/IcebergTableFileScanHelper.java +++ b/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 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); + } } diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/scan/KeyedTableFileScanHelper.java b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/scan/KeyedTableFileScanHelper.java index a69a45e58b..46e07dcec1 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/scan/KeyedTableFileScanHelper.java +++ b/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; + } } diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/scan/TableFileScanHelper.java b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/scan/TableFileScanHelper.java index 60a0c53b7c..5ad1f8668d 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/scan/TableFileScanHelper.java +++ b/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> deleteFiles() { CloseableIterable scan(); TableFileScanHelper withPartitionFilter(Expression partitionFilter); + + PartitionSpec getSpec(int specId); }