Skip to content

[core][spark] Support switching MAP storage layouts #8941

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JingsongLi merged 1 commit into from
Jul 30, 2026
Merged

[core][spark] Support switching MAP storage layouts #8941

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
118 changes: 103 additions & 15 deletions paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlan.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,14 +45,17 @@

import static org.apache.paimon.utils.Preconditions.checkArgument;

/** Read plan that rebuilds logical MAP values from shared-shredding physical ROW values. */
/**
* Read plan that rebuilds logical MAP values from shared-shredding physical ROW values and projects
* selected keys from legacy default-layout MAP values.
*/
public class MapSharedShreddingReadPlan implements ShreddingReadPlan {

private static final int FIELD_MAPPING_POSITION = 0;

private final RowType logicalType;
private final RowType physicalType;
private final Map<Integer, SharedShreddingContext> contextByFieldIndex;
private final Map<Integer, MapReadContext> contextByFieldIndex;

public MapSharedShreddingReadPlan(
RowType logicalType, Map<String, MapSharedShreddingFieldMeta> fieldMetas) {
Expand Down Expand Up @@ -81,15 +84,21 @@ public ShreddingBatchAssembler batchAssembler() {
return new MapSharedShreddingBatchAssembler();
}

private static Map<Integer, SharedShreddingContext> createContexts(
private static Map<Integer, MapReadContext> createContexts(
RowType logicalType,
RowType physicalType,
Map<String, MapSharedShreddingFieldMeta> fieldMetas) {
Map<Integer, SharedShreddingContext> contexts = new LinkedHashMap<>();
Map<Integer, MapReadContext> 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) {
if (MapSelectedKeysMetadataUtils.isMapSelectedKeysField(field)) {
contexts.put(
i,
new NormalMapSelectedKeysContext(
(RowType) field.type(), field.description()));
}
continue;
}

Expand Down Expand Up @@ -117,14 +126,13 @@ private class MapSharedShreddingBatchAssembler implements ShreddingBatchAssemble
public VectorizedColumnBatch assemble(VectorizedColumnBatch physicalBatch) {
ColumnVector[] logicalVectors = new ColumnVector[logicalType.getFieldCount()];
for (int i = 0; i < logicalVectors.length; i++) {
SharedShreddingContext context = contextByFieldIndex.get(i);
MapReadContext context = contextByFieldIndex.get(i);
if (context == null) {
logicalVectors[i] = physicalBatch.columns[i];
} else {
logicalVectors[i] =
context.materialize(
(RowColumnVector) physicalBatch.columns[i],
physicalBatch.getNumRows());
physicalBatch.columns[i], physicalBatch.getNumRows());
}
}
return physicalBatch.copy(logicalVectors);
Expand Down Expand Up @@ -208,6 +216,37 @@ private static ColumnVector materializeSelectedKeysRowVector(
return ColumnVectorUtils.createReadableColumnVector(selectedKeysType, rowVector);
}

private static ColumnVector materializeSelectedKeysRowVector(
MapColumnVector physicalVector, NormalMapSelectedKeysContext context, int rowCount) {
RowType selectedKeysType = context.selectedKeysType;
WritableColumnVector[] selectedKeyVectors =
new WritableColumnVector[selectedKeysType.getFieldCount()];
for (int i = 0; i < selectedKeyVectors.length; i++) {
selectedKeyVectors[i] =
ColumnVectorUtils.createWritableColumnVector(
rowCount, selectedKeysType.getTypeAt(i));
}
HeapRowVector rowVector = new HeapRowVector(rowCount, selectedKeyVectors);

for (int row = 0; row < rowCount; row++) {
if (physicalVector.isNullAt(row)) {
rowVector.appendNull();
continue;
}

InternalMap map = physicalVector.getMap(row);
InternalArray keys = map.keyArray();
InternalArray values = map.valueArray();
for (int ordinal = 0; ordinal < selectedKeyVectors.length; ordinal++) {
appendSelectedKeyValue(
keys, values, map.size(), context, ordinal, selectedKeyVectors[ordinal]);
}
rowVector.appendRow();
}

return ColumnVectorUtils.createReadableColumnVector(selectedKeysType, rowVector);
}

private static ColumnVector[] physicalChildren(RowColumnVector physicalVector) {
ColumnVector[] physicalChildren = physicalVector.getChildren();
if (physicalChildren != null) {
Expand Down Expand Up @@ -329,6 +368,23 @@ private static void appendSelectedKeyValue(
valueVector.appendNull();
}

private static void appendSelectedKeyValue(
InternalArray keys,
InternalArray values,
int size,
NormalMapSelectedKeysContext context,
int ordinal,
WritableColumnVector valueVector) {
BinaryString selectedKey = context.selectedKeys[ordinal];
for (int i = 0; i < size; i++) {
if (!keys.isNullAt(i) && selectedKey.equals(keys.getString(i))) {
appendValue(context.selectedValueConverters[ordinal], values, i, valueVector);
return;
}
}
valueVector.appendNull();
}

private static void appendValue(
RowToColumnConverter.ElementConverter converter,
ColumnVector source,
Expand Down Expand Up @@ -401,9 +457,9 @@ private static InternalMap overflowMap(
return overflowVector.isNullAt(row) ? null : ((MapColumnVector) overflowVector).getMap(row);
}

private interface SharedShreddingContext {
private interface MapReadContext {

ColumnVector materialize(RowColumnVector physicalVector, int rowCount);
ColumnVector materialize(ColumnVector physicalVector, int rowCount);
}

private static class SharedPhysicalContext {
Expand Down Expand Up @@ -475,7 +531,7 @@ private static int physicalColumnIndex(String fieldName) {
}
}

private static class FullMapContext implements SharedShreddingContext {
private static class FullMapContext implements MapReadContext {

private final SharedPhysicalContext physical;
private final MapType mapType;
Expand All @@ -491,12 +547,12 @@ private FullMapContext(SharedPhysicalContext physical, MapType mapType) {
}

@Override
public ColumnVector materialize(RowColumnVector physicalVector, int rowCount) {
return materializeLogicalMapVector(physicalVector, this, rowCount);
public ColumnVector materialize(ColumnVector physicalVector, int rowCount) {
return materializeLogicalMapVector((RowColumnVector) physicalVector, this, rowCount);
}
}

private static class SelectedKeysContext implements SharedShreddingContext {
private static class SelectedKeysContext implements MapReadContext {

private final SharedPhysicalContext physical;
private final RowType selectedKeysType;
Expand Down Expand Up @@ -537,8 +593,9 @@ private SelectedKeysContext(
}

@Override
public ColumnVector materialize(RowColumnVector physicalVector, int rowCount) {
return materializeSelectedKeysRowVector(physicalVector, this, rowCount);
public ColumnVector materialize(ColumnVector physicalVector, int rowCount) {
return materializeSelectedKeysRowVector(
(RowColumnVector) physicalVector, this, rowCount);
}

private static int[] candidateColumns(MapSharedShreddingFieldMeta fieldMeta, int fieldId) {
Expand All @@ -553,4 +610,35 @@ private static int[] candidateColumns(MapSharedShreddingFieldMeta fieldMeta, int
return result;
}
}

private static class NormalMapSelectedKeysContext implements MapReadContext {

private final RowType selectedKeysType;
private final BinaryString[] selectedKeys;
private final RowToColumnConverter.ElementConverter[] selectedValueConverters;

private NormalMapSelectedKeysContext(
RowType selectedKeysType, String selectedKeysMetadata) {
this.selectedKeysType = selectedKeysType;
List<String> keys = MapSelectedKeysMetadataUtils.selectedKeys(selectedKeysMetadata);
checkArgument(
keys.size() == selectedKeysType.getFieldCount(),
"Selected-key metadata size %s does not match selected ROW field count %s.",
keys.size(),
selectedKeysType.getFieldCount());
this.selectedKeys = new BinaryString[keys.size()];
this.selectedValueConverters = new RowToColumnConverter.ElementConverter[keys.size()];
for (int i = 0; i < keys.size(); i++) {
this.selectedKeys[i] = BinaryString.fromString(keys.get(i));
this.selectedValueConverters[i] =
RowToColumnConverter.createElementConverter(selectedKeysType.getTypeAt(i));
}
}

@Override
public ColumnVector materialize(ColumnVector physicalVector, int rowCount) {
return materializeSelectedKeysRowVector(
(MapColumnVector) physicalVector, this, rowCount);
}
}
}
5 changes: 3 additions & 2 deletions ...mon/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlanFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@
import java.util.LinkedHashMap;
import java.util.Map;

/** Creates per-file shared-shredding MAP read plans. */
/** Creates per-file MAP read plans for shared-shredding and selected-key fallback reads. */
public class MapSharedShreddingReadPlanFactory implements ShreddingReadPlanFactory {

private final RowType logicalRowType;
Expand All @@ -46,7 +46,8 @@ public boolean shouldCreateReadPlan(
Map<String, Map<String, String>> fieldMetadata, @Nullable Object fileSchema) {
for (DataField field : logicalRowType.getFields()) {
Map<String, String> metadata = fieldMetadata.get(field.name());
if (MapSharedShreddingUtils.hasShreddingMetadata(metadata)) {
if (MapSharedShreddingUtils.hasShreddingMetadata(metadata)
|| MapSelectedKeysMetadataUtils.isMapSelectedKeysField(field)) {
return true;
}
}
Expand Down
21 changes: 14 additions & 7 deletions paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypeRoot;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.IntType;
import org.apache.paimon.types.MapType;
import org.apache.paimon.types.RowType;
Expand Down Expand Up @@ -108,24 +109,30 @@ public static RowType logicalToPhysicalSchema(
public static RowType buildPhysicalReadType(
RowType logicalReadType,
Map<String, MapSharedShreddingFieldMeta> sharedShreddingFieldMetas) {
if (sharedShreddingFieldMetas.isEmpty()) {
return logicalReadType;
}

List<DataField> physicalReadFields = new ArrayList<>();
boolean converted = false;
for (DataField logicalReadField : logicalReadType.getFields()) {
MapSharedShreddingFieldMeta fieldMeta =
sharedShreddingFieldMetas.get(logicalReadField.name());
if (fieldMeta == null) {
physicalReadFields.add(logicalReadField);
if (MapSelectedKeysMetadataUtils.isMapSelectedKeysField(logicalReadField)) {
// Read a legacy/default-layout file while selected-key pushdown is active.
DataType valueType = selectedKeysValueType((RowType) logicalReadField.type());
DataType physicalType =
DataTypes.MAP(DataTypes.STRING().notNull(), valueType)
.copy(logicalReadField.type().isNullable());
physicalReadFields.add(logicalReadField.newType(physicalType));
converted = true;
} else {
physicalReadFields.add(logicalReadField);
}
continue;
}

DataType valueType;
DataType physicalType;
if (MapSelectedKeysMetadataUtils.isMapSelectedKeysField(logicalReadField)) {
// recall partial key with shared shredding pushdown
// Read only selected keys from a shared-shredding ROW.
valueType = selectedKeysValueType((RowType) logicalReadField.type());
physicalType =
buildSpecificPhysicalStructType(
Expand All @@ -134,7 +141,7 @@ public static RowType buildPhysicalReadType(
selectedKeysIncludeOverflow(logicalReadField, fieldMeta))
.copy(logicalReadField.type().isNullable());
} else {
// recall whole field without shared shredding pushdown
// Rebuild the whole MAP from a shared-shredding ROW.
valueType = ((MapType) logicalReadField.type()).getValueType();
physicalType =
buildPhysicalStructType(valueType, fieldMeta.numColumns())
Expand Down
56 changes: 54 additions & 2 deletions ...common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlanTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,11 +28,13 @@
import org.apache.paimon.data.columnar.RowColumnVector;
import org.apache.paimon.data.columnar.VectorizedColumnBatch;
import org.apache.paimon.data.columnar.heap.HeapArrayVector;
import org.apache.paimon.data.columnar.heap.HeapBytesVector;
import org.apache.paimon.data.columnar.heap.HeapIntVector;
import org.apache.paimon.data.columnar.heap.HeapLongVector;
import org.apache.paimon.data.columnar.heap.HeapMapVector;
import org.apache.paimon.data.columnar.heap.HeapRowVector;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.MapType;
import org.apache.paimon.types.RowType;

import org.junit.jupiter.api.Test;
Expand Down Expand Up @@ -185,6 +187,48 @@ void testReadSelectedKeysFromPrunedPhysicalColumns() {
assertThat(selectedKeys.isNullAt(2)).isTrue();
}

@Test
void testReadSelectedKeysFromNormalMap() {
MapSharedShreddingReadPlan readPlan =
new MapSharedShreddingReadPlan(selectedKeysLogicalType(), Collections.emptyMap());
assertThat(readPlan.physicalRowType().getTypeAt(0)).isInstanceOf(MapType.class);

HeapBytesVector keys = new HeapBytesVector(3);
appendString(keys, "key2");
appendString(keys, "key1");
appendString(keys, "key1");
HeapLongVector values = new HeapLongVector(3);
values.appendLong(20L);
values.appendLong(10L);
values.appendNull();
HeapMapVector normalMaps = new HeapMapVector(3, keys, values);
normalMaps.putOffsetLength(0, 0, 2);
normalMaps.putOffsetLength(1, 2, 1);
normalMaps.setNullAt(2);
normalMaps.putOffsetLength(2, 3, 0);

RowColumnVector selectedKeysVector = assembleSelectedKeysVector(readPlan, normalMaps, 3);
InternalRow first = selectedKeysVector.getRow(0);
assertThat(first.getLong(0)).isEqualTo(10L);
assertThat(first.getLong(1)).isEqualTo(20L);
assertThat(first.isNullAt(2)).isTrue();

InternalRow second = selectedKeysVector.getRow(1);
assertThat(second.isNullAt(0)).isTrue();
assertThat(second.isNullAt(1)).isTrue();
assertThat(second.isNullAt(2)).isTrue();
assertThat(selectedKeysVector.isNullAt(2)).isTrue();
}

@Test
void testFactoryCreatesSelectedKeysPlanForNormalMap() {
MapSharedShreddingReadPlanFactory factory =
new MapSharedShreddingReadPlanFactory(selectedKeysLogicalType());

assertThat(factory.shouldCreateReadPlan(Collections.emptyMap(), null)).isTrue();
assertThat(factory.createReadPlan(Collections.emptyMap(), null).isIdentity()).isFalse();
}

private static InternalMap readMap(
MapSharedShreddingFieldMeta fieldMeta, HeapRowVector physicalMap) {
return assembleMapVector(fieldMeta, physicalMap).getMap(0);
Expand All @@ -205,11 +249,14 @@ private static MapColumnVector assembleMapVector(

private static RowColumnVector assembleSelectedKeysVector(
MapSharedShreddingReadPlan readPlan, HeapRowVector physicalMap) {
assertThat(readPlan.physicalRowType().getTypeAt(0)).isInstanceOf(RowType.class);
return assembleSelectedKeysVector(readPlan, physicalMap, 1);
}

private static RowColumnVector assembleSelectedKeysVector(
MapSharedShreddingReadPlan readPlan, ColumnVector physicalMap, int rowCount) {
VectorizedColumnBatch physicalBatch =
new VectorizedColumnBatch(new ColumnVector[] {physicalMap});
physicalBatch.setNumRows(1);
physicalBatch.setNumRows(rowCount);
VectorizedColumnBatch logicalBatch = readPlan.batchAssembler().assemble(physicalBatch);
return (RowColumnVector) logicalBatch.columns[0];
}
Expand Down Expand Up @@ -282,6 +329,11 @@ private static HeapLongVector longVector(Long value) {
return vector;
}

private static void appendString(HeapBytesVector vector, String value) {
byte[] bytes = BinaryString.fromString(value).toBytes();
vector.appendByteArray(bytes, 0, bytes.length);
}

private static HeapMapVector overflowMap(int key, Long value) {
HeapIntVector keys = new HeapIntVector(1);
keys.appendInt(key);
Expand Down
Loading
Loading