github-actions[bot] commented on code in PR #69671: URL: https://github.com/apache/doris/pull/69671#discussion_r4386053768
########## fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPartitionFilterEvaluator.java: ########## @@ -0,0 +1,225 @@ +// 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.connector.iceberg; + +import org.apache.doris.connector.spi.ConnectorType; +import org.apache.doris.connector.spi.pushdown.ConnectorColumnRef; +import org.apache.doris.connector.spi.pushdown.ConnectorComparison; +import org.apache.doris.connector.spi.pushdown.ConnectorExpression; +import org.apache.doris.connector.spi.pushdown.ConnectorIn; +import org.apache.doris.connector.spi.pushdown.ConnectorIsNull; +import org.apache.doris.connector.spi.pushdown.ConnectorLiteral; + +import org.apache.iceberg.PartitionData; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.PartitionSpecParser; +import org.apache.iceberg.Schema; +import org.apache.iceberg.expressions.Expression; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.expressions.ResidualEvaluator; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types; + +import java.io.Serializable; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.time.ZoneOffset; +import java.util.Collections; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** Per-file proof that an entire static predicate is guaranteed by partition metadata. */ +final class IcebergPartitionFilterEvaluator implements Serializable { + private static final long serialVersionUID = 1L; + private static final int MIN_DATE_DAYS = Math.toIntExact(LocalDate.of(1, 1, 1).toEpochDay()); + private static final int MAX_DATE_DAYS = Math.toIntExact(LocalDate.of(9999, 12, 31).toEpochDay()); + private static final long MIN_TIMESTAMP_MICROS = + LocalDateTime.of(1, 1, 1, 0, 0).toEpochSecond(ZoneOffset.UTC) * 1_000_000L; + private static final long MAX_TIMESTAMP_MICROS = + LocalDateTime.of(9999, 12, 31, 23, 59, 59).toEpochSecond(ZoneOffset.UTC) * 1_000_000L + 999_999L; + + private final Schema schema; + private final PartitionSpec spec; + private final PartitionData partition; + private final ZoneId zone; + private transient Map<Integer, ConnectorExpression> cachedConjuncts; + private transient Set<Integer> cachedEnforced; + private transient boolean rebound; + private transient PartitionSpec reboundSpec; + + IcebergPartitionFilterEvaluator(Schema schema, PartitionSpec spec, PartitionData partition, ZoneId zone) { + this.schema = schema; + this.partition = partition.copy(); + this.zone = zone; + this.spec = spec; + } + + private PartitionSpec reboundSpec() { + if (rebound) { + return reboundSpec; + } + // Partition evolution retains old schemas and names. Rebind by field id only when the + // source types still agree; a DATE partition must not prove a promoted timestamp bound. + boolean compatible = spec.fields().stream().allMatch(field -> + schema.findType(field.sourceId()) != null + && schema.findType(field.sourceId()).equals(spec.schema().findType(field.sourceId()))); + reboundSpec = compatible ? PartitionSpecParser.fromJson(schema, PartitionSpecParser.toJson(spec)) : null; + rebound = true; + return reboundSpec; + } + + synchronized Set<Integer> enforced(Map<Integer, ConnectorExpression> conjuncts) { + if (spec.isUnpartitioned() || conjuncts.isEmpty()) { + return Collections.emptySet(); + } + if (conjuncts.equals(cachedConjuncts)) { + return cachedEnforced; + } + Set<Integer> result = new LinkedHashSet<>(); + for (Map.Entry<Integer, ConnectorExpression> entry : conjuncts.entrySet()) { + Expression predicate = exactPredicate(entry.getValue()); + if (predicate != null && reboundSpec() != null && ResidualEvaluator.of(reboundSpec, predicate, true) Review Comment: [P1] Retain residuals for overflowing numeric truncate partitions For an INT `id` partitioned by `truncate(10, id)`, Iceberg 1.12.0 computes the partition for `id = -2147483648` as `2147483646` because its [truncate implementation](https://github.com/apache/iceberg/blob/apache-iceberg-1.12.0/api/src/main/java/org/apache/iceberg/util/TruncateUtil.java#L51-L55) wraps. Its [strict projection](https://github.com/apache/iceberg/blob/apache-iceberg-1.12.0/api/src/main/java/org/apache/iceberg/transforms/ProjectionUtil.java#L57-L68) converts `id >= 0` into `partition > truncate(-1)`, i.e. `partition > -10`, so this evaluator records the predicate as enforced. With `write.metadata.metrics.default=none`, partition pruning retains that file and absent column bounds cannot reject it. ```text Project(id) Filter(id >= 0) IcebergScan[truncate(10, id); row id = -2147483648] ``` The new enforced ID removes `Filter(id >= 0)` from both TableReader and the final scanner filter, so this query returns the negative row instead of an empty result. Selecting `id` keeps the physical column projected, so this fails independently of placeholder elimination. BIGINT has the analogous signed-minimum case. Please retain these predicates unless numeric truncate wraparound is excluded by the proof, and add signed-minimum coverage with missing file metrics. -- 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]
