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 428c18fa83 [core] Validate unsupported MAP shared-shredding
configurations (#8487)
428c18fa83 is described below
commit 428c18fa836c7d5f5e29e54dc68a41cc19195568
Author: lxy <[email protected]>
AuthorDate: Tue Jul 7 15:20:18 2026 +0800
[core] Validate unsupported MAP shared-shredding configurations (#8487)
This is a subtask of [MAP shared-shredding
support](https://cwiki.apache.org/confluence/display/PAIMON/PIP-43%3A+Columnar+Storage+Optimization+for+MAP+Type+in+Paimon).
This PR splits out the validation and common compatibility work for MAP
shared-shredding so it can be reviewed independently before the
read/write path is wired.
It adds validation for unsupported MAP shared-shredding configurations,
including table modes, file formats, compression settings, nested file
index usage, and unsupported value types such as Variant, BLOB, and
MULTISET.
It also includes small common-layer fixes needed by later
shared-shredding materialization, such as field metadata compression
persistence and `RowToColumnConverter` handling for
CHAR/VARCHAR/BINARY/VARBINARY.
This PR does not yet enable reading or writing MAP shared-shredding
tables. If configured, the option is validated but the functional append
read/write path will be added in the next PR, together with append-table
E2E tests.
---
paimon-common/pom.xml | 1 +
.../paimon/data/columnar/RowToColumnConverter.java | 256 +++++++++++++-
.../data/shredding/MapSharedShreddingDefine.java | 2 +
.../data/shredding/MapSharedShreddingUtils.java | 45 ++-
.../data/columnar/RowToColumnConverterTest.java | 388 +++++++++++++++++++++
.../shredding/MapSharedShreddingUtilsTest.java | 61 +++-
.../shredding/MapSharedShreddingWritePlanTest.java | 4 +-
.../org/apache/paimon/schema/SchemaValidation.java | 126 ++++++-
.../apache/paimon/schema/SchemaValidationTest.java | 241 +++++++++++++
9 files changed, 1095 insertions(+), 29 deletions(-)
diff --git a/paimon-common/pom.xml b/paimon-common/pom.xml
index 961559c9b8..4ba3aac399 100644
--- a/paimon-common/pom.xml
+++ b/paimon-common/pom.xml
@@ -300,6 +300,7 @@ under the License.
<artifactSet>
<includes combine.children="append">
<include>org.antlr:antlr4-runtime</include>
+ <include>io.airlift:aircompressor</include>
<include>org.codehaus.janino:*</include>
<include>it.unimi.dsi:fastutil</include>
<include>org.roaringbitmap:RoaringBitmap</include>
diff --git
a/paimon-common/src/main/java/org/apache/paimon/data/columnar/RowToColumnConverter.java
b/paimon-common/src/main/java/org/apache/paimon/data/columnar/RowToColumnConverter.java
index f3fdba7b09..e35c7d77b1 100644
---
a/paimon-common/src/main/java/org/apache/paimon/data/columnar/RowToColumnConverter.java
+++
b/paimon-common/src/main/java/org/apache/paimon/data/columnar/RowToColumnConverter.java
@@ -18,6 +18,8 @@
package org.apache.paimon.data.columnar;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.Blob;
import org.apache.paimon.data.DataGetters;
import org.apache.paimon.data.Decimal;
import org.apache.paimon.data.InternalArray;
@@ -39,6 +41,7 @@ import
org.apache.paimon.data.columnar.writable.WritableIntVector;
import org.apache.paimon.data.columnar.writable.WritableLongVector;
import org.apache.paimon.data.columnar.writable.WritableShortVector;
import org.apache.paimon.data.columnar.writable.WritableTimestampVector;
+import org.apache.paimon.data.variant.GenericVariant;
import org.apache.paimon.data.variant.Variant;
import org.apache.paimon.types.ArrayType;
import org.apache.paimon.types.BigIntType;
@@ -88,6 +91,237 @@ public class RowToColumnConverter {
}
}
+ /** Create a reusable element converter for materializing column vectors.
*/
+ public static ElementConverter createElementConverter(DataType type) {
+ return new ElementConverter(TypeConverter.getConverterForType(type));
+ }
+
+ /** Converts one element into a writable column vector. */
+ public static class ElementConverter implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final TypeConverter converter;
+
+ private ElementConverter(TypeConverter converter) {
+ this.converter = converter;
+ }
+
+ public void append(DataGetters getters, int pos, WritableColumnVector
vector) {
+ converter.append(getters, pos, vector);
+ }
+
+ public void append(ColumnVector source, int rowId,
WritableColumnVector vector) {
+ converter.append(new ColumnVectorDataGetters(source, rowId), 0,
vector);
+ }
+
+ public void append(Object value, WritableColumnVector vector) {
+ converter.append(new ObjectDataGetters(value), 0, vector);
+ }
+ }
+
+ private static class ColumnVectorDataGetters implements DataGetters {
+
+ private final ColumnVector vector;
+ private final int rowId;
+
+ private ColumnVectorDataGetters(ColumnVector vector, int rowId) {
+ this.vector = vector;
+ this.rowId = rowId;
+ }
+
+ @Override
+ public boolean isNullAt(int pos) {
+ return vector.isNullAt(rowId);
+ }
+
+ @Override
+ public boolean getBoolean(int pos) {
+ return ((BooleanColumnVector) vector).getBoolean(rowId);
+ }
+
+ @Override
+ public byte getByte(int pos) {
+ return ((ByteColumnVector) vector).getByte(rowId);
+ }
+
+ @Override
+ public short getShort(int pos) {
+ return ((ShortColumnVector) vector).getShort(rowId);
+ }
+
+ @Override
+ public int getInt(int pos) {
+ return ((IntColumnVector) vector).getInt(rowId);
+ }
+
+ @Override
+ public long getLong(int pos) {
+ return ((LongColumnVector) vector).getLong(rowId);
+ }
+
+ @Override
+ public float getFloat(int pos) {
+ return ((FloatColumnVector) vector).getFloat(rowId);
+ }
+
+ @Override
+ public double getDouble(int pos) {
+ return ((DoubleColumnVector) vector).getDouble(rowId);
+ }
+
+ @Override
+ public BinaryString getString(int pos) {
+ BytesColumnVector.Bytes bytes = ((BytesColumnVector)
vector).getBytes(rowId);
+ return BinaryString.fromBytes(bytes.data, bytes.offset, bytes.len);
+ }
+
+ @Override
+ public Decimal getDecimal(int pos, int precision, int scale) {
+ return ((DecimalColumnVector) vector).getDecimal(rowId, precision,
scale);
+ }
+
+ @Override
+ public Timestamp getTimestamp(int pos, int precision) {
+ return ((TimestampColumnVector) vector).getTimestamp(rowId,
precision);
+ }
+
+ @Override
+ public byte[] getBinary(int pos) {
+ return ((BytesColumnVector) vector).getBytes(rowId).getBytes();
+ }
+
+ @Override
+ public Variant getVariant(int pos) {
+ InternalRow row = getRow(pos, 2);
+ return new GenericVariant(row.getBinary(0), row.getBinary(1));
+ }
+
+ @Override
+ public Blob getBlob(int pos) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public InternalArray getArray(int pos) {
+ return ((ArrayColumnVector) vector).getArray(rowId);
+ }
+
+ @Override
+ public InternalVector getVector(int pos) {
+ return ((VecColumnVector) vector).getVector(rowId);
+ }
+
+ @Override
+ public InternalMap getMap(int pos) {
+ return ((MapColumnVector) vector).getMap(rowId);
+ }
+
+ @Override
+ public InternalRow getRow(int pos, int numFields) {
+ return ((RowColumnVector) vector).getRow(rowId);
+ }
+ }
+
+ private static class ObjectDataGetters implements DataGetters {
+
+ private final Object value;
+
+ private ObjectDataGetters(Object value) {
+ this.value = value;
+ }
+
+ @Override
+ public boolean isNullAt(int pos) {
+ return value == null;
+ }
+
+ @Override
+ public boolean getBoolean(int pos) {
+ return (Boolean) value;
+ }
+
+ @Override
+ public byte getByte(int pos) {
+ return (Byte) value;
+ }
+
+ @Override
+ public short getShort(int pos) {
+ return (Short) value;
+ }
+
+ @Override
+ public int getInt(int pos) {
+ return (Integer) value;
+ }
+
+ @Override
+ public long getLong(int pos) {
+ return (Long) value;
+ }
+
+ @Override
+ public float getFloat(int pos) {
+ return (Float) value;
+ }
+
+ @Override
+ public double getDouble(int pos) {
+ return (Double) value;
+ }
+
+ @Override
+ public BinaryString getString(int pos) {
+ return (BinaryString) value;
+ }
+
+ @Override
+ public Decimal getDecimal(int pos, int precision, int scale) {
+ return (Decimal) value;
+ }
+
+ @Override
+ public Timestamp getTimestamp(int pos, int precision) {
+ return (Timestamp) value;
+ }
+
+ @Override
+ public byte[] getBinary(int pos) {
+ return (byte[]) value;
+ }
+
+ @Override
+ public Variant getVariant(int pos) {
+ return (Variant) value;
+ }
+
+ @Override
+ public Blob getBlob(int pos) {
+ return (Blob) value;
+ }
+
+ @Override
+ public InternalArray getArray(int pos) {
+ return (InternalArray) value;
+ }
+
+ @Override
+ public InternalVector getVector(int pos) {
+ return (InternalVector) value;
+ }
+
+ @Override
+ public InternalMap getMap(int pos) {
+ return (InternalMap) value;
+ }
+
+ @Override
+ public InternalRow getRow(int pos, int numFields) {
+ return (InternalRow) value;
+ }
+ }
+
private interface TypeConverter extends Serializable {
void append(DataGetters row, int column, WritableColumnVector cv);
@@ -107,20 +341,12 @@ public class RowToColumnConverter {
@Override
public TypeConverter visit(CharType charType) {
- return createConverter(
- charType.isNullable(),
- (row, column, cv) ->
- ((WritableByteVector)
cv).appendByte(row.getByte(column)));
+ return stringConverter(charType.isNullable());
}
@Override
public TypeConverter visit(VarCharType varCharType) {
- return createConverter(
- varCharType.isNullable(),
- (row, column, cv) -> {
- byte[] bytes = row.getString(column).toBytes();
- ((WritableBytesVector) cv).appendByteArray(bytes,
0, bytes.length);
- });
+ return stringConverter(varCharType.isNullable());
}
@Override
@@ -246,6 +472,7 @@ public class RowToColumnConverter {
return createConverter(
variantType.isNullable(),
(row, column, cv) -> {
+ ((HeapRowVector) cv).appendRow();
WritableBytesVector valueVector =
(WritableBytesVector) cv.getChildren()[0];
WritableBytesVector metaDataVector =
@@ -383,6 +610,15 @@ public class RowToColumnConverter {
});
}
+ private static TypeConverter stringConverter(boolean nullable) {
+ return createConverter(
+ nullable,
+ (row, column, cv) -> {
+ byte[] bytes = row.getString(column).toBytes();
+ ((WritableBytesVector) cv).appendByteArray(bytes,
0, bytes.length);
+ });
+ }
+
private static TypeConverter timestampConverter(boolean nullable,
int precision) {
return createConverter(
nullable,
diff --git
a/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingDefine.java
b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingDefine.java
index 6b5cce93d6..eb00505481 100644
---
a/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingDefine.java
+++
b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingDefine.java
@@ -24,6 +24,8 @@ public class MapSharedShreddingDefine {
public static final String VERSION = "paimon.map.shared-shredding.version";
public static final int CURRENT_VERSION = 1;
public static final String FIELD_DICT =
"paimon.map.shared-shredding.field-dict";
+ public static final String FIELD_DICT_COMPRESSION =
+ "paimon.map.shared-shredding.field-dict-compression";
public static final String FIELD_DICT_ORIGINAL_SIZE =
"paimon.map.shared-shredding.field-dict-original-size";
public static final String FIELD_COLUMNS =
"paimon.map.shared-shredding.field-columns";
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 851edb01d8..82f191f624 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
@@ -43,6 +43,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
@@ -115,12 +116,14 @@ public class MapSharedShreddingUtils {
MapSharedShreddingFieldMeta fieldMeta,
@Nullable String compression,
Map<String, String> metadata) {
+ String fieldDictCompression =
normalizeFieldDictCompression(compression);
metadata.put(
MapShreddingDefine.STORAGE_LAYOUT,
MapShreddingDefine.STORAGE_LAYOUT_SHARED_SHREDDING);
metadata.put(
MapSharedShreddingDefine.VERSION,
String.valueOf(MapSharedShreddingDefine.CURRENT_VERSION));
+ metadata.put(MapSharedShreddingDefine.FIELD_DICT_COMPRESSION,
fieldDictCompression);
String fieldDictJson = toJson(new TreeMap<>(fieldMeta.nameToId()));
metadata.put(
@@ -129,7 +132,9 @@ public class MapSharedShreddingUtils {
metadata.put(
MapSharedShreddingDefine.FIELD_DICT,
bytesToString(
-
compress(fieldDictJson.getBytes(StandardCharsets.UTF_8), compression)));
+ compress(
+ fieldDictJson.getBytes(StandardCharsets.UTF_8),
+ fieldDictCompression)));
metadata.put(
MapSharedShreddingDefine.FIELD_COLUMNS,
toJson(sortedFieldColumns(fieldMeta.fieldToColumns())));
@@ -143,6 +148,18 @@ public class MapSharedShreddingUtils {
public static MapSharedShreddingFieldMeta deserializeMetadata(
@Nullable Map<String, String> metadata, @Nullable String
compression) {
+ return deserializeMetadata(metadata, compression, true);
+ }
+
+ public static MapSharedShreddingFieldMeta deserializeMetadata(
+ @Nullable Map<String, String> metadata) {
+ return deserializeMetadata(metadata, null, false);
+ }
+
+ private static MapSharedShreddingFieldMeta deserializeMetadata(
+ @Nullable Map<String, String> metadata,
+ @Nullable String fallbackCompression,
+ boolean useFallbackCompression) {
if (!hasShreddingMetadata(metadata)) {
throw new IllegalArgumentException(
"metadata is null or storage layout is not
shared-shredding");
@@ -158,6 +175,13 @@ public class MapSharedShreddingUtils {
int originalLength =
requiredInt(metadata,
MapSharedShreddingDefine.FIELD_DICT_ORIGINAL_SIZE);
+ String compression =
+ normalizeFieldDictCompression(
+ metadata.getOrDefault(
+
MapSharedShreddingDefine.FIELD_DICT_COMPRESSION,
+ useFallbackCompression
+ ? fallbackCompression
+ :
MapSharedShreddingDefine.DEFAULT_DICT_COMPRESSION));
byte[] fieldDictBytes =
decompress(
stringToBytes(requiredValue(metadata,
MapSharedShreddingDefine.FIELD_DICT)),
@@ -272,6 +296,25 @@ public class MapSharedShreddingUtils {
return compression == null || "none".equalsIgnoreCase(compression);
}
+ public static String normalizeFieldDictCompression(@Nullable String
compression) {
+ if (compression == null) {
+ return MapSharedShreddingDefine.DEFAULT_DICT_COMPRESSION;
+ }
+
+ String normalized = compression.toLowerCase(Locale.ROOT);
+ switch (normalized) {
+ case "none":
+ case "lz4":
+ case "zstd":
+ return normalized;
+ default:
+ throw new IllegalArgumentException(
+ "MAP shared-shredding only supports none/lz4/zstd
compression, but is "
+ + compression
+ + ".");
+ }
+ }
+
private static String bytesToString(byte[] bytes) {
return new String(bytes, StandardCharsets.ISO_8859_1);
}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/data/columnar/RowToColumnConverterTest.java
b/paimon-common/src/test/java/org/apache/paimon/data/columnar/RowToColumnConverterTest.java
index 95c20fc971..4318a54516 100644
---
a/paimon-common/src/test/java/org/apache/paimon/data/columnar/RowToColumnConverterTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/data/columnar/RowToColumnConverterTest.java
@@ -24,6 +24,9 @@ import org.apache.paimon.data.Decimal;
import org.apache.paimon.data.GenericArray;
import org.apache.paimon.data.GenericMap;
import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalArray;
+import org.apache.paimon.data.InternalMap;
+import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.InternalVector;
import org.apache.paimon.data.Timestamp;
import org.apache.paimon.data.columnar.heap.HeapArrayVector;
@@ -37,13 +40,16 @@ 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.data.columnar.heap.HeapShortVector;
+import org.apache.paimon.data.columnar.heap.HeapTimestampVector;
import org.apache.paimon.data.columnar.heap.HeapVectorColumnVector;
import org.apache.paimon.data.columnar.writable.WritableColumnVector;
+import org.apache.paimon.data.variant.GenericVariant;
import org.apache.paimon.types.ArrayType;
import org.apache.paimon.types.BigIntType;
import org.apache.paimon.types.BinaryType;
import org.apache.paimon.types.BooleanType;
import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.DateType;
import org.apache.paimon.types.DecimalType;
@@ -62,8 +68,12 @@ import org.junit.jupiter.api.Test;
import java.lang.reflect.Proxy;
import java.math.BigDecimal;
+import java.util.Arrays;
import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
+import java.util.function.BiConsumer;
+import java.util.function.Supplier;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -233,6 +243,57 @@ public class RowToColumnConverterTest {
assertThat(new
String(bytesVector.getBytes(1).getBytes())).isEqualTo("world");
}
+ @Test
+ public void testConvertCharType() {
+ RowType rowType = RowType.of(new DataField(0, "f",
DataTypes.CHAR(10)));
+ RowToColumnConverter converter = new RowToColumnConverter(rowType);
+
+ GenericRow row1 = GenericRow.of(BinaryString.fromString("hello"));
+ GenericRow row2 = GenericRow.of(BinaryString.fromString("world"));
+
+ HeapBytesVector bytesVector = new HeapBytesVector(2);
+ WritableColumnVector[] vectors = new WritableColumnVector[]
{bytesVector};
+
+ converter.convert(row1, vectors);
+ converter.convert(row2, vectors);
+
+ assertThat(new
String(bytesVector.getBytes(0).getBytes())).isEqualTo("hello");
+ assertThat(new
String(bytesVector.getBytes(1).getBytes())).isEqualTo("world");
+ }
+
+ @Test
+ public void testElementConverterAppendFromObjectAndColumnVector() {
+ RowToColumnConverter.ElementConverter converter =
+
RowToColumnConverter.createElementConverter(DataTypes.STRING());
+ HeapBytesVector sourceVector = new HeapBytesVector(1);
+ sourceVector.appendByteArray("from-vector".getBytes(), 0,
"from-vector".length());
+
+ HeapBytesVector targetVector = new HeapBytesVector(2);
+ converter.append(BinaryString.fromString("from-object"), targetVector);
+ converter.append(sourceVector, 0, targetVector);
+
+ assertThat(new
String(targetVector.getBytes(0).getBytes())).isEqualTo("from-object");
+ assertThat(new
String(targetVector.getBytes(1).getBytes())).isEqualTo("from-vector");
+ }
+
+ @Test
+ public void testElementConverterSupportsAllSupportedTypes() {
+ for (ElementCase elementCase : elementCases()) {
+ elementCase.verify();
+ }
+ }
+
+ @Test
+ public void testElementConverterRejectsUnsupportedTypes() {
+ assertThatThrownBy(() ->
RowToColumnConverter.createElementConverter(DataTypes.BLOB()))
+ .isInstanceOf(UnsupportedOperationException.class);
+ assertThatThrownBy(
+ () ->
+ RowToColumnConverter.createElementConverter(
+ DataTypes.MULTISET(DataTypes.INT())))
+ .isInstanceOf(UnsupportedOperationException.class);
+ }
+
@Test
public void testConvertBinaryType() {
RowType rowType = RowType.of(new DataField(0, "f", new BinaryType(5)));
@@ -251,6 +312,23 @@ public class RowToColumnConverterTest {
assertThat(new
String(bytesVector.getBytes(1).getBytes())).isEqualTo("world");
}
+ @Test
+ public void testConvertVariantType() {
+ RowType rowType = RowType.of(new DataField(0, "f",
DataTypes.VARIANT()));
+ RowToColumnConverter converter = new RowToColumnConverter(rowType);
+ GenericVariant variant = GenericVariant.fromJson("{\"a\":1}");
+
+ HeapRowVector rowVector =
+ new HeapRowVector(1, new HeapBytesVector(1), new
HeapBytesVector(1));
+ WritableColumnVector[] vectors = new WritableColumnVector[]
{rowVector};
+
+ converter.convert(GenericRow.of(variant), vectors);
+
+ assertThat(rowVector.isNullAt(0)).isFalse();
+
assertThat(rowVector.getRow(0).getBinary(0)).isEqualTo(variant.value());
+
assertThat(rowVector.getRow(0).getBinary(1)).isEqualTo(variant.metadata());
+ }
+
@Test
public void testConvertVarBinaryType() {
RowType rowType = RowType.of(new DataField(0, "f", new
VarBinaryType(10)));
@@ -608,4 +686,314 @@ public class RowToColumnConverterTest {
});
return (InternalVector) proxy;
}
+
+ private static List<ElementCase> elementCases() {
+ Decimal smallDecimal = Decimal.fromBigDecimal(new BigDecimal("12.34"),
5, 2);
+ Decimal longDecimal = Decimal.fromBigDecimal(new
BigDecimal("1234567890.12"), 12, 2);
+ Decimal bytesDecimal =
+ Decimal.fromBigDecimal(new
BigDecimal("12345678901234567890.12"), 22, 2);
+ Timestamp millisTimestamp = Timestamp.fromEpochMillis(1234567890L);
+ Timestamp microsTimestamp = Timestamp.fromMicros(1234567890123456L);
+ Timestamp nanosTimestamp = Timestamp.fromEpochMillis(1234567890L,
123456);
+ GenericVariant variant = GenericVariant.fromJson("{\"a\":1}");
+
+ Map<BinaryString, Integer> map = new LinkedHashMap<>();
+ map.put(BinaryString.fromString("k1"), 1);
+ map.put(BinaryString.fromString("k2"), 2);
+
+ return Arrays.asList(
+ new ElementCase(
+ DataTypes.BOOLEAN(),
+ true,
+ () -> new HeapBooleanVector(3),
+ (vector, rowId) ->
+ assertThat(((HeapBooleanVector)
vector).getBoolean(rowId))
+ .isTrue()),
+ new ElementCase(
+ DataTypes.TINYINT(),
+ (byte) 12,
+ () -> new HeapByteVector(3),
+ (vector, rowId) ->
+ assertThat(((HeapByteVector)
vector).getByte(rowId))
+ .isEqualTo((byte) 12)),
+ new ElementCase(
+ DataTypes.SMALLINT(),
+ (short) 123,
+ () -> new HeapShortVector(3),
+ (vector, rowId) ->
+ assertThat(((HeapShortVector)
vector).getShort(rowId))
+ .isEqualTo((short) 123)),
+ new ElementCase(
+ DataTypes.INT(),
+ 123,
+ () -> new HeapIntVector(3),
+ (vector, rowId) ->
+ assertThat(((HeapIntVector)
vector).getInt(rowId)).isEqualTo(123)),
+ new ElementCase(
+ DataTypes.BIGINT(),
+ 123L,
+ () -> new HeapLongVector(3),
+ (vector, rowId) ->
+ assertThat(((HeapLongVector)
vector).getLong(rowId))
+ .isEqualTo(123L)),
+ new ElementCase(
+ DataTypes.FLOAT(),
+ 1.25F,
+ () -> new HeapFloatVector(3),
+ (vector, rowId) ->
+ assertThat(((HeapFloatVector)
vector).getFloat(rowId))
+ .isEqualTo(1.25F)),
+ new ElementCase(
+ DataTypes.DOUBLE(),
+ 2.5D,
+ () -> new HeapDoubleVector(3),
+ (vector, rowId) ->
+ assertThat(((HeapDoubleVector)
vector).getDouble(rowId))
+ .isEqualTo(2.5D)),
+ new ElementCase(
+ DataTypes.DATE(),
+ 12345,
+ () -> new HeapIntVector(3),
+ (vector, rowId) ->
+ assertThat(((HeapIntVector)
vector).getInt(rowId))
+ .isEqualTo(12345)),
+ new ElementCase(
+ DataTypes.TIME(),
+ 67890,
+ () -> new HeapIntVector(3),
+ (vector, rowId) ->
+ assertThat(((HeapIntVector)
vector).getInt(rowId))
+ .isEqualTo(67890)),
+ new ElementCase(
+ DataTypes.CHAR(10),
+ BinaryString.fromString("char"),
+ () -> new HeapBytesVector(3),
+ (vector, rowId) ->
+ assertThat(bytes((HeapBytesVector) vector,
rowId))
+ .isEqualTo("char")),
+ new ElementCase(
+ DataTypes.VARCHAR(10),
+ BinaryString.fromString("varchar"),
+ () -> new HeapBytesVector(3),
+ (vector, rowId) ->
+ assertThat(bytes((HeapBytesVector) vector,
rowId))
+ .isEqualTo("varchar")),
+ new ElementCase(
+ DataTypes.BINARY(6),
+ "binary".getBytes(),
+ () -> new HeapBytesVector(3),
+ (vector, rowId) ->
+ assertThat(bytes((HeapBytesVector) vector,
rowId))
+ .isEqualTo("binary")),
+ new ElementCase(
+ DataTypes.VARBINARY(9),
+ "varbinary".getBytes(),
+ () -> new HeapBytesVector(3),
+ (vector, rowId) ->
+ assertThat(bytes((HeapBytesVector) vector,
rowId))
+ .isEqualTo("varbinary")),
+ new ElementCase(
+ DataTypes.DECIMAL(5, 2),
+ smallDecimal,
+ () -> new HeapIntVector(3),
+ decimalSource(smallDecimal),
+ (vector, rowId) ->
+ assertThat(((HeapIntVector)
vector).getInt(rowId))
+ .isEqualTo((int)
smallDecimal.toUnscaledLong())),
+ new ElementCase(
+ DataTypes.DECIMAL(12, 2),
+ longDecimal,
+ () -> new HeapLongVector(3),
+ decimalSource(longDecimal),
+ (vector, rowId) ->
+ assertThat(((HeapLongVector)
vector).getLong(rowId))
+
.isEqualTo(longDecimal.toUnscaledLong())),
+ new ElementCase(
+ DataTypes.DECIMAL(22, 2),
+ bytesDecimal,
+ () -> new HeapBytesVector(3),
+ decimalSource(bytesDecimal),
+ (vector, rowId) ->
+ assertThat(((HeapBytesVector)
vector).getBytes(rowId).getBytes())
+
.isEqualTo(bytesDecimal.toUnscaledBytes())),
+ new ElementCase(
+ DataTypes.TIMESTAMP(3),
+ millisTimestamp,
+ () -> new HeapLongVector(3),
+ timestampSource(millisTimestamp),
+ (vector, rowId) ->
+ assertThat(((HeapLongVector)
vector).getLong(rowId))
+
.isEqualTo(millisTimestamp.getMillisecond())),
+ new ElementCase(
+ DataTypes.TIMESTAMP(6),
+ microsTimestamp,
+ () -> new HeapLongVector(3),
+ timestampSource(microsTimestamp),
+ (vector, rowId) ->
+ assertThat(((HeapLongVector)
vector).getLong(rowId))
+
.isEqualTo(microsTimestamp.toMicros())),
+ new ElementCase(
+ DataTypes.TIMESTAMP(9),
+ nanosTimestamp,
+ () -> new HeapTimestampVector(3),
+ (vector, rowId) ->
+ assertThat(((HeapTimestampVector)
vector).getTimestamp(rowId, 9))
+ .isEqualTo(nanosTimestamp)),
+ new ElementCase(
+ DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(3),
+ millisTimestamp,
+ () -> new HeapLongVector(3),
+ timestampSource(millisTimestamp),
+ (vector, rowId) ->
+ assertThat(((HeapLongVector)
vector).getLong(rowId))
+
.isEqualTo(millisTimestamp.getMillisecond())),
+ new ElementCase(
+ DataTypes.VARIANT(),
+ variant,
+ () -> new HeapRowVector(3, new HeapBytesVector(3), new
HeapBytesVector(3)),
+ (vector, rowId) -> {
+ InternalRow row = ((HeapRowVector)
vector).getRow(rowId);
+
assertThat(row.getBinary(0)).isEqualTo(variant.value());
+
assertThat(row.getBinary(1)).isEqualTo(variant.metadata());
+ }),
+ new ElementCase(
+ DataTypes.ARRAY(DataTypes.INT()),
+ new GenericArray(new Object[] {1, 2, 3}),
+ () -> new HeapArrayVector(3, new HeapIntVector(9)),
+ (vector, rowId) -> {
+ InternalArray array = ((HeapArrayVector)
vector).getArray(rowId);
+ assertThat(array.size()).isEqualTo(3);
+ assertThat(array.getInt(0)).isEqualTo(1);
+ assertThat(array.getInt(1)).isEqualTo(2);
+ assertThat(array.getInt(2)).isEqualTo(3);
+ }),
+ new ElementCase(
+ DataTypes.MAP(DataTypes.STRING(), DataTypes.INT()),
+ new GenericMap(map),
+ () -> new HeapMapVector(3, new HeapBytesVector(6), new
HeapIntVector(6)),
+ (vector, rowId) -> {
+ InternalMap internalMap = ((HeapMapVector)
vector).getMap(rowId);
+ assertThat(internalMap.size()).isEqualTo(2);
+
assertThat(internalMap.keyArray().getString(0).toString())
+ .isEqualTo("k1");
+
assertThat(internalMap.valueArray().getInt(0)).isEqualTo(1);
+
assertThat(internalMap.keyArray().getString(1).toString())
+ .isEqualTo("k2");
+
assertThat(internalMap.valueArray().getInt(1)).isEqualTo(2);
+ }),
+ new ElementCase(
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "id", DataTypes.INT()),
+ DataTypes.FIELD(1, "name",
DataTypes.STRING())),
+ GenericRow.of(1, BinaryString.fromString("Alice")),
+ () -> new HeapRowVector(3, new HeapIntVector(3), new
HeapBytesVector(3)),
+ (vector, rowId) -> {
+ InternalRow row = ((HeapRowVector)
vector).getRow(rowId);
+ assertThat(row.getInt(0)).isEqualTo(1);
+
assertThat(row.getString(1).toString()).isEqualTo("Alice");
+ }),
+ new ElementCase(
+ DataTypes.VECTOR(3, DataTypes.FLOAT()),
+ BinaryVector.fromPrimitiveArray(new float[] {1.0F,
2.0F, 3.0F}),
+ () -> new HeapVectorColumnVector(3, new
HeapFloatVector(9), 3),
+ (vector, rowId) ->
+ assertThat(
+ ((HeapVectorColumnVector)
vector)
+ .getVector(rowId)
+ .toFloatArray())
+ .containsExactly(1.0F, 2.0F, 3.0F)));
+ }
+
+ private static String bytes(HeapBytesVector vector, int rowId) {
+ return new String(vector.getBytes(rowId).getBytes());
+ }
+
+ private static Supplier<ColumnVector> decimalSource(Decimal decimal) {
+ return () ->
+ new DecimalColumnVector() {
+ @Override
+ public boolean isNullAt(int i) {
+ return false;
+ }
+
+ @Override
+ public Decimal getDecimal(int i, int precision, int scale)
{
+ return decimal;
+ }
+ };
+ }
+
+ private static Supplier<ColumnVector> timestampSource(Timestamp timestamp)
{
+ return () ->
+ new TimestampColumnVector() {
+ @Override
+ public boolean isNullAt(int i) {
+ return false;
+ }
+
+ @Override
+ public Timestamp getTimestamp(int i, int precision) {
+ return timestamp;
+ }
+ };
+ }
+
+ private static class ElementCase {
+
+ private final DataType type;
+ private final Object value;
+ private final Supplier<WritableColumnVector> vectorSupplier;
+ private final Supplier<ColumnVector> sourceSupplier;
+ private final BiConsumer<ColumnVector, Integer> assertion;
+
+ private ElementCase(
+ DataType type,
+ Object value,
+ Supplier<WritableColumnVector> vectorSupplier,
+ BiConsumer<ColumnVector, Integer> assertion) {
+ this(
+ type,
+ value,
+ vectorSupplier,
+ () -> createSourceVector(type, value, vectorSupplier),
+ assertion);
+ }
+
+ private ElementCase(
+ DataType type,
+ Object value,
+ Supplier<WritableColumnVector> vectorSupplier,
+ Supplier<ColumnVector> sourceSupplier,
+ BiConsumer<ColumnVector, Integer> assertion) {
+ this.type = type;
+ this.value = value;
+ this.vectorSupplier = vectorSupplier;
+ this.sourceSupplier = sourceSupplier;
+ this.assertion = assertion;
+ }
+
+ private void verify() {
+ RowToColumnConverter.ElementConverter converter =
+ RowToColumnConverter.createElementConverter(type);
+
+ WritableColumnVector objectTarget = vectorSupplier.get();
+ converter.append(value, objectTarget);
+ assertion.accept(objectTarget, 0);
+
+ WritableColumnVector vectorTarget = vectorSupplier.get();
+ converter.append(sourceSupplier.get(), 0, vectorTarget);
+ assertion.accept(vectorTarget, 0);
+
+ WritableColumnVector nullTarget = vectorSupplier.get();
+ converter.append(null, nullTarget);
+ assertThat(nullTarget.isNullAt(0)).isTrue();
+ }
+
+ private static ColumnVector createSourceVector(
+ DataType type, Object value, Supplier<WritableColumnVector>
vectorSupplier) {
+ WritableColumnVector source = vectorSupplier.get();
+ RowToColumnConverter.createElementConverter(type).append(value,
source);
+ return source;
+ }
+ }
}
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 e9d8d03f09..9c86109efa 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
@@ -204,7 +204,7 @@ class MapSharedShreddingUtilsTest {
new MapSharedShreddingFieldMeta(nameToId, fieldToColumns,
overflowSet, 3, 2);
String expectedDict = "{\"age\":0,\"name\":1}";
- for (String compression : Arrays.<String>asList(null, "none", "NONE"))
{
+ for (String compression : Arrays.asList("none", "NONE")) {
Map<String, String> metadata = new HashMap<>();
MapSharedShreddingUtils.serializeMetadata(original, compression,
metadata);
@@ -212,6 +212,8 @@ class MapSharedShreddingUtilsTest {
assertThat(metadata.get(MapShreddingDefine.STORAGE_LAYOUT))
.isEqualTo("shared-shredding");
assertThat(metadata.get(MapSharedShreddingDefine.VERSION)).isEqualTo("1");
+
assertThat(metadata.get(MapSharedShreddingDefine.FIELD_DICT_COMPRESSION))
+ .isEqualTo("none");
assertThat(metadata.get(MapSharedShreddingDefine.NUM_COLUMNS)).isEqualTo("3");
assertThat(metadata.get(MapSharedShreddingDefine.MAX_ROW_WIDTH)).isEqualTo("2");
@@ -221,9 +223,14 @@ class MapSharedShreddingUtilsTest {
assertThat(metadata.get(MapSharedShreddingDefine.FIELD_COLUMNS))
.isEqualTo("{\"0\":[0],\"1\":[1,2]}");
assertThat(metadata.get(MapSharedShreddingDefine.OVERFLOW_SET)).isEqualTo("[1,5]");
- assertThat(MapSharedShreddingUtils.deserializeMetadata(metadata,
compression))
- .isEqualTo(original);
+
assertThat(MapSharedShreddingUtils.deserializeMetadata(metadata)).isEqualTo(original);
}
+
+ Map<String, String> metadataWithoutCompression = new HashMap<>();
+ MapSharedShreddingUtils.serializeMetadata(original, "zstd",
metadataWithoutCompression);
+
metadataWithoutCompression.remove(MapSharedShreddingDefine.FIELD_DICT_COMPRESSION);
+
assertThat(MapSharedShreddingUtils.deserializeMetadata(metadataWithoutCompression))
+ .isEqualTo(original);
}
@Test
@@ -246,8 +253,9 @@ class MapSharedShreddingUtilsTest {
for (String compression : Arrays.asList("none", "lz4", "zstd")) {
Map<String, String> metadata = new HashMap<>();
MapSharedShreddingUtils.serializeMetadata(original, compression,
metadata);
- assertThat(MapSharedShreddingUtils.deserializeMetadata(metadata,
compression))
- .isEqualTo(original);
+
assertThat(metadata.get(MapSharedShreddingDefine.FIELD_DICT_COMPRESSION))
+ .isEqualTo(compression);
+
assertThat(MapSharedShreddingUtils.deserializeMetadata(metadata)).isEqualTo(original);
}
}
@@ -260,32 +268,56 @@ class MapSharedShreddingUtilsTest {
for (String compression : Arrays.asList("none", "lz4", "zstd")) {
Map<String, String> metadata = new HashMap<>();
MapSharedShreddingUtils.serializeMetadata(original, compression,
metadata);
- assertThat(MapSharedShreddingUtils.deserializeMetadata(metadata,
compression))
- .isEqualTo(original);
+
assertThat(metadata.get(MapSharedShreddingDefine.FIELD_DICT_COMPRESSION))
+ .isEqualTo(compression);
+
assertThat(MapSharedShreddingUtils.deserializeMetadata(metadata)).isEqualTo(original);
}
}
+ @Test
+ void testMetadataRejectsUnknownCompression() {
+ Map<String, Integer> nameToId = new TreeMap<>();
+ nameToId.put("age", 0);
+ MapSharedShreddingFieldMeta original =
+ new MapSharedShreddingFieldMeta(nameToId, new TreeMap<>(), new
HashSet<>(), 1, 1);
+
+ assertThatThrownBy(
+ () ->
+ MapSharedShreddingUtils.serializeMetadata(
+ original, "snappy", new HashMap<>()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(
+ "MAP shared-shredding only supports none/lz4/zstd
compression");
+
+ Map<String, String> metadata = new HashMap<>();
+ MapSharedShreddingUtils.serializeMetadata(original, "zstd", metadata);
+ metadata.put(MapSharedShreddingDefine.FIELD_DICT_COMPRESSION,
"snappy");
+ assertThatThrownBy(() ->
MapSharedShreddingUtils.deserializeMetadata(metadata))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(
+ "MAP shared-shredding only supports none/lz4/zstd
compression");
+ }
+
@Test
void testDeserializeMetadataErrors() {
- assertThatThrownBy(() ->
MapSharedShreddingUtils.deserializeMetadata(null, "none"))
+ assertThatThrownBy(() ->
MapSharedShreddingUtils.deserializeMetadata(null))
.hasMessageContaining("metadata is null or storage layout is
not shared-shredding");
Map<String, String> missingLayout = new HashMap<>();
missingLayout.put("some_key", "some_value");
- assertThatThrownBy(() ->
MapSharedShreddingUtils.deserializeMetadata(missingLayout, "none"))
+ assertThatThrownBy(() ->
MapSharedShreddingUtils.deserializeMetadata(missingLayout))
.hasMessageContaining("metadata is null or storage layout is
not shared-shredding");
Map<String, String> metadata = new HashMap<>();
metadata.put(MapShreddingDefine.STORAGE_LAYOUT, "default");
- assertThatThrownBy(() ->
MapSharedShreddingUtils.deserializeMetadata(metadata, "none"))
+ assertThatThrownBy(() ->
MapSharedShreddingUtils.deserializeMetadata(metadata))
.hasMessageContaining("metadata is null or storage layout is
not shared-shredding");
Map<String, String> missingVersion = new HashMap<>();
missingVersion.put(
MapShreddingDefine.STORAGE_LAYOUT,
MapShreddingDefine.STORAGE_LAYOUT_SHARED_SHREDDING);
- assertThatThrownBy(
- () ->
MapSharedShreddingUtils.deserializeMetadata(missingVersion, "none"))
+ assertThatThrownBy(() ->
MapSharedShreddingUtils.deserializeMetadata(missingVersion))
.hasMessageContaining(
"missing shredding metadata key:
paimon.map.shared-shredding.version");
@@ -296,7 +328,7 @@ class MapSharedShreddingUtilsTest {
wrongVersion.put(MapSharedShreddingDefine.VERSION, "999");
wrongVersion.put(MapSharedShreddingDefine.FIELD_DICT_ORIGINAL_SIZE,
"2");
wrongVersion.put(MapSharedShreddingDefine.FIELD_DICT, "{}");
- assertThatThrownBy(() ->
MapSharedShreddingUtils.deserializeMetadata(wrongVersion, "none"))
+ assertThatThrownBy(() ->
MapSharedShreddingUtils.deserializeMetadata(wrongVersion))
.hasMessageContaining("unsupported shared-shredding metadata
version: 999");
Map<String, String> missingFieldDict = new HashMap<>();
@@ -305,8 +337,7 @@ class MapSharedShreddingUtilsTest {
MapShreddingDefine.STORAGE_LAYOUT_SHARED_SHREDDING);
missingFieldDict.put(MapSharedShreddingDefine.VERSION, "1");
missingFieldDict.put(MapSharedShreddingDefine.FIELD_DICT_ORIGINAL_SIZE, "2");
- assertThatThrownBy(
- () ->
MapSharedShreddingUtils.deserializeMetadata(missingFieldDict, "none"))
+ assertThatThrownBy(() ->
MapSharedShreddingUtils.deserializeMetadata(missingFieldDict))
.hasMessageContaining(
"missing shredding metadata key:
paimon.map.shared-shredding.field-dict");
}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingWritePlanTest.java
b/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingWritePlanTest.java
index 69010c2594..2cfde7d648 100644
---
a/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingWritePlanTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingWritePlanTest.java
@@ -65,8 +65,10 @@ class MapSharedShreddingWritePlanTest {
Map<String, Map<String, String>> fieldMetadata =
writePlan.fieldMetadata("none");
assertThat(fieldMetadata).containsOnlyKeys("tags");
+ assertThat(fieldMetadata.get("tags"))
+
.containsEntry(MapSharedShreddingDefine.FIELD_DICT_COMPRESSION, "none");
MapSharedShreddingFieldMeta fieldMeta =
-
MapSharedShreddingUtils.deserializeMetadata(fieldMetadata.get("tags"), "none");
+
MapSharedShreddingUtils.deserializeMetadata(fieldMetadata.get("tags"));
assertThat(fieldMeta.nameToId()).containsEntry("a",
0).containsEntry("b", 1);
assertThat(fieldMeta.numColumns()).isEqualTo(4);
assertThat(fieldMeta.maxRowWidth()).isEqualTo(3);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index 491da71ff1..56b4676dca 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -45,6 +45,7 @@ import org.apache.paimon.types.MapType;
import org.apache.paimon.types.MultisetType;
import org.apache.paimon.types.RowType;
import org.apache.paimon.types.TimestampType;
+import org.apache.paimon.types.VariantType;
import org.apache.paimon.types.VectorType;
import org.apache.paimon.utils.Preconditions;
import org.apache.paimon.utils.SetUtils;
@@ -59,6 +60,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.function.Predicate;
import java.util.stream.Collectors;
import static org.apache.paimon.CoreOptions.BUCKET_KEY;
@@ -350,6 +352,8 @@ public class SchemaValidation {
validateMergeFunctionFactory(schema);
+ validateMapStorageLayout(schema, options);
+
validateFileIndex(schema);
validateRowTracking(schema, options);
@@ -363,8 +367,6 @@ public class SchemaValidation {
validatePkClusteringOverride(options);
validateManifestSort(schema, options);
-
- validateMapStorageLayout(schema, options);
}
public static void validateFallbackBranch(SchemaManager schemaManager,
TableSchema schema) {
@@ -621,6 +623,7 @@ public class SchemaValidation {
fieldMap.put(field.name(), field);
}
+ boolean hasSharedShredding = false;
for (String key : options.toMap().keySet()) {
if (!key.startsWith(FIELDS_PREFIX + ".") ||
!key.endsWith(layoutSuffix)) {
continue;
@@ -650,6 +653,7 @@ public class SchemaValidation {
continue;
}
+ hasSharedShredding = true;
if (!MapSharedShreddingUtils.isShreddingKeyMap(fieldType)) {
throw new IllegalArgumentException(
String.format(
@@ -658,6 +662,119 @@ public class SchemaValidation {
}
options.mapSharedShreddingMaxColumns(fieldName);
}
+
+ if (hasSharedShredding) {
+ validateMapSharedShreddingFileFormats(options);
+ validateMapSharedShreddingCompressions(options);
+ validateUnsupportedTypesWithMapSharedShredding(schema);
+ if (!schema.primaryKeys().isEmpty()) {
+ throw new IllegalArgumentException(
+ "MAP shared-shredding currently only supports
append-only tables.");
+ }
+ if (options.bucket() != -1 && !options.writeOnly()) {
+ throw new IllegalArgumentException(
+ "MAP shared-shredding currently requires bucket = -1
or write-only = true because rewrite/compaction is not supported.");
+ }
+ }
+ }
+
+ private static void
validateUnsupportedTypesWithMapSharedShredding(TableSchema schema) {
+ RowType rowType = new RowType(schema.fields());
+ if (containsType(rowType, type -> type instanceof VariantType)) {
+ throw new IllegalArgumentException(
+ "MAP shared-shredding currently cannot be used with
Variant fields.");
+ }
+ if (containsType(rowType, type -> type.is(DataTypeRoot.BLOB))) {
+ throw new IllegalArgumentException(
+ "MAP shared-shredding currently cannot be used with BLOB
fields.");
+ }
+ if (containsType(rowType, type -> type instanceof MultisetType)) {
+ throw new IllegalArgumentException(
+ "MAP shared-shredding currently cannot be used with
MULTISET fields.");
+ }
+ if (containsType(rowType, type -> type instanceof VectorType)) {
+ throw new IllegalArgumentException(
+ "MAP shared-shredding currently cannot be used with VECTOR
fields.");
+ }
+ }
+
+ private static boolean containsType(DataType dataType, Predicate<DataType>
predicate) {
+ if (predicate.test(dataType)) {
+ return true;
+ }
+ if (dataType instanceof RowType) {
+ for (DataField field : ((RowType) dataType).getFields()) {
+ if (containsType(field.type(), predicate)) {
+ return true;
+ }
+ }
+ } else if (dataType instanceof ArrayType) {
+ return containsType(((ArrayType) dataType).getElementType(),
predicate);
+ } else if (dataType instanceof MultisetType) {
+ return containsType(((MultisetType) dataType).getElementType(),
predicate);
+ } else if (dataType instanceof MapType) {
+ MapType mapType = (MapType) dataType;
+ return containsType(mapType.getKeyType(), predicate)
+ || containsType(mapType.getValueType(), predicate);
+ } else if (dataType instanceof VectorType) {
+ return containsType(((VectorType) dataType).getElementType(),
predicate);
+ }
+ return false;
+ }
+
+ private static void validateMapSharedShreddingFileFormats(CoreOptions
options) {
+ validateMapSharedShreddingFileFormat(
+ CoreOptions.FILE_FORMAT.key(), options.fileFormatString());
+ for (Map.Entry<Integer, String> entry :
options.fileFormatPerLevel().entrySet()) {
+ validateMapSharedShreddingFileFormat(
+ CoreOptions.FILE_FORMAT_PER_LEVEL.key() + "." +
entry.getKey(),
+ entry.getValue());
+ }
+ validateMapSharedShreddingFileFormat(
+ CoreOptions.CHANGELOG_FILE_FORMAT.key(),
options.changelogFileFormat());
+ validateMapSharedShreddingFileFormat(
+ CoreOptions.VECTOR_FILE_FORMAT.key(),
options.vectorFileFormatString());
+ }
+
+ private static void validateMapSharedShreddingCompressions(CoreOptions
options) {
+ validateMapSharedShreddingCompression(
+ CoreOptions.FILE_COMPRESSION.key(), options.fileCompression());
+ for (Map.Entry<Integer, String> entry :
options.fileCompressionPerLevel().entrySet()) {
+ validateMapSharedShreddingCompression(
+ CoreOptions.FILE_COMPRESSION_PER_LEVEL.key() + "." +
entry.getKey(),
+ entry.getValue());
+ }
+ validateMapSharedShreddingCompression(
+ CoreOptions.CHANGELOG_FILE_COMPRESSION.key(),
options.changelogFileCompression());
+ }
+
+ private static void validateMapSharedShreddingCompression(
+ String optionKey, String compression) {
+ if (StringUtils.isEmpty(compression)) {
+ return;
+ }
+ try {
+ MapSharedShreddingUtils.normalizeFieldDictCompression(compression);
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException(
+ String.format(
+ "MAP shared-shredding only supports none/lz4/zstd
compression, but %s is %s.",
+ optionKey, compression),
+ e);
+ }
+ }
+
+ private static void validateMapSharedShreddingFileFormat(String optionKey,
String format) {
+ if (StringUtils.isEmpty(format)) {
+ return;
+ }
+ if (!CoreOptions.FILE_FORMAT_PARQUET.equals(format)
+ && !CoreOptions.FILE_FORMAT_ORC.equals(format)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "MAP shared-shredding only supports parquet/orc
file formats, but %s is %s.",
+ optionKey, format));
+ }
}
private static void validateFileIndex(TableSchema schema) {
@@ -700,6 +817,11 @@ public class SchemaValidation {
+ "Only CHAR/VARCHAR/STRING is supported.",
columnName,
keyType);
+ checkArgument(
+ options.mapStorageLayout(columnName) !=
MapStorageLayout.SHARED_SHREDDING,
+ "Column '%s' is configured with
map.storage-layout=shared-shredding, "
+ + "which does not support nested file index.",
+ columnName);
}
for (String indexType : entry.getValue().keySet()) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
index f08c852c97..c1d673b168 100644
---
a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
@@ -20,6 +20,7 @@ package org.apache.paimon.schema;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
import org.assertj.core.api.ThrowableAssert;
@@ -313,6 +314,192 @@ class SchemaValidationTest {
.hasMessageContaining("options
map.shared-shredding.max-columns must > 0");
}
+ @Test
+ public void testMapSharedShreddingCannotCombineWithUnsupportedTypes() {
+ List<DataField> variantFields =
topLevelPayloadFields(DataTypes.VARIANT());
+ String variantMessage =
+ "MAP shared-shredding currently cannot be used with Variant
fields.";
+ assertMapSharedShreddingValidationFailed(
+ variantFields, mapSharedShreddingOptions(), variantMessage);
+
+ Map<String, String> inferVariantOptions = mapSharedShreddingOptions();
+
inferVariantOptions.put(CoreOptions.VARIANT_INFER_SHREDDING_SCHEMA.key(),
"true");
+ assertMapSharedShreddingValidationFailed(
+ variantFields, inferVariantOptions, variantMessage);
+
+ Map<String, String> explicitVariantOptions =
mapSharedShreddingOptions();
+ explicitVariantOptions.put(CoreOptions.VARIANT_SHREDDING_SCHEMA.key(),
"{}");
+ assertMapSharedShreddingValidationFailed(
+ variantFields, explicitVariantOptions, variantMessage);
+
+ assertMapSharedShreddingValidationFailed(
+ mapValueFields(DataTypes.BLOB()),
+ mapSharedShreddingOptions(),
+ "MAP shared-shredding currently cannot be used with BLOB
fields.");
+ assertMapSharedShreddingValidationFailed(
+ nestedMapValueFields(DataTypes.BLOB()),
+ mapSharedShreddingOptions(),
+ "MAP shared-shredding currently cannot be used with BLOB
fields.");
+ assertMapSharedShreddingValidationFailed(
+ topLevelPayloadFields(DataTypes.BLOB()),
+ mapSharedShreddingOptions(),
+ "MAP shared-shredding currently cannot be used with BLOB
fields.");
+
+ DataType multisetType = DataTypes.MULTISET(DataTypes.INT());
+ assertMapSharedShreddingValidationFailed(
+ mapValueFields(multisetType),
+ mapSharedShreddingOptions(),
+ "MAP shared-shredding currently cannot be used with MULTISET
fields.");
+ assertMapSharedShreddingValidationFailed(
+ nestedMapValueFields(multisetType),
+ mapSharedShreddingOptions(),
+ "MAP shared-shredding currently cannot be used with MULTISET
fields.");
+ assertMapSharedShreddingValidationFailed(
+ topLevelPayloadFields(multisetType),
+ mapSharedShreddingOptions(),
+ "MAP shared-shredding currently cannot be used with MULTISET
fields.");
+
+ DataType vectorType = DataTypes.VECTOR(3, DataTypes.FLOAT());
+ assertMapSharedShreddingValidationFailed(
+ mapValueFields(vectorType),
+ mapSharedShreddingOptions(),
+ "MAP shared-shredding currently cannot be used with VECTOR
fields.");
+ assertMapSharedShreddingValidationFailed(
+ topLevelPayloadFields(vectorType),
+ mapSharedShreddingOptions(),
+ "MAP shared-shredding currently cannot be used with VECTOR
fields.");
+ }
+
+ @Test
+ public void testMapSharedShreddingFileFormatValidation() {
+ Map<String, String> fileFormatOptions = mapSharedShreddingOptions();
+ fileFormatOptions.put(CoreOptions.FILE_FORMAT.key(), "avro");
+ assertThatThrownBy(
+ () ->
+ validateTableSchema(
+
mapSharedShreddingSchema(fileFormatOptions, emptyList())))
+ .hasMessageContaining(
+ "MAP shared-shredding only supports parquet/orc file
formats, but file.format is avro.");
+
+ Map<String, String> levelFormatOptions = mapSharedShreddingOptions();
+ levelFormatOptions.put(CoreOptions.FILE_FORMAT_PER_LEVEL.key(),
"0:avro");
+ assertThatThrownBy(
+ () ->
+ validateTableSchema(
+
mapSharedShreddingSchema(levelFormatOptions, emptyList())))
+ .hasMessageContaining(
+ "MAP shared-shredding only supports parquet/orc file
formats, but file.format.per.level.0 is avro.");
+
+ Map<String, String> changelogFormatOptions =
mapSharedShreddingOptions();
+ changelogFormatOptions.put(CoreOptions.CHANGELOG_FILE_FORMAT.key(),
"avro");
+ assertThatThrownBy(
+ () ->
+ validateTableSchema(
+ mapSharedShreddingSchema(
+ changelogFormatOptions,
emptyList())))
+ .hasMessageContaining(
+ "MAP shared-shredding only supports parquet/orc file
formats, but changelog-file.format is avro.");
+
+ Map<String, String> vectorFormatOptions = mapSharedShreddingOptions();
+ vectorFormatOptions.put(DATA_EVOLUTION_ENABLED.key(), "true");
+ vectorFormatOptions.put(CoreOptions.ROW_TRACKING_ENABLED.key(),
"true");
+ vectorFormatOptions.put(CoreOptions.VECTOR_FILE_FORMAT.key(), "json");
+ assertThatThrownBy(
+ () ->
+ validateTableSchema(
+
mapSharedShreddingSchema(vectorFormatOptions, emptyList())))
+ .hasMessageContaining(
+ "MAP shared-shredding only supports parquet/orc file
formats, but vector.file.format is json.");
+ }
+
+ @Test
+ public void testMapSharedShreddingFileCompressionValidation() {
+ Map<String, String> fileCompressionOptions =
mapSharedShreddingOptions();
+ fileCompressionOptions.put(CoreOptions.FILE_COMPRESSION.key(),
"snappy");
+ assertThatThrownBy(
+ () ->
+ validateTableSchema(
+ mapSharedShreddingSchema(
+ fileCompressionOptions,
emptyList())))
+ .hasMessageContaining(
+ "MAP shared-shredding only supports none/lz4/zstd
compression, but file.compression is snappy.");
+
+ Map<String, String> levelCompressionOptions =
mapSharedShreddingOptions();
+
levelCompressionOptions.put(CoreOptions.FILE_COMPRESSION_PER_LEVEL.key(),
"0:snappy");
+ assertThatThrownBy(
+ () ->
+ validateTableSchema(
+ mapSharedShreddingSchema(
+ levelCompressionOptions,
emptyList())))
+ .hasMessageContaining(
+ "MAP shared-shredding only supports none/lz4/zstd
compression, but file.compression.per.level.0 is snappy.");
+
+ Map<String, String> changelogCompressionOptions =
mapSharedShreddingOptions();
+
changelogCompressionOptions.put(CoreOptions.CHANGELOG_FILE_COMPRESSION.key(),
"snappy");
+ assertThatThrownBy(
+ () ->
+ validateTableSchema(
+ mapSharedShreddingSchema(
+ changelogCompressionOptions,
emptyList())))
+ .hasMessageContaining(
+ "MAP shared-shredding only supports none/lz4/zstd
compression, but changelog-file.compression is snappy.");
+
+ for (String compression : Arrays.asList("none", "lz4", "zstd",
"ZSTD")) {
+ Map<String, String> supportedOptions = mapSharedShreddingOptions();
+ supportedOptions.put(CoreOptions.FILE_COMPRESSION.key(),
compression);
+ assertThatNoException()
+ .isThrownBy(
+ () ->
+ validateTableSchema(
+ mapSharedShreddingSchema(
+ supportedOptions,
emptyList())));
+ }
+ }
+
+ @Test
+ public void testMapSharedShreddingTableModeValidation() {
+ Map<String, String> primaryKeyOptions = mapSharedShreddingOptions();
+ primaryKeyOptions.put(BUCKET.key(), "-1");
+ assertThatThrownBy(
+ () ->
+ validateTableSchema(
+ mapSharedShreddingSchema(
+ primaryKeyOptions,
singletonList("id"))))
+ .hasMessageContaining(
+ "MAP shared-shredding currently only supports
append-only tables.");
+
+ Map<String, String> fixedBucketOptions = mapSharedShreddingOptions();
+ fixedBucketOptions.put(BUCKET.key(), "1");
+ fixedBucketOptions.put(CoreOptions.BUCKET_KEY.key(), "id");
+ assertThatThrownBy(
+ () ->
+ validateTableSchema(
+
mapSharedShreddingSchema(fixedBucketOptions, emptyList())))
+ .hasMessageContaining(
+ "MAP shared-shredding currently requires bucket = -1
or write-only = true because rewrite/compaction is not supported.");
+
+ Map<String, String> writeOnlyOptions = mapSharedShreddingOptions();
+ writeOnlyOptions.put(BUCKET.key(), "1");
+ writeOnlyOptions.put(CoreOptions.BUCKET_KEY.key(), "id");
+ writeOnlyOptions.put(CoreOptions.WRITE_ONLY.key(), "true");
+ assertThatNoException()
+ .isThrownBy(
+ () ->
+ validateTableSchema(
+
mapSharedShreddingSchema(writeOnlyOptions, emptyList())));
+ }
+
+ @Test
+ public void testMapSharedShreddingNestedFileIndexValidation() {
+ Map<String, String> options = mapSharedShreddingOptions();
+ options.put("file-index.bloom-filter.columns", "metrics[k]");
+
+ assertThatThrownBy(
+ () ->
validateTableSchema(mapSharedShreddingSchema(options, emptyList())))
+ .hasMessageContaining(
+ "Column 'metrics' is configured with
map.storage-layout=shared-shredding, which does not support nested file
index.");
+ }
+
@Test
public void testChainTableAllowsNonDeduplicateMergeEngine() {
Map<String, String> options = new HashMap<>();
@@ -1041,6 +1228,60 @@ class SchemaValidationTest {
assertThatCode(() ->
validateTableSchemaExec(options)).doesNotThrowAnyException();
}
+ private Map<String, String> mapSharedShreddingOptions() {
+ Map<String, String> options = new HashMap<>();
+ options.put(BUCKET.key(), "-1");
+ options.put(CoreOptions.FILE_FORMAT.key(), "parquet");
+ options.put("fields.metrics.map.storage-layout", "shared-shredding");
+ return options;
+ }
+
+ private TableSchema mapSharedShreddingSchema(
+ Map<String, String> options, List<String> primaryKeys) {
+ return new TableSchema(
+ 1, mapValueFields(DataTypes.INT()), 10, emptyList(),
primaryKeys, options, "");
+ }
+
+ private void assertMapSharedShreddingValidationFailed(
+ List<DataField> fields, Map<String, String> options, String
message) {
+ assertThatThrownBy(
+ () ->
+ validateTableSchema(
+ new TableSchema(
+ 1,
+ fields,
+ 10,
+ emptyList(),
+ emptyList(),
+ options,
+ "")))
+ .hasMessageContaining(message);
+ }
+
+ private List<DataField> mapValueFields(DataType valueType) {
+ return Arrays.asList(
+ new DataField(0, "id", DataTypes.INT()),
+ new DataField(1, "metrics", DataTypes.MAP(DataTypes.STRING(),
valueType)));
+ }
+
+ private List<DataField> nestedMapValueFields(DataType valueType) {
+ return Arrays.asList(
+ new DataField(0, "id", DataTypes.INT()),
+ new DataField(
+ 1,
+ "metrics",
+ DataTypes.MAP(
+ DataTypes.STRING(),
+ DataTypes.ROW(DataTypes.FIELD(0, "payload",
valueType)))));
+ }
+
+ private List<DataField> topLevelPayloadFields(DataType payloadType) {
+ return Arrays.asList(
+ new DataField(0, "id", DataTypes.INT()),
+ new DataField(1, "metrics", DataTypes.MAP(DataTypes.STRING(),
DataTypes.INT())),
+ new DataField(2, "payload", payloadType));
+ }
+
private TableSchema vectorTypeSchema(
List<String> partitionKeys, List<String> primaryKeys, List<String>
upsertKeys) {
List<DataField> fields =
