github-actions[bot] commented on code in PR #69681: URL: https://github.com/apache/doris/pull/69681#discussion_r4386805378
########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataDelete.java: ########## @@ -0,0 +1,355 @@ +// 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.doris.datasource.iceberg; + +import org.apache.doris.common.UserException; +import org.apache.doris.datasource.iceberg.IcebergExternalMetaCache.WritableTableLease; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; +import org.apache.iceberg.BaseTable; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.DeleteFiles; +import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.HasTableOperations; +import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.ManifestFiles; +import org.apache.iceberg.ManifestReader; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.SnapshotSummary; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableMetadata; +import org.apache.iceberg.TableOperations; +import org.apache.iceberg.TableScan; +import org.apache.iceberg.Transaction; +import org.apache.iceberg.encryption.EncryptionManager; +import org.apache.iceberg.exceptions.ValidationException; +import org.apache.iceberg.expressions.Evaluator; +import org.apache.iceberg.expressions.Expression; +import org.apache.iceberg.expressions.ExpressionUtil; +import org.apache.iceberg.expressions.Projections; +import org.apache.iceberg.expressions.StrictMetricsEvaluator; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.io.LocationProvider; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; +import java.util.OptionalLong; +import java.util.concurrent.ExecutorService; +import java.util.function.BooleanSupplier; + +/** + * Whole-file metadata delete for Iceberg tables. + * + * <p>The branch-4.2 path is intentionally implemented over the existing Iceberg metadata, + * transaction, and cache-generation machinery. Candidate files must all be proven to match in + * full; any partial match, existing delete file, cancellation, or concurrent snapshot change falls + * back to the unchanged row-level DELETE path. + */ +public final class IcebergMetadataDelete { + private static final Logger LOG = LogManager.getLogger(IcebergMetadataDelete.class); + + private IcebergMetadataDelete() { + } + + /** + * Attempts the fast path against a statement-bound table generation. + * + * @return the number of rows removed when committed, otherwise empty to run row-level DELETE + */ + public static OptionalLong execute(IcebergExternalTable dorisTable, Table targetTable, + Expression deleteExpression, BooleanSupplier isCancelled) throws UserException { + IcebergMetadataOps expectedOps = (IcebergMetadataOps) dorisTable.getCatalog().getMetadataOps(); + try (WritableTableLease lease = IcebergUtils.acquireWritableIcebergTable(dorisTable, expectedOps)) { + Table writableTable = lease.getTable(); + if (IcebergSnapshotCacheValue.isRetainedGeneration(targetTable)) { + writableTable = IcebergSnapshotCacheValue.createWritableTable(targetTable, writableTable); + } + Table table = writableTable; + return lease.getAuthenticator().execute(() -> executeOnTable( + table, targetTable, lease.getOps().getThreadPoolWithPreAuth(), + deleteExpression, isCancelled)); + } catch (ValidationException e) { + LOG.info("Metadata delete of {} is no longer whole-file ({}); falling back.", + dorisTable.getName(), e.getMessage()); + return OptionalLong.empty(); + } catch (UserException e) { + throw e; + } catch (Exception e) { + throw new UserException("Failed to metadata-delete Iceberg table " + + dorisTable.getName() + ": " + e.getMessage(), e); + } + } + + /** + * Core operation over a loaded writable table. This remains separate from catalog acquisition so + * it can be exercised against an InMemoryCatalog in unit tests. + */ + @VisibleForTesting + static OptionalLong executeOnTable(Table table, Table statementTable, ExecutorService executor, + Expression deleteExpression, BooleanSupplier isCancelled) throws UserException { + if (isCancelled.getAsBoolean()) { + return OptionalLong.empty(); + } + if (!(table instanceof BaseTable) + || ((BaseTable) table).operations().current().formatVersion() < 2) { + return OptionalLong.empty(); + } + validateTableIdentity(statementTable, table); + + Snapshot base = table.currentSnapshot(); + if (base == null || hasLiveDeleteFiles(table, base)) { + return OptionalLong.empty(); + } + + if (!matchesWholeFilesOnly(table, deleteExpression) || isCancelled.getAsBoolean()) { + return OptionalLong.empty(); + } + + // Re-read immediately before commit. A racing writer can add a candidate file or a delete + // file; the required snapshot check below then aborts instead of changing the deleted set. + table.refresh(); + validateTableIdentity(statementTable, table); + Snapshot refreshed = table.currentSnapshot(); + if (refreshed == null || hasLiveDeleteFiles(table, refreshed)) { + return OptionalLong.empty(); + } + long requiredSnapshotId = refreshed.snapshotId(); Review Comment: [P1] Fence the snapshot used by the statement and preflight `table` starts from the retained statement snapshot S0, but `refresh()` can adopt S1 and this line fences S1 without comparing it with `base.snapshotId()`. For a v2 table with `delete_isolation_level=snapshot`, let S0 contain a 10-row `region='us'` file and let another writer append a 5-row matching file before this refresh. Identity/no-delete-file checks still pass; the transaction's row filter now deletes both files and reports 15 rows. The existing scan/position-delete path stays on S0 and preserves the appended file under snapshot isolation. The later CAS fence only detects changes after S1. Keep S0 as the required snapshot and fall back whenever refresh sees a different snapshot. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataDeletePredicate.java: ########## @@ -0,0 +1,373 @@ +// 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.doris.datasource.iceberg; + +import org.apache.doris.catalog.Column; +import org.apache.doris.catalog.TableIf; +import org.apache.doris.common.UserException; +import org.apache.doris.nereids.trees.expressions.And; +import org.apache.doris.nereids.trees.expressions.Between; +import org.apache.doris.nereids.trees.expressions.EqualTo; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.GreaterThan; +import org.apache.doris.nereids.trees.expressions.GreaterThanEqual; +import org.apache.doris.nereids.trees.expressions.InPredicate; +import org.apache.doris.nereids.trees.expressions.IsNull; +import org.apache.doris.nereids.trees.expressions.LessThan; +import org.apache.doris.nereids.trees.expressions.LessThanEqual; +import org.apache.doris.nereids.trees.expressions.Not; +import org.apache.doris.nereids.trees.expressions.Or; +import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.trees.expressions.SlotReference; +import org.apache.doris.nereids.trees.expressions.literal.Literal; +import org.apache.doris.nereids.trees.expressions.literal.NullLiteral; +import org.apache.doris.nereids.trees.plans.Plan; +import org.apache.doris.nereids.trees.plans.logical.LogicalCatalogRelation; +import org.apache.doris.nereids.trees.plans.logical.LogicalCheckPolicy; +import org.apache.doris.nereids.trees.plans.logical.LogicalFilter; +import org.apache.doris.nereids.trees.plans.logical.LogicalProject; +import org.apache.doris.nereids.trees.plans.logical.LogicalSink; +import org.apache.doris.nereids.trees.plans.logical.LogicalSubQueryAlias; + +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types.NestedField; + +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.Set; + +/** + * Builds the complete statement predicate for an Iceberg whole-file metadata delete. + * + * <p>Unlike the conflict-detection helper, this path may be destructive: every conjunct must be + * representable exactly. A missing conjunct would widen the delete and can drop a file that still + * contains live rows, so any uncertainty rejects the whole fast path. + */ +public final class IcebergMetadataDeletePredicate { + private IcebergMetadataDeletePredicate() { + } + + /** + * Returns the exact Iceberg predicate for {@code plan}, or empty when the plan is not provably a + * single-table DELETE whose complete predicate is representable. + */ + public static Optional<org.apache.iceberg.expressions.Expression> build( + Table icebergTable, Plan plan, long targetTableId) { + List<Expression> conjuncts = new ArrayList<>(); + if (!collectCompletePredicate(plan, targetTableId, conjuncts)) { + return Optional.empty(); + } + if (conjuncts.isEmpty()) { + return Optional.of(Expressions.alwaysTrue()); + } + + Schema schema = icebergTable.schema(); + Set<Integer> unsafeNanosFields = IcebergTimestampUtils.unsafeNanosFields(icebergTable, schema); + org.apache.iceberg.expressions.Expression result = null; + for (Expression conjunct : conjuncts) { + if (IcebergTimestampUtils.referencesUnsafe(conjunct, schema, unsafeNanosFields)) { + return Optional.empty(); + } + Optional<org.apache.iceberg.expressions.Expression> converted = + convertExact(conjunct, schema); + if (!converted.isPresent()) { + return Optional.empty(); + } + result = result == null ? converted.get() : Expressions.and(result, converted.get()); + } + return Optional.of(result); + } + + /** + * Walks the single-input analyzed plan and returns false as soon as a node can change which + * target rows are produced without exposing that restriction as a LogicalFilter. + */ + private static boolean collectCompletePredicate( + Plan plan, long targetTableId, List<Expression> output) { + if (plan instanceof LogicalFilter) { + for (Expression conjunct : ((LogicalFilter<?>) plan).getConjuncts()) { + if (!isExactPhysicalTargetPredicate(conjunct, targetTableId)) { + return false; + } + output.add(conjunct); + } + } else if (!(plan instanceof LogicalSink) && !(plan instanceof LogicalProject) + && !(plan instanceof LogicalCheckPolicy) && !(plan instanceof LogicalSubQueryAlias) + && !(plan instanceof LogicalCatalogRelation)) { + return false; + } + + List<Plan> children = plan.children(); + if (plan instanceof LogicalCatalogRelation) { + return children.isEmpty() && ((LogicalCatalogRelation) plan).getTable().getId() == targetTableId; + } + if (children.size() != 1) { + return false; + } + return collectCompletePredicate(children.get(0), targetTableId, output); + } + + /** + * A conjunct must target only the DELETE table, use physical column names, and contain no + * synthetic Iceberg metadata columns. + */ + private static boolean isExactPhysicalTargetPredicate(Expression predicate, long targetTableId) { + if (predicate == null) { + return false; + } + Set<Slot> slots = predicate.getInputSlots(); + if (slots.isEmpty()) { + return false; + } + for (Slot slot : slots) { + if (!(slot instanceof SlotReference)) { + return false; + } + SlotReference reference = (SlotReference) slot; + Optional<TableIf> originalTable = reference.getOriginalTable(); + if (!originalTable.isPresent() || originalTable.get().getId() != targetTableId) { + return false; + } + Optional<Column> originalColumn = reference.getOriginalColumn(); + if (!originalColumn.isPresent() || !reference.getName().equals(originalColumn.get().getName())) { + return false; + } + if (Column.ICEBERG_ROWID_COL.equalsIgnoreCase(reference.getName()) + || IcebergMetadataColumn.isMetadataColumn(reference.getName()) + || !reference.getSubPath().isEmpty()) { + return false; + } + } + return true; + } + + private static Optional<org.apache.iceberg.expressions.Expression> convertExact( + Expression expression, Schema schema) { + if (expression instanceof And) { + org.apache.iceberg.expressions.Expression combined = null; + for (Expression child : expression.children()) { + Optional<org.apache.iceberg.expressions.Expression> converted = + convertExact(child, schema); + if (!converted.isPresent()) { + return Optional.empty(); + } + combined = combined == null ? converted.get() + : Expressions.and(combined, converted.get()); + } + return combined == null ? Optional.empty() : Optional.of(combined); + } + if (expression instanceof Or) { + org.apache.iceberg.expressions.Expression combined = null; + for (Expression child : expression.children()) { + Optional<org.apache.iceberg.expressions.Expression> converted = + convertExact(child, schema); + if (!converted.isPresent()) { + return Optional.empty(); + } + combined = combined == null ? converted.get() + : Expressions.or(combined, converted.get()); + } + return combined == null ? Optional.empty() : Optional.of(combined); + } + if (expression instanceof Not) { + Expression child = ((Not) expression).child(); + // Iceberg's notEqual/notIn match NULL while SQL NOT(UNKNOWN) does not. IS NOT NULL is + // the only negation whose SQL semantics are preserved by this conversion. + if (!(child instanceof IsNull)) { + return Optional.empty(); + } + Optional<org.apache.iceberg.expressions.Expression> converted = + convertIsNull((IsNull) child, schema); + return converted.map(Expressions::not); + } + if (expression instanceof EqualTo + || expression instanceof GreaterThan + || expression instanceof GreaterThanEqual + || expression instanceof LessThan + || expression instanceof LessThanEqual) { + return convertComparison(expression, schema); + } + if (expression instanceof InPredicate) { + return convertIn((InPredicate) expression, schema); + } + if (expression instanceof IsNull) { + return convertIsNull((IsNull) expression, schema); + } + if (expression instanceof Between) { + return convertBetween((Between) expression, schema); + } + return Optional.empty(); + } + + private static Optional<org.apache.iceberg.expressions.Expression> convertComparison( + Expression expression, Schema schema) { + Optional<NestedField> field = resolveSingleField(expression, schema); + if (!field.isPresent() || !isSupportedType(field.get().type())) { + return Optional.empty(); + } + + Expression left = expression.child(0); + Expression right = expression.child(1); + Literal literal; + boolean literalOnLeft; + if (left instanceof Slot && right instanceof Literal) { + literal = (Literal) right; + literalOnLeft = false; + } else if (left instanceof Literal && right instanceof Slot) { + literal = (Literal) left; + literalOnLeft = true; + } else { + return Optional.empty(); + } + if (literal instanceof NullLiteral) { + // SQL `col = NULL` does not match rows; do not silently lower it to IS NULL. + return Optional.empty(); + } + + Object value; + try { + value = IcebergNereidsUtils.extractNereidsLiteralValue(literal, field.get().type()); Review Comment: [P1] Preserve fractional milliseconds in timestamp delete literals For `Filter(ts = '2023-01-02 03:04:05.123456') -> Scan(ts)` on an Iceberg timestamp_ntz column, analysis produces a bare slot and scale-6 `DateTimeV2Literal`, so this conversion is admitted. `extractNereidsLiteralValue` gets a `LocalDateTime` that already contains the microseconds, then computes `toEpochMilli() * 1000 + microSecond`: `.123456` becomes `.246456`. With identity partitions at those two timestamps and no delete files, the fast path deletes the `.246456` partition and leaves the requested one. The row scanner's `IcebergUtils.extractDorisLiteral` does not double-count. Use an exact epoch-microsecond conversion for comparison, IN and BETWEEN before permitting metadata deletion. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataDeletePredicate.java: ########## @@ -0,0 +1,373 @@ +// 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.doris.datasource.iceberg; + +import org.apache.doris.catalog.Column; +import org.apache.doris.catalog.TableIf; +import org.apache.doris.common.UserException; +import org.apache.doris.nereids.trees.expressions.And; +import org.apache.doris.nereids.trees.expressions.Between; +import org.apache.doris.nereids.trees.expressions.EqualTo; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.GreaterThan; +import org.apache.doris.nereids.trees.expressions.GreaterThanEqual; +import org.apache.doris.nereids.trees.expressions.InPredicate; +import org.apache.doris.nereids.trees.expressions.IsNull; +import org.apache.doris.nereids.trees.expressions.LessThan; +import org.apache.doris.nereids.trees.expressions.LessThanEqual; +import org.apache.doris.nereids.trees.expressions.Not; +import org.apache.doris.nereids.trees.expressions.Or; +import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.trees.expressions.SlotReference; +import org.apache.doris.nereids.trees.expressions.literal.Literal; +import org.apache.doris.nereids.trees.expressions.literal.NullLiteral; +import org.apache.doris.nereids.trees.plans.Plan; +import org.apache.doris.nereids.trees.plans.logical.LogicalCatalogRelation; +import org.apache.doris.nereids.trees.plans.logical.LogicalCheckPolicy; +import org.apache.doris.nereids.trees.plans.logical.LogicalFilter; +import org.apache.doris.nereids.trees.plans.logical.LogicalProject; +import org.apache.doris.nereids.trees.plans.logical.LogicalSink; +import org.apache.doris.nereids.trees.plans.logical.LogicalSubQueryAlias; + +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types.NestedField; + +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.Set; + +/** + * Builds the complete statement predicate for an Iceberg whole-file metadata delete. + * + * <p>Unlike the conflict-detection helper, this path may be destructive: every conjunct must be + * representable exactly. A missing conjunct would widen the delete and can drop a file that still + * contains live rows, so any uncertainty rejects the whole fast path. + */ +public final class IcebergMetadataDeletePredicate { + private IcebergMetadataDeletePredicate() { + } + + /** + * Returns the exact Iceberg predicate for {@code plan}, or empty when the plan is not provably a + * single-table DELETE whose complete predicate is representable. + */ + public static Optional<org.apache.iceberg.expressions.Expression> build( + Table icebergTable, Plan plan, long targetTableId) { + List<Expression> conjuncts = new ArrayList<>(); + if (!collectCompletePredicate(plan, targetTableId, conjuncts)) { + return Optional.empty(); + } + if (conjuncts.isEmpty()) { + return Optional.of(Expressions.alwaysTrue()); + } + + Schema schema = icebergTable.schema(); + Set<Integer> unsafeNanosFields = IcebergTimestampUtils.unsafeNanosFields(icebergTable, schema); + org.apache.iceberg.expressions.Expression result = null; + for (Expression conjunct : conjuncts) { + if (IcebergTimestampUtils.referencesUnsafe(conjunct, schema, unsafeNanosFields)) { + return Optional.empty(); + } + Optional<org.apache.iceberg.expressions.Expression> converted = + convertExact(conjunct, schema); + if (!converted.isPresent()) { + return Optional.empty(); + } + result = result == null ? converted.get() : Expressions.and(result, converted.get()); + } + return Optional.of(result); + } + + /** + * Walks the single-input analyzed plan and returns false as soon as a node can change which + * target rows are produced without exposing that restriction as a LogicalFilter. + */ + private static boolean collectCompletePredicate( + Plan plan, long targetTableId, List<Expression> output) { + if (plan instanceof LogicalFilter) { + for (Expression conjunct : ((LogicalFilter<?>) plan).getConjuncts()) { + if (!isExactPhysicalTargetPredicate(conjunct, targetTableId)) { + return false; + } + output.add(conjunct); + } + } else if (!(plan instanceof LogicalSink) && !(plan instanceof LogicalProject) + && !(plan instanceof LogicalCheckPolicy) && !(plan instanceof LogicalSubQueryAlias) + && !(plan instanceof LogicalCatalogRelation)) { + return false; + } + + List<Plan> children = plan.children(); + if (plan instanceof LogicalCatalogRelation) { + return children.isEmpty() && ((LogicalCatalogRelation) plan).getTable().getId() == targetTableId; + } + if (children.size() != 1) { + return false; + } + return collectCompletePredicate(children.get(0), targetTableId, output); + } + + /** + * A conjunct must target only the DELETE table, use physical column names, and contain no + * synthetic Iceberg metadata columns. + */ + private static boolean isExactPhysicalTargetPredicate(Expression predicate, long targetTableId) { + if (predicate == null) { + return false; + } + Set<Slot> slots = predicate.getInputSlots(); + if (slots.isEmpty()) { + return false; + } + for (Slot slot : slots) { + if (!(slot instanceof SlotReference)) { + return false; + } + SlotReference reference = (SlotReference) slot; + Optional<TableIf> originalTable = reference.getOriginalTable(); + if (!originalTable.isPresent() || originalTable.get().getId() != targetTableId) { + return false; + } + Optional<Column> originalColumn = reference.getOriginalColumn(); + if (!originalColumn.isPresent() || !reference.getName().equals(originalColumn.get().getName())) { + return false; + } + if (Column.ICEBERG_ROWID_COL.equalsIgnoreCase(reference.getName()) + || IcebergMetadataColumn.isMetadataColumn(reference.getName()) + || !reference.getSubPath().isEmpty()) { + return false; + } + } + return true; + } + + private static Optional<org.apache.iceberg.expressions.Expression> convertExact( + Expression expression, Schema schema) { + if (expression instanceof And) { + org.apache.iceberg.expressions.Expression combined = null; + for (Expression child : expression.children()) { + Optional<org.apache.iceberg.expressions.Expression> converted = + convertExact(child, schema); + if (!converted.isPresent()) { + return Optional.empty(); + } + combined = combined == null ? converted.get() + : Expressions.and(combined, converted.get()); + } + return combined == null ? Optional.empty() : Optional.of(combined); + } + if (expression instanceof Or) { + org.apache.iceberg.expressions.Expression combined = null; + for (Expression child : expression.children()) { + Optional<org.apache.iceberg.expressions.Expression> converted = + convertExact(child, schema); + if (!converted.isPresent()) { + return Optional.empty(); + } + combined = combined == null ? converted.get() + : Expressions.or(combined, converted.get()); + } + return combined == null ? Optional.empty() : Optional.of(combined); + } + if (expression instanceof Not) { + Expression child = ((Not) expression).child(); + // Iceberg's notEqual/notIn match NULL while SQL NOT(UNKNOWN) does not. IS NOT NULL is + // the only negation whose SQL semantics are preserved by this conversion. + if (!(child instanceof IsNull)) { + return Optional.empty(); + } + Optional<org.apache.iceberg.expressions.Expression> converted = + convertIsNull((IsNull) child, schema); + return converted.map(Expressions::not); + } + if (expression instanceof EqualTo + || expression instanceof GreaterThan + || expression instanceof GreaterThanEqual + || expression instanceof LessThan + || expression instanceof LessThanEqual) { + return convertComparison(expression, schema); + } + if (expression instanceof InPredicate) { + return convertIn((InPredicate) expression, schema); + } + if (expression instanceof IsNull) { + return convertIsNull((IsNull) expression, schema); + } + if (expression instanceof Between) { + return convertBetween((Between) expression, schema); + } + return Optional.empty(); + } + + private static Optional<org.apache.iceberg.expressions.Expression> convertComparison( + Expression expression, Schema schema) { + Optional<NestedField> field = resolveSingleField(expression, schema); + if (!field.isPresent() || !isSupportedType(field.get().type())) { + return Optional.empty(); + } + + Expression left = expression.child(0); + Expression right = expression.child(1); + Literal literal; + boolean literalOnLeft; + if (left instanceof Slot && right instanceof Literal) { + literal = (Literal) right; + literalOnLeft = false; + } else if (left instanceof Literal && right instanceof Slot) { + literal = (Literal) left; + literalOnLeft = true; + } else { + return Optional.empty(); + } + if (literal instanceof NullLiteral) { + // SQL `col = NULL` does not match rows; do not silently lower it to IS NULL. + return Optional.empty(); + } + + Object value; + try { + value = IcebergNereidsUtils.extractNereidsLiteralValue(literal, field.get().type()); + } catch (UserException e) { + return Optional.empty(); + } + if (value == null) { + return Optional.empty(); + } + + String columnName = field.get().name(); + if (expression instanceof EqualTo) { + return Optional.of(Expressions.equal(columnName, value)); + } + if (expression instanceof GreaterThan) { + return Optional.of(literalOnLeft + ? Expressions.lessThan(columnName, value) + : Expressions.greaterThan(columnName, value)); + } + if (expression instanceof GreaterThanEqual) { + return Optional.of(literalOnLeft + ? Expressions.lessThanOrEqual(columnName, value) + : Expressions.greaterThanOrEqual(columnName, value)); + } + if (expression instanceof LessThan) { + return Optional.of(literalOnLeft + ? Expressions.greaterThan(columnName, value) + : Expressions.lessThan(columnName, value)); Review Comment: [P1] Preserve SQL signed-zero ordering before dropping floating files `Filter(d < '0') -> Scan(d)` for a DOUBLE column binds to a bare slot and `DoubleLiteral(+0.0)` and reaches this branch. A file containing only `-0.0`, with complete metrics and zero nulls/NaNs, passes Iceberg's strict `d < +0.0` proof: its [floating comparators](https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/api/src/main/java/org/apache/iceberg/types/Comparators.java#L32-L38) distinguish signed zero, and [StrictMetricsEvaluator.lt](https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/api/src/main/java/org/apache/iceberg/expressions/StrictMetricsEvaluator.java#L185-L201) accepts the negative-zero upper bound. Doris `LessOp`/`Compare::less<double>` uses native comparison for non-NaNs, where `-0.0 < +0.0` is false. Metadata DELETE removes the entire file whereas row-level DELETE removes no rows. Reject floating value predicates until the zero-boundary semantics are translated exactly. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataDeletePredicate.java: ########## @@ -0,0 +1,373 @@ +// 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.doris.datasource.iceberg; + +import org.apache.doris.catalog.Column; +import org.apache.doris.catalog.TableIf; +import org.apache.doris.common.UserException; +import org.apache.doris.nereids.trees.expressions.And; +import org.apache.doris.nereids.trees.expressions.Between; +import org.apache.doris.nereids.trees.expressions.EqualTo; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.GreaterThan; +import org.apache.doris.nereids.trees.expressions.GreaterThanEqual; +import org.apache.doris.nereids.trees.expressions.InPredicate; +import org.apache.doris.nereids.trees.expressions.IsNull; +import org.apache.doris.nereids.trees.expressions.LessThan; +import org.apache.doris.nereids.trees.expressions.LessThanEqual; +import org.apache.doris.nereids.trees.expressions.Not; +import org.apache.doris.nereids.trees.expressions.Or; +import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.trees.expressions.SlotReference; +import org.apache.doris.nereids.trees.expressions.literal.Literal; +import org.apache.doris.nereids.trees.expressions.literal.NullLiteral; +import org.apache.doris.nereids.trees.plans.Plan; +import org.apache.doris.nereids.trees.plans.logical.LogicalCatalogRelation; +import org.apache.doris.nereids.trees.plans.logical.LogicalCheckPolicy; +import org.apache.doris.nereids.trees.plans.logical.LogicalFilter; +import org.apache.doris.nereids.trees.plans.logical.LogicalProject; +import org.apache.doris.nereids.trees.plans.logical.LogicalSink; +import org.apache.doris.nereids.trees.plans.logical.LogicalSubQueryAlias; + +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types.NestedField; + +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.Set; + +/** + * Builds the complete statement predicate for an Iceberg whole-file metadata delete. + * + * <p>Unlike the conflict-detection helper, this path may be destructive: every conjunct must be + * representable exactly. A missing conjunct would widen the delete and can drop a file that still + * contains live rows, so any uncertainty rejects the whole fast path. + */ +public final class IcebergMetadataDeletePredicate { + private IcebergMetadataDeletePredicate() { + } + + /** + * Returns the exact Iceberg predicate for {@code plan}, or empty when the plan is not provably a + * single-table DELETE whose complete predicate is representable. + */ + public static Optional<org.apache.iceberg.expressions.Expression> build( + Table icebergTable, Plan plan, long targetTableId) { + List<Expression> conjuncts = new ArrayList<>(); + if (!collectCompletePredicate(plan, targetTableId, conjuncts)) { + return Optional.empty(); + } + if (conjuncts.isEmpty()) { + return Optional.of(Expressions.alwaysTrue()); + } + + Schema schema = icebergTable.schema(); + Set<Integer> unsafeNanosFields = IcebergTimestampUtils.unsafeNanosFields(icebergTable, schema); + org.apache.iceberg.expressions.Expression result = null; + for (Expression conjunct : conjuncts) { + if (IcebergTimestampUtils.referencesUnsafe(conjunct, schema, unsafeNanosFields)) { + return Optional.empty(); + } + Optional<org.apache.iceberg.expressions.Expression> converted = + convertExact(conjunct, schema); + if (!converted.isPresent()) { + return Optional.empty(); + } + result = result == null ? converted.get() : Expressions.and(result, converted.get()); + } + return Optional.of(result); + } + + /** + * Walks the single-input analyzed plan and returns false as soon as a node can change which + * target rows are produced without exposing that restriction as a LogicalFilter. + */ + private static boolean collectCompletePredicate( + Plan plan, long targetTableId, List<Expression> output) { + if (plan instanceof LogicalFilter) { + for (Expression conjunct : ((LogicalFilter<?>) plan).getConjuncts()) { + if (!isExactPhysicalTargetPredicate(conjunct, targetTableId)) { + return false; + } + output.add(conjunct); + } + } else if (!(plan instanceof LogicalSink) && !(plan instanceof LogicalProject) + && !(plan instanceof LogicalCheckPolicy) && !(plan instanceof LogicalSubQueryAlias) + && !(plan instanceof LogicalCatalogRelation)) { + return false; + } + + List<Plan> children = plan.children(); + if (plan instanceof LogicalCatalogRelation) { + return children.isEmpty() && ((LogicalCatalogRelation) plan).getTable().getId() == targetTableId; + } + if (children.size() != 1) { + return false; + } + return collectCompletePredicate(children.get(0), targetTableId, output); + } + + /** + * A conjunct must target only the DELETE table, use physical column names, and contain no + * synthetic Iceberg metadata columns. + */ + private static boolean isExactPhysicalTargetPredicate(Expression predicate, long targetTableId) { + if (predicate == null) { + return false; + } + Set<Slot> slots = predicate.getInputSlots(); + if (slots.isEmpty()) { + return false; + } + for (Slot slot : slots) { + if (!(slot instanceof SlotReference)) { + return false; + } + SlotReference reference = (SlotReference) slot; + Optional<TableIf> originalTable = reference.getOriginalTable(); + if (!originalTable.isPresent() || originalTable.get().getId() != targetTableId) { + return false; + } + Optional<Column> originalColumn = reference.getOriginalColumn(); + if (!originalColumn.isPresent() || !reference.getName().equals(originalColumn.get().getName())) { + return false; + } + if (Column.ICEBERG_ROWID_COL.equalsIgnoreCase(reference.getName()) + || IcebergMetadataColumn.isMetadataColumn(reference.getName()) + || !reference.getSubPath().isEmpty()) { + return false; + } + } + return true; + } + + private static Optional<org.apache.iceberg.expressions.Expression> convertExact( + Expression expression, Schema schema) { + if (expression instanceof And) { + org.apache.iceberg.expressions.Expression combined = null; + for (Expression child : expression.children()) { + Optional<org.apache.iceberg.expressions.Expression> converted = + convertExact(child, schema); + if (!converted.isPresent()) { + return Optional.empty(); + } + combined = combined == null ? converted.get() + : Expressions.and(combined, converted.get()); + } + return combined == null ? Optional.empty() : Optional.of(combined); + } + if (expression instanceof Or) { + org.apache.iceberg.expressions.Expression combined = null; + for (Expression child : expression.children()) { + Optional<org.apache.iceberg.expressions.Expression> converted = + convertExact(child, schema); + if (!converted.isPresent()) { + return Optional.empty(); + } + combined = combined == null ? converted.get() + : Expressions.or(combined, converted.get()); + } + return combined == null ? Optional.empty() : Optional.of(combined); + } + if (expression instanceof Not) { + Expression child = ((Not) expression).child(); + // Iceberg's notEqual/notIn match NULL while SQL NOT(UNKNOWN) does not. IS NOT NULL is + // the only negation whose SQL semantics are preserved by this conversion. + if (!(child instanceof IsNull)) { + return Optional.empty(); + } + Optional<org.apache.iceberg.expressions.Expression> converted = + convertIsNull((IsNull) child, schema); + return converted.map(Expressions::not); + } + if (expression instanceof EqualTo + || expression instanceof GreaterThan + || expression instanceof GreaterThanEqual + || expression instanceof LessThan + || expression instanceof LessThanEqual) { + return convertComparison(expression, schema); + } + if (expression instanceof InPredicate) { + return convertIn((InPredicate) expression, schema); + } + if (expression instanceof IsNull) { + return convertIsNull((IsNull) expression, schema); + } + if (expression instanceof Between) { + return convertBetween((Between) expression, schema); + } + return Optional.empty(); + } + + private static Optional<org.apache.iceberg.expressions.Expression> convertComparison( + Expression expression, Schema schema) { + Optional<NestedField> field = resolveSingleField(expression, schema); + if (!field.isPresent() || !isSupportedType(field.get().type())) { + return Optional.empty(); + } + + Expression left = expression.child(0); + Expression right = expression.child(1); + Literal literal; + boolean literalOnLeft; + if (left instanceof Slot && right instanceof Literal) { + literal = (Literal) right; + literalOnLeft = false; + } else if (left instanceof Literal && right instanceof Slot) { + literal = (Literal) left; + literalOnLeft = true; + } else { + return Optional.empty(); + } + if (literal instanceof NullLiteral) { + // SQL `col = NULL` does not match rows; do not silently lower it to IS NULL. + return Optional.empty(); + } + + Object value; + try { + value = IcebergNereidsUtils.extractNereidsLiteralValue(literal, field.get().type()); + } catch (UserException e) { + return Optional.empty(); + } + if (value == null) { + return Optional.empty(); + } + + String columnName = field.get().name(); + if (expression instanceof EqualTo) { + return Optional.of(Expressions.equal(columnName, value)); + } + if (expression instanceof GreaterThan) { + return Optional.of(literalOnLeft + ? Expressions.lessThan(columnName, value) + : Expressions.greaterThan(columnName, value)); + } + if (expression instanceof GreaterThanEqual) { + return Optional.of(literalOnLeft + ? Expressions.lessThanOrEqual(columnName, value) + : Expressions.greaterThanOrEqual(columnName, value)); + } + if (expression instanceof LessThan) { + return Optional.of(literalOnLeft + ? Expressions.greaterThan(columnName, value) + : Expressions.lessThan(columnName, value)); + } + if (expression instanceof LessThanEqual) { + return Optional.of(literalOnLeft + ? Expressions.greaterThanOrEqual(columnName, value) + : Expressions.lessThanOrEqual(columnName, value)); + } + return Optional.empty(); + } + + private static Optional<org.apache.iceberg.expressions.Expression> convertIn( + InPredicate predicate, Schema schema) { + Optional<NestedField> field = resolveSingleField(predicate, schema); + if (!field.isPresent() || !isSupportedType(field.get().type())) { + return Optional.empty(); + } + if (!(predicate.child(0) instanceof Slot) + || predicate.optionsContainsNullLiteral()) { + return Optional.empty(); + } + + List<Object> values = new ArrayList<>(); + try { + for (int i = 1; i < predicate.children().size(); i++) { + Expression child = predicate.child(i); + if (!(child instanceof Literal)) { + return Optional.empty(); + } + Object value = IcebergNereidsUtils.extractNereidsLiteralValue( + (Literal) child, field.get().type()); + if (value == null) { + return Optional.empty(); + } + values.add(value); + } + } catch (UserException e) { + return Optional.empty(); + } + return values.isEmpty() ? Optional.empty() + : Optional.of(Expressions.in(field.get().name(), values)); + } + + private static Optional<org.apache.iceberg.expressions.Expression> convertIsNull( + IsNull predicate, Schema schema) { + Optional<NestedField> field = resolveSingleField(predicate, schema); + if (!field.isPresent() || !isSupportedType(field.get().type())) { + return Optional.empty(); + } + return Optional.of(Expressions.isNull(field.get().name())); Review Comment: [P1] Require a bare column for IS NULL conversion The captured analyzed plan can be: ```text DeleteSink Project(operation, rowid) Filter(coalesce(region, 'us') IS NULL) Scan(region) ``` `resolveSingleField` checks input slots, so it accepts this computed child; this line then replaces the predicate with `region IS NULL`. On a v2 table identity-partitioned by nullable `region`, with no delete files, the NULL partition passes the whole-file proof and is dropped even though the SQL predicate matches no rows. Capture occurs before rewrite simplification, so a later empty physical plan does not protect it. Require `predicate.child()` to be a physical Slot, including the IS NOT NULL caller, or fall back. ########## regression-test/suites/external_table_p2/maxcompute/test_max_compute_partition_prune.groovy: ########## @@ -105,191 +101,153 @@ suite("test_max_compute_partition_prune", "p2,external,maxcompute,external_remot String mc_catalog_name = "test_max_compute_partition_prune" - for (String enable_profile : ["true","false"] ) { - sql """set enable_profile = ${enable_profile} """; - sql """set fetch_splits_max_wait_time_ms = 10000 """; - - for (String num_partitions : ["1","10","100"] ) { - sql "set num_partitions_in_batch_mode = ${num_partitions} " - - for (String cross_partition : ["true","false"] ) { - - sql """drop catalog if exists ${mc_catalog_name};""" - sql """ - create catalog if not exists ${mc_catalog_name} properties ( - "type" = "max_compute", - "mc.default.project" = "${mc_db}", - "mc.access_key" = "${ak}", - "mc.secret_key" = "${sk}", - "mc.endpoint" = "http://service.cn-beijing-vpc.maxcompute.aliyun-inc.com/api";, - "mc.split_cross_partition" = "${cross_partition}" - ); - """ - sql """ switch ${mc_catalog_name} """ - sql """ use ${mc_db}""" - - qt_one_partition_1_1 one_partition_1_1 - explain { - sql("${one_partition_1_1}") - contains "partition=1/2" - } - - qt_one_partition_2_1 one_partition_2_1 - explain { - sql("${one_partition_2_1}") - contains "partition=1/2" - } - - qt_one_partition_3_all one_partition_3_all - explain { - sql("${one_partition_3_all}") - contains "partition=2/2" - } - - qt_one_partition_4_all one_partition_4_all - explain { - sql("${one_partition_4_all}") - contains "partition=2/2" - } - - qt_one_partition_5_1 one_partition_5_1 - explain { - sql("${one_partition_5_1}") - contains "partition=1/2" - } - - - qt_two_partition_1_1 two_partition_1_1 - explain { - sql("${two_partition_1_1}") - contains "partition=1/4" - } - - qt_two_partition_2_1 two_partition_2_1 - explain { - sql("${two_partition_2_1}") - contains "partition=1/4" - } - - qt_two_partition_3_2 two_partition_3_2 - explain { - sql("${two_partition_3_2}") - contains "partition=2/4" - } - - qt_two_partition_4_all two_partition_4_all - explain { - sql("${two_partition_4_all}") - contains "partition=4/4" - } - - qt_two_partition_5_1 two_partition_5_1 - explain { - sql("${two_partition_5_1}") - contains "partition=1/4" - } - - qt_two_partition_6_1 two_partition_6_1 - explain { - sql("${two_partition_6_1}") - contains "partition=1/4" - } - - - - qt_three_partition_1_1 three_partition_1_1 - explain { - sql("${three_partition_1_1}") - contains "partition=1/10" - } - - qt_three_partition_2_1 three_partition_2_1 - explain { - sql("${three_partition_2_1}") - contains "partition=1/10" - } - - qt_three_partition_3_3 three_partition_3_3 - explain { - sql("${three_partition_3_3}") - contains "partition=3/10" - } - - qt_three_partition_4_2 three_partition_4_2 - explain { - sql("${three_partition_4_2}") - contains "partition=2/10" - } - - qt_three_partition_5_all three_partition_5_all - explain { - sql("${three_partition_5_all}") - contains "partition=10/10" - } - - qt_three_partition_6_1 three_partition_6_1 - explain { - sql("${three_partition_6_1}") - contains "partition=1/10" - } - - qt_three_partition_7_7 three_partition_7_7 - explain { - sql("${three_partition_7_7}") - contains "partition=7/10" - } - - qt_three_partition_8_2 three_partition_8_2 - explain { - sql("${three_partition_8_2}") - contains "partition=2/10" - } - - - // 0 partitions - def one_partition_6_0 = """SELECT * FROM one_partition_tb WHERE part1 = 2023 ORDER BY id;""" - qt_one_partition_6_0 one_partition_6_0 - explain { - sql("${one_partition_6_0}") - contains "partition=0/2" - } - - def two_partition_7_0 = """SELECT * FROM two_partition_tb WHERE part1 = 'CN' AND part2 = 1 ORDER BY id;""" - qt_two_partition_7_0 two_partition_7_0 - explain { - sql("${two_partition_7_0}") - contains "partition=0/4" - } - - def two_partition_8_0 = """SELECT * FROM two_partition_tb WHERE part1 = 'US' AND part2 = 3 ORDER BY id;""" - qt_two_partition_8_0 two_partition_8_0 - explain { - sql("${two_partition_8_0}") - contains "partition=0/4" - } - - def three_partition_9_0 = """SELECT * FROM three_partition_tb WHERE part1 = 'US' AND part2 = 2023 AND part3 = 'Q1' ORDER BY id;""" - qt_three_partition_9_0 three_partition_9_0 - explain { - sql("${three_partition_9_0}") - contains "partition=0/10" - } - - def three_partition_10_0 = """SELECT * FROM three_partition_tb WHERE part1 = 'EU' AND part2 = 2024 AND part3 = 'Q4' ORDER BY id;""" - qt_three_partition_10_0 three_partition_10_0 - explain { - sql("${three_partition_10_0}") - contains "partition=0/10" - } - - def three_partition_11_0 = """SELECT * FROM three_partition_tb WHERE part1 = 'AS' AND part2 = 2025 AND part3 = 'Q4' ORDER BY id;""" - qt_three_partition_11_0 three_partition_11_0 - explain { - sql("${three_partition_11_0}") - contains "partition=0/10" - } - } - } + sql """drop catalog if exists ${mc_catalog_name};""" + sql """ + create catalog if not exists ${mc_catalog_name} properties ( + "type" = "max_compute", + "mc.default.project" = "${mc_db}", + "mc.access_key" = "${ak}", + "mc.secret_key" = "${sk}", + "mc.endpoint" = "http://service.cn-beijing-vpc.maxcompute.aliyun-inc.com/api"; + ); + """ + sql """ switch ${mc_catalog_name} """ + sql """ use ${mc_db}""" + + qt_one_partition_1_1 one_partition_1_1 + explain { + sql("${one_partition_1_1}") + contains "partition=1/2" + // VPluginDrivenScanNode surfaces the backing connector/catalog type + contains "CONNECTOR: max_compute" Review Comment: [P2] Assert the EXPLAIN format implemented on branch-4.2 This branch's `MaxComputeScanNode` extends `FileQueryScanNode` and uses the inherited `FileScanNode.getNodeExplainString`, which never emits a `CONNECTOR:` line; there is no `VPluginDrivenScanNode` implementation here. Whenever MaxCompute tests are enabled, the first EXPLAIN therefore fails this assertion even with a correct plan, preventing all subsequent pruning checks and the non-partitioned-table truth gate from running. Remove this master-only assertion or check the actual `MCScanNode` output. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataDeletePredicate.java: ########## @@ -0,0 +1,373 @@ +// 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.doris.datasource.iceberg; + +import org.apache.doris.catalog.Column; +import org.apache.doris.catalog.TableIf; +import org.apache.doris.common.UserException; +import org.apache.doris.nereids.trees.expressions.And; +import org.apache.doris.nereids.trees.expressions.Between; +import org.apache.doris.nereids.trees.expressions.EqualTo; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.GreaterThan; +import org.apache.doris.nereids.trees.expressions.GreaterThanEqual; +import org.apache.doris.nereids.trees.expressions.InPredicate; +import org.apache.doris.nereids.trees.expressions.IsNull; +import org.apache.doris.nereids.trees.expressions.LessThan; +import org.apache.doris.nereids.trees.expressions.LessThanEqual; +import org.apache.doris.nereids.trees.expressions.Not; +import org.apache.doris.nereids.trees.expressions.Or; +import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.trees.expressions.SlotReference; +import org.apache.doris.nereids.trees.expressions.literal.Literal; +import org.apache.doris.nereids.trees.expressions.literal.NullLiteral; +import org.apache.doris.nereids.trees.plans.Plan; +import org.apache.doris.nereids.trees.plans.logical.LogicalCatalogRelation; +import org.apache.doris.nereids.trees.plans.logical.LogicalCheckPolicy; +import org.apache.doris.nereids.trees.plans.logical.LogicalFilter; +import org.apache.doris.nereids.trees.plans.logical.LogicalProject; +import org.apache.doris.nereids.trees.plans.logical.LogicalSink; +import org.apache.doris.nereids.trees.plans.logical.LogicalSubQueryAlias; + +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types.NestedField; + +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.Set; + +/** + * Builds the complete statement predicate for an Iceberg whole-file metadata delete. + * + * <p>Unlike the conflict-detection helper, this path may be destructive: every conjunct must be + * representable exactly. A missing conjunct would widen the delete and can drop a file that still + * contains live rows, so any uncertainty rejects the whole fast path. + */ +public final class IcebergMetadataDeletePredicate { + private IcebergMetadataDeletePredicate() { + } + + /** + * Returns the exact Iceberg predicate for {@code plan}, or empty when the plan is not provably a + * single-table DELETE whose complete predicate is representable. + */ + public static Optional<org.apache.iceberg.expressions.Expression> build( + Table icebergTable, Plan plan, long targetTableId) { + List<Expression> conjuncts = new ArrayList<>(); + if (!collectCompletePredicate(plan, targetTableId, conjuncts)) { + return Optional.empty(); + } + if (conjuncts.isEmpty()) { + return Optional.of(Expressions.alwaysTrue()); + } + + Schema schema = icebergTable.schema(); + Set<Integer> unsafeNanosFields = IcebergTimestampUtils.unsafeNanosFields(icebergTable, schema); + org.apache.iceberg.expressions.Expression result = null; + for (Expression conjunct : conjuncts) { + if (IcebergTimestampUtils.referencesUnsafe(conjunct, schema, unsafeNanosFields)) { + return Optional.empty(); + } + Optional<org.apache.iceberg.expressions.Expression> converted = + convertExact(conjunct, schema); + if (!converted.isPresent()) { + return Optional.empty(); + } + result = result == null ? converted.get() : Expressions.and(result, converted.get()); + } + return Optional.of(result); + } + + /** + * Walks the single-input analyzed plan and returns false as soon as a node can change which + * target rows are produced without exposing that restriction as a LogicalFilter. + */ + private static boolean collectCompletePredicate( + Plan plan, long targetTableId, List<Expression> output) { + if (plan instanceof LogicalFilter) { + for (Expression conjunct : ((LogicalFilter<?>) plan).getConjuncts()) { + if (!isExactPhysicalTargetPredicate(conjunct, targetTableId)) { + return false; + } + output.add(conjunct); + } + } else if (!(plan instanceof LogicalSink) && !(plan instanceof LogicalProject) + && !(plan instanceof LogicalCheckPolicy) && !(plan instanceof LogicalSubQueryAlias) + && !(plan instanceof LogicalCatalogRelation)) { + return false; + } + + List<Plan> children = plan.children(); + if (plan instanceof LogicalCatalogRelation) { + return children.isEmpty() && ((LogicalCatalogRelation) plan).getTable().getId() == targetTableId; + } + if (children.size() != 1) { + return false; + } + return collectCompletePredicate(children.get(0), targetTableId, output); + } + + /** + * A conjunct must target only the DELETE table, use physical column names, and contain no + * synthetic Iceberg metadata columns. + */ + private static boolean isExactPhysicalTargetPredicate(Expression predicate, long targetTableId) { + if (predicate == null) { + return false; + } + Set<Slot> slots = predicate.getInputSlots(); + if (slots.isEmpty()) { + return false; + } + for (Slot slot : slots) { + if (!(slot instanceof SlotReference)) { + return false; + } + SlotReference reference = (SlotReference) slot; + Optional<TableIf> originalTable = reference.getOriginalTable(); + if (!originalTable.isPresent() || originalTable.get().getId() != targetTableId) { + return false; + } + Optional<Column> originalColumn = reference.getOriginalColumn(); + if (!originalColumn.isPresent() || !reference.getName().equals(originalColumn.get().getName())) { + return false; + } + if (Column.ICEBERG_ROWID_COL.equalsIgnoreCase(reference.getName()) + || IcebergMetadataColumn.isMetadataColumn(reference.getName()) + || !reference.getSubPath().isEmpty()) { + return false; + } + } + return true; + } + + private static Optional<org.apache.iceberg.expressions.Expression> convertExact( + Expression expression, Schema schema) { + if (expression instanceof And) { + org.apache.iceberg.expressions.Expression combined = null; + for (Expression child : expression.children()) { + Optional<org.apache.iceberg.expressions.Expression> converted = + convertExact(child, schema); + if (!converted.isPresent()) { + return Optional.empty(); + } + combined = combined == null ? converted.get() + : Expressions.and(combined, converted.get()); + } + return combined == null ? Optional.empty() : Optional.of(combined); + } + if (expression instanceof Or) { + org.apache.iceberg.expressions.Expression combined = null; + for (Expression child : expression.children()) { + Optional<org.apache.iceberg.expressions.Expression> converted = + convertExact(child, schema); + if (!converted.isPresent()) { + return Optional.empty(); + } + combined = combined == null ? converted.get() + : Expressions.or(combined, converted.get()); + } + return combined == null ? Optional.empty() : Optional.of(combined); + } + if (expression instanceof Not) { + Expression child = ((Not) expression).child(); + // Iceberg's notEqual/notIn match NULL while SQL NOT(UNKNOWN) does not. IS NOT NULL is + // the only negation whose SQL semantics are preserved by this conversion. + if (!(child instanceof IsNull)) { + return Optional.empty(); + } + Optional<org.apache.iceberg.expressions.Expression> converted = + convertIsNull((IsNull) child, schema); + return converted.map(Expressions::not); + } + if (expression instanceof EqualTo + || expression instanceof GreaterThan + || expression instanceof GreaterThanEqual + || expression instanceof LessThan + || expression instanceof LessThanEqual) { + return convertComparison(expression, schema); + } + if (expression instanceof InPredicate) { + return convertIn((InPredicate) expression, schema); + } + if (expression instanceof IsNull) { + return convertIsNull((IsNull) expression, schema); + } + if (expression instanceof Between) { + return convertBetween((Between) expression, schema); + } + return Optional.empty(); + } + + private static Optional<org.apache.iceberg.expressions.Expression> convertComparison( + Expression expression, Schema schema) { + Optional<NestedField> field = resolveSingleField(expression, schema); + if (!field.isPresent() || !isSupportedType(field.get().type())) { + return Optional.empty(); + } + + Expression left = expression.child(0); + Expression right = expression.child(1); + Literal literal; + boolean literalOnLeft; + if (left instanceof Slot && right instanceof Literal) { + literal = (Literal) right; + literalOnLeft = false; + } else if (left instanceof Literal && right instanceof Slot) { + literal = (Literal) left; + literalOnLeft = true; + } else { + return Optional.empty(); + } + if (literal instanceof NullLiteral) { + // SQL `col = NULL` does not match rows; do not silently lower it to IS NULL. + return Optional.empty(); + } + + Object value; + try { + value = IcebergNereidsUtils.extractNereidsLiteralValue(literal, field.get().type()); + } catch (UserException e) { + return Optional.empty(); + } + if (value == null) { + return Optional.empty(); + } + + String columnName = field.get().name(); + if (expression instanceof EqualTo) { + return Optional.of(Expressions.equal(columnName, value)); Review Comment: [P1] Keep analyzed TIMESTAMPTZ literals in UTC With TIMESTAMPTZ mapping enabled and session time zone `+08:00`, `Filter(ts = '2023-01-02 08:00:00.000001') -> Scan(ts)` contains a `TimestampTzLiteral` whose fields are already `2023-01-02T00:00:00.000001Z` (`TypeCoercionUtils.characterLiteralTypeCoercion` calls `fromSessionTimeZone`). The reused extractor treats those fields as session-local again, producing `2023-01-01T16:00:00.000001Z`. On an identity-partitioned v2 table with both instants and no delete files, this expression therefore drops the wrong partition. The normal scanner explicitly converts already-normalized TIMESTAMPTZ fields with UTC. Preserve that distinction here and in the IN/BETWEEN callers; fixing fractional arithmetic alone does not fix this shift. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
