Gabriel39 commented on code in PR #69671: URL: https://github.com/apache/doris/pull/69671#discussion_r4386141942
########## 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: Fixed in d3d17aab62. The evaluator now conservatively retains numeric truncate value predicates, covering overflow of both partition values and literal boundaries. Nullness proofs and proofs on other partition fields remain eligible. Added INT/BIGINT signed-minimum tests using real truncate transforms and DataFile metadata without column bounds. They confirm that the actual row fails the predicate while inclusive pruning, metrics pruning, and Iceberg's raw TRUE residual retain the file. Also covered overflowing literal boundaries, NOT IN, and unaffected nullness/day proofs. The new tests failed before the fix and pass afterward. Validation: 271 Iceberg planner/range/proof tests and 6 SPI surface tests passed; FE Checkstyle passed. Extended the Spark regression with metrics disabled and projected/filter-only queries across scanner and batch settings; Groovy compilation passed, cluster execution is pending CI. -- 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]
