This is an automated email from the ASF dual-hosted git repository. github-actions[bot] pushed a commit to branch cherry-pick-c07c7004-to-branch-1.3 in repository https://gitbox.apache.org/repos/asf/gravitino.git
commit 61472ac6432a3c51574ab97d04c30cbf8327a96e Author: MaSai <[email protected]> AuthorDate: Thu Nov 5 10:03:10 2026 +0800 [#14536] fix(iceberg-rest): honor use-snapshot-schema and plan dropped partition sources (#14552) ### What changes were proposed in this pull request? - When `use-snapshot-schema` is false (REST default) and a `snapshot-id` is set, rewrite filter/select/stats field names from the current table schema to the snapshot schema by Iceberg field id before `useSnapshot`, so branch scans bind against current names. - Sanitize `specs-by-id` used in plan responses when an older partition spec's source column was dropped (`partitionType` contains `unknown`), restoring types from `PartitionData` so file-scan-task serialization succeeds. ### Why are the changes needed? Fixes #14536. Iceberg's Java REST client sends `use-snapshot-schema: false` for branch scans; ignoring it caused 500 `Cannot find field` after renames. Separately, plan serialization threw `Type: unknown is not supported` as text/plain 400 for valid tables after partition evolution + drop. ### Does this PR introduce _any_ user-facing change? Yes. `POST .../plan` with `use-snapshot-schema: false` now binds filter/select to the current schema as specified. Tables whose historical specs reference dropped columns can be planned successfully instead of failing with a bare text/plain 400. ### How was this patch tested? - `./gradlew :iceberg:iceberg-rest-server:test --tests 'org.apache.gravitino.iceberg.service.TestScanPlanSupport' -PskipITs -PskipWeb=true` --------- Co-authored-by: Cursor <[email protected]> # Conflicts: # iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java --- .../iceberg/service/CatalogWrapperForREST.java | 102 ++- .../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, 1883 insertions(+), 12 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..f680e639d9 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 { @@ -483,7 +486,69 @@ public class CatalogWrapperForREST extends IcebergCatalogWrapper { scanPlanCache.get(ScanPlanCacheKey.create(tableIdentifier, table, scanRequest)); if (cachedResponse.isPresent()) { LOG.info("Using cached scan plan for table: {}", tableIdentifier); +<<<<<<< HEAD return cachedResponse.get(); +======= + response = cachedResponse.get(); + } else { + 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); + } + } catch (IOException e) { + 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( + "Scan planning returned no tasks for table: {}. Table may be empty or fully filtered.", + tableIdentifier); + } + + checkDeleteFilesCanBeReferenced(tableIdentifier, fileScanTasks); + try { + response = buildCompletedPlanTableScanResponse(table, fileScanTasks); + } catch (Exception e) { + LOG.error("Failed to build scan plan response for table: {}", tableIdentifier, e); + throw new RuntimeException( + String.format( + "Failed to build scan plan response for table: %s. Error: %s", + tableIdentifier, e.getMessage()), + e); + } + + scanPlanCache.put(ScanPlanCacheKey.create(tableIdentifier, table, scanRequest), response); +>>>>>>> c07c70047 ([#14536] fix(iceberg-rest): honor use-snapshot-schema and plan dropped partition sources (#14552)) } List<FileScanTask> fileScanTasks = new ArrayList<>(); @@ -555,7 +620,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 +632,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 +669,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 +710,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() {
