You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "lxy-9602 (via GitHub)" <gi...@apache.org> on 2026/06/30 08:23:05 UTC
[PR] [core] Support append read/write for MAP shared-shredding [paimon]
lxy-9602 opened a new pull request, #8392:
URL: https://github.com/apache/paimon/pull/8392
## What is changed
This PR adds append-only read/write support for MAP shared-shredding layout. MAP fields configured with `map.storage-layout=shared-shredding` are written with a physical shredded layout and per-field metadata in ORC/Parquet footers, and are reconstructed back to logical MAP values during reads.
The read path is also refactored to use a unified shredding read-plan model for both Variant and MAP shared-shredding. ORC/Parquet readers now create a `ShreddingReadPlan` from file metadata/schema, read the corresponding physical row type, and wrap the raw reader with a common `ShreddingFormatReader` to assemble physical batches back into logical batches. Variant and MAP shared-shredding therefore share the same high-level read flow instead of having separate format-specific conversion models.
## Supported scope
- Append-only tables.
- ORC and Parquet data files.
- MAP shared-shredding write path with footer metadata.
- MAP shared-shredding read path with schema evolution/projection.
- Variant read materialization moved into the shared shredding read-plan framework.
## Limitations
Rewrite/compaction paths are not supported for MAP shared-shredding yet. Unsupported table modes and incompatible combinations, such as primary-key tables or combining Variant with MAP shared-shredding, are rejected by validation or fail fast in write paths.
## Tests
Added coverage for MAP shared-shredding metadata, write plans, read plans, append table end-to-end read/write, projection, null values, overflow values, adaptive column restoration, layout switching, data evolution interaction, and validation of unsupported configurations. Existing Variant shredding tests continue to cover the refactored Variant read 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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Support append read/write for MAP shared-shredding [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #8392:
URL: https://github.com/apache/paimon/pull/8392#issuecomment-4880271741
This PR has become too large, can we split some PRs out.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Support append read/write for MAP shared-shredding [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #8392:
URL: https://github.com/apache/paimon/pull/8392#discussion_r3503128846
##########
paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlan.java:
##########
@@ -0,0 +1,211 @@
+/*
+ * 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.data.shredding;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericMap;
+import org.apache.paimon.data.InternalArray;
+import org.apache.paimon.data.InternalMap;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.columnar.ColumnVector;
+import org.apache.paimon.data.columnar.MapColumnVector;
+import org.apache.paimon.data.columnar.RowColumnVector;
+import org.apache.paimon.data.columnar.VectorizedColumnBatch;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.RowType;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Read plan that rebuilds logical MAP values from shared-shredding physical ROW values. */
+public class MapSharedShreddingReadPlan implements ShreddingReadPlan {
+
+ private final RowType logicalType;
+ private final RowType physicalType;
+ private final Map<Integer, SharedShreddingContext> contextByFieldIndex;
+
+ public MapSharedShreddingReadPlan(
+ RowType logicalType, Map<String, MapSharedShreddingFieldMeta> fieldMetas) {
+ this.logicalType = logicalType;
+ this.physicalType = MapSharedShreddingUtils.buildPhysicalReadType(logicalType, fieldMetas);
+ this.contextByFieldIndex = createContexts(logicalType, fieldMetas);
+ }
+
+ @Override
+ public RowType logicalRowType() {
+ return logicalType;
+ }
+
+ @Override
+ public RowType physicalRowType() {
+ return physicalType;
+ }
+
+ @Override
+ public boolean isIdentity() {
+ return contextByFieldIndex.isEmpty();
+ }
+
+ @Override
+ public ShreddingBatchAssembler batchAssembler() {
+ return new MapSharedShreddingBatchAssembler();
+ }
+
+ private static Map<Integer, SharedShreddingContext> createContexts(
+ RowType logicalType, Map<String, MapSharedShreddingFieldMeta> fieldMetas) {
+ Map<Integer, SharedShreddingContext> contexts = new LinkedHashMap<>();
+ for (int i = 0; i < logicalType.getFieldCount(); i++) {
+ DataField field = logicalType.getFields().get(i);
+ MapSharedShreddingFieldMeta fieldMeta = fieldMetas.get(field.name());
+ if (fieldMeta != null && field.type() instanceof MapType) {
+ contexts.put(i, new SharedShreddingContext(fieldMeta, field.type()));
+ }
+ }
+ return contexts;
+ }
+
+ private static InternalMap rebuildLogicalMap(
+ InternalRow physicalRow, SharedShreddingContext context) {
+ InternalArray fieldMapping = physicalRow.getArray(0);
+ checkArgument(
+ fieldMapping.size() == context.numColumns,
+ "Shared-shredding field mapping size %s does not match metadata num columns %s.",
+ fieldMapping.size(),
+ context.numColumns);
+ Map<Object, Object> result = new LinkedHashMap<>();
+ for (int column = 0; column < context.numColumns; column++) {
+ int fieldId = fieldMapping.isNullAt(column) ? -1 : fieldMapping.getInt(column);
+ if (fieldId < 0) {
+ continue;
+ }
+ BinaryString fieldName = context.nameById.get(fieldId);
+ if (fieldName == null) {
+ continue;
+ }
+ Object value = null;
+ int valuePosition = column + 1;
+ if (valuePosition < physicalRow.getFieldCount()
+ && !physicalRow.isNullAt(valuePosition)) {
+ value = context.valueGetters[column].getFieldOrNull(physicalRow);
+ }
+ result.put(fieldName, value);
+ }
+ if (context.overflowPosition < physicalRow.getFieldCount()
+ && !physicalRow.isNullAt(context.overflowPosition)) {
+ InternalMap overflow = physicalRow.getMap(context.overflowPosition);
+ InternalArray keys = overflow.keyArray();
+ InternalArray values = overflow.valueArray();
+ for (int i = 0; i < overflow.size(); i++) {
+ int fieldId = keys.getInt(i);
+ BinaryString fieldName = context.nameById.get(fieldId);
+ if (fieldName != null) {
+ result.put(fieldName, context.overflowValueGetter.getElementOrNull(values, i));
+ }
+ }
+ }
+ return new GenericMap(result);
+ }
+
+ private class MapSharedShreddingBatchAssembler implements ShreddingBatchAssembler {
+
+ @Override
+ public VectorizedColumnBatch assemble(VectorizedColumnBatch physicalBatch) {
+ ColumnVector[] logicalVectors = new ColumnVector[logicalType.getFieldCount()];
+ for (int i = 0; i < logicalVectors.length; i++) {
+ SharedShreddingContext context = contextByFieldIndex.get(i);
+ if (context == null) {
+ logicalVectors[i] = physicalBatch.columns[i];
+ } else {
+ logicalVectors[i] =
+ new SharedShreddingMapColumnVector(
+ (RowColumnVector) physicalBatch.columns[i], context);
+ }
+ }
+ return physicalBatch.copy(logicalVectors);
+ }
+ }
+
+ private static class SharedShreddingMapColumnVector implements MapColumnVector {
+
+ private final RowColumnVector physicalVector;
+ private final SharedShreddingContext context;
+
+ private SharedShreddingMapColumnVector(
+ RowColumnVector physicalVector, SharedShreddingContext context) {
+ this.physicalVector = physicalVector;
+ this.context = context;
+ }
+
+ @Override
+ public InternalMap getMap(int i) {
+ if (physicalVector.isNullAt(i)) {
+ return null;
+ }
+ // TODO (xinyu.lxy): Move this per-row map assemble path to a real batch-level
+ // physical-to-logical conversion.
+ return rebuildLogicalMap(physicalVector.getRow(i), context);
+ }
+
+ @Override
+ public boolean isNullAt(int i) {
+ return physicalVector.isNullAt(i);
+ }
+
+ @Override
+ public int getCapacity() {
+ return physicalVector.getCapacity();
+ }
+
+ @Override
+ public ColumnVector[] getChildren() {
Review Comment:
This vector is still exposed through the vectorized read path: `ShreddingFormatReader` copies the delegate `VectorizedRowIterator`, so callers such as `ArrowVectorizedBatchConverter` can consume the returned batch directly. The Arrow map writer calls `mapColumnVector.getChildren()[0/1]`, so returning null here makes a shared-shredding MAP column fail with an NPE when exported through the vectorized Arrow path. Could we either materialize key/value child vectors for this wrapper or force this reader to fall back to a row iterator for shared-shredding maps?
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Refactor Variant Parquet reads onto a unified shredding read path [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #8392:
URL: https://github.com/apache/paimon/pull/8392
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Support append read/write for MAP shared-shredding [paimon]
Posted by "lxy-9602 (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Refactor Variant Parquet reads onto a unified shredding read path [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #8392:
URL: https://github.com/apache/paimon/pull/8392#issuecomment-4922052933
cc @Zouxxyy @leaves12138
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Refactor Variant Parquet reads onto a unified shredding read path [paimon]
Posted by "lxy-9602 (via GitHub)" <gi...@apache.org>.
lxy-9602 commented on PR #8392:
URL: https://github.com/apache/paimon/pull/8392#issuecomment-4900166064
> This PR has become too large, can we split some PRs out.
Thanks for the detailed review. To make the changes easier to review, I have split the original PR into smaller independent parts:
1. Refactor the Variant read path to use `ShreddingReadPlan`, so Variant materialization is no longer tightly coupled with the Parquet reader (this pr).
2. Add MAP shared-shredding schema validation and common utility updates (#8487).
3. Add append-table MAP shared-shredding read/write support, including the full append E2E test coverage.
This should make each part more focused and avoid mixing framework refactoring, validation, and the MAP shared-shredding read/write implementation in one large PR. Would be grateful if you could take another look when you have time.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Support append read/write for MAP shared-shredding [paimon]
Posted by "lxy-9602 (via GitHub)" <gi...@apache.org>.
lxy-9602 commented on code in PR #8392:
URL: https://github.com/apache/paimon/pull/8392#discussion_r3520465976
##########
paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java:
##########
@@ -658,6 +663,109 @@ private static void validateMapStorageLayout(TableSchema schema, CoreOptions opt
}
options.mapSharedShreddingMaxColumns(fieldName);
}
+
+ if (hasSharedShredding) {
+ validateMapSharedShreddingFileFormats(options);
+ validateMapSharedShreddingCompressions(options);
+ validateNoVariantOrBlobWithMapSharedShredding(schema);
Review Comment:
Thanks a lot for your review and time! I learned a lot from the feedback and have made a few more fixes.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Support append read/write for MAP shared-shredding [paimon]
Posted by "lxy-9602 (via GitHub)" <gi...@apache.org>.
lxy-9602 commented on code in PR #8392:
URL: https://github.com/apache/paimon/pull/8392#discussion_r3519806001
##########
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:
Thanks for the review! I’ve updated the validation according to the comments.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Support append read/write for MAP shared-shredding [paimon]
Posted by "lxy-9602 (via GitHub)" <gi...@apache.org>.
lxy-9602 commented on code in PR #8392:
URL: https://github.com/apache/paimon/pull/8392#discussion_r3504029700
##########
paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlan.java:
##########
@@ -0,0 +1,211 @@
+/*
+ * 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.data.shredding;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericMap;
+import org.apache.paimon.data.InternalArray;
+import org.apache.paimon.data.InternalMap;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.columnar.ColumnVector;
+import org.apache.paimon.data.columnar.MapColumnVector;
+import org.apache.paimon.data.columnar.RowColumnVector;
+import org.apache.paimon.data.columnar.VectorizedColumnBatch;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.RowType;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Read plan that rebuilds logical MAP values from shared-shredding physical ROW values. */
+public class MapSharedShreddingReadPlan implements ShreddingReadPlan {
+
+ private final RowType logicalType;
+ private final RowType physicalType;
+ private final Map<Integer, SharedShreddingContext> contextByFieldIndex;
+
+ public MapSharedShreddingReadPlan(
+ RowType logicalType, Map<String, MapSharedShreddingFieldMeta> fieldMetas) {
+ this.logicalType = logicalType;
+ this.physicalType = MapSharedShreddingUtils.buildPhysicalReadType(logicalType, fieldMetas);
+ this.contextByFieldIndex = createContexts(logicalType, fieldMetas);
+ }
+
+ @Override
+ public RowType logicalRowType() {
+ return logicalType;
+ }
+
+ @Override
+ public RowType physicalRowType() {
+ return physicalType;
+ }
+
+ @Override
+ public boolean isIdentity() {
+ return contextByFieldIndex.isEmpty();
+ }
+
+ @Override
+ public ShreddingBatchAssembler batchAssembler() {
+ return new MapSharedShreddingBatchAssembler();
+ }
+
+ private static Map<Integer, SharedShreddingContext> createContexts(
+ RowType logicalType, Map<String, MapSharedShreddingFieldMeta> fieldMetas) {
+ Map<Integer, SharedShreddingContext> contexts = new LinkedHashMap<>();
+ for (int i = 0; i < logicalType.getFieldCount(); i++) {
+ DataField field = logicalType.getFields().get(i);
+ MapSharedShreddingFieldMeta fieldMeta = fieldMetas.get(field.name());
+ if (fieldMeta != null && field.type() instanceof MapType) {
+ contexts.put(i, new SharedShreddingContext(fieldMeta, field.type()));
+ }
+ }
+ return contexts;
+ }
+
+ private static InternalMap rebuildLogicalMap(
+ InternalRow physicalRow, SharedShreddingContext context) {
+ InternalArray fieldMapping = physicalRow.getArray(0);
+ checkArgument(
+ fieldMapping.size() == context.numColumns,
+ "Shared-shredding field mapping size %s does not match metadata num columns %s.",
+ fieldMapping.size(),
+ context.numColumns);
+ Map<Object, Object> result = new LinkedHashMap<>();
+ for (int column = 0; column < context.numColumns; column++) {
+ int fieldId = fieldMapping.isNullAt(column) ? -1 : fieldMapping.getInt(column);
+ if (fieldId < 0) {
+ continue;
+ }
+ BinaryString fieldName = context.nameById.get(fieldId);
+ if (fieldName == null) {
+ continue;
+ }
+ Object value = null;
+ int valuePosition = column + 1;
+ if (valuePosition < physicalRow.getFieldCount()
+ && !physicalRow.isNullAt(valuePosition)) {
+ value = context.valueGetters[column].getFieldOrNull(physicalRow);
+ }
+ result.put(fieldName, value);
+ }
+ if (context.overflowPosition < physicalRow.getFieldCount()
+ && !physicalRow.isNullAt(context.overflowPosition)) {
+ InternalMap overflow = physicalRow.getMap(context.overflowPosition);
+ InternalArray keys = overflow.keyArray();
+ InternalArray values = overflow.valueArray();
+ for (int i = 0; i < overflow.size(); i++) {
+ int fieldId = keys.getInt(i);
+ BinaryString fieldName = context.nameById.get(fieldId);
+ if (fieldName != null) {
+ result.put(fieldName, context.overflowValueGetter.getElementOrNull(values, i));
+ }
+ }
+ }
+ return new GenericMap(result);
+ }
+
+ private class MapSharedShreddingBatchAssembler implements ShreddingBatchAssembler {
+
+ @Override
+ public VectorizedColumnBatch assemble(VectorizedColumnBatch physicalBatch) {
+ ColumnVector[] logicalVectors = new ColumnVector[logicalType.getFieldCount()];
+ for (int i = 0; i < logicalVectors.length; i++) {
+ SharedShreddingContext context = contextByFieldIndex.get(i);
+ if (context == null) {
+ logicalVectors[i] = physicalBatch.columns[i];
+ } else {
+ logicalVectors[i] =
+ new SharedShreddingMapColumnVector(
+ (RowColumnVector) physicalBatch.columns[i], context);
+ }
+ }
+ return physicalBatch.copy(logicalVectors);
+ }
+ }
+
+ private static class SharedShreddingMapColumnVector implements MapColumnVector {
+
+ private final RowColumnVector physicalVector;
+ private final SharedShreddingContext context;
+
+ private SharedShreddingMapColumnVector(
+ RowColumnVector physicalVector, SharedShreddingContext context) {
+ this.physicalVector = physicalVector;
+ this.context = context;
+ }
+
+ @Override
+ public InternalMap getMap(int i) {
+ if (physicalVector.isNullAt(i)) {
+ return null;
+ }
+ // TODO (xinyu.lxy): Move this per-row map assemble path to a real batch-level
+ // physical-to-logical conversion.
+ return rebuildLogicalMap(physicalVector.getRow(i), context);
+ }
+
+ @Override
+ public boolean isNullAt(int i) {
+ return physicalVector.isNullAt(i);
+ }
+
+ @Override
+ public int getCapacity() {
+ return physicalVector.getCapacity();
+ }
+
+ @Override
+ public ColumnVector[] getChildren() {
Review Comment:
Thanks for the suggestion! Updated `MapSharedShreddingReadPlan` to materialize MAPs columnarly with key/value child vectors. Also added an e2e append read/write test covering shared-shredding MAP values with the supported nested types.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Support append read/write for MAP shared-shredding [paimon]
Posted by "lxy-9602 (via GitHub)" <gi...@apache.org>.
lxy-9602 commented on code in PR #8392:
URL: https://github.com/apache/paimon/pull/8392#discussion_r3519815438
##########
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:
Thanks for the review, addressed now.
##########
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:
Thanks for the review, fixed now.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Support append read/write for MAP shared-shredding [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Support append read/write for MAP shared-shredding [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #8392:
URL: https://github.com/apache/paimon/pull/8392#discussion_r3520155090
##########
paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlan.java:
##########
@@ -0,0 +1,304 @@
+/*
+ * 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.data.shredding;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.InternalArray;
+import org.apache.paimon.data.InternalMap;
+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.MapColumnVector;
+import org.apache.paimon.data.columnar.RowColumnVector;
+import org.apache.paimon.data.columnar.RowToColumnConverter;
+import org.apache.paimon.data.columnar.VectorizedColumnBatch;
+import org.apache.paimon.data.columnar.heap.CastedMapColumnVector;
+import org.apache.paimon.data.columnar.heap.HeapMapVector;
+import org.apache.paimon.data.columnar.writable.WritableColumnVector;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.RowType;
+
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Read plan that rebuilds logical MAP values from shared-shredding physical ROW values. */
+public class MapSharedShreddingReadPlan implements ShreddingReadPlan {
+
+ private final RowType logicalType;
+ private final RowType physicalType;
+ private final Map<Integer, SharedShreddingContext> contextByFieldIndex;
+
+ public MapSharedShreddingReadPlan(
+ RowType logicalType, Map<String, MapSharedShreddingFieldMeta> fieldMetas) {
+ this.logicalType = logicalType;
+ this.physicalType = MapSharedShreddingUtils.buildPhysicalReadType(logicalType, fieldMetas);
+ this.contextByFieldIndex = createContexts(logicalType, fieldMetas);
+ }
+
+ @Override
+ public RowType logicalRowType() {
+ return logicalType;
+ }
+
+ @Override
+ public RowType physicalRowType() {
+ return physicalType;
+ }
+
+ @Override
+ public boolean isIdentity() {
+ return contextByFieldIndex.isEmpty();
+ }
+
+ @Override
+ public ShreddingBatchAssembler batchAssembler() {
+ return new MapSharedShreddingBatchAssembler();
+ }
+
+ private static Map<Integer, SharedShreddingContext> createContexts(
+ RowType logicalType, Map<String, MapSharedShreddingFieldMeta> fieldMetas) {
+ Map<Integer, SharedShreddingContext> contexts = new LinkedHashMap<>();
+ for (int i = 0; i < logicalType.getFieldCount(); i++) {
+ DataField field = logicalType.getFields().get(i);
+ MapSharedShreddingFieldMeta fieldMeta = fieldMetas.get(field.name());
+ if (fieldMeta != null && field.type() instanceof MapType) {
+ contexts.put(i, new SharedShreddingContext(fieldMeta, field.type()));
+ }
+ }
+ return contexts;
+ }
+
+ private class MapSharedShreddingBatchAssembler implements ShreddingBatchAssembler {
+
+ @Override
+ public VectorizedColumnBatch assemble(VectorizedColumnBatch physicalBatch) {
+ ColumnVector[] logicalVectors = new ColumnVector[logicalType.getFieldCount()];
+ for (int i = 0; i < logicalVectors.length; i++) {
+ SharedShreddingContext context = contextByFieldIndex.get(i);
+ if (context == null) {
+ logicalVectors[i] = physicalBatch.columns[i];
+ } else {
+ logicalVectors[i] =
+ materializeLogicalMapVector(
+ (RowColumnVector) physicalBatch.columns[i],
+ context,
+ physicalBatch.getNumRows());
+ }
+ }
+ return physicalBatch.copy(logicalVectors);
+ }
+ }
+
+ private static CastedMapColumnVector materializeLogicalMapVector(
+ RowColumnVector physicalVector, SharedShreddingContext context, int rowCount) {
+ ColumnVector[] physicalChildren = physicalChildren(physicalVector);
+ int totalElements =
+ countLogicalMapElements(physicalVector, physicalChildren, context, rowCount);
+
+ WritableColumnVector keyVector =
+ ColumnVectorUtils.createWritableColumnVector(
+ totalElements, context.mapType.getKeyType());
+ WritableColumnVector valueVector =
+ ColumnVectorUtils.createWritableColumnVector(
+ totalElements, context.mapType.getValueType());
+ HeapMapVector mapVector = new HeapMapVector(rowCount, keyVector, valueVector);
+
+ int offset = 0;
+ for (int row = 0; row < rowCount; row++) {
+ if (physicalVector.isNullAt(row)) {
+ mapVector.setNullAt(row);
+ mapVector.putOffsetLength(row, offset, 0);
+ continue;
+ }
+
+ int elementCount =
+ appendLogicalMapElements(
+ physicalChildren, row, context, keyVector, valueVector);
+ mapVector.putOffsetLength(row, offset, elementCount);
+ offset += elementCount;
+ }
+ mapVector.addElementsAppended(rowCount);
+
+ return new CastedMapColumnVector(
+ mapVector,
+ ColumnVectorUtils.createReadableColumnVectors(
+ Arrays.asList(context.mapType.getKeyType(), context.mapType.getValueType()),
+ new WritableColumnVector[] {keyVector, valueVector}));
+ }
+
+ private static ColumnVector[] physicalChildren(RowColumnVector physicalVector) {
+ ColumnVector[] physicalChildren = physicalVector.getChildren();
+ if (physicalChildren != null) {
+ return physicalChildren;
+ }
+ return physicalVector.getBatch().columns;
+ }
+
+ private static int countLogicalMapElements(
+ RowColumnVector physicalVector,
+ ColumnVector[] physicalChildren,
+ SharedShreddingContext context,
+ int rowCount) {
+ int totalElements = 0;
+ for (int row = 0; row < rowCount; row++) {
+ if (!physicalVector.isNullAt(row)) {
+ totalElements += countLogicalMapElements(physicalChildren, row, context);
+ }
+ }
+ return totalElements;
+ }
+
+ private static int countLogicalMapElements(
+ ColumnVector[] physicalChildren, int row, SharedShreddingContext context) {
+ int count = 0;
+ InternalArray fieldMapping = fieldMapping(physicalChildren, row, context);
+ for (int column = 0; column < context.numColumns; column++) {
+ if (logicalFieldName(fieldMapping, column, context) != null) {
+ count++;
+ }
+ }
+
+ InternalMap overflow = overflowMap(physicalChildren, row, context);
+ if (overflow != null) {
+ InternalArray keys = overflow.keyArray();
+ for (int i = 0; i < overflow.size(); i++) {
+ if (context.nameById.containsKey(keys.getInt(i))) {
+ count++;
+ }
+ }
+ }
+ return count;
+ }
+
+ private static int appendLogicalMapElements(
+ ColumnVector[] physicalChildren,
+ int row,
+ SharedShreddingContext context,
+ WritableColumnVector keyVector,
+ WritableColumnVector valueVector) {
+ int count = 0;
+ InternalArray fieldMapping = fieldMapping(physicalChildren, row, context);
+ for (int column = 0; column < context.numColumns; column++) {
+ BinaryString fieldName = logicalFieldName(fieldMapping, column, context);
+ if (fieldName == null) {
+ continue;
+ }
+
+ context.keyConverter.append(fieldName, keyVector);
+ ColumnVector valueColumn = physicalChildren[column + 1];
+ appendValue(context, valueColumn, row, valueVector);
+ count++;
+ }
+
+ InternalMap overflow = overflowMap(physicalChildren, row, context);
+ if (overflow != null) {
+ InternalArray keys = overflow.keyArray();
+ InternalArray values = overflow.valueArray();
+ for (int i = 0; i < overflow.size(); i++) {
+ BinaryString fieldName = context.nameById.get(keys.getInt(i));
+ if (fieldName != null) {
+ context.keyConverter.append(fieldName, keyVector);
+ appendValue(context, values, i, valueVector);
+ count++;
+ }
+ }
+ }
+ return count;
+ }
+
+ private static void appendValue(
+ SharedShreddingContext context,
+ ColumnVector source,
+ int row,
+ WritableColumnVector valueVector) {
+ if (source.isNullAt(row)) {
+ valueVector.appendNull();
+ } else {
+ context.valueConverter.append(source, row, valueVector);
+ }
+ }
+
+ private static void appendValue(
+ SharedShreddingContext context,
+ InternalArray source,
+ int pos,
+ WritableColumnVector valueVector) {
+ if (source.isNullAt(pos)) {
+ valueVector.appendNull();
+ } else {
+ context.valueConverter.append(source, pos, valueVector);
+ }
+ }
+
+ private static InternalArray fieldMapping(
+ ColumnVector[] physicalChildren, int row, SharedShreddingContext context) {
+ InternalArray fieldMapping = ((ArrayColumnVector) physicalChildren[0]).getArray(row);
+ checkArgument(
+ fieldMapping.size() == context.numColumns,
+ "Shared-shredding field mapping size %s does not match metadata num columns %s.",
+ fieldMapping.size(),
+ context.numColumns);
+ return fieldMapping;
+ }
+
+ private static BinaryString logicalFieldName(
+ InternalArray fieldMapping, int column, SharedShreddingContext context) {
+ int fieldId = fieldMapping.isNullAt(column) ? -1 : fieldMapping.getInt(column);
+ return fieldId < 0 ? null : context.nameById.get(fieldId);
+ }
+
+ private static InternalMap overflowMap(
+ ColumnVector[] physicalChildren, int row, SharedShreddingContext context) {
+ if (context.overflowPosition >= physicalChildren.length) {
+ return null;
+ }
+
+ ColumnVector overflowVector = physicalChildren[context.overflowPosition];
+ return overflowVector.isNullAt(row) ? null : ((MapColumnVector) overflowVector).getMap(row);
+ }
+
+ private static class SharedShreddingContext {
+
+ private final MapType mapType;
+ private final Map<Integer, BinaryString> nameById;
+ private final RowToColumnConverter.ElementConverter keyConverter;
+ private final RowToColumnConverter.ElementConverter valueConverter;
+ private final int numColumns;
+ private final int overflowPosition;
+
+ private SharedShreddingContext(MapSharedShreddingFieldMeta fieldMeta, DataType fieldType) {
+ this.mapType = (MapType) fieldType;
+ this.nameById = new LinkedHashMap<>();
+ for (Map.Entry<String, Integer> entry : fieldMeta.nameToId().entrySet()) {
+ this.nameById.put(entry.getValue(), BinaryString.fromString(entry.getKey()));
+ }
+ this.keyConverter =
+ RowToColumnConverter.createElementConverter(this.mapType.getKeyType());
+ this.valueConverter =
+ RowToColumnConverter.createElementConverter(this.mapType.getValueType());
Review Comment:
This path allows `MAP<STRING, CHAR>` values to reach `RowToColumnConverter`. The shared converter currently handles `CharType` with `row.getByte(...)` and `WritableByteVector`, while `ColumnVectorUtils.createWritableColumnVector` creates a `HeapBytesVector` for `CHAR` (same as `VARCHAR`). As a result, materializing a shared-shredding MAP whose value type contains `CHAR` will fail or use the wrong accessor. Please either fix `CharType` in `RowToColumnConverter` to use the string/bytes path and add a regression, or reject `CHAR` in the shared-shredding validation until it is supported.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Support append read/write for MAP shared-shredding [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #8392:
URL: https://github.com/apache/paimon/pull/8392#discussion_r3520159016
##########
paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java:
##########
@@ -658,6 +663,109 @@ private static void validateMapStorageLayout(TableSchema schema, CoreOptions opt
}
options.mapSharedShreddingMaxColumns(fieldName);
}
+
+ if (hasSharedShredding) {
+ validateMapSharedShreddingFileFormats(options);
+ validateMapSharedShreddingCompressions(options);
+ validateNoVariantOrBlobWithMapSharedShredding(schema);
Review Comment:
This validation still allows shared-shredding maps whose value type contains `MULTISET`, for example `MAP<STRING, MULTISET<INT>>`. The write converter can carry that value as an `InternalMap`, but read materialization later calls `RowToColumnConverter.createElementConverter(valueType)`, and `RowToColumnConverter.visit(MultisetType)` currently throws `UnsupportedOperationException`. That means such a table can be created and written but fails when the shared-shredding map is read back. Please either implement multiset conversion in the row-to-column path or reject `MultisetType` recursively here with the other unsupported types.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org