juntaozhang commented on code in PR #9389: URL: https://github.com/apache/paimon/pull/9389#discussion_r3864773348
########## paimon-format/src/main/java/org/apache/paimon/format/parquet/VariantShreddingTypePruner.java: ########## @@ -0,0 +1,262 @@ +/* + * 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.paimon.format.parquet; + +import org.apache.paimon.data.variant.PaimonShreddingUtils; +import org.apache.paimon.data.variant.VariantMetadataUtils; +import org.apache.paimon.data.variant.VariantPathSegment; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.RowType; + +import org.apache.parquet.schema.GroupType; +import org.apache.parquet.schema.LogicalTypeAnnotation; +import org.apache.parquet.schema.Type; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Set; + +import static org.apache.paimon.format.parquet.ParquetSchemaConverter.parquetListElementType; +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** + * Prunes a shredded Variant Parquet type according to a logical Variant projection. + * + * <p>It builds a trie from the requested object/array paths and recursively removes unneeded fields + * from the {@code typed_value} (and list element) groups while preserving {@code value} fallbacks + * when a requested path cannot be satisfied from typed columns. + */ +public class VariantShreddingTypePruner { + private static final String LIST_WRAPPER_NAME = "list"; + private static final String LIST_ELEMENT_NAME = "element"; + + private final boolean caseSensitive; + @Nullable private final PathNode root; + + VariantShreddingTypePruner(RowType variantRowType, boolean caseSensitive) { + this.caseSensitive = caseSensitive; + this.root = buildPathTree(variantRowType); + } + + /** + * Clips the given Parquet Variant type to only include fields needed for {@code + * variantRowType}. + * + * @param variantRowType the logical Variant projection row type + * @param parquetType the physical Parquet Variant type + * @param caseSensitive whether field name matching is case-sensitive + * @return a clipped Parquet type + */ + public static Type clip(RowType variantRowType, GroupType parquetType, boolean caseSensitive) { + return new VariantShreddingTypePruner(variantRowType, caseSensitive).clip(parquetType); + } + + private Type clip(GroupType parquetType) { + return clipShreddingRow(parquetType, root); + } + + /** A projection trie for Variant object paths and array element paths. */ + private static class PathNode { + private final Map<String, PathNode> children = new HashMap<>(); + private PathNode arrayElement; + private boolean keepAll; + + private PathNode getOrCreateChild(String key) { + return children.computeIfAbsent(key, k -> new PathNode()); + } + } + + @Nullable + private PathNode buildPathTree(RowType variantRowType) { + PathNode root = new PathNode(); + for (DataField field : variantRowType.getFields()) { + String path = VariantMetadataUtils.path(field.description()); + VariantPathSegment[] segments = VariantPathSegment.parse(path); + if (segments.length == 0) { + return null; + } + + PathNode node = root; + for (VariantPathSegment segment : segments) { + if (segment instanceof VariantPathSegment.ArrayExtraction) { + // Array indices cannot prune individual elements at the Parquet level, + // but we can still prune nested fields inside each array element. + if (node.arrayElement == null) { + node.arrayElement = new PathNode(); + } + node = node.arrayElement; + } else if (segment instanceof VariantPathSegment.ObjectExtraction) { + String key = ((VariantPathSegment.ObjectExtraction) segment).getKey(); + if (!caseSensitive) { + key = key.toLowerCase(Locale.ROOT); Review Comment: Thanks for the guidance, I have removed the case-insensitivity support. PTAL ########## paimon-format/src/main/java/org/apache/paimon/format/parquet/VariantShreddingTypePruner.java: ########## @@ -0,0 +1,262 @@ +/* + * 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.paimon.format.parquet; + +import org.apache.paimon.data.variant.PaimonShreddingUtils; +import org.apache.paimon.data.variant.VariantMetadataUtils; +import org.apache.paimon.data.variant.VariantPathSegment; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.RowType; + +import org.apache.parquet.schema.GroupType; +import org.apache.parquet.schema.LogicalTypeAnnotation; +import org.apache.parquet.schema.Type; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Set; + +import static org.apache.paimon.format.parquet.ParquetSchemaConverter.parquetListElementType; +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** + * Prunes a shredded Variant Parquet type according to a logical Variant projection. + * + * <p>It builds a trie from the requested object/array paths and recursively removes unneeded fields + * from the {@code typed_value} (and list element) groups while preserving {@code value} fallbacks + * when a requested path cannot be satisfied from typed columns. + */ +public class VariantShreddingTypePruner { + private static final String LIST_WRAPPER_NAME = "list"; + private static final String LIST_ELEMENT_NAME = "element"; + + private final boolean caseSensitive; + @Nullable private final PathNode root; + + VariantShreddingTypePruner(RowType variantRowType, boolean caseSensitive) { + this.caseSensitive = caseSensitive; + this.root = buildPathTree(variantRowType); + } + + /** + * Clips the given Parquet Variant type to only include fields needed for {@code + * variantRowType}. + * + * @param variantRowType the logical Variant projection row type + * @param parquetType the physical Parquet Variant type + * @param caseSensitive whether field name matching is case-sensitive + * @return a clipped Parquet type + */ + public static Type clip(RowType variantRowType, GroupType parquetType, boolean caseSensitive) { + return new VariantShreddingTypePruner(variantRowType, caseSensitive).clip(parquetType); + } + + private Type clip(GroupType parquetType) { + return clipShreddingRow(parquetType, root); + } + + /** A projection trie for Variant object paths and array element paths. */ + private static class PathNode { + private final Map<String, PathNode> children = new HashMap<>(); + private PathNode arrayElement; + private boolean keepAll; + + private PathNode getOrCreateChild(String key) { + return children.computeIfAbsent(key, k -> new PathNode()); + } + } + + @Nullable + private PathNode buildPathTree(RowType variantRowType) { + PathNode root = new PathNode(); + for (DataField field : variantRowType.getFields()) { + String path = VariantMetadataUtils.path(field.description()); + VariantPathSegment[] segments = VariantPathSegment.parse(path); + if (segments.length == 0) { + return null; + } + + PathNode node = root; + for (VariantPathSegment segment : segments) { + if (segment instanceof VariantPathSegment.ArrayExtraction) { + // Array indices cannot prune individual elements at the Parquet level, + // but we can still prune nested fields inside each array element. + if (node.arrayElement == null) { + node.arrayElement = new PathNode(); + } + node = node.arrayElement; + } else if (segment instanceof VariantPathSegment.ObjectExtraction) { + String key = ((VariantPathSegment.ObjectExtraction) segment).getKey(); + if (!caseSensitive) { + key = key.toLowerCase(Locale.ROOT); + } + node = node.getOrCreateChild(key); + } else { + return null; + } + } + node.keepAll = true; + } + return root; + } + + private Type clipShreddingRow(Type type, PathNode node) { + if (type.isPrimitive() || node == null) { + return type; + } + + GroupType group = type.asGroupType(); + if (node.keepAll || !group.containsField(PaimonShreddingUtils.TYPED_VALUE_FIELD_NAME)) { + return group; + } + + List<Type> newFields = new ArrayList<>(); + if (group.containsField(PaimonShreddingUtils.METADATA_FIELD_NAME)) { + newFields.add(group.getType(PaimonShreddingUtils.METADATA_FIELD_NAME)); + } + + Type typedValue = group.getType(PaimonShreddingUtils.TYPED_VALUE_FIELD_NAME); + if (isCanonicalList(typedValue) && node.arrayElement != null) { + return clipListShreddingRow(group, node, newFields); + } else if (isObjectGroup(typedValue)) { + return clipObjectShreddingRow(group, node, newFields); + } else { + return group; + } + } + + private GroupType clipObjectShreddingRow(GroupType group, PathNode node, List<Type> newFields) { + Type typedValueType = group.getType(PaimonShreddingUtils.TYPED_VALUE_FIELD_NAME); + GroupType typedValue = typedValueType.asGroupType(); + // typed_value is an object group: prune by object key. + boolean needValue = false; Review Comment: Thanks for pointing out this issue, I have fixed it. PTAL -- 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]
