This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new bbe084067c [core][spark] Support switching MAP storage layouts (#8941)
bbe084067c is described below
commit bbe084067c5e135c226644c4b53f9e530dd77aeb
Author: lszskye <[email protected]>
AuthorDate: Thu Jul 30 08:01:06 2026 -0700
[core][spark] Support switching MAP storage layouts (#8941)
---
.../data/shredding/MapSharedShreddingReadPlan.java | 118 ++++++++--
.../MapSharedShreddingReadPlanFactory.java | 5 +-
.../data/shredding/MapSharedShreddingUtils.java | 21 +-
.../shredding/MapSharedShreddingReadPlanTest.java | 56 ++++-
.../shredding/MapSharedShreddingUtilsTest.java | 105 +++++++++
.../org/apache/paimon/schema/SchemaManager.java | 46 ----
.../apache/paimon/schema/SchemaManagerTest.java | 67 +++---
.../paimon/table/MapSharedShreddingTableTest.java | 241 ++++++++++++++++++---
...MapSelectedKeysSharedShreddingE2ETestBase.scala | 93 ++++++++
9 files changed, 621 insertions(+), 131 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlan.java
b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlan.java
index f09c58d685..6cc8dae4e3 100644
---
a/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlan.java
+++
b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlan.java
@@ -45,14 +45,17 @@ 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. */
+/**
+ * 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) {
@@ -81,15 +84,21 @@ public class MapSharedShreddingReadPlan implements
ShreddingReadPlan {
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;
}
@@ -117,14 +126,13 @@ public class MapSharedShreddingReadPlan implements
ShreddingReadPlan {
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);
@@ -208,6 +216,37 @@ public class MapSharedShreddingReadPlan implements
ShreddingReadPlan {
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) {
@@ -329,6 +368,23 @@ public class MapSharedShreddingReadPlan implements
ShreddingReadPlan {
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,
@@ -401,9 +457,9 @@ public class MapSharedShreddingReadPlan implements
ShreddingReadPlan {
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 {
@@ -475,7 +531,7 @@ public class MapSharedShreddingReadPlan implements
ShreddingReadPlan {
}
}
- private static class FullMapContext implements SharedShreddingContext {
+ private static class FullMapContext implements MapReadContext {
private final SharedPhysicalContext physical;
private final MapType mapType;
@@ -491,12 +547,12 @@ public class MapSharedShreddingReadPlan implements
ShreddingReadPlan {
}
@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;
@@ -537,8 +593,9 @@ public class MapSharedShreddingReadPlan implements
ShreddingReadPlan {
}
@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) {
@@ -553,4 +610,35 @@ public class MapSharedShreddingReadPlan implements
ShreddingReadPlan {
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);
+ }
+ }
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlanFactory.java
b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlanFactory.java
index 36ee35b8f7..aeedf4057e 100644
---
a/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlanFactory.java
+++
b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlanFactory.java
@@ -27,7 +27,7 @@ import javax.annotation.Nullable;
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;
@@ -46,7 +46,8 @@ public class MapSharedShreddingReadPlanFactory implements
ShreddingReadPlanFacto
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;
}
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingUtils.java
b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingUtils.java
index 6fc2d13b5a..caf4bbc887 100644
---
a/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingUtils.java
+++
b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingUtils.java
@@ -28,6 +28,7 @@ import org.apache.paimon.types.ArrayType;
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;
@@ -108,24 +109,30 @@ public class MapSharedShreddingUtils {
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(
@@ -134,7 +141,7 @@ public class MapSharedShreddingUtils {
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())
diff --git
a/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlanTest.java
b/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlanTest.java
index 6b781be417..b376a11d95 100644
---
a/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlanTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlanTest.java
@@ -28,11 +28,13 @@ 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.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;
@@ -185,6 +187,48 @@ class MapSharedShreddingReadPlanTest {
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);
@@ -205,11 +249,14 @@ class MapSharedShreddingReadPlanTest {
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];
}
@@ -282,6 +329,11 @@ class MapSharedShreddingReadPlanTest {
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);
diff --git
a/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingUtilsTest.java
b/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingUtilsTest.java
index dd628bc5a3..12acb8b099 100644
---
a/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingUtilsTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingUtilsTest.java
@@ -27,6 +27,7 @@ import org.apache.paimon.types.RowType;
import org.junit.jupiter.api.Test;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -187,6 +188,110 @@ class MapSharedShreddingUtilsTest {
.isEqualTo(logical);
}
+ @Test
+ void testBuildPhysicalReadTypeWithoutSharedShreddingMetadata() {
+ RowType fullMapReadType =
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "id", DataTypes.INT()),
+ DataTypes.FIELD(
+ 1,
+ "metrics",
+ DataTypes.MAP(
+ DataTypes.STRING().notNull(),
+ DataTypes.BIGINT().notNull())));
+ assertThat(
+ MapSharedShreddingUtils.buildPhysicalReadType(
+ fullMapReadType, Collections.emptyMap()))
+ .isSameAs(fullMapReadType);
+
+ DataField selectedKeysField =
+ MapSelectedKeysMetadataUtils.withSelectedKeys(
+ fullMapReadType.getField("metrics"),
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "0",
DataTypes.BIGINT().notNull()),
+ DataTypes.FIELD(1, "1",
DataTypes.BIGINT().notNull()),
+ DataTypes.FIELD(2, "2",
DataTypes.BIGINT().notNull())),
+ Arrays.asList("key1", "key2", "missing"));
+ RowType selectedKeysReadType =
+ new RowType(
+ false, Arrays.asList(fullMapReadType.getField("id"),
selectedKeysField));
+
+ RowType physicalReadType =
+ MapSharedShreddingUtils.buildPhysicalReadType(
+ selectedKeysReadType, Collections.emptyMap());
+ assertThat(physicalReadType.isNullable()).isFalse();
+
assertThat(physicalReadType.getField("id")).isSameAs(fullMapReadType.getField("id"));
+ DataField physicalMapField = physicalReadType.getField("metrics");
+ assertThat(physicalMapField.id()).isEqualTo(selectedKeysField.id());
+
assertThat(physicalMapField.description()).isEqualTo(selectedKeysField.description());
+ assertThat(physicalMapField.type())
+ .isEqualTo(
+ DataTypes.MAP(DataTypes.STRING().notNull(),
DataTypes.BIGINT().notNull()));
+ }
+
+ @Test
+ void testBuildPhysicalReadTypeWithSharedShreddingMetadata() {
+ Map<String, Integer> nameToId = new TreeMap<>();
+ nameToId.put("key1", 0);
+ nameToId.put("key2", 1);
+ Map<Integer, List<Integer>> fieldToColumns = new TreeMap<>();
+ fieldToColumns.put(0, Collections.singletonList(2));
+ MapSharedShreddingFieldMeta fieldMeta =
+ new MapSharedShreddingFieldMeta(
+ nameToId,
+ fieldToColumns,
+ new TreeSet<Integer>(Collections.singletonList(1)),
+ 4,
+ 2);
+ Map<String, MapSharedShreddingFieldMeta> fieldMetas = new HashMap<>();
+ fieldMetas.put("metrics", fieldMeta);
+
+ DataField fullMapField =
+ DataTypes.FIELD(
+ 1,
+ "metrics",
+ DataTypes.MAP(DataTypes.STRING().notNull(),
DataTypes.BIGINT()).notNull());
+ RowType fullMapReadType =
+ new RowType(
+ false,
+ Arrays.asList(DataTypes.FIELD(0, "id",
DataTypes.INT()), fullMapField));
+ RowType fullPhysicalReadType =
+ MapSharedShreddingUtils.buildPhysicalReadType(fullMapReadType,
fieldMetas);
+ DataField fullPhysicalMapField =
fullPhysicalReadType.getField("metrics");
+ assertThat(fullPhysicalMapField.id()).isEqualTo(fullMapField.id());
+ assertThat(fullPhysicalMapField.type().isNullable()).isFalse();
+ assertThat(((RowType) fullPhysicalMapField.type()).getFieldNames())
+ .containsExactly(
+ "__field_mapping",
+ "__col_0",
+ "__col_1",
+ "__col_2",
+ "__col_3",
+ "__overflow");
+
+ DataField selectedKeysField =
+ MapSelectedKeysMetadataUtils.withSelectedKeys(
+ fullMapField,
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "0",
DataTypes.BIGINT()),
+ DataTypes.FIELD(1, "1",
DataTypes.BIGINT()),
+ DataTypes.FIELD(2, "2",
DataTypes.BIGINT()))
+ .notNull(),
+ Arrays.asList("key1", "key2", "missing"));
+ RowType selectedKeysReadType =
+ new RowType(
+ false, Arrays.asList(fullMapReadType.getField("id"),
selectedKeysField));
+ RowType selectedPhysicalReadType =
+
MapSharedShreddingUtils.buildPhysicalReadType(selectedKeysReadType, fieldMetas);
+ DataField selectedPhysicalMapField =
selectedPhysicalReadType.getField("metrics");
+
assertThat(selectedPhysicalMapField.id()).isEqualTo(selectedKeysField.id());
+ assertThat(selectedPhysicalMapField.description())
+ .isEqualTo(selectedKeysField.description());
+ assertThat(selectedPhysicalMapField.type().isNullable()).isFalse();
+ assertThat(((RowType) selectedPhysicalMapField.type()).getFieldNames())
+ .containsExactly("__field_mapping", "__col_2", "__overflow");
+ }
+
@Test
void testBuildSpecificPhysicalStructType() {
RowType physicalType =
diff --git
a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
index 5eadd040c7..4c238781b9 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
@@ -637,56 +637,10 @@ public class SchemaManager implements Serializable {
newSchema.primaryKeys(),
newSchema.options(),
newSchema.comment());
- checkMapStorageLayoutUnchanged(oldTableSchema, newTableSchema);
SchemaValidation.validateTableSchema(newTableSchema);
return newTableSchema;
}
- private static void checkMapStorageLayoutUnchanged(
- TableSchema oldTableSchema, TableSchema newTableSchema) {
- Map<Integer, CoreOptions.MapStorageLayout> oldLayouts =
- mapStorageLayoutByFieldId(oldTableSchema);
- Map<Integer, CoreOptions.MapStorageLayout> newLayouts =
- mapStorageLayoutByFieldId(newTableSchema);
- Map<Integer, String> oldNames = fieldNameById(oldTableSchema);
- Map<Integer, String> newNames = fieldNameById(newTableSchema);
-
- for (Map.Entry<Integer, CoreOptions.MapStorageLayout> oldLayout :
oldLayouts.entrySet()) {
- Integer fieldId = oldLayout.getKey();
- CoreOptions.MapStorageLayout newLayout = newLayouts.get(fieldId);
- if (newLayout == null || oldLayout.getValue() == newLayout) {
- continue;
- }
-
- throw new UnsupportedOperationException(
- String.format(
- "Cannot change map storage layout for field id %s
('%s' -> '%s') from '%s' to '%s'.",
- fieldId,
- oldNames.get(fieldId),
- newNames.get(fieldId),
- oldLayout.getValue(),
- newLayout));
- }
- }
-
- private static Map<Integer, CoreOptions.MapStorageLayout>
mapStorageLayoutByFieldId(
- TableSchema schema) {
- CoreOptions options = new CoreOptions(schema.options());
- Map<Integer, CoreOptions.MapStorageLayout> layouts = new HashMap<>();
- for (DataField field : schema.fields()) {
- layouts.put(field.id(), options.mapStorageLayout(field.name()));
- }
- return layouts;
- }
-
- private static Map<Integer, String> fieldNameById(TableSchema schema) {
- Map<Integer, String> names = new HashMap<>();
- for (DataField field : schema.fields()) {
- names.put(field.id(), field.name());
- }
- return names;
- }
-
// gets the rootType at the defined depth
// ex: ARRAY<MAP<STRING, ARRAY<INT>>>
// if we want to update ARRAY<INT> -> ARRAY<BIGINT>
diff --git
a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java
b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java
index 2b0f4f18bf..aed3a9a309 100644
--- a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java
@@ -172,41 +172,52 @@ public class SchemaManagerTest {
}
@Test
- public void testCannotChangeMapStorageLayoutForExistingField() throws
Exception {
+ public void testChangeMapStorageLayoutForExistingField() throws Exception {
retryArtificialException(() ->
manager.createTable(mapStorageLayoutSchema("default")));
- assertThatThrownBy(
- () ->
- retryArtificialException(
- () ->
- manager.commitChanges(
- SchemaChange.setOption(
-
"fields.metrics.map.storage-layout",
-
"shared-shredding"))))
- .isInstanceOf(UnsupportedOperationException.class)
- .hasMessageContaining(
- "Cannot change map storage layout for field id 1
('metrics' -> 'metrics') from 'default' to 'shared-shredding'.");
+ retryArtificialException(
+ () ->
+ manager.commitChanges(
+ SchemaChange.setOption(
+ "fields.metrics.map.storage-layout",
"shared-shredding")));
+ Optional<TableSchema> sharedShredding = retryArtificialException(() ->
manager.latest());
+ assertThat(sharedShredding).isPresent();
+ assertThat(sharedShredding.get().options())
+ .containsEntry("fields.metrics.map.storage-layout",
"shared-shredding")
+
.containsEntry("fields.metrics.map.shared-shredding.max-columns", "2");
+
+ retryArtificialException(
+ () ->
+ manager.commitChanges(
+ SchemaChange.setOption(
+ "fields.metrics.map.storage-layout",
"default")));
+ Optional<TableSchema> defaultLayout = retryArtificialException(() ->
manager.latest());
+ assertThat(defaultLayout).isPresent();
+ assertThat(defaultLayout.get().options())
+ .containsEntry("fields.metrics.map.storage-layout", "default")
+
.containsEntry("fields.metrics.map.shared-shredding.max-columns", "2");
}
@Test
- public void testCannotChangeMapStorageLayoutByRenameColumn() throws
Exception {
+ public void testChangeMapStorageLayoutByRenameColumn() throws Exception {
retryArtificialException(() ->
manager.createTable(mapStorageLayoutSchema(null)));
- assertThatThrownBy(
- () ->
- retryArtificialException(
- () ->
- manager.commitChanges(
- Arrays.asList(
-
SchemaChange.renameColumn(
-
"metrics",
-
"renamed_metrics"),
-
SchemaChange.setOption(
-
"fields.renamed_metrics.map.storage-layout",
-
"shared-shredding")))))
- .isInstanceOf(UnsupportedOperationException.class)
- .hasMessageContaining(
- "Cannot change map storage layout for field id 1
('metrics' -> 'renamed_metrics') from 'default' to 'shared-shredding'.");
+ retryArtificialException(
+ () ->
+ manager.commitChanges(
+ Arrays.asList(
+ SchemaChange.renameColumn("metrics",
"renamed_metrics"),
+ SchemaChange.setOption(
+
"fields.renamed_metrics.map.storage-layout",
+ "shared-shredding"))));
+
+ Optional<TableSchema> latest = retryArtificialException(() ->
manager.latest());
+ assertThat(latest).isPresent();
+ assertThat(latest.get().fields().get(1).id()).isEqualTo(1);
+
assertThat(latest.get().fields().get(1).name()).isEqualTo("renamed_metrics");
+ assertThat(latest.get().options())
+ .doesNotContainKey("fields.metrics.map.storage-layout")
+ .containsEntry("fields.renamed_metrics.map.storage-layout",
"shared-shredding");
}
@ParameterizedTest
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/MapSharedShreddingTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/MapSharedShreddingTableTest.java
index c300bb4a70..2a7adce1c4 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/MapSharedShreddingTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/MapSharedShreddingTableTest.java
@@ -76,7 +76,6 @@ import java.util.stream.Stream;
import static
org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Table-level tests for MAP shared-shredding. */
public class MapSharedShreddingTableTest extends TableTestBase {
@@ -248,6 +247,45 @@ public class MapSharedShreddingTableTest extends
TableTestBase {
assertThat(actual).containsExactly(Arrays.asList(10L, null, null));
}
+ @ParameterizedTest
+ @ValueSource(strings = {"orc", "parquet"})
+ public void
testReadSelectedKeysAfterDefaultToSharedAndValueTypeEvolution(String format)
+ throws Exception {
+ catalog.createTable(
+ identifier(format),
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column(
+ "metrics",
+ DataTypes.MAP(DataTypes.STRING().notNull(),
DataTypes.INT()))
+ .option("bucket", "-1")
+ .option("file.format", format)
+ .option(CoreOptions.WRITE_ONLY.key(), "true")
+ .build(),
+ true);
+ Table table = catalog.getTable(identifier(format));
+ Map<BinaryString, Integer> oldValues = new LinkedHashMap<>();
+ oldValues.put(BinaryString.fromString("key1"), 10);
+ write(table, GenericRow.of(1, new GenericMap(oldValues)));
+
+ catalog.alterTable(
+ identifier(format),
+ Arrays.asList(
+ SchemaChange.updateColumnType(
+ new String[] {"metrics", "value"},
DataTypes.BIGINT(), false),
+ SchemaChange.setOption(
+ "fields.metrics.map.storage-layout",
"shared-shredding"),
+ SchemaChange.setOption(
+
"fields.metrics.map.shared-shredding.max-columns", "1")),
+ false);
+ table = catalog.getTable(identifier(format));
+ write(table, GenericRow.of(2, mapOf("key2", 20L)));
+
+ assertThat(readSelectedKeysById(table))
+ .containsEntry(1, Arrays.asList(10L, null, null))
+ .containsEntry(2, Arrays.asList(null, 20L, null));
+ }
+
@ParameterizedTest
@ValueSource(strings = {"orc", "parquet"})
public void testReadSelectedKeysAfterRenameColumn(String format) throws
Exception {
@@ -285,6 +323,36 @@ public class MapSharedShreddingTableTest extends
TableTestBase {
.containsEntry(2, Arrays.asList(null, 30L, null));
}
+ @ParameterizedTest
+ @ValueSource(strings = {"orc", "parquet"})
+ public void testReadSelectedKeysAfterDefaultToSharedAndRenameColumn(String
format)
+ throws Exception {
+ Table table =
+ createTableWithBucket(
+ format,
+ 1,
+ "1",
+ Collections.singletonList("metrics"),
+ Collections.emptyList());
+ write(table, GenericRow.of(1, mapOf("key1", 10L, "key2", 20L)));
+
+ catalog.alterTable(
+ identifier(format),
+ Arrays.asList(
+ SchemaChange.renameColumn("metrics",
"renamed_metrics"),
+ SchemaChange.setOption(
+ "fields.renamed_metrics.map.storage-layout",
"shared-shredding"),
+ SchemaChange.setOption(
+
"fields.renamed_metrics.map.shared-shredding.max-columns", "1")),
+ false);
+ table = catalog.getTable(identifier(format));
+ write(table, GenericRow.of(2, mapOf("key2", 30L)));
+
+ assertThat(readSelectedKeysById(table, "renamed_metrics"))
+ .containsEntry(1, Arrays.asList(10L, 20L, null))
+ .containsEntry(2, Arrays.asList(null, 30L, null));
+ }
+
@ParameterizedTest
@ValueSource(strings = {"orc", "parquet"})
public void testAppendOnlyTableReadWriteWithTwoMapFields(String format)
throws Exception {
@@ -857,29 +925,89 @@ public class MapSharedShreddingTableTest extends
TableTestBase {
@ParameterizedTest
@ValueSource(strings = {"orc", "parquet"})
- public void
testCannotSwitchMapLayoutAndUseMaxColumnsWithoutMetadata(String format)
- throws Exception {
- createTableWithBucket(
- format, 4, "1", Arrays.asList("metrics", "labels"),
Arrays.asList("labels"));
+ public void testSwitchMapLayoutAndInferColumns(String format) throws
Exception {
+ Table table =
+ createTableWithBucket(
+ format,
+ 4,
+ "1",
+ Arrays.asList("metrics", "labels"),
+ Arrays.asList("labels"));
- assertThatThrownBy(
- () ->
- catalog.alterTable(
- identifier(format),
- Arrays.asList(
- SchemaChange.setOption(
-
"fields.metrics.map.storage-layout",
- "shared-shredding"),
- SchemaChange.setOption(
-
"fields.metrics.map.shared-shredding.max-columns",
- "3"),
- SchemaChange.setOption(
-
"fields.labels.map.storage-layout",
- "default")),
- false))
- .isInstanceOf(UnsupportedOperationException.class)
- .hasMessageContaining(
- "Cannot change map storage layout for field id 1
('metrics' -> 'metrics') from 'default' to 'shared-shredding'.");
+ write(table, GenericRow.of(1, mapOf("a", 11L, "b", 12L), mapOf("x",
21L)));
+
+ catalog.alterTable(
+ identifier(format),
+ Arrays.asList(
+ SchemaChange.setOption(
+ "fields.metrics.map.storage-layout",
"shared-shredding"),
+ SchemaChange.setOption(
+
"fields.metrics.map.shared-shredding.max-columns", "3"),
+
SchemaChange.setOption("fields.labels.map.storage-layout", "default")),
+ false);
+ table = catalog.getTable(identifier(format));
+
+ write(table, GenericRow.of(2, mapOf("c", 31L), mapOf("y", 41L, "z",
42L)));
+
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+ List<DataFileWithSplit> files = currentDataFiles(fileStoreTable);
+ files.sort(Comparator.comparingLong(file ->
file.dataFile.minSequenceNumber()));
+ assertThat(files).hasSize(2);
+
+ MapSharedShreddingFieldMeta metricsMeta =
+ readSharedShreddingFieldMeta(fileStoreTable, files.get(1),
"metrics");
+ assertThat(metricsMeta.numColumns()).isEqualTo(3);
+ assertThat(metricsMeta.maxRowWidth()).isEqualTo(1);
+
+ Map<Integer, List<Map<String, Long>>> actual = new LinkedHashMap<>();
+ for (InternalRow row : read(table)) {
+ actual.put(
+ row.getInt(0),
+ Arrays.asList(
+ row.isNullAt(1) ? null : toJavaMap(row.getMap(1)),
+ row.isNullAt(2) ? null :
toJavaMap(row.getMap(2))));
+ }
+
+ assertThat(actual)
+ .containsEntry(1, Arrays.asList(javaMapOf("a", 11L, "b", 12L),
javaMapOf("x", 21L)))
+ .containsEntry(
+ 2, Arrays.asList(javaMapOf("c", 31L), javaMapOf("y",
41L, "z", 42L)));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"orc", "parquet"})
+ public void testSwitchSharedShreddingToDefaultMap(String format) throws
Exception {
+ Table table =
+ createTableWithBucket(
+ format,
+ 1,
+ "1",
+ Collections.singletonList("metrics"),
+ Collections.singletonList("metrics"));
+ write(table, GenericRow.of(1, mapOf("key1", 10L, "key2", 20L)));
+
+ catalog.alterTable(
+ identifier(format),
+ Collections.singletonList(
+
SchemaChange.setOption("fields.metrics.map.storage-layout", "default")),
+ false);
+ table = catalog.getTable(identifier(format));
+ write(table, GenericRow.of(2, mapOf("key2", 30L, "cold", 40L)));
+
+ assertThat(readMapsById(table.newReadBuilder()))
+ .containsOnlyKeys(1, 2)
+ .containsEntry(1, javaMapOf("key1", 10L, "key2", 20L))
+ .containsEntry(2, javaMapOf("key2", 30L, "cold", 40L));
+ assertThat(countSharedShreddingFiles((FileStoreTable)
table)).isEqualTo(1);
+
+ table = enableForcedCompaction(table);
+ compact(table, BinaryRow.EMPTY_ROW, 0);
+ assertThat(countSharedShreddingFiles((FileStoreTable) table)).isZero();
+ assertThat(currentDataFiles((FileStoreTable) table)).hasSize(1);
+ assertThat(readMapsById(table.newReadBuilder()))
+ .containsOnlyKeys(1, 2)
+ .containsEntry(1, javaMapOf("key1", 10L, "key2", 20L))
+ .containsEntry(2, javaMapOf("key2", 30L, "cold", 40L));
}
@ParameterizedTest
@@ -1099,6 +1227,13 @@ public class MapSharedShreddingTableTest extends
TableTestBase {
return createTable(format, 2, sharedShreddingFields);
}
+ private Table enableForcedCompaction(Table table) {
+ Map<String, String> options = new LinkedHashMap<>();
+ options.put(CoreOptions.WRITE_ONLY.key(), "false");
+ options.put(CoreOptions.COMPACTION_FORCE_REWRITE_ALL_FILES.key(),
"true");
+ return table.copy(options);
+ }
+
private Table createTable(String format, int maxColumns, String...
sharedShreddingFields)
throws Exception {
return createTableWithBucket(format, maxColumns, "-1",
sharedShreddingFields);
@@ -1537,21 +1672,65 @@ public class MapSharedShreddingTableTest extends
TableTestBase {
return files;
}
+ private int countSharedShreddingFiles(FileStoreTable table) throws
Exception {
+ int count = 0;
+ for (DataFileWithSplit file : currentDataFiles(table)) {
+ Map<String, Map<String, String>> fieldMetadata =
readFieldMetadata(table, file);
+ if
(MapSharedShreddingUtils.hasShreddingMetadata(fieldMetadata.get("metrics"))) {
+ count++;
+ }
+ }
+ return count;
+ }
+
private MapSharedShreddingFieldMeta readSharedShreddingFieldMeta(
FileStoreTable table, DataFileWithSplit file, String fieldName)
throws Exception {
+ return MapSharedShreddingUtils.deserializeMetadata(
+ readFieldMetadata(table, file).get(fieldName));
+ }
+
+ private Map<String, Map<String, String>> readFieldMetadata(
+ FileStoreTable table, DataFileWithSplit file) throws Exception {
DataFilePathFactory pathFactory =
table.store().pathFactory().createDataFilePathFactory(file.partition,
file.bucket);
FileFormat fileFormat =
FileFormatDiscover.of(new CoreOptions(table.options()))
.discover(file.dataFile.fileFormat());
- Map<String, Map<String, String>> fieldMetadata =
- ((SupportsFieldMetadata) fileFormat)
- .readFieldMetadata(
- new FormatReaderContext(
- table.fileIO(),
- pathFactory.toPath(file.dataFile),
- file.dataFile.fileSize()));
- return
MapSharedShreddingUtils.deserializeMetadata(fieldMetadata.get(fieldName));
+ return ((SupportsFieldMetadata) fileFormat)
+ .readFieldMetadata(
+ new FormatReaderContext(
+ table.fileIO(),
+ pathFactory.toPath(file.dataFile),
+ file.dataFile.fileSize()));
+ }
+
+ private Map<Integer, List<Long>> readSelectedKeysById(Table table) throws
Exception {
+ return readSelectedKeysById(table, "metrics");
+ }
+
+ private Map<Integer, List<Long>> readSelectedKeysById(Table table, String
fieldName)
+ throws Exception {
+ ReadBuilder readBuilder =
+
table.newReadBuilder().withReadType(selectedKeysReadType(fieldName));
+ Map<Integer, List<Long>> actual = new LinkedHashMap<>();
+ try (RecordReader<InternalRow> reader =
+
readBuilder.newRead().createReader(readBuilder.newScan().plan())) {
+ reader.forEachRemaining(
+ row -> {
+ if (row.isNullAt(1)) {
+ actual.put(row.getInt(0), null);
+ return;
+ }
+ InternalRow selectedKeys = row.getRow(1, 3);
+ actual.put(
+ row.getInt(0),
+ Arrays.asList(
+ selectedKeys.isNullAt(0) ? null :
selectedKeys.getLong(0),
+ selectedKeys.isNullAt(1) ? null :
selectedKeys.getLong(1),
+ selectedKeys.isNullAt(2) ? null :
selectedKeys.getLong(2)));
+ });
+ }
+ return actual;
}
private GenericMap mapOf(Object... entries) {
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/MapSelectedKeysSharedShreddingE2ETestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/MapSelectedKeysSharedShreddingE2ETestBase.scala
index f2351ac965..fa3f49b77c 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/MapSelectedKeysSharedShreddingE2ETestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/MapSelectedKeysSharedShreddingE2ETestBase.scala
@@ -64,6 +64,88 @@ abstract class MapSelectedKeysSharedShreddingE2ETestBase
extends PaimonSparkTest
}
}
+ test(s"switch map storage layout in both directions for $format") {
+ withTable("T") {
+ sql(s"""
+ |CREATE TABLE T (id INT, attrs MAP<STRING, BIGINT>)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'file.format' = '$format'
+ |)
+ |""".stripMargin)
+ sql("""
+ |INSERT INTO T VALUES
+ | (1, map('key1', CAST(10 AS BIGINT), 'key2', CAST(20 AS
BIGINT)))
+ |""".stripMargin)
+
+ sql("""
+ |ALTER TABLE T SET TBLPROPERTIES (
+ | 'fields.attrs.map.storage-layout' = 'shared-shredding',
+ | 'fields.attrs.map.shared-shredding.max-columns' = '1'
+ |)
+ |""".stripMargin)
+ sql("""
+ |INSERT INTO T VALUES
+ | (2, map('key2', CAST(30 AS BIGINT), 'cold', CAST(40 AS
BIGINT)))
+ |""".stripMargin)
+
+ val firstSharedQuery =
+ sql("SELECT id, attrs['key1'], attrs['key2'] FROM T ORDER BY id")
+ assert(
+ pushedMapSelectedKeys(firstSharedQuery).contains("attrs"),
+ s"Expected selected-key pushdown after switching $format to
shared-shredding")
+ checkAnswer(firstSharedQuery, Row(1, 10L, 20L) :: Row(2, null, 30L)
:: Nil)
+
+ sql("""
+ |ALTER TABLE T SET TBLPROPERTIES (
+ | 'fields.attrs.map.storage-layout' = 'default'
+ |)
+ |""".stripMargin)
+ sql("""
+ |INSERT INTO T VALUES
+ | (3, map('key1', CAST(50 AS BIGINT)))
+ |""".stripMargin)
+
+ val defaultQuery =
+ sql("SELECT id, attrs['key1'], attrs['key2'] FROM T ORDER BY id")
+ assert(
+ pushedMapSelectedKeys(defaultQuery).isEmpty,
+ s"Expected selected-key pushdown to be disabled after switching
$format to default")
+ checkAnswer(
+ defaultQuery,
+ Row(1, 10L, 20L) :: Row(2, null, 30L) :: Row(3, 50L, null) :: Nil)
+
+ sql("""
+ |ALTER TABLE T SET TBLPROPERTIES (
+ | 'fields.attrs.map.storage-layout' = 'shared-shredding'
+ |)
+ |""".stripMargin)
+ sql("""
+ |INSERT INTO T VALUES
+ | (4, map('key2', CAST(60 AS BIGINT)))
+ |""".stripMargin)
+
+ val secondSharedQuery =
+ sql("SELECT id, attrs['key1'], attrs['key2'] FROM T ORDER BY id")
+ assert(
+ pushedMapSelectedKeys(secondSharedQuery).contains("attrs"),
+ s"Expected selected-key pushdown after switching $format back to
shared-shredding")
+ checkAnswer(
+ secondSharedQuery,
+ Row(1, 10L, 20L) ::
+ Row(2, null, 30L) ::
+ Row(3, 50L, null) ::
+ Row(4, null, 60L) :: Nil)
+ checkAnswer(
+ sql("SELECT id, attrs FROM T ORDER BY id"),
+ Row(1, Map("key1" -> 10L, "key2" -> 20L)) ::
+ Row(2, Map("key2" -> 30L, "cold" -> 40L)) ::
+ Row(3, Map("key1" -> 50L)) ::
+ Row(4, Map("key2" -> 60L)) :: Nil
+ )
+ }
+ }
+
Seq(false, true).foreach {
thinMode =>
test(s"skip selected-key pushdown for merge_map from $format with
thin mode $thinMode") {
@@ -212,4 +294,15 @@ abstract class MapSelectedKeysSharedShreddingE2ETestBase
extends PaimonSparkTest
checkAnswer(query, Row(1, 10L, 30L) :: Row(2, 50L, null) :: Nil)
}
}
+
+ private def pushedMapSelectedKeys(
+ query: org.apache.spark.sql.DataFrame): Map[String, Seq[String]] = {
+ val sparkPlan = query.queryExecution.sparkPlan
+ sparkPlan
+ .collectFirst {
+ case scan: BatchScanExec if scan.scan.isInstanceOf[PaimonScan] =>
+ scan.scan.asInstanceOf[PaimonScan].pushedMapSelectedKeys
+ }
+ .getOrElse(fail(s"Expected a Paimon scan in physical plan:\n$sparkPlan"))
+ }
}
