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 877c9e42e3 [core][spark] Support selected-key pushdown for
shared-shredding MAP (#8782)
877c9e42e3 is described below
commit 877c9e42e3bb6f6477ff3a17f5c55194f73c2f7d
Author: lszskye <[email protected]>
AuthorDate: Tue Jul 28 02:44:23 2026 -0700
[core][spark] Support selected-key pushdown for shared-shredding MAP (#8782)
---
.../data/shredding/MapSharedShreddingReadPlan.java | 342 ++++++++++++++++++---
.../data/shredding/MapSharedShreddingUtils.java | 102 ++++--
.../shredding/MapSharedShreddingReadPlanTest.java | 125 ++++++++
.../shredding/MapSharedShreddingUtilsTest.java | 28 ++
.../org/apache/paimon/schema/SchemaManager.java | 46 +++
.../apache/paimon/utils/FormatReaderMapping.java | 73 ++++-
.../compact/aggregate/TestMapOnlyAggFactory.java | 46 +++
.../apache/paimon/schema/SchemaManagerTest.java | 92 ++++++
.../paimon/table/MapSharedShreddingTableTest.java | 212 ++++++++++---
.../paimon/utils/FormatReaderMappingTest.java | 217 +++++++++++++
.../services/org.apache.paimon.factories.Factory | 3 +-
.../MapSelectedKeysSharedShreddingE2ETest.scala | 21 ++
.../MapSelectedKeysSharedShreddingE2ETest.scala | 21 ++
.../MapSelectedKeysSharedShreddingE2ETest.scala | 21 ++
.../MapSelectedKeysSharedShreddingE2ETest.scala | 21 ++
.../MapSelectedKeysSharedShreddingE2ETest.scala | 21 ++
.../MapSelectedKeysSharedShreddingE2ETest.scala | 21 ++
.../optimizer/PushDownMapSelectedKeys.scala | 11 +-
.../paimon/spark/execution/PaimonStrategy.scala | 15 +-
.../extensions/PaimonSparkSessionExtensions.scala | 2 -
...MapSelectedKeysSharedShreddingE2ETestBase.scala | 215 +++++++++++++
21 files changed, 1537 insertions(+), 118 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 521dbd5b55..f09c58d685 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
@@ -30,14 +30,17 @@ 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.heap.HeapRowVector;
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 javax.annotation.Nullable;
+
import java.util.Arrays;
import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -45,6 +48,8 @@ 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 static final int FIELD_MAPPING_POSITION = 0;
+
private final RowType logicalType;
private final RowType physicalType;
private final Map<Integer, SharedShreddingContext> contextByFieldIndex;
@@ -53,7 +58,7 @@ public class MapSharedShreddingReadPlan implements
ShreddingReadPlan {
RowType logicalType, Map<String, MapSharedShreddingFieldMeta>
fieldMetas) {
this.logicalType = logicalType;
this.physicalType =
MapSharedShreddingUtils.buildPhysicalReadType(logicalType, fieldMetas);
- this.contextByFieldIndex = createContexts(logicalType, fieldMetas);
+ this.contextByFieldIndex = createContexts(logicalType, physicalType,
fieldMetas);
}
@Override
@@ -77,13 +82,30 @@ public class MapSharedShreddingReadPlan implements
ShreddingReadPlan {
}
private static Map<Integer, SharedShreddingContext> createContexts(
- RowType logicalType, Map<String, MapSharedShreddingFieldMeta>
fieldMetas) {
+ RowType logicalType,
+ RowType physicalType,
+ 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()));
+ if (fieldMeta == null) {
+ continue;
+ }
+
+ RowType physicalStructType = (RowType) physicalType.getTypeAt(i);
+ SharedPhysicalContext physicalContext =
+ new SharedPhysicalContext(fieldMeta, physicalStructType);
+ if (field.type() instanceof MapType) {
+ contexts.put(i, new FullMapContext(physicalContext, (MapType)
field.type()));
+ } else if
(MapSelectedKeysMetadataUtils.isMapSelectedKeysField(field)) {
+ contexts.put(
+ i,
+ new SelectedKeysContext(
+ physicalContext,
+ fieldMeta,
+ (RowType) field.type(),
+ field.description()));
}
}
return contexts;
@@ -100,9 +122,8 @@ public class MapSharedShreddingReadPlan implements
ShreddingReadPlan {
logicalVectors[i] = physicalBatch.columns[i];
} else {
logicalVectors[i] =
- materializeLogicalMapVector(
+ context.materialize(
(RowColumnVector) physicalBatch.columns[i],
- context,
physicalBatch.getNumRows());
}
}
@@ -111,7 +132,7 @@ public class MapSharedShreddingReadPlan implements
ShreddingReadPlan {
}
private static CastedMapColumnVector materializeLogicalMapVector(
- RowColumnVector physicalVector, SharedShreddingContext context,
int rowCount) {
+ RowColumnVector physicalVector, FullMapContext context, int
rowCount) {
ColumnVector[] physicalChildren = physicalChildren(physicalVector);
int totalElements =
countLogicalMapElements(physicalVector, physicalChildren,
context, rowCount);
@@ -147,6 +168,46 @@ public class MapSharedShreddingReadPlan implements
ShreddingReadPlan {
new WritableColumnVector[] {keyVector, valueVector}));
}
+ private static ColumnVector materializeSelectedKeysRowVector(
+ RowColumnVector physicalVector, SelectedKeysContext context, int
rowCount) {
+ ColumnVector[] physicalChildren = physicalChildren(physicalVector);
+ 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;
+ }
+
+ InternalArray fieldMapping = fieldMapping(physicalChildren, row,
context.physical);
+ InternalMap overflow = null;
+ for (int ordinal = 0; ordinal < selectedKeyVectors.length;
ordinal++) {
+ if (context.selectedKeyOverflow[ordinal] && overflow == null) {
+ overflow = overflowMap(physicalChildren, row,
context.physical);
+ }
+ appendSelectedKeyValue(
+ physicalChildren,
+ fieldMapping,
+ context.selectedKeyOverflow[ordinal] ? overflow : null,
+ row,
+ context,
+ ordinal,
+ selectedKeyVectors[ordinal]);
+ }
+ rowVector.appendRow();
+ }
+
+ return ColumnVectorUtils.createReadableColumnVector(selectedKeysType,
rowVector);
+ }
+
private static ColumnVector[] physicalChildren(RowColumnVector
physicalVector) {
ColumnVector[] physicalChildren = physicalVector.getChildren();
if (physicalChildren != null) {
@@ -158,7 +219,7 @@ public class MapSharedShreddingReadPlan implements
ShreddingReadPlan {
private static int countLogicalMapElements(
RowColumnVector physicalVector,
ColumnVector[] physicalChildren,
- SharedShreddingContext context,
+ FullMapContext context,
int rowCount) {
int totalElements = 0;
for (int row = 0; row < rowCount; row++) {
@@ -170,20 +231,20 @@ public class MapSharedShreddingReadPlan implements
ShreddingReadPlan {
}
private static int countLogicalMapElements(
- ColumnVector[] physicalChildren, int row, SharedShreddingContext
context) {
+ ColumnVector[] physicalChildren, int row, FullMapContext context) {
int count = 0;
- InternalArray fieldMapping = fieldMapping(physicalChildren, row,
context);
- for (int column = 0; column < context.numColumns; column++) {
- if (logicalFieldName(fieldMapping, column, context) != null) {
+ InternalArray fieldMapping = fieldMapping(physicalChildren, row,
context.physical);
+ for (int column = 0; column < context.physical.numColumns; column++) {
+ if (logicalFieldName(fieldMapping, column, context.physical) !=
null) {
count++;
}
}
- InternalMap overflow = overflowMap(physicalChildren, row, context);
+ InternalMap overflow = overflowMap(physicalChildren, row,
context.physical);
if (overflow != null) {
InternalArray keys = overflow.keyArray();
for (int i = 0; i < overflow.size(); i++) {
- if (context.nameById.containsKey(keys.getInt(i))) {
+ if (context.physical.nameById.containsKey(keys.getInt(i))) {
count++;
}
}
@@ -194,32 +255,32 @@ public class MapSharedShreddingReadPlan implements
ShreddingReadPlan {
private static int appendLogicalMapElements(
ColumnVector[] physicalChildren,
int row,
- SharedShreddingContext context,
+ FullMapContext 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);
+ InternalArray fieldMapping = fieldMapping(physicalChildren, row,
context.physical);
+ for (int column = 0; column < context.physical.numColumns; column++) {
+ BinaryString fieldName = logicalFieldName(fieldMapping, column,
context.physical);
if (fieldName == null) {
continue;
}
context.keyConverter.append(fieldName, keyVector);
- ColumnVector valueColumn = physicalChildren[column + 1];
- appendValue(context, valueColumn, row, valueVector);
+ ColumnVector valueColumn = physicalColumn(physicalChildren,
column, context.physical);
+ appendValue(context.valueConverter, valueColumn, row, valueVector);
count++;
}
- InternalMap overflow = overflowMap(physicalChildren, row, context);
+ InternalMap overflow = overflowMap(physicalChildren, row,
context.physical);
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));
+ BinaryString fieldName =
context.physical.nameById.get(keys.getInt(i));
if (fieldName != null) {
context.keyConverter.append(fieldName, keyVector);
- appendValue(context, values, i, valueVector);
+ appendValue(context.valueConverter, values, i,
valueVector);
count++;
}
}
@@ -227,50 +288,112 @@ public class MapSharedShreddingReadPlan implements
ShreddingReadPlan {
return count;
}
+ private static void appendSelectedKeyValue(
+ ColumnVector[] physicalChildren,
+ InternalArray fieldMapping,
+ @Nullable InternalMap overflow,
+ int row,
+ SelectedKeysContext context,
+ int ordinal,
+ WritableColumnVector valueVector) {
+ int fieldId = context.selectedKeyFieldIds[ordinal];
+ if (fieldId < 0) {
+ valueVector.appendNull();
+ return;
+ }
+
+ int[] candidateColumns = context.selectedKeyColumns[ordinal];
+ for (int i = 0; i < candidateColumns.length; i++) {
+ int column = candidateColumns[i];
+ if (mappedFieldId(fieldMapping, column) == fieldId) {
+ appendValue(
+ context.selectedValueConverters[ordinal],
+ physicalColumn(physicalChildren, column,
context.physical),
+ row,
+ valueVector);
+ return;
+ }
+ }
+
+ if (overflow != null) {
+ InternalArray keys = overflow.keyArray();
+ InternalArray values = overflow.valueArray();
+ for (int i = 0; i < overflow.size(); i++) {
+ if (!keys.isNullAt(i) && keys.getInt(i) == fieldId) {
+ appendValue(context.selectedValueConverters[ordinal],
values, i, valueVector);
+ return;
+ }
+ }
+ }
+
+ valueVector.appendNull();
+ }
+
private static void appendValue(
- SharedShreddingContext context,
+ RowToColumnConverter.ElementConverter converter,
ColumnVector source,
int row,
WritableColumnVector valueVector) {
if (source.isNullAt(row)) {
valueVector.appendNull();
} else {
- context.valueConverter.append(source, row, valueVector);
+ converter.append(source, row, valueVector);
}
}
private static void appendValue(
- SharedShreddingContext context,
+ RowToColumnConverter.ElementConverter converter,
InternalArray source,
int pos,
WritableColumnVector valueVector) {
if (source.isNullAt(pos)) {
valueVector.appendNull();
} else {
- context.valueConverter.append(source, pos, valueVector);
+ converter.append(source, pos, valueVector);
}
}
private static InternalArray fieldMapping(
- ColumnVector[] physicalChildren, int row, SharedShreddingContext
context) {
- InternalArray fieldMapping = ((ArrayColumnVector)
physicalChildren[0]).getArray(row);
+ ColumnVector[] physicalChildren, int row, SharedPhysicalContext
context) {
+ InternalArray fieldMapping =
+ ((ArrayColumnVector)
physicalChildren[FIELD_MAPPING_POSITION]).getArray(row);
checkArgument(
fieldMapping.size() == context.numColumns,
"Shared-shredding field mapping size %s does not match
metadata num columns %s.",
fieldMapping.size(),
context.numColumns);
+ for (int column = 0; column < fieldMapping.size(); column++) {
+ checkArgument(
+ !fieldMapping.isNullAt(column),
+ "Shared-shredding field mapping must not contain null at
column %s.",
+ column);
+ }
return fieldMapping;
}
private static BinaryString logicalFieldName(
- InternalArray fieldMapping, int column, SharedShreddingContext
context) {
- int fieldId = fieldMapping.isNullAt(column) ? -1 :
fieldMapping.getInt(column);
+ InternalArray fieldMapping, int column, SharedPhysicalContext
context) {
+ int fieldId = mappedFieldId(fieldMapping, column);
return fieldId < 0 ? null : context.nameById.get(fieldId);
}
+ private static int mappedFieldId(InternalArray fieldMapping, int column) {
+ return fieldMapping.getInt(column);
+ }
+
+ private static ColumnVector physicalColumn(
+ ColumnVector[] physicalChildren, int column, SharedPhysicalContext
context) {
+ int position = context.physicalColumnPositions[column];
+ checkArgument(
+ position >= 0,
+ "Shared-shredding physical column %s was not included in the
read schema.",
+ MapSharedShreddingDefine.physicalColumnName(column));
+ return physicalChildren[position];
+ }
+
private static InternalMap overflowMap(
- ColumnVector[] physicalChildren, int row, SharedShreddingContext
context) {
- if (context.overflowPosition >= physicalChildren.length) {
+ ColumnVector[] physicalChildren, int row, SharedPhysicalContext
context) {
+ if (context.overflowPosition < 0 || context.overflowPosition >=
physicalChildren.length) {
return null;
}
@@ -278,27 +401,156 @@ public class MapSharedShreddingReadPlan implements
ShreddingReadPlan {
return overflowVector.isNullAt(row) ? null : ((MapColumnVector)
overflowVector).getMap(row);
}
- private static class SharedShreddingContext {
+ private interface SharedShreddingContext {
+
+ ColumnVector materialize(RowColumnVector physicalVector, int rowCount);
+ }
+
+ private static class SharedPhysicalContext {
- 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[] physicalColumnPositions;
private final int overflowPosition;
- private SharedShreddingContext(MapSharedShreddingFieldMeta fieldMeta,
DataType fieldType) {
- this.mapType = (MapType) fieldType;
+ private SharedPhysicalContext(
+ MapSharedShreddingFieldMeta fieldMeta, RowType
physicalStructType) {
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());
this.numColumns = fieldMeta.numColumns();
- this.overflowPosition = fieldMeta.numColumns() + 1;
+ checkArgument(
+ physicalStructType.getFieldCount() > 0
+ && MapSharedShreddingDefine.FIELD_MAPPING.equals(
+
physicalStructType.getFieldNames().get(FIELD_MAPPING_POSITION)),
+ "Shared-shredding physical struct must start with %s.",
+ MapSharedShreddingDefine.FIELD_MAPPING);
+ this.physicalColumnPositions =
physicalColumnPositions(physicalStructType, numColumns);
+ this.overflowPosition = overflowPosition(physicalStructType);
+ }
+
+ private static int[] physicalColumnPositions(RowType
physicalStructType, int numColumns) {
+ int[] positions = new int[numColumns];
+ Arrays.fill(positions, -1);
+ int physicalColumnEnd = overflowPosition(physicalStructType);
+ if (physicalColumnEnd < 0) {
+ physicalColumnEnd = physicalStructType.getFieldCount();
+ }
+ for (int i = FIELD_MAPPING_POSITION + 1; i < physicalColumnEnd;
i++) {
+ String fieldName = physicalStructType.getFieldNames().get(i);
+ int column = physicalColumnIndex(fieldName);
+ checkArgument(
+ column >= 0,
+ "Unexpected shared-shredding physical field %s at
position %s.",
+ fieldName,
+ i);
+ checkArgument(
+ column < numColumns,
+ "Shared-shredding physical column %s exceeds metadata
num columns %s.",
+ fieldName,
+ numColumns);
+ positions[column] = i;
+ }
+ return positions;
+ }
+
+ private static int overflowPosition(RowType physicalStructType) {
+ int lastPosition = physicalStructType.getFieldCount() - 1;
+ if (lastPosition <= FIELD_MAPPING_POSITION) {
+ return -1;
+ }
+ return MapSharedShreddingDefine.OVERFLOW.equals(
+
physicalStructType.getFieldNames().get(lastPosition))
+ ? lastPosition
+ : -1;
+ }
+
+ private static int physicalColumnIndex(String fieldName) {
+ String prefix = "__col_";
+ if (!fieldName.startsWith(prefix)) {
+ return -1;
+ }
+ return Integer.parseInt(fieldName.substring(prefix.length()));
+ }
+ }
+
+ private static class FullMapContext implements SharedShreddingContext {
+
+ private final SharedPhysicalContext physical;
+ private final MapType mapType;
+ private final RowToColumnConverter.ElementConverter keyConverter;
+ private final RowToColumnConverter.ElementConverter valueConverter;
+
+ private FullMapContext(SharedPhysicalContext physical, MapType
mapType) {
+ this.physical = physical;
+ this.mapType = mapType;
+ this.keyConverter =
RowToColumnConverter.createElementConverter(mapType.getKeyType());
+ this.valueConverter =
+
RowToColumnConverter.createElementConverter(mapType.getValueType());
+ }
+
+ @Override
+ public ColumnVector materialize(RowColumnVector physicalVector, int
rowCount) {
+ return materializeLogicalMapVector(physicalVector, this, rowCount);
+ }
+ }
+
+ private static class SelectedKeysContext implements SharedShreddingContext
{
+
+ private final SharedPhysicalContext physical;
+ private final RowType selectedKeysType;
+ private final int[] selectedKeyFieldIds;
+ private final int[][] selectedKeyColumns;
+ private final boolean[] selectedKeyOverflow;
+ private final RowToColumnConverter.ElementConverter[]
selectedValueConverters;
+
+ private SelectedKeysContext(
+ SharedPhysicalContext physical,
+ MapSharedShreddingFieldMeta fieldMeta,
+ RowType selectedKeysType,
+ String selectedKeysMetadata) {
+ this.physical = physical;
+ this.selectedKeysType = selectedKeysType;
+ List<String> selectedKeys =
+
MapSelectedKeysMetadataUtils.selectedKeys(selectedKeysMetadata);
+ checkArgument(
+ selectedKeys.size() == selectedKeysType.getFieldCount(),
+ "Selected-key metadata size %s does not match selected ROW
field count %s.",
+ selectedKeys.size(),
+ selectedKeysType.getFieldCount());
+ this.selectedKeyFieldIds = new int[selectedKeys.size()];
+ this.selectedKeyColumns = new int[selectedKeys.size()][];
+ this.selectedKeyOverflow = new boolean[selectedKeys.size()];
+ this.selectedValueConverters =
+ new
RowToColumnConverter.ElementConverter[selectedKeys.size()];
+ for (int i = 0; i < selectedKeys.size(); i++) {
+ Integer fieldId =
fieldMeta.nameToId().get(selectedKeys.get(i));
+ this.selectedKeyFieldIds[i] = fieldId == null ? -1 : fieldId;
+ this.selectedKeyColumns[i] =
+ fieldId == null ? new int[0] :
candidateColumns(fieldMeta, fieldId);
+ this.selectedKeyOverflow[i] =
+ fieldId != null &&
fieldMeta.overflowFieldSet().contains(fieldId);
+ this.selectedValueConverters[i] =
+
RowToColumnConverter.createElementConverter(selectedKeysType.getTypeAt(i));
+ }
+ }
+
+ @Override
+ public ColumnVector materialize(RowColumnVector physicalVector, int
rowCount) {
+ return materializeSelectedKeysRowVector(physicalVector, this,
rowCount);
+ }
+
+ private static int[] candidateColumns(MapSharedShreddingFieldMeta
fieldMeta, int fieldId) {
+ List<Integer> columns = fieldMeta.fieldToColumns().get(fieldId);
+ if (columns == null || columns.isEmpty()) {
+ return new int[0];
+ }
+ int[] result = new int[columns.size()];
+ for (int i = 0; i < columns.size(); i++) {
+ result[i] = columns.get(i);
+ }
+ return result;
}
}
}
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 c2526b03bb..6fc2d13b5a 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
@@ -50,6 +50,8 @@ import java.util.TreeMap;
import java.util.TreeSet;
import java.util.stream.Collectors;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
/**
* Utility functions for the shared-shredding MAP storage layout.
*
@@ -115,18 +117,30 @@ public class MapSharedShreddingUtils {
for (DataField logicalReadField : logicalReadType.getFields()) {
MapSharedShreddingFieldMeta fieldMeta =
sharedShreddingFieldMetas.get(logicalReadField.name());
- if (fieldMeta == null || !(logicalReadField.type() instanceof
MapType)) {
+ if (fieldMeta == null) {
physicalReadFields.add(logicalReadField);
continue;
}
- MapType mapType = (MapType) logicalReadField.type();
- DataType physicalType =
- buildSpecificPhysicalStructType(
- mapType.getValueType(),
- fieldMeta.numColumns(),
- !fieldMeta.overflowFieldSet().isEmpty())
- .copy(logicalReadField.type().isNullable());
+ DataType valueType;
+ DataType physicalType;
+ if
(MapSelectedKeysMetadataUtils.isMapSelectedKeysField(logicalReadField)) {
+ // recall partial key with shared shredding pushdown
+ valueType = selectedKeysValueType((RowType)
logicalReadField.type());
+ physicalType =
+ buildSpecificPhysicalStructType(
+ valueType,
+
selectedPhysicalColumnIds(logicalReadField, fieldMeta),
+
selectedKeysIncludeOverflow(logicalReadField, fieldMeta))
+ .copy(logicalReadField.type().isNullable());
+ } else {
+ // recall whole field without shared shredding pushdown
+ valueType = ((MapType) logicalReadField.type()).getValueType();
+ physicalType =
+ buildPhysicalStructType(valueType,
fieldMeta.numColumns())
+ .copy(logicalReadField.type().isNullable());
+ }
+
physicalReadFields.add(logicalReadField.newType(physicalType));
converted = true;
}
@@ -135,6 +149,19 @@ public class MapSharedShreddingUtils {
: logicalReadType;
}
+ private static DataType selectedKeysValueType(RowType selectedKeysType) {
+ checkArgument(
+ selectedKeysType.getFieldCount() > 0,
+ "Selected-key MAP read type must contain at least one field.");
+ DataType valueType = selectedKeysType.getTypeAt(0);
+ for (int i = 1; i < selectedKeysType.getFieldCount(); i++) {
+ checkArgument(
+
selectedKeysType.getTypeAt(i).equalsIgnoreNullable(valueType),
+ "Selected-key MAP fields must have the same value type.");
+ }
+ return valueType;
+ }
+
public static Map<String, Integer> buildColumnToNumColumns(
List<String> shreddingFieldNames, CoreOptions options) {
Map<String, Integer> fieldToNumColumns = new HashMap<>();
@@ -245,21 +272,21 @@ public class MapSharedShreddingUtils {
}
private static RowType buildPhysicalStructType(DataType valueType, int
numColumns) {
- RowType.Builder builder = RowType.builder();
- builder.field(MapSharedShreddingDefine.FIELD_MAPPING, new
ArrayType(new IntType()));
- for (int i = 0; i < numColumns; i++) {
- builder.field(MapSharedShreddingDefine.physicalColumnName(i),
valueType);
- }
- builder.field(MapSharedShreddingDefine.OVERFLOW, new MapType(new
IntType(), valueType));
- return builder.build();
+ return buildSpecificPhysicalStructType(valueType,
physicalColumnIds(numColumns), true);
}
- private static RowType buildSpecificPhysicalStructType(
- DataType valueType, int numColumns, boolean includeOverflow) {
+ public static RowType buildSpecificPhysicalStructType(
+ DataType valueType, Set<Integer> physicalColumnIds, boolean
includeOverflow) {
+ return innerBuildSpecificPhysicalStructType(
+ valueType, new ArrayList<>(new TreeSet<>(physicalColumnIds)),
includeOverflow);
+ }
+
+ private static RowType innerBuildSpecificPhysicalStructType(
+ DataType valueType, List<Integer> sortedColumns, boolean
includeOverflow) {
RowType.Builder builder = RowType.builder();
builder.field(MapSharedShreddingDefine.FIELD_MAPPING, new
ArrayType(new IntType()));
- for (int i = 0; i < numColumns; i++) {
- builder.field(MapSharedShreddingDefine.physicalColumnName(i),
valueType);
+ for (Integer column : sortedColumns) {
+ builder.field(MapSharedShreddingDefine.physicalColumnName(column),
valueType);
}
if (includeOverflow) {
builder.field(MapSharedShreddingDefine.OVERFLOW, new MapType(new
IntType(), valueType));
@@ -267,6 +294,43 @@ public class MapSharedShreddingUtils {
return builder.build();
}
+ private static Set<Integer> physicalColumnIds(int numColumns) {
+ Set<Integer> physicalColumnIds = new TreeSet<>();
+ for (int i = 0; i < numColumns; i++) {
+ physicalColumnIds.add(i);
+ }
+ return physicalColumnIds;
+ }
+
+ private static Set<Integer> selectedPhysicalColumnIds(
+ DataField selectedKeysField, MapSharedShreddingFieldMeta
fieldMeta) {
+ Set<Integer> selectedColumns = new TreeSet<>();
+ for (String selectedKey :
+
MapSelectedKeysMetadataUtils.selectedKeys(selectedKeysField.description())) {
+ Integer fieldId = fieldMeta.nameToId().get(selectedKey);
+ if (fieldId == null) {
+ continue;
+ }
+ List<Integer> columns = fieldMeta.fieldToColumns().get(fieldId);
+ if (columns != null) {
+ selectedColumns.addAll(columns);
+ }
+ }
+ return selectedColumns;
+ }
+
+ private static boolean selectedKeysIncludeOverflow(
+ DataField selectedKeysField, MapSharedShreddingFieldMeta
fieldMeta) {
+ for (String selectedKey :
+
MapSelectedKeysMetadataUtils.selectedKeys(selectedKeysField.description())) {
+ Integer fieldId = fieldMeta.nameToId().get(selectedKey);
+ if (fieldId != null &&
fieldMeta.overflowFieldSet().contains(fieldId)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
private static Map<Integer, List<Integer>> sortedFieldColumns(
Map<Integer, List<Integer>> fieldToColumns) {
Map<Integer, List<Integer>> result = new TreeMap<>();
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 943bf94dbf..6b781be417 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
@@ -20,10 +20,12 @@ package org.apache.paimon.data.shredding;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.InternalMap;
+import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.columnar.BytesColumnVector;
import org.apache.paimon.data.columnar.ColumnVector;
import org.apache.paimon.data.columnar.LongColumnVector;
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.HeapIntVector;
@@ -35,13 +37,16 @@ import org.apache.paimon.types.RowType;
import org.junit.jupiter.api.Test;
+import java.util.Arrays;
import java.util.Collections;
import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.TreeSet;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link MapSharedShreddingReadPlan}. */
class MapSharedShreddingReadPlanTest {
@@ -88,6 +93,19 @@ class MapSharedShreddingReadPlanTest {
assertThat(restored.valueArray().getLong(0)).isEqualTo(30L);
}
+ @Test
+ void testFieldMappingRejectsNullElement() {
+ MapSharedShreddingFieldMeta fieldMeta =
+ new MapSharedShreddingFieldMeta(
+ nameToId("a", 0), Collections.emptyMap(), new
TreeSet<Integer>(), 2, 1);
+ HeapRowVector physicalMap =
+ rowVector(fieldMappingWithNull(0, null), longVector(10L),
longVector(null));
+
+ assertThatThrownBy(() -> readMap(fieldMeta, physicalMap))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("field mapping must not contain null");
+ }
+
@Test
void testAssembledMapVectorExposesKeyValueChildren() {
MapSharedShreddingFieldMeta fieldMeta =
@@ -114,6 +132,59 @@ class MapSharedShreddingReadPlanTest {
assertThat(mapVector.getMap(0).size()).isEqualTo(2);
}
+ @Test
+ void testReadSelectedKeysAsLogicalRowDirectly() {
+ MapSharedShreddingFieldMeta fieldMeta =
+ new MapSharedShreddingFieldMeta(
+ nameToId("key1", 0, "key2", 1, "cold", 2),
+ fieldToColumns(
+ 0, Collections.singletonList(0), 1,
Collections.singletonList(1)),
+ new TreeSet<Integer>(Collections.singletonList(1)),
+ 2,
+ 2);
+ HeapRowVector physicalMap =
+ rowVector(
+ fieldMapping(0, -1),
+ longVector(10L),
+ longVector(null),
+ overflowMap(1, 20L));
+
+ MapSharedShreddingReadPlan readPlan = selectedKeysReadPlan(fieldMeta);
+ RowColumnVector selectedKeysVector =
assembleSelectedKeysVector(readPlan, physicalMap);
+ InternalRow selectedKeys = selectedKeysVector.getRow(0);
+
+ assertThat(selectedKeys.getLong(0)).isEqualTo(10L);
+ assertThat(selectedKeys.getLong(1)).isEqualTo(20L);
+ assertThat(selectedKeys.isNullAt(2)).isTrue();
+ }
+
+ @Test
+ void testReadSelectedKeysFromPrunedPhysicalColumns() {
+ MapSharedShreddingFieldMeta fieldMeta =
+ new MapSharedShreddingFieldMeta(
+ nameToId("key1", 0, "key2", 1, "cold", 2),
+ fieldToColumns(
+ 0, Collections.singletonList(2), 2,
Collections.singletonList(0)),
+ new TreeSet<Integer>(Collections.singletonList(1)),
+ 4,
+ 2);
+
+ MapSharedShreddingReadPlan readPlan = selectedKeysReadPlan(fieldMeta);
+ RowType physicalMapType = (RowType)
readPlan.physicalRowType().getTypeAt(0);
+ assertThat(physicalMapType.getFieldNames())
+ .containsExactly("__field_mapping", "__col_2", "__overflow");
+
+ HeapRowVector physicalMap =
+ rowVector(fieldMapping(-1, -1, 0, -1), longVector(10L),
overflowMap(1, 20L));
+
+ RowColumnVector selectedKeysVector =
assembleSelectedKeysVector(readPlan, physicalMap);
+ InternalRow selectedKeys = selectedKeysVector.getRow(0);
+
+ assertThat(selectedKeys.getLong(0)).isEqualTo(10L);
+ assertThat(selectedKeys.getLong(1)).isEqualTo(20L);
+ assertThat(selectedKeys.isNullAt(2)).isTrue();
+ }
+
private static InternalMap readMap(
MapSharedShreddingFieldMeta fieldMeta, HeapRowVector physicalMap) {
return assembleMapVector(fieldMeta, physicalMap).getMap(0);
@@ -132,6 +203,24 @@ class MapSharedShreddingReadPlanTest {
return (MapColumnVector) logicalBatch.columns[0];
}
+ private static RowColumnVector assembleSelectedKeysVector(
+ MapSharedShreddingReadPlan readPlan, HeapRowVector physicalMap) {
+
assertThat(readPlan.physicalRowType().getTypeAt(0)).isInstanceOf(RowType.class);
+
+ VectorizedColumnBatch physicalBatch =
+ new VectorizedColumnBatch(new ColumnVector[] {physicalMap});
+ physicalBatch.setNumRows(1);
+ VectorizedColumnBatch logicalBatch =
readPlan.batchAssembler().assemble(physicalBatch);
+ return (RowColumnVector) logicalBatch.columns[0];
+ }
+
+ private static MapSharedShreddingReadPlan selectedKeysReadPlan(
+ MapSharedShreddingFieldMeta fieldMeta) {
+ Map<String, MapSharedShreddingFieldMeta> fieldMetas = new
LinkedHashMap<>();
+ fieldMetas.put("metrics", fieldMeta);
+ return new MapSharedShreddingReadPlan(selectedKeysLogicalType(),
fieldMetas);
+ }
+
private static RowType logicalType() {
return DataTypes.ROW(
DataTypes.FIELD(
@@ -140,6 +229,19 @@ class MapSharedShreddingReadPlanTest {
DataTypes.MAP(DataTypes.STRING().notNull(),
DataTypes.BIGINT())));
}
+ private static RowType selectedKeysLogicalType() {
+ RowType selectedKeysType =
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "0", DataTypes.BIGINT()),
+ DataTypes.FIELD(1, "1", DataTypes.BIGINT()),
+ DataTypes.FIELD(2, "2", DataTypes.BIGINT()));
+ return DataTypes.ROW(
+ MapSelectedKeysMetadataUtils.withSelectedKeys(
+ DataTypes.FIELD(0, "metrics", selectedKeysType),
+ selectedKeysType,
+ Arrays.asList("key1", "key2", "missing")));
+ }
+
private static HeapRowVector rowVector(ColumnVector... children) {
HeapRowVector vector = new HeapRowVector(1, children);
vector.appendRow();
@@ -156,6 +258,20 @@ class MapSharedShreddingReadPlanTest {
return vector;
}
+ private static HeapArrayVector fieldMappingWithNull(Integer... ids) {
+ HeapIntVector child = new HeapIntVector(ids.length);
+ for (Integer id : ids) {
+ if (id == null) {
+ child.appendNull();
+ } else {
+ child.appendInt(id);
+ }
+ }
+ HeapArrayVector vector = new HeapArrayVector(1, child);
+ vector.putOffsetLength(0, 0, ids.length);
+ return vector;
+ }
+
private static HeapLongVector longVector(Long value) {
HeapLongVector vector = new HeapLongVector(1);
if (value == null) {
@@ -182,4 +298,13 @@ class MapSharedShreddingReadPlanTest {
}
return result;
}
+
+ @SuppressWarnings("unchecked")
+ private static Map<Integer, List<Integer>> fieldToColumns(Object... pairs)
{
+ Map<Integer, List<Integer>> result = new TreeMap<>();
+ for (int i = 0; i < pairs.length; i += 2) {
+ result.put((Integer) pairs[i], (List<Integer>) pairs[i + 1]);
+ }
+ return result;
+ }
}
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 9c86109efa..dd628bc5a3 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
@@ -32,6 +32,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
+import java.util.TreeSet;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -186,6 +187,33 @@ class MapSharedShreddingUtilsTest {
.isEqualTo(logical);
}
+ @Test
+ void testBuildSpecificPhysicalStructType() {
+ RowType physicalType =
+ MapSharedShreddingUtils.buildSpecificPhysicalStructType(
+ DataTypes.BIGINT().notNull(),
+ new TreeSet<Integer>(Arrays.asList(3, 1)),
+ true);
+
+ assertThat(physicalType.getFieldNames())
+ .containsExactly("__field_mapping", "__col_1", "__col_3",
"__overflow");
+
assertThat(physicalType.getFields()).extracting(DataField::id).containsExactly(0,
1, 2, 3);
+
assertThat(physicalType.getField("__col_1").type()).isEqualTo(DataTypes.BIGINT().notNull());
+ assertThat(physicalType.getField("__overflow").type())
+ .isEqualTo(DataTypes.MAP(DataTypes.INT(),
DataTypes.BIGINT().notNull()));
+ }
+
+ @Test
+ void testBuildSpecificPhysicalStructTypeWithoutOverflow() {
+ RowType physicalType =
+ MapSharedShreddingUtils.buildSpecificPhysicalStructType(
+ DataTypes.STRING(), new
TreeSet<Integer>(Arrays.asList(3)), false);
+
+
assertThat(physicalType.getFieldNames()).containsExactly("__field_mapping",
"__col_3");
+
assertThat(physicalType.getFields()).extracting(DataField::id).containsExactly(0,
1);
+
assertThat(physicalType.getField("__col_3").type()).isEqualTo(DataTypes.STRING());
+ }
+
@Test
void testMetadataRoundtrip() {
Map<String, Integer> nameToId = new TreeMap<>();
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 4c238781b9..5eadd040c7 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,10 +637,56 @@ 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/main/java/org/apache/paimon/utils/FormatReaderMapping.java
b/paimon-core/src/main/java/org/apache/paimon/utils/FormatReaderMapping.java
index 77395cd63a..e583f389d0 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/FormatReaderMapping.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/FormatReaderMapping.java
@@ -18,7 +18,10 @@
package org.apache.paimon.utils;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.CoreOptions.MapStorageLayout;
import org.apache.paimon.casting.CastFieldGetter;
+import org.apache.paimon.data.shredding.MapSelectedKeysMetadataUtils;
import org.apache.paimon.data.variant.VariantMetadataUtils;
import org.apache.paimon.format.FileFormatDiscover;
import org.apache.paimon.format.FormatReaderFactory;
@@ -44,10 +47,12 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Set;
import java.util.function.Function;
import static
org.apache.paimon.predicate.PredicateBuilder.excludePredicateWithFields;
import static org.apache.paimon.table.SpecialFields.KEY_FIELD_ID_START;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
/** Class with index mapping and format reader. */
public class FormatReaderMapping {
@@ -205,7 +210,9 @@ public class FormatReaderMapping {
new ArrayList<>(fieldsExtractor.apply(dataSchema));
Map<String, Integer> systemFields =
findSystemFields(expectedFields);
- List<DataField> readDataFields =
readDataFields(allDataFieldsInFile, expectedFields);
+ Set<Integer> selectedKeysFieldIds =
selectedKeysFieldIds(tableSchema, expectedFields);
+ List<DataField> readDataFields =
+ readDataFields(allDataFieldsInFile, expectedFields,
selectedKeysFieldIds);
IndexCastMapping indexCastMapping =
SchemaEvolutionUtil.createIndexCastMapping(expectedFields,
readDataFields);
@@ -314,7 +321,9 @@ public class FormatReaderMapping {
}
private List<DataField> readDataFields(
- List<DataField> allDataFields, List<DataField> expectedFields)
{
+ List<DataField> allDataFields,
+ List<DataField> expectedFields,
+ Set<Integer> selectedKeysFieldIds) {
List<DataField> readDataFields = new ArrayList<>();
for (DataField dataField : allDataFields) {
expectedFields.stream()
@@ -322,6 +331,12 @@ public class FormatReaderMapping {
.findFirst()
.ifPresent(
field -> {
+ if
(selectedKeysFieldIds.contains(field.id())) {
+ checkSelectedKeysDataField(dataField);
+
readDataFields.add(selectedKeysDataField(field, dataField));
+ return;
+ }
+
DataType prunedType =
pruneDataType(field.type(),
dataField.type());
if (prunedType != null) {
@@ -332,6 +347,60 @@ public class FormatReaderMapping {
return readDataFields;
}
+ private DataField selectedKeysDataField(DataField expectedField,
DataField dataField) {
+ RowType selectedKeysType = (RowType) expectedField.type();
+ DataType dataValueType = ((MapType)
dataField.type()).getValueType();
+ List<DataField> selectedKeysDataFields = new ArrayList<>();
+ for (DataField selectedKeyField : selectedKeysType.getFields()) {
+ selectedKeysDataFields.add(
+ selectedKeyField.newType(
+
dataValueType.copy(selectedKeyField.type().isNullable())));
+ }
+ return dataField
+ .newType(selectedKeysType.copy(selectedKeysDataFields))
+ .newDescription(expectedField.description());
+ }
+
+ private void checkSelectedKeysDataField(DataField dataField) {
+ checkArgument(
+ dataField.type() instanceof MapType,
+ "Selected-key MAP field %s should be MAP type in data
schema.",
+ dataField.name());
+ }
+
+ private Set<Integer> selectedKeysFieldIds(
+ TableSchema tableSchema, List<DataField> expectedFields) {
+ CoreOptions options = CoreOptions.fromMap(tableSchema.options());
+ Map<Integer, DataField> tableFields = tableSchema.idToFieldMap();
+ Set<Integer> selectedKeysFieldIds = new HashSet<>();
+ for (DataField expectedField : expectedFields) {
+ DataField tableField = tableFields.get(expectedField.id());
+ if
(MapSelectedKeysMetadataUtils.isMapSelectedKeysField(expectedField)
+ && tableField != null
+ && tableField.type() instanceof MapType) {
+ checkArgument(
+ options.mapStorageLayout(tableField.name())
+ == MapStorageLayout.SHARED_SHREDDING,
+ "Selected-key MAP pushdown only supports top-level
shared-shredding MAP field: %s.",
+ tableField.name());
+ validateSelectedKeyValueTypes(expectedField, tableField);
+ selectedKeysFieldIds.add(expectedField.id());
+ }
+ }
+ return selectedKeysFieldIds;
+ }
+
+ private void validateSelectedKeyValueTypes(DataField expectedField,
DataField tableField) {
+ RowType selectedKeysType = (RowType) expectedField.type();
+ DataType mapValueType = ((MapType)
tableField.type()).getValueType();
+ for (DataField selectedKeyField : selectedKeysType.getFields()) {
+ checkArgument(
+
selectedKeyField.type().equalsIgnoreNullable(mapValueType),
+ "Selected-key MAP pushdown does not support pruning
MAP value fields: %s.",
+ tableField.name());
+ }
+ }
+
@Nullable
private DataType pruneDataType(DataType readType, DataType dataType) {
switch (readType.getTypeRoot()) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/TestMapOnlyAggFactory.java
b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/TestMapOnlyAggFactory.java
new file mode 100644
index 0000000000..e6b4296a95
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/TestMapOnlyAggFactory.java
@@ -0,0 +1,46 @@
+/*
+ * 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.mergetree.compact.aggregate;
+
+import org.apache.paimon.CoreOptions;
+import
org.apache.paimon.mergetree.compact.aggregate.factory.FieldAggregatorFactory;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.MapType;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** A custom MAP-only aggregator factory for testing {@link
FieldAggregatorFactory} discovery. */
+public class TestMapOnlyAggFactory implements FieldAggregatorFactory {
+
+ public static final String NAME = "my_merge_map";
+
+ @Override
+ public FieldAggregator create(DataType fieldType, CoreOptions options,
String field) {
+ checkArgument(
+ fieldType instanceof MapType,
+ "Data type for custom merge map column must be 'MAP' but was
'%s'",
+ fieldType);
+ return new FieldMergeMapAgg(identifier(), (MapType) fieldType);
+ }
+
+ @Override
+ public String identifier() {
+ return NAME;
+ }
+}
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 1b3140a616..2b0f4f18bf 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
@@ -54,6 +54,7 @@ import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.params.provider.ValueSource;
import java.io.File;
import java.io.IOException;
@@ -170,6 +171,97 @@ public class SchemaManagerTest {
assertThat(latest.get().options()).containsEntry("new_k", "new_v");
}
+ @Test
+ public void testCannotChangeMapStorageLayoutForExistingField() 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'.");
+ }
+
+ @Test
+ public void testCannotChangeMapStorageLayoutByRenameColumn() 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'.");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"plain", "sequential"})
+ public void testRenameColumnKeepsMapStorageLayoutOptions(String
placementPolicy)
+ throws Exception {
+ retryArtificialException(
+ () ->
+ manager.createTable(
+ mapStorageLayoutSchema("shared-shredding",
placementPolicy)));
+
+ retryArtificialException(
+ () ->
manager.commitChanges(SchemaChange.renameColumn("metrics", "renamed")));
+
+ Optional<TableSchema> latest = retryArtificialException(() ->
manager.latest());
+ assertThat(latest.isPresent()).isTrue();
+ assertThat(latest.get().options())
+ .doesNotContainKeys(
+ "fields.metrics.map.storage-layout",
+ "fields.metrics.map.shared-shredding.max-columns",
+
"fields.metrics.map.shared-shredding.column-placement-policy")
+ .containsEntry("fields.renamed.map.storage-layout",
"shared-shredding")
+
.containsEntry("fields.renamed.map.shared-shredding.max-columns", "2")
+ .containsEntry(
+
"fields.renamed.map.shared-shredding.column-placement-policy",
+ placementPolicy);
+ }
+
+ private Schema mapStorageLayoutSchema(String layout) {
+ return mapStorageLayoutSchema(layout, null);
+ }
+
+ private Schema mapStorageLayoutSchema(String layout, String
placementPolicy) {
+ Map<String, String> options = new HashMap<>();
+ if (layout != null) {
+ options.put("fields.metrics.map.storage-layout", layout);
+ options.put("fields.metrics.map.shared-shredding.max-columns",
"2");
+ }
+ if (placementPolicy != null) {
+ options.put(
+
"fields.metrics.map.shared-shredding.column-placement-policy", placementPolicy);
+ }
+ return new Schema(
+ Arrays.asList(
+ new DataField(0, "id", DataTypes.INT()),
+ new DataField(
+ 1,
+ "metrics",
+ DataTypes.MAP(DataTypes.STRING().notNull(),
DataTypes.BIGINT()))),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ options,
+ "");
+ }
+
@Test
public void testRejectRenamePrimaryKeyVectorIndexColumn() throws Exception
{
Map<String, String> options = new HashMap<>();
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 fe06144044..c300bb4a70 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
@@ -31,6 +31,7 @@ import org.apache.paimon.data.InternalArray;
import org.apache.paimon.data.InternalMap;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.data.shredding.MapSelectedKeysMetadataUtils;
import org.apache.paimon.data.shredding.MapSharedShreddingFieldMeta;
import org.apache.paimon.data.shredding.MapSharedShreddingUtils;
import org.apache.paimon.format.FileFormat;
@@ -75,6 +76,7 @@ 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 {
@@ -159,6 +161,130 @@ public class MapSharedShreddingTableTest extends
TableTestBase {
.containsEntry(4, javaMapOf("a", 70L, "b", 80L, "c", 90L, "d",
100L));
}
+ @ParameterizedTest
+ @ValueSource(strings = {"orc", "parquet"})
+ public void testReadSelectedKeysAsRow(String format) throws Exception {
+ Table table = createTable(format, 1, "metrics");
+
+ write(
+ table,
+ GenericRow.of(1, mapOf("key1", 10L, "key2", 20L)),
+ GenericRow.of(2, mapOf("key2", 30L, "cold", 40L)),
+ GenericRow.of(3, null));
+
+ Map<Integer, List<Long>> actual = new LinkedHashMap<>();
+ ReadBuilder readBuilder =
table.newReadBuilder().withReadType(selectedKeysReadType());
+ RecordReader<InternalRow> reader =
+
readBuilder.newRead().createReader(readBuilder.newScan().plan());
+ reader.forEachRemaining(
+ row -> {
+ if (row.isNullAt(1)) {
+ actual.put(row.getInt(0), null);
+ } else {
+ 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)));
+ }
+ });
+
+ assertThat(actual)
+ .containsEntry(1, Arrays.asList(10L, 20L, null))
+ .containsEntry(2, Arrays.asList(null, 30L, null))
+ .containsEntry(3, null);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"orc", "parquet"})
+ public void testReadSelectedKeysAfterMapValueTypeEvolution(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")
+ .option("fields.metrics.map.storage-layout",
"shared-shredding")
+
.option("fields.metrics.map.shared-shredding.max-columns", "1")
+ .build(),
+ true);
+ Table table = catalog.getTable(identifier(format));
+ Map<BinaryString, Integer> values = new LinkedHashMap<>();
+ values.put(BinaryString.fromString("key1"), 10);
+ write(table, GenericRow.of(1, new GenericMap(values)));
+
+ catalog.alterTable(
+ identifier(format),
+ Collections.singletonList(
+ SchemaChange.updateColumnType(
+ new String[] {"metrics", "value"},
DataTypes.BIGINT(), false)),
+ false);
+ table = catalog.getTable(identifier(format));
+
+ ReadBuilder readBuilder =
table.newReadBuilder().withReadType(selectedKeysReadType());
+ List<List<Long>> actual = new ArrayList<>();
+ readBuilder
+ .newRead()
+ .createReader(readBuilder.newScan().plan())
+ .forEachRemaining(
+ row -> {
+ InternalRow selectedKeys = row.getRow(1, 3);
+ actual.add(
+ Arrays.asList(
+ selectedKeys.getLong(0),
+ selectedKeys.isNullAt(1)
+ ? null
+ : selectedKeys.getLong(1),
+ selectedKeys.isNullAt(2)
+ ? null
+ :
selectedKeys.getLong(2)));
+ });
+ assertThat(actual).containsExactly(Arrays.asList(10L, null, null));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"orc", "parquet"})
+ public void testReadSelectedKeysAfterRenameColumn(String format) throws
Exception {
+ Table table = createTable(format, 1, "metrics");
+ write(
+ table,
+ GenericRow.of(1, mapOf("key1", 10L, "key2", 20L)),
+ GenericRow.of(2, mapOf("key2", 30L)));
+
+ catalog.alterTable(
+ identifier(format),
+ Collections.singletonList(SchemaChange.renameColumn("metrics",
"renamed_metrics")),
+ false);
+ table = catalog.getTable(identifier(format));
+
+ ReadBuilder readBuilder =
+
table.newReadBuilder().withReadType(selectedKeysReadType("renamed_metrics"));
+ Map<Integer, List<Long>> actual = new LinkedHashMap<>();
+ try (RecordReader<InternalRow> reader =
+
readBuilder.newRead().createReader(readBuilder.newScan().plan())) {
+ reader.forEachRemaining(
+ row -> {
+ 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)));
+ });
+ }
+
+ assertThat(actual)
+ .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 {
@@ -731,53 +857,29 @@ public class MapSharedShreddingTableTest extends
TableTestBase {
@ParameterizedTest
@ValueSource(strings = {"orc", "parquet"})
- public void testSwitchMapLayoutAndInferColumns(String format) throws
Exception {
- Table table =
- createTableWithBucket(
- format,
- 4,
- "1",
- Arrays.asList("metrics", "labels"),
- Arrays.asList("labels"));
-
- 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))));
- }
+ public void
testCannotSwitchMapLayoutAndUseMaxColumnsWithoutMetadata(String format)
+ throws Exception {
+ createTableWithBucket(
+ format, 4, "1", Arrays.asList("metrics", "labels"),
Arrays.asList("labels"));
- 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)));
+ 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'.");
}
@ParameterizedTest
@@ -1372,6 +1474,24 @@ public class MapSharedShreddingTableTest extends
TableTestBase {
return schemaWithBucket(format, maxColumns, "-1",
sharedShreddingFields);
}
+ private RowType selectedKeysReadType() {
+ return selectedKeysReadType("metrics");
+ }
+
+ private RowType selectedKeysReadType(String fieldName) {
+ RowType selectedKeysType =
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "0", DataTypes.BIGINT()),
+ DataTypes.FIELD(1, "1", DataTypes.BIGINT()),
+ DataTypes.FIELD(2, "2", DataTypes.BIGINT()));
+ return DataTypes.ROW(
+ DataTypes.FIELD(0, "id", DataTypes.INT()),
+ MapSelectedKeysMetadataUtils.withSelectedKeys(
+ DataTypes.FIELD(1, fieldName, selectedKeysType),
+ selectedKeysType,
+ Arrays.asList("key1", "key2", "missing")));
+ }
+
private Schema schemaWithBucket(
String format, int maxColumns, String bucket, String...
sharedShreddingFields) {
return schemaWithBucket(
diff --git
a/paimon-core/src/test/java/org/apache/paimon/utils/FormatReaderMappingTest.java
b/paimon-core/src/test/java/org/apache/paimon/utils/FormatReaderMappingTest.java
index 7b2ad9898d..88f8d94366 100644
---
a/paimon-core/src/test/java/org/apache/paimon/utils/FormatReaderMappingTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/utils/FormatReaderMappingTest.java
@@ -18,8 +18,12 @@
package org.apache.paimon.utils;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.shredding.MapSelectedKeysMetadataUtils;
import org.apache.paimon.schema.IndexCastMapping;
import org.apache.paimon.schema.SchemaEvolutionUtil;
+import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.SpecialFields;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
@@ -28,9 +32,14 @@ import org.apache.paimon.types.RowType;
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.lang.reflect.Method;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
+import java.util.Set;
/** Test for {@link FormatReaderMapping.Builder}. */
public class FormatReaderMappingTest {
@@ -147,4 +156,212 @@ public class FormatReaderMappingTest {
Assertions.assertThat(trimmed.get(2).id()).isEqualTo(3);
Assertions.assertThat(trimmed.size()).isEqualTo(3);
}
+
+ @Test
+ public void testMapSelectedKeysKeepsSelectedRowTypeForFormatReader()
throws Exception {
+ RowType dataValueType =
+ DataTypes.ROW(
+ DataTypes.FIELD(10, "a", DataTypes.INT()),
+ DataTypes.FIELD(11, "b", DataTypes.INT()),
+ DataTypes.FIELD(12, "c", DataTypes.INT()));
+ RowType selectedKeysType =
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "0", dataValueType),
+ DataTypes.FIELD(1, "1", dataValueType));
+
+ DataField dataField =
+ DataTypes.FIELD(2, "attrs", DataTypes.MAP(DataTypes.STRING(),
dataValueType));
+ DataField expectedField =
+ MapSelectedKeysMetadataUtils.withSelectedKeys(
+ DataTypes.FIELD(2, "attrs", selectedKeysType),
+ selectedKeysType,
+ Arrays.asList("key1", "key2"));
+
+ List<DataField> dataFields = Collections.singletonList(dataField);
+ List<DataField> expectedFields =
Collections.singletonList(expectedField);
+
+ DataField readField = invokeDataFields(dataFields,
expectedFields).get(0);
+
+ Assertions.assertThat(readField.type()).isEqualTo(selectedKeysType);
+
Assertions.assertThat(MapSelectedKeysMetadataUtils.isMapSelectedKeysField(readField))
+ .isTrue();
+ }
+
+ @Test
+ public void testMapSelectedKeysUsesDataValueTypeForSchemaEvolution()
throws Exception {
+ DataField dataField =
+ DataTypes.FIELD(2, "attrs", DataTypes.MAP(DataTypes.STRING(),
DataTypes.INT()));
+ DataField expectedField = selectedKeysField(2, "attrs",
DataTypes.BIGINT());
+
+ DataField readField =
+ invokeDataFields(
+ Collections.singletonList(dataField),
+ Collections.singletonList(expectedField))
+ .get(0);
+ RowType readType = (RowType) readField.type();
+
Assertions.assertThat(readType.getTypeAt(0)).isEqualTo(DataTypes.INT());
+
Assertions.assertThat(MapSelectedKeysMetadataUtils.isMapSelectedKeysField(readField))
+ .isTrue();
+
+ IndexCastMapping mapping =
+ SchemaEvolutionUtil.createIndexCastMapping(
+ Collections.singletonList(expectedField),
+ Collections.singletonList(readField));
+ Assertions.assertThat(mapping.getCastMapping()).isNotNull();
+ InternalRow selectedKeys =
+
mapping.getCastMapping()[0].getFieldOrNull(GenericRow.of(GenericRow.of(10)));
+ Assertions.assertThat(selectedKeys.getLong(0)).isEqualTo(10L);
+ }
+
+ @Test
+ public void testMapSelectedKeysUsesDataFieldNameAfterRename() throws
Exception {
+ DataField dataField =
+ DataTypes.FIELD(
+ 2, "old_attrs", DataTypes.MAP(DataTypes.STRING(),
DataTypes.BIGINT()));
+ DataField expectedField = selectedKeysField(2, "new_attrs",
DataTypes.BIGINT());
+
+ DataField readField =
+ invokeDataFields(
+ Collections.singletonList(dataField),
+ Collections.singletonList(expectedField))
+ .get(0);
+
+ Assertions.assertThat(readField.name()).isEqualTo("old_attrs");
+
Assertions.assertThat(readField.description()).isEqualTo(expectedField.description());
+
Assertions.assertThat(MapSelectedKeysMetadataUtils.isMapSelectedKeysField(readField))
+ .isTrue();
+ }
+
+ @Test
+ public void testRejectMapSelectedKeysValuePartialProjection() {
+ RowType tableValueType =
+ DataTypes.ROW(
+ DataTypes.FIELD(10, "a", DataTypes.INT()),
+ DataTypes.FIELD(11, "b", DataTypes.INT()));
+ RowType selectedValueType = DataTypes.ROW(DataTypes.FIELD(10, "a",
DataTypes.INT()));
+ DataField tableField =
+ DataTypes.FIELD(2, "attrs", DataTypes.MAP(DataTypes.STRING(),
tableValueType));
+ DataField expectedField = selectedKeysField(2, "attrs",
selectedValueType);
+
+ Map<String, String> options = new HashMap<>();
+ options.put("fields.attrs.map.storage-layout", "shared-shredding");
+
+ Assertions.assertThatThrownBy(
+ () ->
+ invokeSelectedKeysFieldIds(
+
tableSchema(Collections.singletonList(tableField), options),
+
Collections.singletonList(expectedField)))
+ .hasRootCauseMessage(
+ "Selected-key MAP pushdown does not support pruning
MAP value fields: attrs.");
+ }
+
+ @Test
+ public void testMapSelectedKeysOnlySupportsTopLevelSharedShreddingMap()
throws Exception {
+ DataField tableField =
+ DataTypes.FIELD(2, "attrs", DataTypes.MAP(DataTypes.STRING(),
DataTypes.BIGINT()));
+ DataField expectedField = selectedKeysField(2, "attrs",
DataTypes.BIGINT());
+
+ Map<String, String> options = new HashMap<>();
+ options.put("fields.attrs.map.storage-layout", "shared-shredding");
+ Assertions.assertThat(
+ invokeSelectedKeysFieldIds(
+
tableSchema(Collections.singletonList(tableField), options),
+ Collections.singletonList(expectedField)))
+ .containsExactly(2);
+
+ Assertions.assertThatThrownBy(
+ () ->
+ invokeSelectedKeysFieldIds(
+ tableSchema(
+
Collections.singletonList(tableField),
+ Collections.emptyMap()),
+
Collections.singletonList(expectedField)))
+ .hasRootCauseMessage(
+ "Selected-key MAP pushdown only supports top-level
shared-shredding MAP field: attrs.");
+ }
+
+ @Test
+ public void testNormalRowWithSelectedKeysMetadataComment() throws
Exception {
+ RowType rowType = DataTypes.ROW(DataTypes.FIELD(3, "value",
DataTypes.BIGINT()));
+ DataField rowField =
+ DataTypes.FIELD(2, "payload", rowType)
+
.newDescription(MapSelectedKeysMetadataUtils.METADATA_KEY + "key1");
+ TableSchema tableSchema =
+ tableSchema(Collections.singletonList(rowField),
Collections.emptyMap());
+
+ Set<Integer> selectedKeysFieldIds =
+ invokeSelectedKeysFieldIds(tableSchema,
Collections.singletonList(rowField));
+
+ Assertions.assertThat(selectedKeysFieldIds).isEmpty();
+ Assertions.assertThat(
+ invokeDataFields(
+ Collections.singletonList(rowField),
+ Collections.singletonList(rowField),
+ selectedKeysFieldIds))
+ .containsExactly(rowField);
+ }
+
+ @Test
+ public void testRejectSelectedKeysDataFieldWithNonMapType() {
+ DataField dataField = DataTypes.FIELD(2, "attrs", DataTypes.BIGINT());
+ DataField expectedField = selectedKeysField(2, "attrs",
DataTypes.BIGINT());
+
+ Assertions.assertThatThrownBy(
+ () ->
+ invokeDataFields(
+ Collections.singletonList(dataField),
+
Collections.singletonList(expectedField)))
+ .hasRootCauseMessage(
+ "Selected-key MAP field attrs should be MAP type in
data schema.");
+ }
+
+ @SuppressWarnings("unchecked")
+ private static List<DataField> invokeDataFields(
+ List<DataField> allDataFields, List<DataField> expectedFields)
throws Exception {
+ return invokeDataFields(allDataFields, expectedFields,
Collections.singleton(2));
+ }
+
+ @SuppressWarnings("unchecked")
+ private static List<DataField> invokeDataFields(
+ List<DataField> allDataFields,
+ List<DataField> expectedFields,
+ Set<Integer> selectedKeysFieldIds)
+ throws Exception {
+ FormatReaderMapping.Builder builder =
+ new FormatReaderMapping.Builder(
+ null, Collections.emptyList(), null, null, null, null);
+ Method method =
+ FormatReaderMapping.Builder.class.getDeclaredMethod(
+ "readDataFields", List.class, List.class, Set.class);
+ method.setAccessible(true);
+ return (List<DataField>)
+ method.invoke(builder, allDataFields, expectedFields,
selectedKeysFieldIds);
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Set<Integer> invokeSelectedKeysFieldIds(
+ TableSchema tableSchema, List<DataField> expectedFields) throws
Exception {
+ FormatReaderMapping.Builder builder =
+ new FormatReaderMapping.Builder(
+ null, Collections.emptyList(), null, null, null, null);
+ Method method =
+ FormatReaderMapping.Builder.class.getDeclaredMethod(
+ "selectedKeysFieldIds", TableSchema.class, List.class);
+ method.setAccessible(true);
+ return (Set<Integer>) method.invoke(builder, tableSchema,
expectedFields);
+ }
+
+ private static DataField selectedKeysField(
+ int id, String name, org.apache.paimon.types.DataType valueType) {
+ RowType selectedKeysType = DataTypes.ROW(DataTypes.FIELD(0, "0",
valueType));
+ return MapSelectedKeysMetadataUtils.withSelectedKeys(
+ DataTypes.FIELD(id, name, selectedKeysType),
+ selectedKeysType,
+ Collections.singletonList("key1"));
+ }
+
+ private static TableSchema tableSchema(List<DataField> fields, Map<String,
String> options) {
+ return new TableSchema(
+ 1, fields, 100, Collections.emptyList(),
Collections.emptyList(), options, "");
+ }
}
diff --git
a/paimon-core/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory
b/paimon-core/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory
index b44cf4490a..76ffd7a52f 100644
---
a/paimon-core/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory
+++
b/paimon-core/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory
@@ -14,4 +14,5 @@
# limitations under the License.
org.apache.paimon.mergetree.compact.aggregate.TestCustomAggFactory
-org.apache.paimon.rest.auth.CustomTestDLFTokenLoaderFactory
\ No newline at end of file
+org.apache.paimon.mergetree.compact.aggregate.TestMapOnlyAggFactory
+org.apache.paimon.rest.auth.CustomTestDLFTokenLoaderFactory
diff --git
a/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/MapSelectedKeysSharedShreddingE2ETest.scala
b/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/MapSelectedKeysSharedShreddingE2ETest.scala
new file mode 100644
index 0000000000..1bdf43ce20
--- /dev/null
+++
b/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/MapSelectedKeysSharedShreddingE2ETest.scala
@@ -0,0 +1,21 @@
+/*
+ * 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.spark.sql
+
+class MapSelectedKeysSharedShreddingE2ETest extends
MapSelectedKeysSharedShreddingE2ETestBase {}
diff --git
a/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/MapSelectedKeysSharedShreddingE2ETest.scala
b/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/MapSelectedKeysSharedShreddingE2ETest.scala
new file mode 100644
index 0000000000..1bdf43ce20
--- /dev/null
+++
b/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/MapSelectedKeysSharedShreddingE2ETest.scala
@@ -0,0 +1,21 @@
+/*
+ * 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.spark.sql
+
+class MapSelectedKeysSharedShreddingE2ETest extends
MapSelectedKeysSharedShreddingE2ETestBase {}
diff --git
a/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/MapSelectedKeysSharedShreddingE2ETest.scala
b/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/MapSelectedKeysSharedShreddingE2ETest.scala
new file mode 100644
index 0000000000..1bdf43ce20
--- /dev/null
+++
b/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/MapSelectedKeysSharedShreddingE2ETest.scala
@@ -0,0 +1,21 @@
+/*
+ * 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.spark.sql
+
+class MapSelectedKeysSharedShreddingE2ETest extends
MapSelectedKeysSharedShreddingE2ETestBase {}
diff --git
a/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/MapSelectedKeysSharedShreddingE2ETest.scala
b/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/MapSelectedKeysSharedShreddingE2ETest.scala
new file mode 100644
index 0000000000..1bdf43ce20
--- /dev/null
+++
b/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/MapSelectedKeysSharedShreddingE2ETest.scala
@@ -0,0 +1,21 @@
+/*
+ * 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.spark.sql
+
+class MapSelectedKeysSharedShreddingE2ETest extends
MapSelectedKeysSharedShreddingE2ETestBase {}
diff --git
a/paimon-spark/paimon-spark-4.0/src/test/scala/org/apache/paimon/spark/sql/MapSelectedKeysSharedShreddingE2ETest.scala
b/paimon-spark/paimon-spark-4.0/src/test/scala/org/apache/paimon/spark/sql/MapSelectedKeysSharedShreddingE2ETest.scala
new file mode 100644
index 0000000000..1bdf43ce20
--- /dev/null
+++
b/paimon-spark/paimon-spark-4.0/src/test/scala/org/apache/paimon/spark/sql/MapSelectedKeysSharedShreddingE2ETest.scala
@@ -0,0 +1,21 @@
+/*
+ * 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.spark.sql
+
+class MapSelectedKeysSharedShreddingE2ETest extends
MapSelectedKeysSharedShreddingE2ETestBase {}
diff --git
a/paimon-spark/paimon-spark-4.1/src/test/scala/org/apache/paimon/spark/sql/MapSelectedKeysSharedShreddingE2ETest.scala
b/paimon-spark/paimon-spark-4.1/src/test/scala/org/apache/paimon/spark/sql/MapSelectedKeysSharedShreddingE2ETest.scala
new file mode 100644
index 0000000000..1bdf43ce20
--- /dev/null
+++
b/paimon-spark/paimon-spark-4.1/src/test/scala/org/apache/paimon/spark/sql/MapSelectedKeysSharedShreddingE2ETest.scala
@@ -0,0 +1,21 @@
+/*
+ * 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.spark.sql
+
+class MapSelectedKeysSharedShreddingE2ETest extends
MapSelectedKeysSharedShreddingE2ETestBase {}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownMapSelectedKeys.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownMapSelectedKeys.scala
index 16db25531f..5ec0e9e281 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownMapSelectedKeys.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownMapSelectedKeys.scala
@@ -187,18 +187,25 @@ abstract class PushDownMapSelectedKeysBase extends
Rule[LogicalPlan] {
return false
}
+ val options = CoreOptions.fromMap(scan.table.options())
access.mapType match {
case MapType(StringType, _, _) =>
fieldType(scan.table.rowType(), access.fieldName) match {
case Some(mapType: org.apache.paimon.types.MapType) if
isStringKeyMap(mapType) =>
-
CoreOptions.fromMap(scan.table.options()).mapStorageLayout(access.fieldName) ==
- MapStorageLayout.SHARED_SHREDDING
+ options.mapStorageLayout(access.fieldName) ==
MapStorageLayout.SHARED_SHREDDING &&
+ !hasConfiguredAggregator(options, access.fieldName)
case _ => false
}
case _ => false
}
}
+ private def hasConfiguredAggregator(options: CoreOptions, fieldName:
String): Boolean = {
+ Option(options.fieldAggFunc(fieldName))
+ .orElse(Option(options.fieldsDefaultFunc()))
+ .isDefined
+ }
+
private def canEncodeKey(key: String): Boolean = {
!key.contains(MapSelectedKeysMetadataUtils.KEY_DELIMITER) &&
!key.startsWith(MapSelectedKeysMetadataUtils.METADATA_KEY)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonStrategy.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonStrategy.scala
index 24b32671c9..53a10b0287 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonStrategy.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonStrategy.scala
@@ -27,6 +27,7 @@ import org.apache.paimon.predicate.{Predicate,
PredicateBuilder}
import org.apache.paimon.spark.{PaimonRecordReaderIterator, PaimonScan,
PostponeMergeInputScan, PostponeMergeOnRead, SparkCatalog, SparkGenericCatalog,
SparkTable, SparkUtils}
import org.apache.paimon.spark.catalog.{SparkBaseCatalog, SupportView}
import org.apache.paimon.spark.catalyst.analysis.ResolvedPaimonView
+import org.apache.paimon.spark.catalyst.optimizer.PushDownMapSelectedKeys
import
org.apache.paimon.spark.catalyst.optimizer.RepartitionLateralVectorSearchInput
import
org.apache.paimon.spark.catalyst.plans.logical.{CopyIntoLocationCommand,
CopyIntoLocationSource, CopyIntoTableCommand, CreateOrReplaceTagCommand,
CreatePaimonView, DeleteTagCommand, DropPaimonView, LateralVectorSearch,
PaimonCallCommand, PaimonDropPartitions, PaimonTableValuedFunctions,
RenameTagCommand, ResolvedIdentifier, ShowPaimonViews, ShowTagsCommand,
TruncatePaimonTableWithFilter}
import org.apache.paimon.spark.data.SparkInternalRow
@@ -68,7 +69,19 @@ case class PaimonStrategy(spark: SparkSession)
import DataSourceV2Implicits._
protected lazy val catalogManager = spark.sessionState.catalogManager
- override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+ override def apply(plan: LogicalPlan): Seq[SparkPlan] = {
+ // Spark creates DataSourceV2ScanRelation after injected optimizer rules
have run. Apply this
+ // rewrite during physical planning, when the scan relation is available,
and let the regular
+ // Spark strategies plan the rewritten logical subtree.
+ val rewritten = PushDownMapSelectedKeys(plan)
+ if (!rewritten.fastEquals(plan)) {
+ planLater(rewritten) :: Nil
+ } else {
+ applyWithoutMapSelectedKeysPushDown(plan)
+ }
+ }
+
+ private def applyWithoutMapSelectedKeysPushDown(plan: LogicalPlan):
Seq[SparkPlan] = plan match {
case PhysicalOperation(projects, filters, relation:
DataSourceV2ScanRelation) =>
relation.scan match {
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala
index 95c2657036..e3701a468a 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala
@@ -101,8 +101,6 @@ class PaimonSparkSessionExtensions extends
(SparkSessionExtensions => Unit) {
// optimization rules
extensions.injectOptimizerRule(spark => ReplacePaimonFunctions(spark))
extensions.injectOptimizerRule(spark =>
OptimizeMetadataOnlyDeleteFromPaimonTable(spark))
- // TODO: Enable MAP selected-key pushdown after core reader supports
- // __PAIMON_MAP_SELECTED_KEYS read type.
extensions.injectOptimizerRule(_ => MergePaimonScalarSubqueries)
extensions.injectOptimizerRule(_ => RepartitionLateralVectorSearchInput)
extensions.injectOptimizerRule(_ => PushDownLateralVectorSearchFilter)
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
new file mode 100644
index 0000000000..f2351ac965
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/MapSelectedKeysSharedShreddingE2ETestBase.scala
@@ -0,0 +1,215 @@
+/*
+ * 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.spark.sql
+
+import org.apache.paimon.spark.{PaimonScan, PaimonSparkTestBase}
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+
+abstract class MapSelectedKeysSharedShreddingE2ETestBase extends
PaimonSparkTestBase {
+
+ Seq("parquet", "orc").foreach {
+ format =>
+ test(s"read selected shared-shredding map keys directly from $format") {
+ withTable("T") {
+ sql(s"""
+ |CREATE TABLE T (id INT, attrs MAP<STRING, BIGINT>)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'file.format' = '$format',
+ | 'fields.attrs.map.storage-layout' = 'shared-shredding',
+ | 'fields.attrs.map.shared-shredding.max-columns' = '1'
+ |)
+ |""".stripMargin)
+
+ sql("""
+ |INSERT INTO T VALUES
+ | (1, map('key1', CAST(10 AS BIGINT), 'key2', CAST(20 AS
BIGINT))),
+ | (2, map('key2', CAST(30 AS BIGINT), 'cold', CAST(40 AS
BIGINT)))
+ |""".stripMargin)
+
+ val query =
+ sql("SELECT id, attrs['key1'], attrs['key2'], attrs['missing']
FROM T ORDER BY id")
+ val sparkPlan = query.queryExecution.sparkPlan
+ val pushedMapSelectedKeys = sparkPlan.collectFirst {
+ case scan: BatchScanExec if scan.scan.isInstanceOf[PaimonScan] =>
+ scan.scan.asInstanceOf[PaimonScan].pushedMapSelectedKeys
+ }
+ val expectedPushedMapSelectedKeys = Map("attrs" -> Seq("key1",
"key2", "missing"))
+ assert(
+ pushedMapSelectedKeys.contains(expectedPushedMapSelectedKeys),
+ s"""Expected selected MAP keys to be pushed down.
+ |Physical plan:
+ |$sparkPlan""".stripMargin
+ )
+
+ checkAnswer(query, Row(1, 10L, 20L, null) :: Row(2, null, 30L, null)
:: Nil)
+ }
+ }
+
+ Seq(false, true).foreach {
+ thinMode =>
+ test(s"skip selected-key pushdown for merge_map from $format with
thin mode $thinMode") {
+ checkMapAggregatorRead(format, thinMode,
"fields.attrs.aggregate-function", "merge_map")
+ }
+ }
+
+ Seq("fields.attrs.aggregate-function",
"fields.default-aggregate-function").foreach {
+ aggregateOption =>
+ test(
+ s"skip selected-key pushdown for custom MAP aggregator configured
by $aggregateOption from $format") {
+ checkMapAggregatorRead(format, false, aggregateOption,
"my_merge_map")
+ }
+ }
+
+ test(s"read full shared-shredding map column from $format") {
+ withTable("T") {
+ sql(s"""
+ |CREATE TABLE T (id INT, attrs MAP<STRING, BIGINT>)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'file.format' = '$format',
+ | 'fields.attrs.map.storage-layout' = 'shared-shredding',
+ | 'fields.attrs.map.shared-shredding.max-columns' = '1'
+ |)
+ |""".stripMargin)
+
+ sql("""
+ |INSERT INTO T VALUES
+ | (1, map('key1', CAST(10 AS BIGINT), 'key2', CAST(20 AS
BIGINT))),
+ | (2, map('key2', CAST(30 AS BIGINT), 'cold', CAST(40 AS
BIGINT))),
+ | (3, NULL)
+ |""".stripMargin)
+
+ 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, null) :: Nil)
+ }
+ }
+
+ test(s"read selected normal map keys from $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))),
+ | (2, map('key2', CAST(30 AS BIGINT), 'cold', CAST(40 AS
BIGINT))),
+ | (3, NULL)
+ |""".stripMargin)
+
+ checkAnswer(
+ sql("SELECT id, attrs['key1'], attrs['key2'], attrs['missing']
FROM T ORDER BY id"),
+ Row(1, 10L, 20L, null) ::
+ Row(2, null, 30L, null) ::
+ Row(3, null, null, null) :: Nil
+ )
+ }
+ }
+
+ test(s"read full normal map column from $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))),
+ | (2, map('key2', CAST(30 AS BIGINT), 'cold', CAST(40 AS
BIGINT))),
+ | (3, NULL)
+ |""".stripMargin)
+
+ 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, null) :: Nil)
+ }
+ }
+ }
+
+ private def checkMapAggregatorRead(
+ format: String,
+ thinMode: Boolean,
+ aggregateOption: String,
+ aggregateFunction: String): Unit = {
+ withTable("T") {
+ sql(s"""
+ |CREATE TABLE T (
+ | id INT,
+ | attrs MAP<STRING, BIGINT>
+ |)
+ |TBLPROPERTIES (
+ | 'primary-key' = 'id',
+ | 'bucket' = '1',
+ | 'file.format' = '$format',
+ | 'data-file.thin-mode' = '$thinMode',
+ | 'merge-engine' = 'aggregation',
+ | '$aggregateOption' = '$aggregateFunction',
+ | 'fields.attrs.map.storage-layout' = 'shared-shredding',
+ | 'fields.attrs.map.shared-shredding.max-columns' = '1',
+ | 'write-only' = 'true'
+ |)
+ |""".stripMargin)
+
+ sql("""
+ |INSERT INTO T VALUES
+ | (1, map('key1', CAST(10 AS BIGINT), 'key2', CAST(20 AS
BIGINT))),
+ | (2, NULL)
+ |""".stripMargin)
+ sql("""
+ |INSERT INTO T VALUES
+ | (1, map('key2', CAST(30 AS BIGINT), 'cold', CAST(40 AS
BIGINT))),
+ | (2, map('key1', CAST(50 AS BIGINT)))
+ |""".stripMargin)
+
+ val query =
+ sql("SELECT id, attrs['key1'], attrs['key2'] FROM T ORDER BY id")
+ val sparkPlan = query.queryExecution.sparkPlan
+ val paimonScan = sparkPlan
+ .collectFirst {
+ case scan: BatchScanExec if scan.scan.isInstanceOf[PaimonScan] =>
+ scan.scan.asInstanceOf[PaimonScan]
+ }
+ .getOrElse(fail(s"Expected a Paimon scan in physical
plan:\n$sparkPlan"))
+ assert(
+ !paimonScan.pushedMapSelectedKeys.contains("attrs"),
+ s"""Expected selected-key pushdown to be skipped for
$aggregateFunction.
+ |Physical plan:
+ |$sparkPlan""".stripMargin
+ )
+
+ checkAnswer(query, Row(1, 10L, 30L) :: Row(2, 50L, null) :: Nil)
+ }
+ }
+}
