github-actions[bot] commented on code in PR #68393: URL: https://github.com/apache/doris/pull/68393#discussion_r4089225051
########## fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/iceberg/Partitioning.java: ########## @@ -0,0 +1,411 @@ +/* + * 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; + +import java.util.Collection; +import java.util.Comparator; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; +import org.apache.iceberg.exceptions.ValidationException; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import org.apache.iceberg.transforms.PartitionSpecVisitor; +import org.apache.iceberg.transforms.Transform; +import org.apache.iceberg.transforms.Transforms; +import org.apache.iceberg.transforms.UnknownTransform; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types.NestedField; +import org.apache.iceberg.types.Types.StructType; + +/** + * Copied from Apache Iceberg 1.11.0's Partitioning.java. + * + * <p>Disambiguates historical partition field names when building a unified type. Keep the remaining + * implementation aligned with iceberg-core. + */ +public class Partitioning { + private Partitioning() {} + + /** + * Check whether the spec contains a bucketed partition field. + * + * @param spec a partition spec + * @return true if the spec has field with a bucket transform + */ + public static boolean hasBucketField(PartitionSpec spec) { + List<Boolean> bucketList = + PartitionSpecVisitor.visit( + spec, + new PartitionSpecVisitor<Boolean>() { + @Override + public Boolean identity(int fieldId, String sourceName, int sourceId) { + return false; + } + + @Override + public Boolean bucket(int fieldId, String sourceName, int sourceId, int width) { + return true; + } + + @Override + public Boolean truncate(int fieldId, String sourceName, int sourceId, int width) { + return false; + } + + @Override + public Boolean year(int fieldId, String sourceName, int sourceId) { + return false; + } + + @Override + public Boolean month(int fieldId, String sourceName, int sourceId) { + return false; + } + + @Override + public Boolean day(int fieldId, String sourceName, int sourceId) { + return false; + } + + @Override + public Boolean hour(int fieldId, String sourceName, int sourceId) { + return false; + } + + @Override + public Boolean alwaysNull(int fieldId, String sourceName, int sourceId) { + return false; + } + + @Override + public Boolean unknown( + int fieldId, String sourceName, int sourceId, String transform) { + return false; + } + }); + + return bucketList.stream().anyMatch(Boolean::booleanValue); + } + + /** + * Create a sort order that will group data for a partition spec. + * + * <p>If the partition spec contains bucket columns, the sort order will also have a field to sort + * by a column that is bucketed in the spec. The column is selected by the highest number of + * buckets in the transform. + * + * @param spec a partition spec + * @return a sort order that will cluster data for the spec + */ + public static SortOrder sortOrderFor(PartitionSpec spec) { + if (spec.isUnpartitioned()) { + return SortOrder.unsorted(); + } + + SortOrder.Builder builder = SortOrder.builderFor(spec.schema()); + SpecToOrderVisitor converter = new SpecToOrderVisitor(builder); + PartitionSpecVisitor.visit(spec, converter); + + // columns used for bucketing are high cardinality; add one to the sort at the end + String bucketColumn = converter.bucketColumn(); + if (bucketColumn != null) { + builder.asc(bucketColumn); + } + + return builder.build(); + } + + private static class SpecToOrderVisitor implements PartitionSpecVisitor<Void> { + private final SortOrder.Builder builder; + private String bucketColumn = null; + private int highestNumBuckets = 0; + + private SpecToOrderVisitor(SortOrder.Builder builder) { + this.builder = builder; + } + + String bucketColumn() { + return bucketColumn; + } + + @Override + public Void identity(int fieldId, String sourceName, int sourceId) { + builder.asc(sourceName); + return null; + } + + @Override + public Void bucket(int fieldId, String sourceName, int sourceId, int numBuckets) { + // the column with highest cardinality is usually the one with the highest number of buckets + if (numBuckets > highestNumBuckets) { + this.highestNumBuckets = numBuckets; + this.bucketColumn = sourceName; + } + builder.asc(Expressions.bucket(sourceName, numBuckets)); + return null; + } + + @Override + public Void truncate(int fieldId, String sourceName, int sourceId, int width) { + builder.asc(Expressions.truncate(sourceName, width)); + return null; + } + + @Override + public Void year(int fieldId, String sourceName, int sourceId) { + builder.asc(Expressions.year(sourceName)); + return null; + } + + @Override + public Void month(int fieldId, String sourceName, int sourceId) { + builder.asc(Expressions.month(sourceName)); + return null; + } + + @Override + public Void day(int fieldId, String sourceName, int sourceId) { + builder.asc(Expressions.day(sourceName)); + return null; + } + + @Override + public Void hour(int fieldId, String sourceName, int sourceId) { + builder.asc(Expressions.hour(sourceName)); + return null; + } + + @Override + public Void alwaysNull(int fieldId, String sourceName, int sourceId) { + // do nothing for alwaysNull, it doesn't need to be added to the sort + return null; + } + } + + /** + * Builds a grouping key type considering the provided schema and specs. + * + * <p>A grouping key defines how data is split between files and consists of partition fields with + * non-void transforms that are present in each provided spec. Iceberg guarantees that records + * with different values for the grouping key are disjoint and are stored in separate files. + * + * <p>If there is only one spec, the grouping key will include all partition fields with non-void + * transforms from that spec. Whenever there are multiple specs, the grouping key will represent + * an intersection of all partition fields with non-void transforms. If a partition field is + * present only in a subset of specs, Iceberg cannot guarantee data distribution on that field. + * That's why it will not be part of the grouping key. Unpartitioned tables or tables with + * non-overlapping specs have empty grouping keys. + * + * <p>When partition fields are dropped in v1 tables, they are replaced with new partition fields + * that have the same field ID but use a void transform under the hood. Such fields cannot be part + * of the grouping key as void transforms always return null. + * + * <p>If the provided schema is not null, this method will only take into account partition fields + * on top of columns present in the schema. Otherwise, all partition fields will be considered. + * + * @param schema a schema specifying a set of source columns to consider (null to consider all) + * @param specs one or many specs + * @return the constructed grouping key type + */ + public static StructType groupingKeyType(Schema schema, Collection<PartitionSpec> specs) { + return buildPartitionProjectionType("grouping key", specs, commonActiveFieldIds(schema, specs)); + } + + /** + * Builds a unified partition type considering all specs in a table. + * + * <p>If there is only one spec, the partition type is that spec's partition type. Whenever there + * are multiple specs, the partition type is a struct containing all fields that have ever been a + * part of any spec in the table. In other words, the struct fields represent a union of all known + * partition fields. + * + * @param table a table with one or many specs + * @return the constructed unified partition type + */ + public static StructType partitionType(Table table) { + Collection<PartitionSpec> specs = table.specs().values(); + return buildPartitionProjectionType( + "table partition", specs, allActiveFieldIds(table.schema(), specs)); + } + + /** + * Checks if any of the specs in a table is partitioned. + * + * @param table the table to check. + * @return {@code true} if the table is partitioned, {@code false} otherwise. + */ + public static boolean isPartitioned(Table table) { + return table.specs().values().stream().anyMatch(PartitionSpec::isPartitioned); + } + + private static StructType buildPartitionProjectionType( + String typeName, Collection<PartitionSpec> specs, Set<Integer> projectedFieldIds) { + + // we currently don't know the output type of unknown transforms + List<Transform<?, ?>> unknownTransforms = collectUnknownTransforms(specs); + ValidationException.check( + unknownTransforms.isEmpty(), + "Cannot build %s type, unknown transforms: %s", + typeName, + unknownTransforms); + + Map<Integer, PartitionField> fieldMap = Maps.newLinkedHashMap(); + Map<Integer, Type> typeMap = Maps.newHashMap(); + Map<Integer, String> nameMap = Maps.newHashMap(); + + // sort specs by ID in descending order to pick up the most recent field names + List<PartitionSpec> sortedSpecs = + specs.stream() + .sorted(Comparator.comparingLong(PartitionSpec::specId).reversed()) + .collect(Collectors.toList()); + + for (PartitionSpec spec : sortedSpecs) { + for (PartitionField field : spec.fields()) { + int fieldId = field.fieldId(); + + if (!projectedFieldIds.contains(fieldId)) { + continue; + } + + NestedField structField = spec.partitionType().field(fieldId); + PartitionField existingField = fieldMap.get(fieldId); + + if (existingField == null) { + fieldMap.put(fieldId, field); + typeMap.put(fieldId, structField.type()); + nameMap.put(fieldId, structField.name()); + + } else { + // verify the fields are compatible as they may conflict in v1 tables + ValidationException.check( + equivalentIgnoringNames(field, existingField), + "Conflicting partition fields: ['%s', '%s']", + field, + existingField); + + // use the correct type for dropped partitions in v1 tables + if (isVoidTransform(existingField) && !isVoidTransform(field)) { + fieldMap.put(fieldId, field); + typeMap.put(fieldId, structField.type()); + } + } + } + } + + // Different field IDs can reuse a name across specs. Keep the newest field's name and + // disambiguate older fields without losing their IDs or historical partition values. + // Reserve all original names first so a generated suffix cannot shadow another real field. + // Doris struct fields and Iceberg's case-insensitive binder fold names with Locale.ROOT. + Set<String> reservedNames = + nameMap.values().stream() + .map(name -> name.toLowerCase(Locale.ROOT)) + .collect(Collectors.toSet()); + Set<String> assignedNames = Sets.newHashSet(); + for (int fieldId : fieldMap.keySet()) { Review Comment: [P1] Keep the active v1 replacement on the unsuffixed name A legal v1 sequence still gives the canonical name to the wrong field: create identity `record_key` (ID 1000), remove it (v1 carries ID 1000 as `void`), then add identity `RECORD_KEY` (ID 1001). Iceberg appends the new active field after the carried void field, so this insertion order lets historical ID 1000 claim folded `record_key` and renames the active field to `RECORD_KEY_1001`. `$files`/`$partitions` current rows are then NULL under their declared current partition name, and predicates on `RECORD_KEY` bind the historical ID. Please track the newest occurrence's active/void state before v1 type recovery, let the active field claim the unsuffixed name, and cover this v1 remove-then-add case-only evolution. -- 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]
