This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new 08f3a2fca2 [Cherry-pick to branch-1.3] [#14536] fix(iceberg-rest):
honor use-snapshot-schema and plan dropped partition sources (#14552) (#14647)
08f3a2fca2 is described below
commit 08f3a2fca24e0b03a6bbd0cac3c507790b9de6fa
Author: MaSai <[email protected]>
AuthorDate: Thu Nov 5 14:07:33 2026 +0800
[Cherry-pick to branch-1.3] [#14536] fix(iceberg-rest): honor
use-snapshot-schema and plan dropped partition sources (#14552) (#14647)
## Purpose
Cherry-pick of #14552 onto `branch-1.3`, replacing bot PR #14646 (closed
with conflicts).
## Changes
- Honor `use-snapshot-schema` and plan tables whose partition sources
were dropped.
- branch-1.3 adaptations from review follow-ups on #14552:
- `rewriteFilterNames` NOT polarity
- `canProjectPartition` for incompatible partition projection
- restore `equalityFieldIds` after `FileMetadata.copy` (Iceberg 1.11)
- include `schemaId` in `ScanPlanCacheKey`
## Closes
Supersedes #14646
Co-authored-by: Cursor <[email protected]>
---
.../iceberg/service/CatalogWrapperForREST.java | 68 +-
.../iceberg/service/IcebergScanPlanHelper.java | 822 +++++++++++++++++++++
.../iceberg/service/cache/ScanPlanCacheKey.java | 21 +-
.../iceberg/service/TestIcebergScanPlanHelper.java | 732 ++++++++++++++++++
.../iceberg/service/TestIcebergScanPlanning.java | 183 ++++-
.../iceberg/service/rest/TestScanPlanCache.java | 35 +
6 files changed, 1848 insertions(+), 13 deletions(-)
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
index 723ab46a45..d96c2f9cba 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
@@ -55,6 +55,7 @@ import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.HasTableOperations;
import org.apache.iceberg.IncrementalAppendScan;
import org.apache.iceberg.Scan;
+import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.TableProperties;
@@ -65,6 +66,7 @@ import org.apache.iceberg.exceptions.BadRequestException;
import org.apache.iceberg.exceptions.ServiceFailureException;
import org.apache.iceberg.exceptions.ServiceUnavailableException;
import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.iceberg.expressions.Expression;
import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.rest.CatalogHandlers;
import org.apache.iceberg.rest.PlanStatus;
@@ -76,6 +78,7 @@ import
org.apache.iceberg.rest.responses.ImmutableLoadCredentialsResponse;
import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
import org.apache.iceberg.rest.responses.LoadTableResponse;
import org.apache.iceberg.rest.responses.PlanTableScanResponse;
+import org.apache.iceberg.util.SnapshotUtil;
/** Process Iceberg REST specific operations, like credential vending. */
public class CatalogWrapperForREST extends IcebergCatalogWrapper {
@@ -486,10 +489,28 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
return cachedResponse.get();
}
+ Schema residualFrom = null;
+ Schema residualTo = null;
+ Expression extraResidual = null;
List<FileScanTask> fileScanTasks = new ArrayList<>();
-
try (CloseableIterable<FileScanTask> scanTasks =
createFilePlanScanTasks(table, tableIdentifier, scanRequest)) {
+ // Recompute whether names were rewritten so residuals can be mapped
back to the request
+ // schema. createFilePlanScanTasks applies the same rewrite rules.
+ if (scanRequest.snapshotId() != null &&
!scanRequest.useSnapshotSchema()) {
+ long snapshotId = scanRequest.snapshotId();
+ Optional<String> branch = IcebergScanPlanHelper.branchAt(table,
snapshotId);
+ if (!branch.isPresent()) {
+ residualFrom = SnapshotUtil.schemaFor(table, snapshotId);
+ residualTo = table.schema();
+ // Keep the complete client filter as a residual. Extracting only
+ // post-snapshot predicates by replacing others with always-true
+ // collapses OR/NOT (e.g. id=1 OR new_col=2 → true).
+ if (scanRequest.filter() != null) {
+ extraResidual = scanRequest.filter();
+ }
+ }
+ }
for (FileScanTask fileScanTask : scanTasks) {
fileScanTasks.add(fileScanTask);
}
@@ -497,6 +518,14 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
LOG.error("Failed to close scan task iterator for table: {}",
tableIdentifier, e);
throw new RuntimeException("Failed to plan scan tasks: " +
e.getMessage(), e);
}
+ fileScanTasks =
+ IcebergScanPlanHelper.prepareTasksForResponse(
+ fileScanTasks,
+ table.specs(),
+ residualTo,
+ residualFrom,
+ extraResidual,
+ scanRequest.caseSensitive());
if (fileScanTasks.isEmpty()) {
LOG.info(
@@ -555,7 +584,9 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
return PlanTableScanResponse.builder()
.withPlanStatus(PlanStatus.COMPLETED)
.withFileScanTasks(fileScanTasks)
- .withSpecsById(table.specs())
+ .withSpecsById(
+ IcebergScanPlanHelper.sanitizeSpecsForSerialization(
+ table.specs(), fileScanTasks, table.schemas().values()))
.build();
}
@@ -565,6 +596,12 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
* <p>If both start and end snapshot IDs are provided, uses
IncrementalAppendScan. Otherwise, uses
* regular TableScan.
*
+ * <p>With {@code use-snapshot-schema} false (the REST default, sent for
branch scans) and a
+ * {@code snapshot-id}, the request is written against the current table
schema. A branch that
+ * points at the snapshot is scanned with {@code useRef}, which binds to the
current schema like a
+ * client-side branch scan. If no branch points there any more, the snapshot
is scanned and the
+ * request's column names are mapped to the snapshot schema by field id.
+ *
* @param table The table to scan
* @param tableIdentifier The table identifier for logging
* @param scanRequest The scan request parameters
@@ -596,12 +633,28 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
return incrementalScan.planFiles();
} else {
TableScan tableScan = table.newScan();
+ PlanTableScanRequest requestToApply = scanRequest;
// Snapshot ID 0 has no special meaning in Iceberg, so we only apply if
not null
if (scanRequest.snapshotId() != null) {
- tableScan = tableScan.useSnapshot(scanRequest.snapshotId());
- LOG.debug("Applied snapshot filter: snapshot-id={}",
scanRequest.snapshotId());
+ long snapshotId = scanRequest.snapshotId();
+ Optional<String> branch =
+ scanRequest.useSnapshotSchema()
+ ? Optional.empty()
+ : IcebergScanPlanHelper.branchAt(table, snapshotId);
+ if (branch.isPresent()) {
+ tableScan = tableScan.useRef(branch.get());
+ LOG.debug("Applied branch {} for snapshot-id={}", branch.get(),
snapshotId);
+ } else {
+ tableScan = tableScan.useSnapshot(snapshotId);
+ if (!scanRequest.useSnapshotSchema()) {
+ requestToApply =
+ IcebergScanPlanHelper.rewriteForSnapshot(
+ scanRequest, table.schema(), SnapshotUtil.schemaFor(table,
snapshotId));
+ }
+ LOG.debug("Applied snapshot filter: snapshot-id={}", snapshotId);
+ }
}
- tableScan = applyScanRequest(tableScan, scanRequest);
+ tableScan = applyScanRequest(tableScan, requestToApply);
return tableScan.planFiles();
}
}
@@ -621,8 +674,9 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
private <T extends Scan> T applyScanFilter(T scan, PlanTableScanRequest
scanRequest) {
if (scanRequest.filter() != null) {
try {
- scan = (T) scan.filter(scanRequest.filter());
- LOG.debug("Applied filter expression: {}", scanRequest.filter());
+ Expression filter = scanRequest.filter();
+ scan = (T) scan.filter(filter);
+ LOG.debug("Applied filter expression: {}", filter);
} catch (Exception e) {
LOG.error("Failed to apply filter expression: {}", e.getMessage(), e);
throw new IllegalArgumentException("Invalid filter expression: " +
e.getMessage(), e);
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergScanPlanHelper.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergScanPlanHelper.java
new file mode 100644
index 0000000000..5843729938
--- /dev/null
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergScanPlanHelper.java
@@ -0,0 +1,822 @@
+/*
+ * 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.gravitino.iceberg.service;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+import org.apache.iceberg.BaseFileScanTask;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileContent;
+import org.apache.iceberg.FileMetadata;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.PartitionData;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.PartitionSpecParser;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SchemaParser;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.expressions.And;
+import org.apache.iceberg.expressions.BoundPredicate;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.ExpressionVisitors;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.expressions.Literal;
+import org.apache.iceberg.expressions.Not;
+import org.apache.iceberg.expressions.Or;
+import org.apache.iceberg.expressions.ResidualEvaluator;
+import org.apache.iceberg.expressions.UnboundPredicate;
+import org.apache.iceberg.expressions.UnboundTerm;
+import org.apache.iceberg.expressions.UnboundTransform;
+import org.apache.iceberg.rest.requests.PlanTableScanRequest;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.types.Types;
+
+/**
+ * Helpers for Iceberg REST server-side scan planning: honour {@code
use-snapshot-schema} and avoid
+ * failing when historical partition specs reference dropped columns.
+ */
+final class IcebergScanPlanHelper {
+
+ private IcebergScanPlanHelper() {}
+
+ /**
+ * Returns a branch that points at the snapshot, preferring {@code main}.
+ *
+ * <p>Iceberg's REST client sends {@code use-snapshot-schema=false} for a
branch scan, and {@code
+ * TableScan#useRef} on a branch reads its snapshot with the current table
schema, which is what
+ * that flag asks for.
+ */
+ static Optional<String> branchAt(Table table, long snapshotId) {
+ SnapshotRef main = table.refs().get(SnapshotRef.MAIN_BRANCH);
+ if (main != null && main.snapshotId() == snapshotId) {
+ return Optional.of(SnapshotRef.MAIN_BRANCH);
+ }
+ return table.refs().entrySet().stream()
+ .filter(entry -> entry.getValue().isBranch() &&
entry.getValue().snapshotId() == snapshotId)
+ .map(Map.Entry::getKey)
+ .sorted()
+ .findFirst();
+ }
+
+ /**
+ * Rewrites filter/select/stats field names from the current table schema to
the snapshot schema
+ * by field id, so a {@code useSnapshot} scan can serve a request written
against the current
+ * schema. Names are resolved on the current schema first, so a name that
only the snapshot schema
+ * has, or that a later column reuses, never binds to the snapshot's field.
Columns added after
+ * the snapshot are dropped from select/stats; filters on them are rewritten
conservatively for
+ * planning with polarity-aware placeholders (never under-prune) and the
full original filter is
+ * reattached as a residual by the caller.
+ */
+ static PlanTableScanRequest rewriteForSnapshot(
+ PlanTableScanRequest request, Schema current, Schema snapshot) {
+ PlanTableScanRequest.Builder builder =
+ PlanTableScanRequest.builder()
+ .withSnapshotId(request.snapshotId())
+ .withCaseSensitive(request.caseSensitive())
+ .withUseSnapshotSchema(request.useSnapshotSchema())
+ .withStartSnapshotId(request.startSnapshotId())
+ .withEndSnapshotId(request.endSnapshotId())
+ .withMinRowsRequested(request.minRowsRequested());
+ if (request.filter() != null) {
+ builder.withFilter(
+ rewriteFilterNames(request.filter(), current, snapshot,
request.caseSensitive()));
+ }
+ if (request.select() != null && !request.select().isEmpty()) {
+ List<String> select =
+ rewriteColumnNames(request.select(), current, snapshot,
request.caseSensitive());
+ if (!select.isEmpty()) {
+ builder.withSelect(select);
+ }
+ }
+ if (request.statsFields() != null && !request.statsFields().isEmpty()) {
+ List<String> statsFields =
+ rewriteColumnNames(request.statsFields(), current, snapshot,
request.caseSensitive());
+ if (!statsFields.isEmpty()) {
+ builder.withStatsFields(statsFields);
+ }
+ }
+ return builder.build();
+ }
+
+ /**
+ * Remaps residual filters that were bound/named against {@code from} onto
{@code to} by field id,
+ * so a response for {@code use-snapshot-schema=false} keeps the request
schema's identities.
+ *
+ * <p>Handles both bound residuals (field id from the scan schema) and
unbound ones (names
+ * resolved on {@code from}).
+ */
+ @VisibleForTesting
+ static Expression remapExpressionNames(
+ Expression expression, Schema from, Schema to, boolean caseSensitive) {
+ if (expression == null) {
+ return null;
+ }
+ return ExpressionVisitors.visit(
+ expression,
+ new ExpressionVisitors.ExpressionVisitor<Expression>() {
+ @Override
+ public Expression alwaysTrue() {
+ return Expressions.alwaysTrue();
+ }
+
+ @Override
+ public Expression alwaysFalse() {
+ return Expressions.alwaysFalse();
+ }
+
+ @Override
+ public Expression not(Expression result) {
+ return Expressions.not(result);
+ }
+
+ @Override
+ public Expression and(Expression left, Expression right) {
+ return Expressions.and(left, right);
+ }
+
+ @Override
+ public Expression or(Expression left, Expression right) {
+ return Expressions.or(left, right);
+ }
+
+ @Override
+ public <T> Expression predicate(BoundPredicate<T> pred) {
+ String toName = to.findColumnName(pred.ref().fieldId());
+ if (toName == null) {
+ return Expressions.alwaysTrue();
+ }
+ return unboundFromBound(pred, toName);
+ }
+
+ @Override
+ public <T> Expression predicate(UnboundPredicate<T> pred) {
+ String fromName = pred.ref().name();
+ String toName = renameByFieldId(fromName, from, to, caseSensitive);
+ if (toName == null) {
+ // Field was dropped from the request schema; residual cannot
name it.
+ return Expressions.alwaysTrue();
+ }
+ if (toName.equals(fromName)) {
+ return pred;
+ }
+ return unboundWithName(pred, toName);
+ }
+ });
+ }
+
+ /**
+ * Remaps residuals (bound or unbound) onto {@code to} by field id. Bound
residuals from a
+ * snapshot scan carry snapshot names; unbound ones do as well after a name
rewrite.
+ */
+ @VisibleForTesting
+ static Expression remapResidualToSchema(Expression residual, Schema to,
boolean caseSensitive) {
+ if (residual == null) {
+ return null;
+ }
+ return ExpressionVisitors.visit(
+ residual,
+ new ExpressionVisitors.BoundExpressionVisitor<Expression>() {
+ @Override
+ public Expression alwaysTrue() {
+ return Expressions.alwaysTrue();
+ }
+
+ @Override
+ public Expression alwaysFalse() {
+ return Expressions.alwaysFalse();
+ }
+
+ @Override
+ public Expression not(Expression result) {
+ return Expressions.not(result);
+ }
+
+ @Override
+ public Expression and(Expression left, Expression right) {
+ return Expressions.and(left, right);
+ }
+
+ @Override
+ public Expression or(Expression left, Expression right) {
+ return Expressions.or(left, right);
+ }
+
+ @Override
+ public <T> Expression predicate(BoundPredicate<T> pred) {
+ String toName = to.findColumnName(pred.ref().fieldId());
+ if (toName == null) {
+ return Expressions.alwaysTrue();
+ }
+ return unboundFromBound(pred, toName);
+ }
+
+ @Override
+ public <T> Expression predicate(UnboundPredicate<T> pred) {
+ // Unbound residual still carries the scan-schema name; resolve
via {@code to} only when
+ // the name already matches a current field, otherwise keep as-is
(tests use unbound
+ // remaps through {@link #remapExpressionNames}).
+ Types.NestedField field =
+ caseSensitive
+ ? to.findField(pred.ref().name())
+ : to.caseInsensitiveFindField(pred.ref().name());
+ if (field != null) {
+ return pred;
+ }
+ return pred;
+ }
+ });
+ }
+
+ /**
+ * Prepares planned tasks for the REST response: remap residuals onto the
request schema and
+ * rewrite tasks whose partition spec still has {@code unknown} onto a
client-parseable spec from
+ * {@code tableSpecs} (prefer unpartitioned, else any spec without {@code
unknown}) so Iceberg
+ * clients can parse them with their local {@code table.specs()} map.
Associated delete files are
+ * rewritten to the same target spec.
+ */
+ static List<FileScanTask> prepareTasksForResponse(
+ List<FileScanTask> fileScanTasks,
+ Map<Integer, PartitionSpec> tableSpecs,
+ @Nullable Schema residualTargetSchema,
+ @Nullable Schema residualSourceSchema,
+ @Nullable Expression extraResidual,
+ boolean caseSensitive) {
+ if (fileScanTasks.isEmpty()) {
+ return fileScanTasks;
+ }
+ PartitionSpec compatibleSpec = findClientCompatibleSpec(tableSpecs);
+ List<FileScanTask> prepared =
Lists.newArrayListWithCapacity(fileScanTasks.size());
+ for (FileScanTask task : fileScanTasks) {
+ Expression residual = task.residual();
+ if (residualTargetSchema != null && residual != null) {
+ if (residualSourceSchema != null) {
+ residual =
+ remapExpressionNames(
+ residual, residualSourceSchema, residualTargetSchema,
caseSensitive);
+ } else {
+ residual = remapResidualToSchema(residual, residualTargetSchema,
caseSensitive);
+ }
+ }
+ if (extraResidual != null) {
+ residual = residual == null ? extraResidual :
Expressions.and(residual, extraResidual);
+ }
+ PartitionSpec spec = tableSpecs.get(task.file().specId());
+ boolean needsSpecRewrite =
+ spec != null && partitionTypeHasUnknown(spec.partitionType()) &&
compatibleSpec != null;
+ if (needsSpecRewrite) {
+ if (!canProjectPartition(task.spec(), compatibleSpec)) {
+ // Inventing NULL for missing target fields makes residuals evaluate
wrong.
+ throw new IllegalArgumentException(
+ String.format(
+ "Cannot rewrite file %s from spec %d to client-compatible
spec %d: "
+ + "target partition fields are missing from the source
file",
+ task.file().location(), task.spec().specId(),
compatibleSpec.specId()));
+ }
+ prepared.add(
+ withCompatibleSpec(
+ task, compatibleSpec, residual, residualTargetSchema,
caseSensitive));
+ } else if (residual != null && residual != task.residual()) {
+ prepared.add(withResidual(task, residual, residualTargetSchema,
caseSensitive));
+ } else {
+ prepared.add(task);
+ }
+ }
+ return prepared;
+ }
+
+ /**
+ * Rebinds partition specs whose {@link PartitionSpec#partitionType()}
contains {@code unknown}
+ * (dropped source columns) so file-scan-task serialization can write
partition values. Source
+ * types are restored from historical table schemas (and identity partition
data when present);
+ * transform result types and a binary guess are never used. Specs that
cannot be restored are
+ * left unchanged so empty plans are not made worse.
+ */
+ @VisibleForTesting
+ static Map<Integer, PartitionSpec> sanitizeSpecsForSerialization(
+ Map<Integer, PartitionSpec> specsById,
+ List<FileScanTask> fileScanTasks,
+ Collection<Schema> historicalSchemas) {
+ Map<Integer, Types.StructType> partitionTypesBySpecId = Maps.newHashMap();
+ for (FileScanTask task : fileScanTasks) {
+ StructLike partition = task.file().partition();
+ if (partition instanceof PartitionData) {
+ partitionTypesBySpecId.putIfAbsent(
+ task.file().specId(), ((PartitionData)
partition).getPartitionType());
+ }
+ }
+
+ Map<Integer, Type> typesBySourceId =
indexTypesByFieldId(historicalSchemas);
+
+ Map<Integer, PartitionSpec> sanitized = Maps.newHashMap();
+ boolean changed = false;
+ for (Map.Entry<Integer, PartitionSpec> entry : specsById.entrySet()) {
+ PartitionSpec spec = entry.getValue();
+ if (!partitionTypeHasUnknown(spec.partitionType())) {
+ sanitized.put(entry.getKey(), spec);
+ continue;
+ }
+ Schema extended =
+ extendSchemaForDroppedSources(
+ spec, partitionTypesBySpecId.get(entry.getKey()),
typesBySourceId);
+ if (extended == spec.schema()) {
+ // Could not restore every dropped source; leave the broken spec alone.
+ sanitized.put(entry.getKey(), spec);
+ continue;
+ }
+ // ignoreMissingFields covers any residual gaps if a type is still
missing.
+ sanitized.put(entry.getKey(), spec.toUnbound().bind(extended, true));
+ changed = true;
+ }
+ return changed ? Collections.unmodifiableMap(sanitized) : specsById;
+ }
+
+ /**
+ * Rewrites filter names onto the snapshot schema for planning.
+ *
+ * <p>Columns added after the snapshot are replaced conservatively with
polarity awareness: under
+ * a positive context they become {@code true} (never under-prune); under
{@code NOT} they become
+ * {@code false} so {@code NOT(old AND new)} stays weak instead of
collapsing to {@code NOT(old)}.
+ * The caller still ANDs the full original filter onto the response residual.
+ */
+ @VisibleForTesting
+ static Expression rewriteFilterNames(
+ Expression filter, Schema current, Schema snapshot, boolean
caseSensitive) {
+ return rewriteFilterNames(filter, current, snapshot, caseSensitive, true);
+ }
+
+ private static Expression rewriteFilterNames(
+ Expression filter, Schema current, Schema snapshot, boolean
caseSensitive, boolean positive) {
+ switch (filter.op()) {
+ case TRUE:
+ return Expressions.alwaysTrue();
+ case FALSE:
+ return Expressions.alwaysFalse();
+ case NOT:
+ Expression child =
+ rewriteFilterNames(((Not) filter).child(), current, snapshot,
caseSensitive, !positive);
+ if (child.op() == Expression.Operation.TRUE) {
+ return Expressions.alwaysFalse();
+ }
+ if (child.op() == Expression.Operation.FALSE) {
+ return Expressions.alwaysTrue();
+ }
+ return Expressions.not(child);
+ case AND:
+ {
+ And and = (And) filter;
+ Expression left =
+ rewriteFilterNames(and.left(), current, snapshot, caseSensitive,
positive);
+ Expression right =
+ rewriteFilterNames(and.right(), current, snapshot,
caseSensitive, positive);
+ if (left.op() == Expression.Operation.FALSE || right.op() ==
Expression.Operation.FALSE) {
+ return Expressions.alwaysFalse();
+ }
+ if (left.op() == Expression.Operation.TRUE) {
+ return right;
+ }
+ if (right.op() == Expression.Operation.TRUE) {
+ return left;
+ }
+ return Expressions.and(left, right);
+ }
+ case OR:
+ {
+ Or or = (Or) filter;
+ Expression left =
+ rewriteFilterNames(or.left(), current, snapshot, caseSensitive,
positive);
+ Expression right =
+ rewriteFilterNames(or.right(), current, snapshot, caseSensitive,
positive);
+ if (left.op() == Expression.Operation.TRUE || right.op() ==
Expression.Operation.TRUE) {
+ return Expressions.alwaysTrue();
+ }
+ if (left.op() == Expression.Operation.FALSE) {
+ return right;
+ }
+ if (right.op() == Expression.Operation.FALSE) {
+ return left;
+ }
+ return Expressions.or(left, right);
+ }
+ default:
+ if (!(filter instanceof UnboundPredicate)) {
+ throw new IllegalArgumentException("Unsupported filter expression: "
+ filter);
+ }
+ return rewriteFilterPredicate(
+ (UnboundPredicate<?>) filter, current, snapshot, caseSensitive,
positive);
+ }
+ }
+
+ private static <T> Expression rewriteFilterPredicate(
+ UnboundPredicate<T> pred,
+ Schema current,
+ Schema snapshot,
+ boolean caseSensitive,
+ boolean positive) {
+ String name = pred.ref().name();
+ String snapshotName = toSnapshotName(name, current, snapshot,
caseSensitive);
+ if (snapshotName == null) {
+ // Post-snapshot column: weaken the planning filter (over-include).
Polarity flips under NOT.
+ return positive ? Expressions.alwaysTrue() : Expressions.alwaysFalse();
+ }
+ if (snapshotName.equals(name)) {
+ return pred;
+ }
+ return unboundWithName(pred, snapshotName);
+ }
+
+ @VisibleForTesting
+ static List<String> rewriteColumnNames(
+ List<String> columns, Schema current, Schema snapshot, boolean
caseSensitive) {
+ List<String> rewritten = Lists.newArrayListWithCapacity(columns.size());
+ for (String column : columns) {
+ if ("*".equals(column)) {
+ // Iceberg schema selection treats "*" as project-all, not a field
name.
+ rewritten.add("*");
+ continue;
+ }
+ String snapshotName = toSnapshotName(column, current, snapshot,
caseSensitive);
+ if (snapshotName != null) {
+ rewritten.add(snapshotName);
+ }
+ }
+ return rewritten;
+ }
+
+ /**
+ * Returns the snapshot schema's name for a current-schema column, or {@code
null} if the column
+ * was added after the snapshot.
+ *
+ * @throws IllegalArgumentException if the current table schema has no such
column
+ */
+ @Nullable
+ private static String toSnapshotName(
+ String name, Schema current, Schema snapshot, boolean caseSensitive) {
+ Types.NestedField field =
+ caseSensitive ? current.findField(name) :
current.caseInsensitiveFindField(name);
+ if (field == null) {
+ throw new IllegalArgumentException(
+ String.format("Cannot find field '%s' in the current table schema",
name));
+ }
+ return snapshot.findColumnName(field.fieldId());
+ }
+
+ @Nullable
+ private static String renameByFieldId(
+ String fromName, Schema from, Schema to, boolean caseSensitive) {
+ Types.NestedField field =
+ caseSensitive ? from.findField(fromName) :
from.caseInsensitiveFindField(fromName);
+ if (field == null) {
+ return null;
+ }
+ return to.findColumnName(field.fieldId());
+ }
+
+ private static boolean partitionTypeHasUnknown(Types.StructType
partitionType) {
+ for (Types.NestedField field : partitionType.fields()) {
+ if (field.type().typeId() == Type.TypeID.UNKNOWN) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static Map<Integer, Type> indexTypesByFieldId(Collection<Schema>
schemas) {
+ Map<Integer, Type> types = Maps.newHashMap();
+ if (schemas == null) {
+ return types;
+ }
+ for (Schema schema : schemas) {
+ for (Types.NestedField field :
TypeUtil.indexById(schema.asStruct()).values()) {
+ types.putIfAbsent(field.fieldId(), field.type());
+ }
+ }
+ return types;
+ }
+
+ private static Schema extendSchemaForDroppedSources(
+ PartitionSpec spec, Types.StructType partitionDataType, Map<Integer,
Type> typesBySourceId) {
+ Schema schema = spec.schema();
+ List<Types.NestedField> extras = Lists.newArrayList();
+ Set<Integer> addedSourceIds = Sets.newHashSet();
+ List<Types.NestedField> partitionFields =
+ partitionDataType == null ? Collections.emptyList() :
partitionDataType.fields();
+ int partitionPos = 0;
+ for (PartitionField field : spec.fields()) {
+ Types.NestedField source = schema.findField(field.sourceId());
+ Type partitionFieldType =
+ partitionPos < partitionFields.size() ?
partitionFields.get(partitionPos).type() : null;
+ partitionPos++;
+ if (source != null || !addedSourceIds.add(field.sourceId())) {
+ continue;
+ }
+ Type restored = typesBySourceId.get(field.sourceId());
+ if (restored == null
+ && field.transform().isIdentity()
+ && partitionFieldType != null
+ && partitionFieldType.typeId() != Type.TypeID.UNKNOWN) {
+ // Identity partition values share the source type; non-identity
transform outputs do not.
+ restored = partitionFieldType;
+ }
+ if (restored == null) {
+ // Cannot restore this source; signal the caller by returning the
original schema.
+ return schema;
+ }
+ extras.add(
+ Types.NestedField.optional(
+ field.sourceId(), "_dropped_source_" + field.sourceId(),
restored));
+ }
+ if (extras.isEmpty()) {
+ return schema;
+ }
+ List<Types.NestedField> all = Lists.newArrayList(schema.columns());
+ all.addAll(extras);
+ return new Schema(all);
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> Expression unboundWithName(UnboundPredicate<T> pred,
String name) {
+ UnboundTerm<T> term = renameTerm(pred.term(), name);
+ if (pred.literals() == null) {
+ return Expressions.predicate(pred.op(), term);
+ }
+ List<T> values =
pred.literals().stream().map(Literal::value).collect(Collectors.toList());
+ return Expressions.predicate(pred.op(), term, values);
+ }
+
+ private static <T> Expression unboundFromBound(BoundPredicate<T> pred,
String name) {
+ if (pred.isUnaryPredicate()) {
+ return Expressions.predicate(pred.op(), name);
+ }
+ if (pred.isLiteralPredicate()) {
+ return Expressions.predicate(pred.op(), name,
pred.asLiteralPredicate().literal().value());
+ }
+ if (pred.isSetPredicate()) {
+ return Expressions.predicate(pred.op(), name,
pred.asSetPredicate().literalSet());
+ }
+ // Fallback: preserve always-true rather than drop an unrecognised shape.
+ return Expressions.alwaysTrue();
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> UnboundTerm<T> renameTerm(UnboundTerm<T> term, String
name) {
+ if (term instanceof UnboundTransform) {
+ return Expressions.transform(name, ((UnboundTransform<?, T>)
term).transform());
+ }
+ return (UnboundTerm<T>) Expressions.ref(name);
+ }
+
+ /**
+ * Prefers an unpartitioned parseable spec; otherwise any parseable spec
(typically the current
+ * default after partition evolution). Iceberg REST clients resolve file
{@code spec-id} against
+ * their local {@code table.specs()}, so the target id must already exist
there without {@code
+ * unknown}.
+ */
+ @Nullable
+ private static PartitionSpec findClientCompatibleSpec(Map<Integer,
PartitionSpec> specsById) {
+ PartitionSpec unpartitioned = null;
+ PartitionSpec anyCompatible = null;
+ int bestId = Integer.MIN_VALUE;
+ for (PartitionSpec spec : specsById.values()) {
+ if (partitionTypeHasUnknown(spec.partitionType())) {
+ continue;
+ }
+ if (spec.isUnpartitioned()) {
+ unpartitioned = spec;
+ }
+ if (spec.specId() >= bestId) {
+ bestId = spec.specId();
+ anyCompatible = spec;
+ }
+ }
+ return unpartitioned != null ? unpartitioned : anyCompatible;
+ }
+
+ private static FileScanTask withResidual(
+ FileScanTask task,
+ Expression residual,
+ @Nullable Schema residualSchema,
+ boolean caseSensitive) {
+ PartitionSpec fileSpec = task.spec();
+ PartitionSpec evalSpec =
+ residualSchema == null ? fileSpec :
fileSpec.toUnbound().bind(residualSchema, true);
+ ResidualEvaluator residuals =
+ ResidualEvaluator.of(
+ evalSpec, residual == null ? Expressions.alwaysTrue() : residual,
caseSensitive);
+ Schema schemaJson = residualSchema != null ? residualSchema :
task.schema();
+ return new BaseFileScanTask(
+ task.file(),
+ task.deletes().toArray(new DeleteFile[0]),
+ SchemaParser.toJson(schemaJson),
+ PartitionSpecParser.toJson(fileSpec),
+ residuals);
+ }
+
+ private static FileScanTask withCompatibleSpec(
+ FileScanTask task,
+ PartitionSpec targetSpec,
+ Expression residual,
+ @Nullable Schema residualSchema,
+ boolean caseSensitive) {
+ // Rebuild under a client-parseable spec id so Iceberg clients can parse
the task (and its
+ // deletes) with their local table.specs() map, which still has
UnknownType for dropped
+ // partition sources on the historical spec id.
+ DataFile rewritten = rewriteDataFile(task.file(), task.spec(), targetSpec);
+ DeleteFile[] deletes = rewriteDeletes(task.deletes(), task.spec(),
targetSpec);
+ Schema evalSchema = residualSchema != null ? residualSchema :
targetSpec.schema();
+ PartitionSpec evalSpec = targetSpec.toUnbound().bind(evalSchema, true);
+ ResidualEvaluator residuals =
+ ResidualEvaluator.of(
+ evalSpec, residual == null ? Expressions.alwaysTrue() : residual,
caseSensitive);
+ return new BaseFileScanTask(
+ rewritten,
+ deletes,
+ SchemaParser.toJson(evalSchema),
+ PartitionSpecParser.toJson(targetSpec),
+ residuals);
+ }
+
+ private static DataFile rewriteDataFile(
+ DataFile file, PartitionSpec fromSpec, PartitionSpec targetSpec) {
+ if (targetSpec.isUnpartitioned()) {
+ return DataFiles.builder(targetSpec).copy(file).build();
+ }
+ if (fromSpec.specId() == targetSpec.specId()) {
+ return DataFiles.builder(targetSpec).copy(file).build();
+ }
+ DataFiles.Builder builder =
+ DataFiles.builder(targetSpec)
+ .withPath(file.location())
+ .withFormat(file.format())
+ .withFileSizeInBytes(file.fileSizeInBytes())
+ .withRecordCount(file.recordCount());
+ if (file.splitOffsets() != null) {
+ builder.withSplitOffsets(file.splitOffsets());
+ }
+ if (file.sortOrderId() != null) {
+ builder.withSortOrderId(file.sortOrderId());
+ }
+ if (file.keyMetadata() != null) {
+ builder.withEncryptionKeyMetadata(file.keyMetadata());
+ }
+ StructLike projected = projectPartition(file.partition(), fromSpec,
targetSpec);
+ if (projected == null) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Cannot project data file %s partition from spec %d onto spec
%d",
+ file.location(), fromSpec.specId(), targetSpec.specId()));
+ }
+ builder.withPartition(projected);
+ return builder.build();
+ }
+
+ private static DeleteFile[] rewriteDeletes(
+ List<DeleteFile> deletes, PartitionSpec fromSpec, PartitionSpec
targetSpec) {
+ if (deletes == null || deletes.isEmpty()) {
+ return new DeleteFile[0];
+ }
+ DeleteFile[] rewritten = new DeleteFile[deletes.size()];
+ for (int i = 0; i < deletes.size(); i++) {
+ rewritten[i] = rewriteDeleteFile(deletes.get(i), fromSpec, targetSpec);
+ }
+ return rewritten;
+ }
+
+ private static DeleteFile rewriteDeleteFile(
+ DeleteFile delete, PartitionSpec fromSpec, PartitionSpec targetSpec) {
+ if (targetSpec.isUnpartitioned() || fromSpec.specId() ==
targetSpec.specId()) {
+ // Iceberg 1.11 FileMetadata.copy does not copy equalityFieldIds;
restore them explicitly.
+ FileMetadata.Builder builder =
FileMetadata.deleteFileBuilder(targetSpec).copy(delete);
+ restoreEqualityFieldIds(builder, delete);
+ return builder.build();
+ }
+ FileMetadata.Builder builder = FileMetadata.deleteFileBuilder(targetSpec);
+ switch (delete.content()) {
+ case POSITION_DELETES:
+ builder.ofPositionDeletes();
+ break;
+ case EQUALITY_DELETES:
+ restoreEqualityFieldIds(builder, delete);
+ break;
+ default:
+ throw new IllegalArgumentException("Unsupported delete content: " +
delete.content());
+ }
+ builder
+ .withPath(delete.location())
+ .withFormat(delete.format())
+ .withFileSizeInBytes(delete.fileSizeInBytes())
+ .withRecordCount(delete.recordCount());
+ if (delete.referencedDataFile() != null) {
+ builder.withReferencedDataFile(delete.referencedDataFile());
+ }
+ if (delete.contentOffset() != null) {
+ builder.withContentOffset(delete.contentOffset());
+ }
+ if (delete.contentSizeInBytes() != null) {
+ builder.withContentSizeInBytes(delete.contentSizeInBytes());
+ }
+ StructLike projected = projectPartition(delete.partition(), fromSpec,
targetSpec);
+ if (projected == null) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Cannot project delete file %s partition from spec %d onto spec
%d",
+ delete.location(), fromSpec.specId(), targetSpec.specId()));
+ }
+ builder.withPartition(projected);
+ return builder.build();
+ }
+
+ private static void restoreEqualityFieldIds(FileMetadata.Builder builder,
DeleteFile delete) {
+ if (delete.content() != FileContent.EQUALITY_DELETES) {
+ return;
+ }
+ List<Integer> equalityFieldIds = delete.equalityFieldIds();
+ int[] eqIds =
+ equalityFieldIds == null
+ ? new int[0]
+ : equalityFieldIds.stream().mapToInt(Integer::intValue).toArray();
+ builder.ofEqualityDeletes(eqIds);
+ }
+
+ /**
+ * True when every field of {@code toSpec} can be copied from {@code
fromSpec} by matching source
+ * id and transform. Unpartitioned targets are always projectable.
+ */
+ private static boolean canProjectPartition(PartitionSpec fromSpec,
PartitionSpec toSpec) {
+ if (toSpec.isUnpartitioned()) {
+ return true;
+ }
+ List<PartitionField> fromFields = fromSpec.fields();
+ for (PartitionField toField : toSpec.fields()) {
+ boolean found = false;
+ for (PartitionField fromField : fromFields) {
+ if (fromField.sourceId() == toField.sourceId()
+ &&
fromField.transform().toString().equals(toField.transform().toString())) {
+ found = true;
+ break;
+ }
+ }
+ if (!found) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ @Nullable
+ private static StructLike projectPartition(
+ StructLike partition, PartitionSpec fromSpec, PartitionSpec toSpec) {
+ if (toSpec.isUnpartitioned() || partition == null) {
+ return null;
+ }
+ if (!canProjectPartition(fromSpec, toSpec)) {
+ return null;
+ }
+ PartitionData out = new PartitionData(toSpec.partitionType());
+ List<PartitionField> fromFields = fromSpec.fields();
+ List<PartitionField> toFields = toSpec.fields();
+ for (int toPos = 0; toPos < toFields.size(); toPos++) {
+ PartitionField toField = toFields.get(toPos);
+ for (int i = 0; i < fromFields.size(); i++) {
+ PartitionField fromField = fromFields.get(i);
+ if (fromField.sourceId() == toField.sourceId()
+ &&
fromField.transform().toString().equals(toField.transform().toString())) {
+ out.set(toPos, partition.get(i, Object.class));
+ break;
+ }
+ }
+ }
+ return out;
+ }
+}
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cache/ScanPlanCacheKey.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cache/ScanPlanCacheKey.java
index 8823d1aaad..24fdfa3090 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cache/ScanPlanCacheKey.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cache/ScanPlanCacheKey.java
@@ -37,6 +37,7 @@ public class ScanPlanCacheKey {
private final String statsFieldsStr;
private final boolean caseSensitive;
private final boolean useSnapshotSchema;
+ private final int schemaId;
private ScanPlanCacheKey(
TableIdentifier tableIdentifier,
@@ -47,7 +48,8 @@ public class ScanPlanCacheKey {
String select,
String statsFields,
boolean caseSensitive,
- boolean useSnapshotSchema) {
+ boolean useSnapshotSchema,
+ int schemaId) {
this.tableIdentifier = tableIdentifier;
this.snapshotId = snapshotId;
this.startSnapshotId = startSnapshotId;
@@ -57,6 +59,7 @@ public class ScanPlanCacheKey {
this.statsFieldsStr = statsFields;
this.caseSensitive = caseSensitive;
this.useSnapshotSchema = useSnapshotSchema;
+ this.schemaId = schemaId;
}
/**
@@ -94,6 +97,9 @@ public class ScanPlanCacheKey {
statsFieldsStr =
scanRequest.statsFields().stream().sorted().collect(Collectors.joining(","));
}
+ // Current-schema planning binds names by field id; a schema-only
evolution must miss cache.
+ int schemaId = table.schema().schemaId();
+
return new ScanPlanCacheKey(
tableIdentifier,
snapshotId,
@@ -103,7 +109,8 @@ public class ScanPlanCacheKey {
selectStr,
statsFieldsStr,
scanRequest.caseSensitive(),
- scanRequest.useSnapshotSchema());
+ scanRequest.useSnapshotSchema(),
+ schemaId);
}
public TableIdentifier getTableIdentifier() {
@@ -125,6 +132,7 @@ public class ScanPlanCacheKey {
ScanPlanCacheKey that = (ScanPlanCacheKey) o;
return caseSensitive == that.caseSensitive
&& useSnapshotSchema == that.useSnapshotSchema
+ && schemaId == that.schemaId
&& Objects.equal(tableIdentifier, that.tableIdentifier)
&& Objects.equal(snapshotId, that.snapshotId)
&& Objects.equal(startSnapshotId, that.startSnapshotId)
@@ -147,14 +155,16 @@ public class ScanPlanCacheKey {
selectStr,
statsFieldsStr,
caseSensitive,
- useSnapshotSchema);
+ useSnapshotSchema,
+ schemaId);
}
@Override
public String toString() {
return String.format(
"ScanPlanCacheKey{table=%s, snapshotId=%s, startSnapshotId=%s,
endSnapshotId=%s, "
- + "filter=%s, select=%s, statsFields=%s, caseSensitive=%s,
useSnapshotSchema=%s}",
+ + "filter=%s, select=%s, statsFields=%s, caseSensitive=%s,
useSnapshotSchema=%s, "
+ + "schemaId=%s}",
tableIdentifier,
snapshotId,
startSnapshotId,
@@ -163,6 +173,7 @@ public class ScanPlanCacheKey {
selectStr.isEmpty() ? "null" : selectStr,
statsFieldsStr.isEmpty() ? "null" : statsFieldsStr,
caseSensitive,
- useSnapshotSchema);
+ useSnapshotSchema,
+ schemaId);
}
}
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergScanPlanHelper.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergScanPlanHelper.java
new file mode 100644
index 0000000000..61ba354586
--- /dev/null
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergScanPlanHelper.java
@@ -0,0 +1,732 @@
+/*
+ * 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.gravitino.iceberg.service;
+
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.ContentFileParser;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.FileMetadata;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.PartitionData;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.PartitionSpecParser;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SchemaParser;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.expressions.Evaluator;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.ExpressionParser;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.expressions.ResidualEvaluator;
+import org.apache.iceberg.rest.requests.PlanTableScanRequest;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.JsonUtil;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestIcebergScanPlanHelper {
+
+ // Field 2 was "l" at the snapshot and is "l2" now; field 3, added later,
reuses the name "l".
+ private static final Schema SNAPSHOT_SCHEMA =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "l", Types.LongType.get()));
+ private static final Schema CURRENT_SCHEMA =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "l2", Types.LongType.get()),
+ Types.NestedField.optional(3, "l", Types.LongType.get()));
+
+ @Test
+ void testRewriteForSnapshotMapsCurrentNames() {
+ PlanTableScanRequest rewritten =
+ IcebergScanPlanHelper.rewriteForSnapshot(
+ PlanTableScanRequest.builder()
+ .withSnapshotId(7L)
+ .withFilter(Expressions.equal("l2", 10L))
+ .withSelect(ImmutableList.of("id", "l2", "l"))
+ .withStatsFields(ImmutableList.of("l2", "l"))
+ .build(),
+ CURRENT_SCHEMA,
+ SNAPSHOT_SCHEMA);
+
+ Assertions.assertEquals(Expressions.equal("l", 10L).toString(),
rewritten.filter().toString());
+ Assertions.assertEquals(ImmutableList.of("id", "l"), rewritten.select());
+ Assertions.assertEquals(ImmutableList.of("l"), rewritten.statsFields());
+ }
+
+ @Test
+ void testRewriteForSnapshotResolvesCaseInsensitively() {
+ PlanTableScanRequest rewritten =
+ IcebergScanPlanHelper.rewriteForSnapshot(
+ PlanTableScanRequest.builder()
+ .withSnapshotId(7L)
+ .withCaseSensitive(false)
+ .withFilter(Expressions.equal("L2", 10L))
+ .build(),
+ CURRENT_SCHEMA,
+ SNAPSHOT_SCHEMA);
+
+ Assertions.assertEquals(Expressions.equal("l", 10L).toString(),
rewritten.filter().toString());
+ }
+
+ @Test
+ void testRewriteForSnapshotKeepsStarSelect() {
+ PlanTableScanRequest rewritten =
+ IcebergScanPlanHelper.rewriteForSnapshot(
+ PlanTableScanRequest.builder()
+ .withSnapshotId(7L)
+ .withSelect(ImmutableList.of("*"))
+ .build(),
+ CURRENT_SCHEMA,
+ SNAPSHOT_SCHEMA);
+
+ Assertions.assertEquals(ImmutableList.of("*"), rewritten.select());
+ }
+
+ @Test
+ void testRewriteFilterNames() {
+ Assertions.assertEquals(
+ Expressions.equal("l", 10L).toString(),
+ IcebergScanPlanHelper.rewriteFilterNames(
+ Expressions.equal("l2", 10L), CURRENT_SCHEMA, SNAPSHOT_SCHEMA,
true)
+ .toString());
+ Assertions.assertEquals(
+ Expressions.equal("id", 1L).toString(),
+ IcebergScanPlanHelper.rewriteFilterNames(
+ Expressions.equal("id", 1L), CURRENT_SCHEMA, SNAPSHOT_SCHEMA,
true)
+ .toString());
+ Assertions.assertEquals(
+ Expressions.and(Expressions.in("l", 1L, 2L),
Expressions.not(Expressions.isNull("l")))
+ .toString(),
+ IcebergScanPlanHelper.rewriteFilterNames(
+ Expressions.and(
+ Expressions.in("l2", 1L, 2L),
Expressions.not(Expressions.isNull("l2"))),
+ CURRENT_SCHEMA,
+ SNAPSHOT_SCHEMA,
+ true)
+ .toString());
+ Assertions.assertEquals(
+ Expressions.equal(Expressions.bucket("l", 4), 1).toString(),
+ IcebergScanPlanHelper.rewriteFilterNames(
+ Expressions.equal(Expressions.bucket("l2", 4), 1),
+ CURRENT_SCHEMA,
+ SNAPSHOT_SCHEMA,
+ true)
+ .toString());
+
+ // Column added after the snapshot: do not prune; callers AND the full
original filter.
+ Assertions.assertEquals(
+ Expressions.alwaysTrue().toString(),
+ IcebergScanPlanHelper.rewriteFilterNames(
+ Expressions.isNull("l"), CURRENT_SCHEMA, SNAPSHOT_SCHEMA, true)
+ .toString());
+ Assertions.assertEquals(
+ Expressions.alwaysTrue().toString(),
+ IcebergScanPlanHelper.rewriteFilterNames(
+ Expressions.equal("l", 10L), CURRENT_SCHEMA, SNAPSHOT_SCHEMA,
true)
+ .toString());
+
+ // Mixed OR: post-snapshot leaf becomes always-true so planning
over-includes; residual keeps
+ // the complete original filter.
+ Assertions.assertEquals(
+ Expressions.alwaysTrue().toString(),
+ IcebergScanPlanHelper.rewriteFilterNames(
+ Expressions.or(Expressions.equal("id", 1L),
Expressions.equal("l", 2L)),
+ CURRENT_SCHEMA,
+ SNAPSHOT_SCHEMA,
+ true)
+ .toString());
+
+ // NOT of a post-snapshot predicate must not become always-false (would
prune every file).
+ Assertions.assertEquals(
+ Expressions.alwaysTrue().toString(),
+ IcebergScanPlanHelper.rewriteFilterNames(
+ Expressions.not(Expressions.equal("l", 1L)), CURRENT_SCHEMA,
SNAPSHOT_SCHEMA, true)
+ .toString());
+ Assertions.assertEquals(
+ Expressions.not(Expressions.equal("id", 1L)).toString(),
+ IcebergScanPlanHelper.rewriteFilterNames(
+ Expressions.not(Expressions.equal("id", 1L)), CURRENT_SCHEMA,
SNAPSHOT_SCHEMA, true)
+ .toString());
+
+ // NOT over mixed AND/OR must stay weak (never collapse to NOT(old) and
under-prune).
+ Assertions.assertEquals(
+ Expressions.alwaysTrue().toString(),
+ IcebergScanPlanHelper.rewriteFilterNames(
+ Expressions.not(
+ Expressions.and(Expressions.equal("id", 1L),
Expressions.notNull("l"))),
+ CURRENT_SCHEMA,
+ SNAPSHOT_SCHEMA,
+ true)
+ .toString());
+ // Under NOT, a post-snapshot OR leaf becomes false, so planning keeps
NOT(old) — correct for
+ // historical rows where the new column is null/absent.
+ Assertions.assertEquals(
+ Expressions.not(Expressions.equal("id", 1L)).toString(),
+ IcebergScanPlanHelper.rewriteFilterNames(
+ Expressions.not(
+ Expressions.or(Expressions.equal("id", 1L),
Expressions.equal("l", 2L))),
+ CURRENT_SCHEMA,
+ SNAPSHOT_SCHEMA,
+ true)
+ .toString());
+
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ IcebergScanPlanHelper.rewriteFilterNames(
+ Expressions.equal("nope", 1L), CURRENT_SCHEMA,
SNAPSHOT_SCHEMA, true));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ IcebergScanPlanHelper.rewriteFilterNames(
+ Expressions.equal("L2", 10L), CURRENT_SCHEMA, SNAPSHOT_SCHEMA,
true));
+ }
+
+ @Test
+ void testRemapResidualKeepsRequestFieldIdentities() {
+ Expression residualOnSnapshot =
+ IcebergScanPlanHelper.rewriteFilterNames(
+ Expressions.equal("l2", 10L), CURRENT_SCHEMA, SNAPSHOT_SCHEMA,
true);
+ Assertions.assertEquals(Expressions.equal("l", 10L).toString(),
residualOnSnapshot.toString());
+
+ Expression residualForClient =
+ IcebergScanPlanHelper.remapExpressionNames(
+ residualOnSnapshot, SNAPSHOT_SCHEMA, CURRENT_SCHEMA, true);
+ Assertions.assertEquals(Expressions.equal("l2", 10L).toString(),
residualForClient.toString());
+
+ // With name reuse, evaluating the remapped residual against the request
row is true; the
+ // snapshot-named residual would bind to the wrong field and evaluate
false.
+ Types.StructType rowType = CURRENT_SCHEMA.asStruct();
+ StructLike row = new ArrayStruct(1L, 10L, null);
+ Assertions.assertTrue(new Evaluator(rowType, residualForClient).eval(row));
+ Assertions.assertFalse(new Evaluator(rowType,
residualOnSnapshot).eval(row));
+ }
+
+ @Test
+ void testRewriteColumnNamesDropsColumnsAddedLater() {
+ Assertions.assertEquals(
+ ImmutableList.of("id", "l"),
+ IcebergScanPlanHelper.rewriteColumnNames(
+ ImmutableList.of("id", "l2", "l"), CURRENT_SCHEMA,
SNAPSHOT_SCHEMA, true));
+ }
+
+ @Test
+ void testSanitizeSpecsForDroppedPartitionSource() {
+ Schema originalSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "s", Types.StringType.get()),
+ Types.NestedField.optional(3, "n", Types.IntegerType.get()));
+ PartitionSpec originalSpec =
+
PartitionSpec.builderFor(originalSchema).withSpecId(0).identity("s").build();
+
+ Schema evolvedSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(3, "n", Types.IntegerType.get()));
+ // allowMissingFields=true mirrors a table whose current schema dropped
the partition source
+ // while the historical spec is still retained in specs-by-id.
+ PartitionSpec staleSpec = originalSpec.toUnbound().bind(evolvedSchema,
true);
+ Assertions.assertTrue(
+ staleSpec.partitionType().fields().stream()
+ .anyMatch(f -> f.type().typeId() == Type.TypeID.UNKNOWN));
+
+ Types.StructType partitionDataType =
+ Types.StructType.of(Types.NestedField.optional(1000, "s",
Types.StringType.get()));
+ PartitionData partitionData = new PartitionData(partitionDataType);
+ partitionData.set(0, "a");
+
+ DataFile dataFile = mock(DataFile.class);
+ doReturn(0).when(dataFile).specId();
+ doReturn(partitionData).when(dataFile).partition();
+
+ FileScanTask task = mock(FileScanTask.class);
+ doReturn(dataFile).when(task).file();
+
+ Map<Integer, PartitionSpec> sanitized =
+ IcebergScanPlanHelper.sanitizeSpecsForSerialization(
+ ImmutableMap.of(0, staleSpec),
+ Collections.singletonList(task),
+ ImmutableList.of(originalSchema, evolvedSchema));
+
+ Assertions.assertFalse(
+ sanitized.get(0).partitionType().fields().stream()
+ .anyMatch(f -> f.type().typeId() == Type.TypeID.UNKNOWN));
+ Assertions.assertEquals(
+ Type.TypeID.STRING,
sanitized.get(0).partitionType().fields().get(0).type().typeId());
+ }
+
+ @Test
+ void testSanitizeSpecsDedupesMultipleTransformsOfOneDroppedSource() {
+ Schema originalSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "s", Types.StringType.get()));
+ PartitionSpec originalSpec =
+
PartitionSpec.builderFor(originalSchema).withSpecId(0).identity("s").bucket("s",
8).build();
+ Schema evolved = new Schema(Types.NestedField.optional(1, "id",
Types.LongType.get()));
+ PartitionSpec stale = originalSpec.toUnbound().bind(evolved, true);
+
+ Map<Integer, PartitionSpec> sanitized =
+ IcebergScanPlanHelper.sanitizeSpecsForSerialization(
+ ImmutableMap.of(0, stale),
+ Collections.emptyList(),
+ ImmutableList.of(originalSchema, evolved));
+
+ Assertions.assertDoesNotThrow(() -> sanitized.get(0).schema());
+ Assertions.assertEquals(
+ 1,
+ sanitized.get(0).schema().columns().stream()
+ .filter(f -> f.name().startsWith("_dropped_source_"))
+ .count());
+ }
+
+ @Test
+ void testSanitizeSpecsDoesNotUseTransformResultTypeAsSource() {
+ Schema originalSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "s", Types.StringType.get()),
+ Types.NestedField.optional(3, "ts",
Types.TimestampType.withoutZone()));
+ PartitionSpec originalSpec =
+
PartitionSpec.builderFor(originalSchema).withSpecId(0).identity("s").hour("ts").build();
+ Schema evolved = new Schema(Types.NestedField.optional(1, "id",
Types.LongType.get()));
+ PartitionSpec stale = originalSpec.toUnbound().bind(evolved, true);
+
+ // Partition data carries transform results (string + int), which must not
become source types.
+ Types.StructType partitionDataType =
+ Types.StructType.of(
+ Types.NestedField.optional(1000, "s", Types.StringType.get()),
+ Types.NestedField.optional(1001, "ts_hour",
Types.IntegerType.get()));
+ PartitionData partitionData = new PartitionData(partitionDataType);
+ partitionData.set(0, "a");
+ partitionData.set(1, 1);
+ DataFile dataFile = mock(DataFile.class);
+ doReturn(0).when(dataFile).specId();
+ doReturn(partitionData).when(dataFile).partition();
+ FileScanTask task = mock(FileScanTask.class);
+ doReturn(dataFile).when(task).file();
+
+ Map<Integer, PartitionSpec> sanitized =
+ IcebergScanPlanHelper.sanitizeSpecsForSerialization(
+ ImmutableMap.of(0, stale),
+ Collections.singletonList(task),
+ ImmutableList.of(originalSchema, evolved));
+
+ Assertions.assertEquals(
+ Type.TypeID.STRING,
sanitized.get(0).schema().findField(2).type().typeId());
+ Assertions.assertEquals(
+ Type.TypeID.TIMESTAMP,
sanitized.get(0).schema().findField(3).type().typeId());
+ }
+
+ @Test
+ void testPrepareTasksRewritesUnknownSpecForClientRoundTrip() {
+ Schema originalSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "s", Types.StringType.get()));
+ PartitionSpec originalSpec =
+
PartitionSpec.builderFor(originalSchema).withSpecId(0).identity("s").build();
+ Schema evolved = new Schema(Types.NestedField.optional(1, "id",
Types.LongType.get()));
+ PartitionSpec unpartitioned =
PartitionSpec.builderFor(evolved).withSpecId(1).build();
+ PartitionSpec stale = originalSpec.toUnbound().bind(evolved, true);
+
+ PartitionData partitionData =
+ new PartitionData(
+ Types.StructType.of(Types.NestedField.optional(1000, "s",
Types.StringType.get())));
+ partitionData.set(0, "a");
+ DataFile dataFile =
+ DataFiles.builder(originalSpec)
+ .withPath("file:///tmp/f.parquet")
+ .withFormat(FileFormat.PARQUET)
+ .withFileSizeInBytes(10)
+ .withRecordCount(1)
+ .withPartition(partitionData)
+ .build();
+
+ // Build a task against the stale (client-visible) spec id.
+ org.apache.iceberg.BaseFileScanTask task =
+ new org.apache.iceberg.BaseFileScanTask(
+
DataFiles.builder(stale).copy(dataFile).withPartition(partitionData).build(),
+ null,
+ SchemaParser.toJson(evolved),
+ PartitionSpecParser.toJson(stale),
+ ResidualEvaluator.of(stale, Expressions.equal("id", 1L), true));
+
+ Map<Integer, PartitionSpec> clientSpecs = ImmutableMap.of(0, stale, 1,
unpartitioned);
+ List<FileScanTask> prepared =
+ IcebergScanPlanHelper.prepareTasksForResponse(
+ Collections.singletonList(task), clientSpecs, evolved, null, null,
true);
+
+ Assertions.assertEquals(1, prepared.get(0).file().specId());
+
Assertions.assertTrue(clientSpecs.get(prepared.get(0).file().specId()).isUnpartitioned());
+
+ // Serialize with sanitized specs (server), parse with the client's
original broken specs.
+ Map<Integer, PartitionSpec> sanitized =
+ IcebergScanPlanHelper.sanitizeSpecsForSerialization(
+ clientSpecs, prepared, ImmutableList.of(originalSchema, evolved));
+ String json =
+ JsonUtil.generate(
+ gen -> ContentFileParser.toJson(prepared.get(0).file(),
sanitized.get(1), gen), false);
+ JsonNode node = JsonUtil.parse(json, n -> n);
+ DataFile parsed = (DataFile) ContentFileParser.fromJson(node, clientSpecs);
+ Assertions.assertEquals("file:///tmp/f.parquet", parsed.location());
+ }
+
+ @Test
+ void testPrepareTasksRewritesToCurrentSpecWhenNeverUnpartitioned() {
+ // Evolution (s, id) → (id) never creates an unpartitioned spec; old files
keep spec 0.
+ Schema originalSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "s", Types.StringType.get()));
+ PartitionSpec originalSpec =
+
PartitionSpec.builderFor(originalSchema).withSpecId(0).identity("s").identity("id").build();
+ Schema evolved =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "s", Types.StringType.get()));
+ PartitionSpec currentSpec =
+ PartitionSpec.builderFor(evolved).withSpecId(1).identity("id").build();
+ Schema droppedS = new Schema(Types.NestedField.optional(1, "id",
Types.LongType.get()));
+ PartitionSpec stale = originalSpec.toUnbound().bind(droppedS, true);
+ PartitionSpec currentOnDropped = currentSpec.toUnbound().bind(droppedS,
true);
+
+ PartitionData partitionData =
+ new PartitionData(
+ Types.StructType.of(
+ Types.NestedField.optional(1000, "s", Types.StringType.get()),
+ Types.NestedField.optional(1001, "id", Types.LongType.get())));
+ partitionData.set(0, "a");
+ partitionData.set(1, 7L);
+ DataFile dataFile =
+ DataFiles.builder(originalSpec)
+ .withPath("file:///tmp/sid.parquet")
+ .withFormat(FileFormat.PARQUET)
+ .withFileSizeInBytes(10)
+ .withRecordCount(1)
+ .withPartition(partitionData)
+ .build();
+
+ org.apache.iceberg.BaseFileScanTask task =
+ new org.apache.iceberg.BaseFileScanTask(
+
DataFiles.builder(stale).copy(dataFile).withPartition(partitionData).build(),
+ null,
+ SchemaParser.toJson(droppedS),
+ PartitionSpecParser.toJson(stale),
+ ResidualEvaluator.of(stale, Expressions.alwaysTrue(), true));
+
+ Map<Integer, PartitionSpec> clientSpecs = ImmutableMap.of(0, stale, 1,
currentOnDropped);
+ List<FileScanTask> prepared =
+ IcebergScanPlanHelper.prepareTasksForResponse(
+ Collections.singletonList(task), clientSpecs, droppedS, null,
null, true);
+
+ Assertions.assertEquals(1, prepared.get(0).file().specId());
+ Assertions.assertFalse(clientSpecs.get(1).isUnpartitioned());
+
+ String json =
+ JsonUtil.generate(
+ gen -> ContentFileParser.toJson(prepared.get(0).file(),
clientSpecs.get(1), gen),
+ false);
+ DataFile parsed =
+ (DataFile) ContentFileParser.fromJson(JsonUtil.parse(json, n -> n),
clientSpecs);
+ Assertions.assertEquals("file:///tmp/sid.parquet", parsed.location());
+ Assertions.assertEquals(1, parsed.specId());
+ }
+
+ @Test
+ void testPrepareTasksFailsWhenTargetPartitionValuesMissing() {
+ // Evolution (s) → (id) with s dropped: old files have no id partition
value and no
+ // unpartitioned spec. Rewriting must not invent NULL for id.
+ Schema originalSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "s", Types.StringType.get()));
+ PartitionSpec originalSpec =
+
PartitionSpec.builderFor(originalSchema).withSpecId(0).identity("s").build();
+ Schema evolved = new Schema(Types.NestedField.optional(1, "id",
Types.LongType.get()));
+ PartitionSpec currentSpec =
+ PartitionSpec.builderFor(evolved).withSpecId(1).identity("id").build();
+ PartitionSpec stale = originalSpec.toUnbound().bind(evolved, true);
+ PartitionSpec currentOnDropped = currentSpec.toUnbound().bind(evolved,
true);
+
+ PartitionData partitionData =
+ new PartitionData(
+ Types.StructType.of(Types.NestedField.optional(1000, "s",
Types.StringType.get())));
+ partitionData.set(0, "a");
+ DataFile dataFile =
+ DataFiles.builder(originalSpec)
+ .withPath("file:///tmp/missing-id.parquet")
+ .withFormat(FileFormat.PARQUET)
+ .withFileSizeInBytes(10)
+ .withRecordCount(1)
+ .withPartition(partitionData)
+ .build();
+
+ org.apache.iceberg.BaseFileScanTask task =
+ new org.apache.iceberg.BaseFileScanTask(
+
DataFiles.builder(stale).copy(dataFile).withPartition(partitionData).build(),
+ null,
+ SchemaParser.toJson(evolved),
+ PartitionSpecParser.toJson(stale),
+ ResidualEvaluator.of(stale, Expressions.equal("id", 1L), true));
+
+ Map<Integer, PartitionSpec> clientSpecs = ImmutableMap.of(0, stale, 1,
currentOnDropped);
+ IllegalArgumentException thrown =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ IcebergScanPlanHelper.prepareTasksForResponse(
+ Collections.singletonList(task), clientSpecs, evolved,
null, null, true));
+ Assertions.assertTrue(
+ thrown.getMessage().contains("missing from the source file"),
thrown.getMessage());
+ }
+
+ @Test
+ void testPrepareTasksRewritesDeleteFilesWithCompatibleSpec() {
+ Schema originalSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "s", Types.StringType.get()));
+ PartitionSpec originalSpec =
+
PartitionSpec.builderFor(originalSchema).withSpecId(0).identity("s").build();
+ Schema evolved = new Schema(Types.NestedField.optional(1, "id",
Types.LongType.get()));
+ PartitionSpec unpartitioned =
PartitionSpec.builderFor(evolved).withSpecId(1).build();
+ PartitionSpec stale = originalSpec.toUnbound().bind(evolved, true);
+
+ PartitionData partitionData =
+ new PartitionData(
+ Types.StructType.of(Types.NestedField.optional(1000, "s",
Types.StringType.get())));
+ partitionData.set(0, "a");
+ DataFile dataFile =
+ DataFiles.builder(originalSpec)
+ .withPath("file:///tmp/data.parquet")
+ .withFormat(FileFormat.PARQUET)
+ .withFileSizeInBytes(10)
+ .withRecordCount(1)
+ .withPartition(partitionData)
+ .build();
+ DeleteFile deleteFile =
+ FileMetadata.deleteFileBuilder(originalSpec)
+ .ofPositionDeletes()
+ .withPath("file:///tmp/data.parquet.pos-deletes")
+ .withFormat(FileFormat.PARQUET)
+ .withFileSizeInBytes(4)
+ .withRecordCount(1)
+ .withPartition(partitionData)
+ .build();
+
+ org.apache.iceberg.BaseFileScanTask task =
+ new org.apache.iceberg.BaseFileScanTask(
+
DataFiles.builder(stale).copy(dataFile).withPartition(partitionData).build(),
+ new DeleteFile[] {
+ FileMetadata.deleteFileBuilder(stale)
+ .copy(deleteFile)
+ .withPartition(partitionData)
+ .build()
+ },
+ SchemaParser.toJson(evolved),
+ PartitionSpecParser.toJson(stale),
+ ResidualEvaluator.of(stale, Expressions.alwaysTrue(), true));
+
+ Map<Integer, PartitionSpec> clientSpecs = ImmutableMap.of(0, stale, 1,
unpartitioned);
+ List<FileScanTask> prepared =
+ IcebergScanPlanHelper.prepareTasksForResponse(
+ Collections.singletonList(task), clientSpecs, evolved, null, null,
true);
+
+ Assertions.assertEquals(1, prepared.get(0).file().specId());
+ Assertions.assertEquals(1, prepared.get(0).deletes().size());
+ Assertions.assertEquals(1, prepared.get(0).deletes().get(0).specId());
+
+ String deleteJson =
+ JsonUtil.generate(
+ gen -> ContentFileParser.toJson(prepared.get(0).deletes().get(0),
unpartitioned, gen),
+ false);
+ DeleteFile parsedDelete =
+ (DeleteFile) ContentFileParser.fromJson(JsonUtil.parse(deleteJson, n
-> n), clientSpecs);
+ Assertions.assertEquals("file:///tmp/data.parquet.pos-deletes",
parsedDelete.location());
+ Assertions.assertEquals(1, parsedDelete.specId());
+ }
+
+ @Test
+ void testPrepareTasksPreservesEqualityFieldIdsOnRewrite() {
+ Schema originalSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "s", Types.StringType.get()));
+ PartitionSpec originalSpec =
+
PartitionSpec.builderFor(originalSchema).withSpecId(0).identity("s").build();
+ Schema evolved = new Schema(Types.NestedField.optional(1, "id",
Types.LongType.get()));
+ PartitionSpec unpartitioned =
PartitionSpec.builderFor(evolved).withSpecId(1).build();
+ PartitionSpec stale = originalSpec.toUnbound().bind(evolved, true);
+
+ PartitionData partitionData =
+ new PartitionData(
+ Types.StructType.of(Types.NestedField.optional(1000, "s",
Types.StringType.get())));
+ partitionData.set(0, "a");
+ DataFile dataFile =
+ DataFiles.builder(originalSpec)
+ .withPath("file:///tmp/eq-data.parquet")
+ .withFormat(FileFormat.PARQUET)
+ .withFileSizeInBytes(10)
+ .withRecordCount(1)
+ .withPartition(partitionData)
+ .build();
+ DeleteFile equalityDelete =
+ FileMetadata.deleteFileBuilder(originalSpec)
+ .ofEqualityDeletes(1)
+ .withPath("file:///tmp/eq-data.parquet.eq-deletes")
+ .withFormat(FileFormat.PARQUET)
+ .withFileSizeInBytes(4)
+ .withRecordCount(1)
+ .withPartition(partitionData)
+ .build();
+
+ org.apache.iceberg.BaseFileScanTask task =
+ new org.apache.iceberg.BaseFileScanTask(
+
DataFiles.builder(stale).copy(dataFile).withPartition(partitionData).build(),
+ new DeleteFile[] {
+ FileMetadata.deleteFileBuilder(stale)
+ .copy(equalityDelete)
+ .ofEqualityDeletes(1)
+ .withPartition(partitionData)
+ .build()
+ },
+ SchemaParser.toJson(evolved),
+ PartitionSpecParser.toJson(stale),
+ ResidualEvaluator.of(stale, Expressions.alwaysTrue(), true));
+
+ Map<Integer, PartitionSpec> clientSpecs = ImmutableMap.of(0, stale, 1,
unpartitioned);
+ List<FileScanTask> prepared =
+ IcebergScanPlanHelper.prepareTasksForResponse(
+ Collections.singletonList(task), clientSpecs, evolved, null, null,
true);
+
+ DeleteFile rewritten = prepared.get(0).deletes().get(0);
+ Assertions.assertEquals(1, rewritten.specId());
+ Assertions.assertEquals(ImmutableList.of(1), rewritten.equalityFieldIds());
+
+ String deleteJson =
+ JsonUtil.generate(gen -> ContentFileParser.toJson(rewritten,
unpartitioned, gen), false);
+ DeleteFile parsedDelete =
+ (DeleteFile) ContentFileParser.fromJson(JsonUtil.parse(deleteJson, n
-> n), clientSpecs);
+ Assertions.assertEquals(ImmutableList.of(1),
parsedDelete.equalityFieldIds());
+ }
+
+ @Test
+ void testPrepareTasksResidualEvaluatorUsesRequestSchema() {
+ // Partitioned by id; snapshot schema names field 2 "l", request schema
renamed it to "l2".
+ Schema snapshotSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "l", Types.LongType.get()));
+ Schema requestSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "l2", Types.LongType.get()));
+ PartitionSpec snapshotSpec =
+
PartitionSpec.builderFor(snapshotSchema).withSpecId(0).identity("id").build();
+ PartitionSpec requestSpec = snapshotSpec.toUnbound().bind(requestSchema,
true);
+
+ PartitionData partitionData =
+ new PartitionData(
+ Types.StructType.of(Types.NestedField.optional(1000, "id",
Types.LongType.get())));
+ partitionData.set(0, 1L);
+ DataFile dataFile =
+ DataFiles.builder(snapshotSpec)
+ .withPath("file:///tmp/part.parquet")
+ .withFormat(FileFormat.PARQUET)
+ .withFileSizeInBytes(10)
+ .withRecordCount(1)
+ .withPartition(partitionData)
+ .build();
+
+ Expression snapshotResidual = Expressions.equal("l", 10L);
+ org.apache.iceberg.BaseFileScanTask task =
+ new org.apache.iceberg.BaseFileScanTask(
+ dataFile,
+ null,
+ SchemaParser.toJson(snapshotSchema),
+ PartitionSpecParser.toJson(snapshotSpec),
+ ResidualEvaluator.of(snapshotSpec, snapshotResidual, true));
+
+ List<FileScanTask> prepared =
+ IcebergScanPlanHelper.prepareTasksForResponse(
+ Collections.singletonList(task),
+ ImmutableMap.of(0, requestSpec),
+ requestSchema,
+ snapshotSchema,
+ null,
+ true);
+
+ Assertions.assertEquals(
+ Expressions.equal("l2", 10L).toString(),
prepared.get(0).residual().toString());
+ // ResidualEvaluator must bind remapped names against the request schema,
not the snapshot.
+ Types.StructType rowType = requestSchema.asStruct();
+ Assertions.assertTrue(
+ new Evaluator(rowType, prepared.get(0).residual()).eval(new
ArrayStruct(1L, 10L)));
+ }
+
+ @Test
+ void testRemapExpressionRoundTripThroughParser() {
+ Expression remapped =
+ IcebergScanPlanHelper.remapExpressionNames(
+ Expressions.equal("l", 10L), SNAPSHOT_SCHEMA, CURRENT_SCHEMA,
true);
+ String json = ExpressionParser.toJson(remapped);
+ Expression parsed = ExpressionParser.fromJson(json);
+ Assertions.assertEquals(Expressions.equal("l2", 10L).toString(),
parsed.toString());
+ }
+
+ private static final class ArrayStruct implements StructLike {
+ private final Object[] values;
+
+ private ArrayStruct(Object... values) {
+ this.values = values;
+ }
+
+ @Override
+ public int size() {
+ return values.length;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T> T get(int pos, Class<T> javaClass) {
+ return (T) values[pos];
+ }
+
+ @Override
+ public <T> void set(int pos, T value) {
+ values[pos] = value;
+ }
+ }
+}
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergScanPlanning.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergScanPlanning.java
index 84b11e5ec1..4f0d1d034f 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergScanPlanning.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergScanPlanning.java
@@ -20,13 +20,18 @@ package org.apache.gravitino.iceberg.service;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.stream.Collectors;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
import org.apache.gravitino.iceberg.common.IcebergConfig;
+import org.apache.iceberg.DataFile;
import org.apache.iceberg.DataFiles;
import org.apache.iceberg.DeleteFile;
import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.Metrics;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
@@ -38,16 +43,20 @@ import org.apache.iceberg.expressions.Expression;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.rest.requests.PlanTableScanRequest;
import org.apache.iceberg.rest.responses.PlanTableScanResponse;
+import org.apache.iceberg.types.Conversions;
import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.function.Executable;
import org.mockito.Mockito;
/** Tests server-side scan planning against the REST spec's request semantics.
*/
public class TestIcebergScanPlanning {
private static final TableIdentifier IDENT =
TableIdentifier.of(Namespace.of("db"), "t");
+ private static final List<String> F1 = ImmutableList.of("/data/f1.parquet");
+ private static final List<String> NONE = ImmutableList.of();
private CatalogWrapperForREST wrapper;
private Table table;
@@ -130,6 +139,101 @@ public class TestIcebergScanPlanning {
Assertions.assertEquals(Expression.Operation.TRUE,
alsoAccepted.filter().op());
}
+ @Test
+ public void testBranchScanBindsToTheCurrentSchema() {
+ EvolvedTable evolved = evolvedTable("branch_scan");
+ // Branch "b" still points at s1, so this is the request Iceberg's client
sends for a branch.
+ evolved.table.manageSnapshots().createBranch("b", evolved.s1).commit();
+
+ Assertions.assertEquals(F1, planBranch(evolved, Expressions.equal("l2",
10L)));
+ Assertions.assertEquals(NONE, planBranch(evolved, Expressions.equal("l2",
99L)));
+ // "l" is the column added after s1. It has no values in f1, so f1 is not
pruned on it,
+ // while binding to the snapshot's "l" (field 2, bounds [10, 10]) would
prune f1.
+ Assertions.assertEquals(F1, planBranch(evolved, Expressions.equal("l",
99L)));
+ assertInvalidScan(() -> planBranch(evolved, Expressions.equal("a", 5L)),
"'a'");
+
+ Assertions.assertEquals(
+ F1,
+ planFiles(
+ evolved.ident,
+ branchRequest(evolved).withSelect(ImmutableList.of("id", "l2",
"l", "a2"))));
+ Assertions.assertEquals(
+ F1,
+ planFiles(
+ evolved.ident,
+ branchRequest(evolved)
+ .withCaseSensitive(false)
+ .withFilter(Expressions.equal("L2", 10L))));
+ }
+
+ @Test
+ public void testSnapshotNoBranchPointsAtBindsToTheCurrentSchema() {
+ EvolvedTable evolved = evolvedTable("moved_branch");
+ // The branch moved past s1 after the client read it, so no branch points
at s1 any more.
+ evolved.table.manageSnapshots().createBranch("b", evolved.s1).commit();
+ evolved
+ .table
+ .newFastAppend()
+ .appendFile(file("/data/f3.parquet", ImmutableMap.of(2, 30L)))
+ .toBranch("b")
+ .commit();
+ evolved.table.refresh();
+
+ Assertions.assertEquals(F1, planBranch(evolved, Expressions.equal("l2",
10L)));
+ Assertions.assertEquals(NONE, planBranch(evolved, Expressions.equal("l2",
99L)));
+ // Same current-schema semantics as when a branch still pointed here: "l"
was added after s1,
+ // so old files are not pruned on it (they read as null under the current
schema).
+ Assertions.assertEquals(F1, planBranch(evolved, Expressions.equal("l",
99L)));
+ assertInvalidScan(() -> planBranch(evolved, Expressions.equal("a", 5L)),
"'a'");
+
+ // Residual must keep current-schema names (l2), not the snapshot name (l)
that now means a
+ // different field id after rename+reuse. The complete client filter is
ANDed on as an
+ // extra residual (may duplicate the remapped scan residual; that is
intentional).
+ PlanTableScanResponse residualResponse =
+ wrapper.planTableScan(
+ evolved.ident,
branchRequest(evolved).withFilter(Expressions.equal("l2", 10L)).build());
+ Assertions.assertEquals(1, residualResponse.fileScanTasks().size());
+ String residual =
residualResponse.fileScanTasks().get(0).residual().toString();
+ Assertions.assertTrue(residual.contains("l2"), residual);
+ Assertions.assertFalse(residual.contains("name=\"l\""), residual);
+
+ Assertions.assertEquals(
+ F1,
+ planFiles(
+ evolved.ident,
+ branchRequest(evolved)
+ .withSelect(ImmutableList.of("id", "l2", "l", "a2"))
+ .withStatsFields(ImmutableList.of("l2", "l"))));
+ Assertions.assertEquals(
+ F1,
+ planFiles(
+ evolved.ident,
+ branchRequest(evolved)
+ .withCaseSensitive(false)
+ .withFilter(Expressions.equal("L2", 10L))));
+ Assertions.assertEquals(
+ F1, planFiles(evolved.ident,
branchRequest(evolved).withSelect(ImmutableList.of("*"))));
+ }
+
+ @Test
+ public void testTimeTravelBindsToTheSnapshotSchema() {
+ EvolvedTable evolved = evolvedTable("time_travel");
+ PlanTableScanRequest.Builder request =
+
PlanTableScanRequest.builder().withSnapshotId(evolved.s1).withUseSnapshotSchema(true);
+
+ Assertions.assertEquals(
+ F1, planFiles(evolved.ident, request.withFilter(Expressions.equal("l",
10L))));
+ assertInvalidScan(
+ () ->
+ planFiles(
+ evolved.ident,
+ PlanTableScanRequest.builder()
+ .withSnapshotId(evolved.s1)
+ .withUseSnapshotSchema(true)
+ .withFilter(Expressions.equal("l2", 10L))),
+ "'l2'");
+ }
+
@Test
public void testDeletionVectorsSharingAFileAreRefused() {
DeleteFile dv1 = deletionVector("/data/dv.puffin", 4L, "/data/f1.parquet");
@@ -174,10 +278,87 @@ public class TestIcebergScanPlanning {
}
private List<String> planFiles(PlanTableScanRequest.Builder request) {
- PlanTableScanResponse response = wrapper.planTableScan(IDENT,
request.build());
+ return planFiles(IDENT, request);
+ }
+
+ private List<String> planFiles(TableIdentifier ident,
PlanTableScanRequest.Builder request) {
+ PlanTableScanResponse response = wrapper.planTableScan(ident,
request.build());
return response.fileScanTasks().stream()
.map(task -> task.file().location())
.sorted()
.collect(Collectors.toList());
}
+
+ private List<String> planBranch(EvolvedTable evolved, Expression filter) {
+ return planFiles(evolved.ident, branchRequest(evolved).withFilter(filter));
+ }
+
+ private static PlanTableScanRequest.Builder branchRequest(EvolvedTable
evolved) {
+ return
PlanTableScanRequest.builder().withSnapshotId(evolved.s1).withUseSnapshotSchema(false);
+ }
+
+ private static void assertInvalidScan(Executable plan, String field) {
+ IllegalArgumentException e =
Assertions.assertThrows(IllegalArgumentException.class, plan);
+ Assertions.assertTrue(e.getMessage().contains(field), e.getMessage());
+ }
+
+ /**
+ * Creates a table whose snapshot s1 holds f1 with l=10 and a=5. Main then
moves on to f2, and the
+ * schema evolves: l is renamed to l2, a new column reuses the name l, and a
is renamed to a2.
+ */
+ private EvolvedTable evolvedTable(String name) {
+ TableIdentifier ident = TableIdentifier.of(Namespace.of("db"), name);
+ Table evolved =
+ wrapper
+ .getCatalog()
+ .createTable(
+ ident,
+ new Schema(
+ Types.NestedField.required(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "l", Types.LongType.get()),
+ Types.NestedField.optional(3, "a", Types.LongType.get())),
+ PartitionSpec.unpartitioned());
+ evolved
+ .newFastAppend()
+ .appendFile(file("/data/f1.parquet", ImmutableMap.of(2, 10L, 3, 5L)))
+ .commit();
+ evolved.refresh();
+ long s1 = evolved.currentSnapshot().snapshotId();
+ evolved.newFastAppend().appendFile(file("/data/f2.parquet",
ImmutableMap.of(2, 20L))).commit();
+ // Rename first so the next update can reuse the freed name "l".
+ evolved.updateSchema().renameColumn("l", "l2").renameColumn("a",
"a2").commit();
+ evolved.updateSchema().addColumn("l", Types.LongType.get()).commit();
+ evolved.refresh();
+ return new EvolvedTable(ident, evolved, s1);
+ }
+
+ private static DataFile file(String path, Map<Integer, Long> values) {
+ Map<Integer, ByteBuffer> bounds = new HashMap<>();
+ Map<Integer, Long> valueCounts = new HashMap<>();
+ Map<Integer, Long> nullCounts = new HashMap<>();
+ values.forEach(
+ (fieldId, value) -> {
+ bounds.put(fieldId, Conversions.toByteBuffer(Types.LongType.get(),
value));
+ valueCounts.put(fieldId, 1L);
+ nullCounts.put(fieldId, 0L);
+ });
+ return DataFiles.builder(PartitionSpec.unpartitioned())
+ .withPath(path)
+ .withFileSizeInBytes(10)
+ .withRecordCount(1)
+ .withMetrics(new Metrics(1L, null, valueCounts, nullCounts, null,
bounds, bounds))
+ .build();
+ }
+
+ private static final class EvolvedTable {
+ private final TableIdentifier ident;
+ private final Table table;
+ private final long s1;
+
+ private EvolvedTable(TableIdentifier ident, Table table, long s1) {
+ this.ident = ident;
+ this.table = table;
+ this.s1 = s1;
+ }
+ }
}
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestScanPlanCache.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestScanPlanCache.java
index 8fb5afec21..39e028d570 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestScanPlanCache.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestScanPlanCache.java
@@ -28,6 +28,7 @@ import java.util.Optional;
import org.apache.gravitino.iceberg.service.cache.LocalScanPlanCache;
import org.apache.gravitino.iceberg.service.cache.ScanPlanCache;
import org.apache.gravitino.iceberg.service.cache.ScanPlanCacheKey;
+import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
import org.apache.iceberg.catalog.Namespace;
@@ -37,6 +38,7 @@ import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.rest.PlanStatus;
import org.apache.iceberg.rest.requests.PlanTableScanRequest;
import org.apache.iceberg.rest.responses.PlanTableScanResponse;
+import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
@@ -61,6 +63,8 @@ public class TestScanPlanCache {
mockTable = mock(Table.class);
when(mockTable.name()).thenReturn("test_table");
when(mockTable.currentSnapshot()).thenReturn(mockSnapshot);
+ when(mockTable.schema())
+ .thenReturn(new Schema(0, Types.NestedField.optional(1, "id",
Types.LongType.get())));
}
@AfterEach
@@ -263,6 +267,37 @@ public class TestScanPlanCache {
cached1.get(), cached2.get(), "Different requests should have
different cache entries");
}
+ @Test
+ public void testCacheMissOnSchemaIdChange() {
+ PlanTableScanRequest scanRequest =
PlanTableScanRequest.builder().withSnapshotId(1L).build();
+ PlanTableScanResponse response =
+ PlanTableScanResponse.builder()
+ .withPlanStatus(PlanStatus.COMPLETED)
+ .withFileScanTasks(Collections.emptyList())
+ .build();
+
+ scanPlanCache.put(ScanPlanCacheKey.create(tableIdentifier, mockTable,
scanRequest), response);
+ Assertions.assertTrue(
+ scanPlanCache
+ .get(ScanPlanCacheKey.create(tableIdentifier, mockTable,
scanRequest))
+ .isPresent());
+
+ Table evolvedTable = mock(Table.class);
+ when(evolvedTable.currentSnapshot()).thenReturn(mockSnapshot);
+ when(evolvedTable.schema())
+ .thenReturn(
+ new Schema(
+ 1,
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "a", Types.LongType.get())));
+
+ Assertions.assertFalse(
+ scanPlanCache
+ .get(ScanPlanCacheKey.create(tableIdentifier, evolvedTable,
scanRequest))
+ .isPresent(),
+ "Schema-only evolution must miss the scan-plan cache");
+ }
+
@Test
public void testExpressionIsEquivalentTo() {
