JingsongLi commented on code in PR #8392:
URL: https://github.com/apache/paimon/pull/8392#discussion_r3519460264
##########
paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java:
##########
@@ -658,6 +661,77 @@ private static void validateMapStorageLayout(TableSchema
schema, CoreOptions opt
}
options.mapSharedShreddingMaxColumns(fieldName);
}
+
+ if (hasSharedShredding) {
+ validateMapSharedShreddingFileFormats(options);
Review Comment:
Schema validation accepts shared-shredding tables with
`file.compression=snappy` or other Parquet/ORC codecs, but
`MapSharedShreddingWritePlan.fieldMetadata(compression)` passes that same codec
into `normalizeFieldDictCompression`, which only allows `zstd`, `lz4`, or
`none`. Such a table can be created successfully and then fail only when the
writer closes after producing data. Please either validate the relevant
data/changelog/per-level compression options here for shared-shredding, or
decouple field-dictionary compression from the data-file codec.
##########
paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java:
##########
@@ -658,6 +661,77 @@ private static void validateMapStorageLayout(TableSchema
schema, CoreOptions opt
}
options.mapSharedShreddingMaxColumns(fieldName);
}
+
+ if (hasSharedShredding) {
+ validateMapSharedShreddingFileFormats(options);
+ validateNoVariantWithMapSharedShredding(schema);
+ if (!schema.primaryKeys().isEmpty()) {
+ throw new IllegalArgumentException(
+ "MAP shared-shredding currently only supports
append-only tables.");
+ }
+ if (options.bucket() != -1 && !options.writeOnly()) {
+ throw new IllegalArgumentException(
+ "MAP shared-shredding currently requires bucket = -1
or write-only = true because rewrite/compaction is not supported.");
+ }
+ }
+ }
+
+ private static void validateNoVariantWithMapSharedShredding(TableSchema
schema) {
+ if (containsVariantFields(new RowType(schema.fields()))) {
Review Comment:
The shared-shredding validation only rejects Variant recursively, but BLOB
can still be nested in a shredding MAP value because the existing BLOB checks
only inspect top-level fields. A schema such as `MAP<STRING, BLOB>` passes
validation and then fails when the read plan creates
`RowToColumnConverter.createElementConverter(mapType.getValueType())`, because
`BlobType` currently throws `UnsupportedOperationException`. Please reject BLOB
recursively for shared-shredding MAP values or implement BLOB materialization,
and add a validation test for `MAP<STRING, BLOB>`.
##########
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:
This read plan is created whenever the requested row type contains any
Variant field, but `buildPhysicalRowType` resolves every top-level field
directly from the Parquet file schema. If schema evolution adds a non-Variant
column after older files were written, that column is absent from `fileSchema`;
the previous Parquet reader path let `clipParquetSchema` synthesize a missing
column and fill NULLs, while this throws here before requested schema creation.
Please preserve the missing-column path, for example by keeping the logical
field type when the file field is absent and letting `createRequestedSchema`
mark it missing, and add a test with a Variant column plus a newly added field
read from old Parquet files.
--
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]
