koodin9 commented on code in PR #18003: URL: https://github.com/apache/iceberg/pull/18003#discussion_r4374820473
########## kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/KeyFilters.java: ########## @@ -0,0 +1,203 @@ +/* + * 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.iceberg.connect.data; + +import java.util.Comparator; +import java.util.List; +import java.util.Set; +import org.apache.iceberg.PartitionField; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.StructLike; +import org.apache.iceberg.expressions.Expression; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.expressions.UnboundTerm; +import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import org.apache.iceberg.transforms.Transform; +import org.apache.iceberg.transforms.UnknownTransform; +import org.apache.iceberg.types.Comparators; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.StructLikeSet; + +/** + * Row filters built from the keys of equality delete files. + * + * <p>The filters are inclusive: every row with a deleted key satisfies them, and rows without a + * deleted key may satisfy them too. {@link ScanKeyPositionResolver} uses them to prune the data + * files it has to read; {@link EqualityDeleteConverter} uses them to detect files that another + * writer added concurrently for the deleted keys. + */ +final class KeyFilters { + + // metrics evaluators stop pruning with an IN list above this many values, so keys are matched + // with several lists of at most this size + @VisibleForTesting static final int IN_PREDICATE_LIMIT = 200; + + // the filters are evaluated once per data file entry during planning; above this many keys the + // lists are replaced by the bounds of the leading key column, two comparisons per entry + @VisibleForTesting static final int MAX_LISTED_KEYS = 1_000; + + private KeyFilters() {} + + /** + * Pins the partition of an equality delete file. Every partition field is expressed through its + * transform, e.g. {@code bucket(16, id) = 3}, so the scan planner prunes to the partition even + * when the key filter is a range that a transform like bucket cannot project. Identity fields + * reference the column directly. Void fields carry no information and unknown transforms cannot + * be evaluated, so both are skipped. + */ + @SuppressWarnings("unchecked") + static Expression partitionFilter(Schema tableSchema, PartitionSpec spec, StructLike partition) { + Expression filter = Expressions.alwaysTrue(); + List<PartitionField> fields = spec.fields(); + for (int pos = 0; pos < fields.size(); pos++) { + PartitionField field = fields.get(pos); + Transform<?, ?> transform = field.transform(); + if (transform.isVoid() || transform instanceof UnknownTransform) { + continue; + } + + String column = tableSchema.findColumnName(field.sourceId()); + UnboundTerm<Object> term = + transform.isIdentity() + ? Expressions.ref(column) + : Expressions.transform(column, (Transform<?, Object>) transform); + Object value = partition.get(pos, Object.class); + filter = + Expressions.and( + filter, value == null ? Expressions.isNull(term) : Expressions.equal(term, value)); + } + + return filter; + } + + /** + * Filter for the deleted keys. Up to {@link #MAX_LISTED_KEYS} keys are matched with {@code IN} + * lists of at most {@link #IN_PREDICATE_LIMIT} keys each, one list per key column. More keys are + * matched with the minimum and maximum of the leading key column, which may also cover files that + * hold no deleted key. + */ + static Expression keyFilter(Schema keySchema, StructLikeSet keys) { + if (keys.size() <= MAX_LISTED_KEYS) { Review Comment: > What envelope (keys per commit, files per partition, key distribution) should the part-2 benchmark cover? - Keys per commit: from a few keys to well beyond the IN-list limit, so that both the IN lists and the min/max range are covered. - Data files per partition: from small partitions to ones where reading the key columns of every file takes longer than the commit interval. - Key distribution, in two parts: - Which rows a commit changes: recent rows, rows scattered across history, and new keys, since upsert mode also deletes the key of every insert. - How keys lie in data files: increasing IDs, UUIDs after a sorting compaction, and unsorted UUIDs. - Layout: unpartitioned, identity-partitioned, and bucketed by the key, each with and without bloom filters on the key columns. For each case, the benchmark measures conversion time against the commit interval, planned data files, and bytes and manifests read, and compares read latency with the same commits written as equality deletes. The envelope is where the conversion still finishes within the commit interval. > What happens when `plannedDataFiles()` goes past it? The conversion still runs. Going past the envelope costs commit time, not correctness. In part 2, the coordinator will: - Report the planned data files and the conversion time of each commit. - Warn when a conversion takes longer than the commit interval. - On a conflict, re-run the conversion (and its scan) against the new snapshot, and fail the commit if it keeps conflicting. The benchmark results also say how to stay inside the envelope or widen it: - Layouts that stay inside it: sorting or bucketing by the key, bloom filters, compaction. These go into the docs. - The first pruning change to make: several ranges instead of one min/max range, one manifest pass for all partition groups, or skipping row groups with bloom filters. - Beyond that, the resolver seam is where an index (#16961) can take over. As you asked, I'll add a short version of this plan to the description. -- 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]
