lxy-9602 commented on code in PR #8392: URL: https://github.com/apache/paimon/pull/8392#discussion_r3519804203
########## paimon-format/src/main/java/org/apache/paimon/format/parquet/VariantShreddingReadPlanFactory.java: ########## @@ -0,0 +1,461 @@ +/* + * 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.columnar.ArrayColumnVector; +import org.apache.paimon.data.columnar.ColumnVector; +import org.apache.paimon.data.columnar.ColumnVectorUtils; +import org.apache.paimon.data.columnar.VecColumnVector; +import org.apache.paimon.data.columnar.VectorizedColumnBatch; +import org.apache.paimon.data.columnar.heap.CastedArrayColumnVector; +import org.apache.paimon.data.columnar.heap.CastedMapColumnVector; +import org.apache.paimon.data.columnar.heap.CastedRowColumnVector; +import org.apache.paimon.data.columnar.heap.CastedVectorColumnVector; +import org.apache.paimon.data.columnar.writable.WritableColumnVector; +import org.apache.paimon.data.shredding.ShreddingBatchAssembler; +import org.apache.paimon.data.shredding.ShreddingReadPlan; +import org.apache.paimon.data.variant.PaimonShreddingUtils; +import org.apache.paimon.data.variant.PaimonShreddingUtils.FieldToExtract; +import org.apache.paimon.data.variant.VariantMetadataUtils; +import org.apache.paimon.data.variant.VariantPathSegment; +import org.apache.paimon.format.shredding.ShreddingReadPlanFactory; +import org.apache.paimon.types.ArrayType; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.MapType; +import org.apache.paimon.types.RowType; +import org.apache.paimon.types.VariantType; +import org.apache.paimon.types.VectorType; +import org.apache.paimon.utils.Pair; +import org.apache.paimon.utils.Preconditions; + +import org.apache.parquet.schema.GroupType; +import org.apache.parquet.schema.Type; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static org.apache.paimon.data.variant.Variant.METADATA; +import static org.apache.paimon.data.variant.Variant.VALUE; +import static org.apache.paimon.data.variant.VariantMetadataUtils.path; +import static org.apache.paimon.format.parquet.ParquetSchemaConverter.parquetListElementType; +import static org.apache.paimon.format.parquet.ParquetSchemaConverter.parquetMapKeyValueType; + +/** + * Parquet read plan factory for Variant physical layouts. + * + * <p>The plan reads the physical Variant row layout from Parquet, then materializes it back to the + * logical Variant or Variant projection row in the shared shredding batch assembly path. + */ +public class VariantShreddingReadPlanFactory implements ShreddingReadPlanFactory { + + private final RowType logicalRowType; + private final boolean caseSensitive; + + public VariantShreddingReadPlanFactory(RowType logicalRowType, boolean caseSensitive) { + this.logicalRowType = logicalRowType; + this.caseSensitive = caseSensitive; + } + + @Override + public RowType logicalRowType() { + return logicalRowType; + } + + @Override + public boolean shouldCreateReadPlan( + Map<String, Map<String, String>> fieldMetadata, @Nullable Object fileSchema) { + return fileSchema instanceof GroupType && containsVariantFields(logicalRowType); + } + + @Override + public ShreddingReadPlan createReadPlan( + Map<String, Map<String, String>> fieldMetadata, @Nullable Object fileSchema) { + return new VariantShreddingReadPlan(logicalRowType, (GroupType) fileSchema, caseSensitive); + } + + /** Extracts the physical Variant file schema from a Parquet field. */ + public static RowType variantFileType(Type fileType) { + boolean isShredded = + fileType.asGroupType().containsField(PaimonShreddingUtils.TYPED_VALUE_FIELD_NAME); + if (isShredded) { + return (RowType) ParquetSchemaConverter.convertToPaimonField(fileType).type(); + } + + List<DataField> dataFields = new ArrayList<>(); + dataFields.add(new DataField(0, VALUE, DataTypes.BYTES().notNull())); + dataFields.add(new DataField(1, METADATA, DataTypes.BYTES().notNull())); + return new RowType(dataFields); + } + + /** Clips a Variant Parquet field according to the logical Variant read type. */ + @Nullable + public static Type clipParquetType(DataType logicalType, Type parquetType) { + if (logicalType instanceof VariantType) { + return parquetType; + } + if (VariantMetadataUtils.isVariantRowType(logicalType)) { + return clipVariantType((RowType) logicalType, parquetType.asGroupType()); + } + return null; + } + + /** Clips a Variant Parquet field according to the logical Variant row read type. */ + public static Type clipVariantType(RowType variantRowType, GroupType parquetType) { + if (!parquetType.containsField(PaimonShreddingUtils.TYPED_VALUE_FIELD_NAME)) { + return parquetType; + } + + boolean canClip = true; + Set<String> fieldsToRead = new HashSet<>(); + for (DataField field : variantRowType.getFields()) { + String path = path(field.description()); + VariantPathSegment[] pathSegments = VariantPathSegment.parse(path); + if (pathSegments.length < 1) { + canClip = false; + break; + } + + // TODO: support nested column pruning. + VariantPathSegment pathSegment = pathSegments[0]; + if (pathSegment instanceof VariantPathSegment.ObjectExtraction) { + fieldsToRead.add(((VariantPathSegment.ObjectExtraction) pathSegment).getKey()); + } else { + canClip = false; + break; + } + } + + if (!canClip) { + return parquetType; + } + + List<Type> typedFieldsToRead = new ArrayList<>(); + GroupType typedValue = + parquetType.getType(PaimonShreddingUtils.TYPED_VALUE_FIELD_NAME).asGroupType(); + for (Type field : typedValue.getFields()) { + if (fieldsToRead.contains(field.getName())) { + typedFieldsToRead.add(field); + fieldsToRead.remove(field.getName()); + } + } + + List<Type> rowGroupFields = new ArrayList<>(); + rowGroupFields.add(parquetType.getType(PaimonShreddingUtils.METADATA_FIELD_NAME)); + if (!fieldsToRead.isEmpty()) { + rowGroupFields.add(parquetType.getType(PaimonShreddingUtils.VARIANT_VALUE_FIELD_NAME)); + } + if (!typedFieldsToRead.isEmpty()) { + rowGroupFields.add(typedValue.withNewFields(typedFieldsToRead)); + } + return parquetType.withNewFields(rowGroupFields); + } + + private static boolean containsVariantFields(DataType dataType) { + if (dataType instanceof VariantType || VariantMetadataUtils.isVariantRowType(dataType)) { + return true; + } + if (dataType instanceof RowType) { + for (DataField field : ((RowType) dataType).getFields()) { + if (containsVariantFields(field.type())) { + return true; + } + } + } + if (dataType instanceof ArrayType) { + return containsVariantFields(((ArrayType) dataType).getElementType()); + } + if (dataType instanceof VectorType) { + return containsVariantFields(((VectorType) dataType).getElementType()); + } + if (dataType instanceof MapType) { + MapType mapType = (MapType) dataType; + return containsVariantFields(mapType.getKeyType()) + || containsVariantFields(mapType.getValueType()); + } + return false; + } + + private static RowType buildPhysicalRowType( + RowType logicalRowType, GroupType parquetType, boolean caseSensitive) { + List<DataField> fields = new ArrayList<>(); + for (DataField field : logicalRowType.getFields()) { + Type fileType = matchParquetField(parquetType, field.name(), caseSensitive); Review Comment: Thanks for pointing this out. I have updated the read plan to be defensive here. One note: in the normal table read path, schema-evolution missing columns should already be handled before reaching the format reader. `FormatReaderMapping` builds the actual format read type from the file data schema, so fields absent from an old file are not included in the `actualReadRowType` passed to Parquet/ORC. Those fields are then represented by `indexMapping = -1` and filled as NULL by `DataFileRecordReader` / `VectorMappingUtils`. That said, keeping the missing-field handling in the Variant read plan makes the format-level code more robust and avoids surprises if this plan is used from a more direct format path. -- 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]
