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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
