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 23fca3cd12 [core] Add shared-shredding map defs and utilities (#8314)
23fca3cd12 is described below
commit 23fca3cd125e93e5dad7141b55a3066ff008ebfd
Author: lxy <[email protected]>
AuthorDate: Mon Jun 22 17:28:45 2026 +0800
[core] Add shared-shredding map defs and utilities (#8314)
This PR adds the shared-shredding MAP schema utilities and option
validation for [PIP-43: Columnar Storage Optimization for MAP Type in
Paimon](https://cwiki.apache.org/confluence/display/PAIMON/PIP-43%3A+Columnar+Storage+Optimization+for+MAP+Type+in+Paimon).
---
.../main/java/org/apache/paimon/CoreOptions.java | 58 ++++
.../data/shredding/MapSharedShreddingDefine.java | 44 +++
.../shredding/MapSharedShreddingFieldMeta.java | 88 ++++++
.../data/shredding/MapSharedShreddingUtils.java | 285 ++++++++++++++++++
.../paimon/data/shredding/MapShreddingDefine.java | 29 ++
.../shredding/MapSharedShreddingUtilsTest.java | 333 +++++++++++++++++++++
.../org/apache/paimon/schema/SchemaValidation.java | 51 ++++
.../java/org/apache/paimon/CoreOptionsTest.java | 40 +++
.../apache/paimon/schema/SchemaValidationTest.java | 103 +++++++
9 files changed, 1031 insertions(+)
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index a884752434..bf85dd66dc 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -90,6 +90,11 @@ public class CoreOptions implements Serializable {
public static final String MERGE_MAP_TS_FIELD = "ts-field";
+ public static final String MAP_STORAGE_LAYOUT = "map.storage-layout";
+
+ public static final String MAP_SHARED_SHREDDING_MAX_COLUMNS =
+ "map.shared-shredding.max-columns";
+
public static final String FILE_INDEX = "file-index";
public static final String COLUMNS = "columns";
@@ -4847,6 +4852,59 @@ public class CoreOptions implements Serializable {
.noDefaultValue());
}
+ public MapStorageLayout mapStorageLayout(String fieldName) {
+ return options.get(
+ key(FIELDS_PREFIX + "." + fieldName + "." + MAP_STORAGE_LAYOUT)
+ .enumType(MapStorageLayout.class)
+ .defaultValue(MapStorageLayout.DEFAULT));
+ }
+
+ public int mapSharedShreddingMaxColumns(String fieldName) {
+ int maxColumns =
+ options.get(
+ key(FIELDS_PREFIX
+ + "."
+ + fieldName
+ + "."
+ + MAP_SHARED_SHREDDING_MAX_COLUMNS)
+ .intType()
+ .defaultValue(256));
+ checkArgument(maxColumns > 0, "options %s must > 0",
MAP_SHARED_SHREDDING_MAX_COLUMNS);
+ return maxColumns;
+ }
+
+ /** MAP storage layout. */
+ public enum MapStorageLayout implements DescribedEnum {
+ DEFAULT(
+ "default",
+ "Store MAP columns with the normal key-value array layout.
This is the compatible "
+ + "layout used when no field-level MAP layout option
is configured."),
+ SHARED_SHREDDING(
+ "shared-shredding",
+ "Store MAP<STRING, T> columns as a physical row with reusable
value columns, a "
+ + "field mapping, and an overflow map. This layout is
intended for maps "
+ + "whose string keys repeat across rows and can
benefit from more columnar "
+ + "storage.");
+
+ private final String value;
+ private final String description;
+
+ MapStorageLayout(String value, String description) {
+ this.value = value;
+ this.description = description;
+ }
+
+ @Override
+ public String toString() {
+ return value;
+ }
+
+ @Override
+ public InlineElement getDescription() {
+ return text(description);
+ }
+ }
+
/**
* Action to take when an UPDATE (e.g. via MERGE INTO) modifies columns
that are covered by a
* global index.
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
new file mode 100644
index 0000000000..6b5cce93d6
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingDefine.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.data.shredding;
+
+/** Constants for the shared-shredding MAP storage layout. */
+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_ORIGINAL_SIZE =
+ "paimon.map.shared-shredding.field-dict-original-size";
+ public static final String FIELD_COLUMNS =
"paimon.map.shared-shredding.field-columns";
+ public static final String OVERFLOW_SET =
"paimon.map.shared-shredding.overflow-set";
+ public static final String NUM_COLUMNS =
"paimon.map.shared-shredding.num-columns";
+ public static final String MAX_ROW_WIDTH =
"paimon.map.shared-shredding.max-row-width";
+
+ public static final String FIELD_MAPPING = "__field_mapping";
+ public static final String OVERFLOW = "__overflow";
+
+ public static final String DEFAULT_DICT_COMPRESSION = "zstd";
+
+ public static String physicalColumnName(int index) {
+ return "__col_" + index;
+ }
+
+ private MapSharedShreddingDefine() {}
+}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingFieldMeta.java
b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingFieldMeta.java
new file mode 100644
index 0000000000..dea1928a34
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingFieldMeta.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.data.shredding;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+/** Parsed file-level metadata for one shared-shredding MAP column. */
+public class MapSharedShreddingFieldMeta {
+
+ private final Map<String, Integer> nameToId;
+ private final Map<Integer, List<Integer>> fieldToColumns;
+ private final Set<Integer> overflowFieldSet;
+ private final int numColumns;
+ private final int maxRowWidth;
+
+ public MapSharedShreddingFieldMeta(
+ Map<String, Integer> nameToId,
+ Map<Integer, List<Integer>> fieldToColumns,
+ Set<Integer> overflowFieldSet,
+ int numColumns,
+ int maxRowWidth) {
+ this.nameToId = nameToId;
+ this.fieldToColumns = fieldToColumns;
+ this.overflowFieldSet = overflowFieldSet;
+ this.numColumns = numColumns;
+ this.maxRowWidth = maxRowWidth;
+ }
+
+ public Map<String, Integer> nameToId() {
+ return nameToId;
+ }
+
+ public Map<Integer, List<Integer>> fieldToColumns() {
+ return fieldToColumns;
+ }
+
+ public Set<Integer> overflowFieldSet() {
+ return overflowFieldSet;
+ }
+
+ public int numColumns() {
+ return numColumns;
+ }
+
+ public int maxRowWidth() {
+ return maxRowWidth;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof MapSharedShreddingFieldMeta)) {
+ return false;
+ }
+ MapSharedShreddingFieldMeta that = (MapSharedShreddingFieldMeta) o;
+ return numColumns == that.numColumns
+ && maxRowWidth == that.maxRowWidth
+ && Objects.equals(nameToId, that.nameToId)
+ && Objects.equals(fieldToColumns, that.fieldToColumns)
+ && Objects.equals(overflowFieldSet, that.overflowFieldSet);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(nameToId, fieldToColumns, overflowFieldSet,
numColumns, maxRowWidth);
+ }
+}
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
new file mode 100644
index 0000000000..6f0cd879c5
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingUtils.java
@@ -0,0 +1,285 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.data.shredding;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.CoreOptions.MapStorageLayout;
+import org.apache.paimon.compression.BlockCompressionFactory;
+import org.apache.paimon.compression.BlockCompressor;
+import org.apache.paimon.compression.BlockDecompressor;
+import org.apache.paimon.compression.CompressOptions;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypeRoot;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.JsonSerdeUtil;
+
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.type.TypeReference;
+
+import javax.annotation.Nullable;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import java.util.stream.Collectors;
+
+/**
+ * Utility functions for the shared-shredding MAP storage layout.
+ *
+ * <p>Shared-shredding can be enabled for {@code MAP<STRING, T>} fields. It
rewrites a logical MAP
+ * field into a physical ROW containing a key-to-column mapping array, a fixed
number of value
+ * columns, and an overflow MAP for keys that cannot be placed in the fixed
columns. This class also
+ * serializes and deserializes the per-field shredding metadata stored with
the physical data,
+ * including optional compression for the field dictionary.
+ */
+public class MapSharedShreddingUtils {
+
+ private MapSharedShreddingUtils() {}
+
+ public static boolean isShreddingKeyMap(DataType dataType) {
+ if (!(dataType instanceof MapType)) {
+ return false;
+ }
+ MapType mapType = (MapType) dataType;
+ return mapType.getKeyType().getTypeRoot() == DataTypeRoot.VARCHAR;
+ }
+
+ public static List<String> detectShreddingColumns(RowType rowType,
CoreOptions options) {
+ List<String> fieldNames = new ArrayList<>();
+ for (DataField field : rowType.getFields()) {
+ if (!isShreddingKeyMap(field.type())) {
+ continue;
+ }
+ if (options.mapStorageLayout(field.name()) ==
MapStorageLayout.SHARED_SHREDDING) {
+ fieldNames.add(field.name());
+ }
+ }
+ return fieldNames;
+ }
+
+ public static RowType logicalToPhysicalSchema(
+ RowType logicalSchema, Map<String, Integer> fieldToNumColumns) {
+ List<DataField> physicalFields = new ArrayList<>();
+ for (DataField field : logicalSchema.getFields()) {
+ Integer numColumns = fieldToNumColumns.get(field.name());
+ if (numColumns == null) {
+ physicalFields.add(field);
+ continue;
+ }
+
+ MapType mapType = (MapType) field.type();
+ DataType physicalType =
+ buildPhysicalStructType(mapType.getValueType(), numColumns)
+ .copy(field.type().isNullable());
+ physicalFields.add(field.newType(physicalType));
+ }
+ return new RowType(logicalSchema.isNullable(), physicalFields);
+ }
+
+ public static Map<String, Integer> buildColumnToNumColumns(
+ List<String> shreddingFieldNames, CoreOptions options) {
+ Map<String, Integer> fieldToNumColumns = new HashMap<>();
+ for (String fieldName : shreddingFieldNames) {
+ fieldToNumColumns.put(fieldName,
options.mapSharedShreddingMaxColumns(fieldName));
+ }
+ return fieldToNumColumns;
+ }
+
+ public static void serializeMetadata(
+ MapSharedShreddingFieldMeta fieldMeta,
+ String compression,
+ Map<String, String> metadata) {
+ metadata.put(
+ MapShreddingDefine.STORAGE_LAYOUT,
+ MapShreddingDefine.STORAGE_LAYOUT_SHARED_SHREDDING);
+ metadata.put(
+ MapSharedShreddingDefine.VERSION,
+ String.valueOf(MapSharedShreddingDefine.CURRENT_VERSION));
+
+ String fieldDictJson = toJson(new TreeMap<>(fieldMeta.nameToId()));
+ metadata.put(
+ MapSharedShreddingDefine.FIELD_DICT_ORIGINAL_SIZE,
+
String.valueOf(fieldDictJson.getBytes(StandardCharsets.UTF_8).length));
+ metadata.put(
+ MapSharedShreddingDefine.FIELD_DICT,
+ bytesToString(
+
compress(fieldDictJson.getBytes(StandardCharsets.UTF_8), compression)));
+ metadata.put(
+ MapSharedShreddingDefine.FIELD_COLUMNS,
+ toJson(sortedFieldColumns(fieldMeta.fieldToColumns())));
+ metadata.put(
+ MapSharedShreddingDefine.OVERFLOW_SET,
+ toJson(new TreeSet<>(fieldMeta.overflowFieldSet())));
+ metadata.put(MapSharedShreddingDefine.NUM_COLUMNS,
String.valueOf(fieldMeta.numColumns()));
+ metadata.put(
+ MapSharedShreddingDefine.MAX_ROW_WIDTH,
String.valueOf(fieldMeta.maxRowWidth()));
+ }
+
+ public static MapSharedShreddingFieldMeta deserializeMetadata(
+ @Nullable Map<String, String> metadata, String compression) {
+ if (!hasShreddingMetadata(metadata)) {
+ throw new IllegalArgumentException(
+ "metadata is null or storage layout is not
shared-shredding");
+ }
+
+ int version = requiredInt(metadata, MapSharedShreddingDefine.VERSION);
+ if (version != MapSharedShreddingDefine.CURRENT_VERSION) {
+ throw new IllegalArgumentException(
+ String.format(
+ "unsupported shared-shredding metadata version:
%s, expected: %s",
+ version,
MapSharedShreddingDefine.CURRENT_VERSION));
+ }
+
+ int originalLength =
+ requiredInt(metadata,
MapSharedShreddingDefine.FIELD_DICT_ORIGINAL_SIZE);
+ byte[] fieldDictBytes =
+ decompress(
+ stringToBytes(requiredValue(metadata,
MapSharedShreddingDefine.FIELD_DICT)),
+ originalLength,
+ compression);
+ Map<String, Integer> nameToId =
+ fromJson(
+ new String(fieldDictBytes, StandardCharsets.UTF_8),
+ new TypeReference<Map<String, Integer>>() {});
+ Map<Integer, List<Integer>> fieldToColumns =
+ parseFieldColumns(requiredValue(metadata,
MapSharedShreddingDefine.FIELD_COLUMNS));
+ Set<Integer> overflowSet =
+ fromJson(
+ requiredValue(metadata,
MapSharedShreddingDefine.OVERFLOW_SET),
+ new TypeReference<Set<Integer>>() {});
+
+ return new MapSharedShreddingFieldMeta(
+ nameToId,
+ fieldToColumns,
+ overflowSet,
+ requiredInt(metadata, MapSharedShreddingDefine.NUM_COLUMNS),
+ requiredInt(metadata, MapSharedShreddingDefine.MAX_ROW_WIDTH));
+ }
+
+ public static boolean hasShreddingMetadata(@Nullable Map<String, String>
metadata) {
+ return metadata != null
+ && MapShreddingDefine.STORAGE_LAYOUT_SHARED_SHREDDING.equals(
+ metadata.get(MapShreddingDefine.STORAGE_LAYOUT));
+ }
+
+ 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();
+ }
+
+ private static Map<Integer, List<Integer>> sortedFieldColumns(
+ Map<Integer, List<Integer>> fieldToColumns) {
+ Map<Integer, List<Integer>> result = new TreeMap<>();
+ for (Map.Entry<Integer, List<Integer>> entry :
fieldToColumns.entrySet()) {
+ List<Integer> columns =
entry.getValue().stream().sorted().collect(Collectors.toList());
+ result.put(entry.getKey(), columns);
+ }
+ return result;
+ }
+
+ private static Map<Integer, List<Integer>> parseFieldColumns(String json) {
+ Map<String, List<Integer>> parsed =
+ fromJson(json, new TypeReference<Map<String, List<Integer>>>()
{});
+ Map<Integer, List<Integer>> result = new TreeMap<>();
+ for (Map.Entry<String, List<Integer>> entry : parsed.entrySet()) {
+ result.put(Integer.parseInt(entry.getKey()), entry.getValue());
+ }
+ return result;
+ }
+
+ private static String requiredValue(Map<String, String> metadata, String
key) {
+ String value = metadata.get(key);
+ if (value == null) {
+ throw new IllegalArgumentException("missing shredding metadata
key: " + key);
+ }
+ return value;
+ }
+
+ private static int requiredInt(Map<String, String> metadata, String key) {
+ try {
+ return Integer.parseInt(requiredValue(metadata, key));
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException(
+ "malformed shredding metadata value for key: " + key, e);
+ }
+ }
+
+ private static byte[] compress(byte[] input, String compression) {
+ BlockCompressionFactory factory =
+ BlockCompressionFactory.create(new
CompressOptions(compression, 1));
+ if (factory == null) {
+ return input;
+ }
+ BlockCompressor compressor = factory.getCompressor();
+ byte[] output = new
byte[compressor.getMaxCompressedSize(input.length)];
+ int actualSize = compressor.compress(input, 0, input.length, output,
0);
+ return Arrays.copyOf(output, actualSize);
+ }
+
+ private static byte[] decompress(byte[] input, int originalLength, String
compression) {
+ BlockCompressionFactory factory =
+ BlockCompressionFactory.create(new
CompressOptions(compression, 1));
+ if (factory == null) {
+ return input;
+ }
+ BlockDecompressor decompressor = factory.getDecompressor();
+ byte[] output = new byte[originalLength];
+ int actualSize = decompressor.decompress(input, 0, input.length,
output, 0);
+ return Arrays.copyOf(output, actualSize);
+ }
+
+ private static String bytesToString(byte[] bytes) {
+ return new String(bytes, StandardCharsets.ISO_8859_1);
+ }
+
+ private static byte[] stringToBytes(String string) {
+ return string.getBytes(StandardCharsets.ISO_8859_1);
+ }
+
+ private static String toJson(Object object) {
+ try {
+ return
JsonSerdeUtil.OBJECT_MAPPER_INSTANCE.writeValueAsString(object);
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException("Failed to serialize shared-shredding
metadata.", e);
+ }
+ }
+
+ private static <T> T fromJson(String json, TypeReference<T> typeReference)
{
+ try {
+ return JsonSerdeUtil.OBJECT_MAPPER_INSTANCE.readValue(json,
typeReference);
+ } catch (JsonProcessingException e) {
+ throw new IllegalArgumentException("malformed shredding metadata",
e);
+ }
+ }
+}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapShreddingDefine.java
b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapShreddingDefine.java
new file mode 100644
index 0000000000..ebc7d9a97e
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapShreddingDefine.java
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.data.shredding;
+
+/** Constants for MAP storage layout marker. */
+public class MapShreddingDefine {
+
+ public static final String STORAGE_LAYOUT = "paimon.map.storage-layout";
+
+ public static final String STORAGE_LAYOUT_SHARED_SHREDDING =
"shared-shredding";
+
+ private MapShreddingDefine() {}
+}
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
new file mode 100644
index 0000000000..29314c6a00
--- /dev/null
+++
b/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingUtilsTest.java
@@ -0,0 +1,333 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.data.shredding;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link MapSharedShreddingUtils}. */
+class MapSharedShreddingUtilsTest {
+
+ @Test
+ void testIsShreddingKeyMap() {
+ assertThat(
+ MapSharedShreddingUtils.isShreddingKeyMap(
+ DataTypes.MAP(DataTypes.STRING(),
DataTypes.INT())))
+ .isTrue();
+ assertThat(
+ MapSharedShreddingUtils.isShreddingKeyMap(
+ DataTypes.MAP(DataTypes.STRING(),
DataTypes.DOUBLE())))
+ .isTrue();
+ assertThat(
+ MapSharedShreddingUtils.isShreddingKeyMap(
+ DataTypes.MAP(
+ DataTypes.STRING(),
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "x",
DataTypes.INT()),
+ DataTypes.FIELD(1, "y",
DataTypes.STRING())))))
+ .isTrue();
+ assertThat(
+ MapSharedShreddingUtils.isShreddingKeyMap(
+ DataTypes.MAP(DataTypes.INT(),
DataTypes.STRING())))
+ .isFalse();
+
assertThat(MapSharedShreddingUtils.isShreddingKeyMap(DataTypes.INT())).isFalse();
+
assertThat(MapSharedShreddingUtils.isShreddingKeyMap(DataTypes.ARRAY(DataTypes.STRING())))
+ .isFalse();
+ }
+
+ @Test
+ void testDetectAndBuildColumnToNumColumns() {
+ RowType rowType =
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "id", DataTypes.INT()),
+ DataTypes.FIELD(
+ 1, "tags", DataTypes.MAP(DataTypes.STRING(),
DataTypes.STRING())),
+ DataTypes.FIELD(
+ 2,
+ "metrics",
+ DataTypes.MAP(DataTypes.STRING(),
DataTypes.BIGINT())),
+ DataTypes.FIELD(
+ 3, "codes", DataTypes.MAP(DataTypes.INT(),
DataTypes.STRING())));
+
+ Options conf = new Options();
+ conf.setString("fields.tags.map.storage-layout", "shared-shredding");
+ conf.setString("fields.metrics.map.storage-layout",
"shared-shredding");
+ conf.setString("fields.tags.map.shared-shredding.max-columns", "128");
+ conf.setString("fields.metrics.map.shared-shredding.max-columns",
"64");
+ conf.setString("fields.codes.map.storage-layout", "shared-shredding");
+
+ CoreOptions options = new CoreOptions(conf);
+ assertThat(MapSharedShreddingUtils.detectShreddingColumns(rowType,
options))
+ .containsExactly("tags", "metrics");
+ assertThat(
+ MapSharedShreddingUtils.buildColumnToNumColumns(
+
MapSharedShreddingUtils.detectShreddingColumns(rowType, options),
+ options))
+ .containsEntry("tags", 128)
+ .containsEntry("metrics", 64);
+
+ Options defaultConf = new Options();
+ CoreOptions defaultOptions = new CoreOptions(defaultConf);
+ assertThat(MapSharedShreddingUtils.detectShreddingColumns(rowType,
defaultOptions))
+ .isEmpty();
+ assertThat(
+ MapSharedShreddingUtils.buildColumnToNumColumns(
+ Arrays.asList("tags"), defaultOptions))
+ .containsEntry("tags", 256);
+ }
+
+ @Test
+ void testLogicalToPhysicalSchema() {
+ RowType logical =
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "id", DataTypes.INT()),
+ DataTypes.FIELD(
+ 1,
+ "metrics",
+ DataTypes.MAP(DataTypes.STRING(),
DataTypes.DOUBLE().notNull())));
+
+ Map<String, Integer> fieldToNumColumns = new HashMap<>();
+ fieldToNumColumns.put("metrics", 2);
+
+ RowType physical =
+ MapSharedShreddingUtils.logicalToPhysicalSchema(logical,
fieldToNumColumns);
+ DataField metrics = physical.getField("metrics");
+ assertThat(metrics.id()).isEqualTo(1);
+ assertThat(metrics.type()).isInstanceOf(RowType.class);
+ RowType metricsPhysicalType = (RowType) metrics.type();
+ assertThat(metricsPhysicalType.getFieldNames())
+ .containsExactly("__field_mapping", "__col_0", "__col_1",
"__overflow");
+ assertThat(metricsPhysicalType.getFields())
+ .extracting(DataField::id)
+ .containsExactly(0, 1, 2, 3);
+ assertThat(metricsPhysicalType.getField("__col_0").type())
+ .isEqualTo(DataTypes.DOUBLE().notNull());
+ assertThat(metricsPhysicalType.getField("__overflow").type())
+ .isEqualTo(DataTypes.MAP(DataTypes.INT(),
DataTypes.DOUBLE().notNull()));
+ }
+
+ @Test
+ void testLogicalToPhysicalSchemaNestedValueAndStableFieldIds() {
+ RowType nestedValue =
+ DataTypes.ROW(
+ DataTypes.FIELD(2, "a", DataTypes.INT()),
+ DataTypes.FIELD(3, "b", DataTypes.STRING()));
+ RowType original =
+ DataTypes.ROW(
+ DataTypes.FIELD(1, "data",
DataTypes.MAP(DataTypes.STRING(), nestedValue)));
+ RowType evolved =
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "new_col", DataTypes.STRING()),
+ DataTypes.FIELD(1, "data",
DataTypes.MAP(DataTypes.STRING(), nestedValue)));
+
+ Map<String, Integer> fieldToNumColumns = new HashMap<>();
+ fieldToNumColumns.put("data", 2);
+
+ RowType physical =
+ MapSharedShreddingUtils.logicalToPhysicalSchema(original,
fieldToNumColumns);
+ assertThat(physical.getField("data").id()).isEqualTo(1);
+ RowType dataPhysicalType = (RowType) physical.getField("data").type();
+ assertThat(dataPhysicalType.getFieldNames())
+ .containsExactly("__field_mapping", "__col_0", "__col_1",
"__overflow");
+ assertThat(dataPhysicalType.getFields())
+ .extracting(DataField::id)
+ .containsExactly(0, 1, 2, 3);
+
assertThat(dataPhysicalType.getField("__col_0").type()).isEqualTo(nestedValue);
+ assertThat(dataPhysicalType.getField("__overflow").type())
+ .isEqualTo(DataTypes.MAP(DataTypes.INT(), nestedValue));
+
+ RowType evolvedPhysical =
+ MapSharedShreddingUtils.logicalToPhysicalSchema(evolved,
fieldToNumColumns);
+ assertThat(evolvedPhysical.getField("data").id()).isEqualTo(1);
+ assertThat(((RowType)
evolvedPhysical.getField("data").type()).getFields())
+ .extracting(DataField::id)
+ .containsExactly(0, 1, 2, 3);
+ }
+
+ @Test
+ void testLogicalToPhysicalSchemaNoShreddingColumns() {
+ RowType logical =
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "id", DataTypes.INT()),
+ DataTypes.FIELD(1, "name", DataTypes.STRING()));
+
+ assertThat(MapSharedShreddingUtils.logicalToPhysicalSchema(logical,
new HashMap<>()))
+ .isEqualTo(logical);
+ }
+
+ @Test
+ void testMetadataRoundtrip() {
+ Map<String, Integer> nameToId = new TreeMap<>();
+ nameToId.put("age", 0);
+ nameToId.put("name", 1);
+
+ Map<Integer, List<Integer>> fieldToColumns = new TreeMap<>();
+ fieldToColumns.put(0, Arrays.asList(0));
+ fieldToColumns.put(1, Arrays.asList(1, 2));
+
+ HashSet<Integer> overflowSet = new HashSet<>();
+ overflowSet.add(1);
+ overflowSet.add(5);
+
+ MapSharedShreddingFieldMeta original =
+ new MapSharedShreddingFieldMeta(nameToId, fieldToColumns,
overflowSet, 3, 2);
+
+ Map<String, String> metadata = new HashMap<>();
+ MapSharedShreddingUtils.serializeMetadata(original, "none", metadata);
+
+
assertThat(MapSharedShreddingUtils.hasShreddingMetadata(metadata)).isTrue();
+
assertThat(metadata.get(MapShreddingDefine.STORAGE_LAYOUT)).isEqualTo("shared-shredding");
+
assertThat(metadata.get(MapSharedShreddingDefine.VERSION)).isEqualTo("1");
+
assertThat(metadata.get(MapSharedShreddingDefine.NUM_COLUMNS)).isEqualTo("3");
+
assertThat(metadata.get(MapSharedShreddingDefine.MAX_ROW_WIDTH)).isEqualTo("2");
+
+ String expectedDict = "{\"age\":0,\"name\":1}";
+
assertThat(metadata.get(MapSharedShreddingDefine.FIELD_DICT)).isEqualTo(expectedDict);
+
assertThat(metadata.get(MapSharedShreddingDefine.FIELD_DICT_ORIGINAL_SIZE))
+ .isEqualTo(String.valueOf(expectedDict.length()));
+ 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,
"none"))
+ .isEqualTo(original);
+ }
+
+ @Test
+ void testMetadataRoundtripCompression() {
+ Map<String, Integer> nameToId = new TreeMap<>();
+ nameToId.put("alpha", 0);
+ nameToId.put("beta", 1);
+ nameToId.put("gamma", 2);
+
+ Map<Integer, List<Integer>> fieldToColumns = new TreeMap<>();
+ fieldToColumns.put(0, Arrays.asList(0, 1, 2));
+ fieldToColumns.put(1, Arrays.asList(3));
+ fieldToColumns.put(2, Arrays.asList(4, 5));
+
+ HashSet<Integer> overflowSet = new HashSet<>();
+ overflowSet.add(2);
+ MapSharedShreddingFieldMeta original =
+ new MapSharedShreddingFieldMeta(nameToId, fieldToColumns,
overflowSet, 6, 3);
+
+ 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);
+ }
+ }
+
+ @Test
+ void testMetadataRoundtripEmptyData() {
+ MapSharedShreddingFieldMeta original =
+ new MapSharedShreddingFieldMeta(
+ new TreeMap<>(), new TreeMap<>(), new HashSet<>(), 0,
0);
+
+ 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);
+ }
+ }
+
+ @Test
+ void testDeserializeMetadataErrors() {
+ assertThatThrownBy(() ->
MapSharedShreddingUtils.deserializeMetadata(null, "none"))
+ .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"))
+ .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"))
+ .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"))
+ .hasMessageContaining(
+ "missing shredding metadata key:
paimon.map.shared-shredding.version");
+
+ Map<String, String> wrongVersion = new HashMap<>();
+ wrongVersion.put(
+ MapShreddingDefine.STORAGE_LAYOUT,
+ MapShreddingDefine.STORAGE_LAYOUT_SHARED_SHREDDING);
+ wrongVersion.put(MapSharedShreddingDefine.VERSION, "999");
+ wrongVersion.put(MapSharedShreddingDefine.FIELD_DICT_ORIGINAL_SIZE,
"2");
+ wrongVersion.put(MapSharedShreddingDefine.FIELD_DICT, "{}");
+ assertThatThrownBy(() ->
MapSharedShreddingUtils.deserializeMetadata(wrongVersion, "none"))
+ .hasMessageContaining("unsupported shared-shredding metadata
version: 999");
+
+ Map<String, String> missingFieldDict = new HashMap<>();
+ missingFieldDict.put(
+ MapShreddingDefine.STORAGE_LAYOUT,
+ MapShreddingDefine.STORAGE_LAYOUT_SHARED_SHREDDING);
+ missingFieldDict.put(MapSharedShreddingDefine.VERSION, "1");
+
missingFieldDict.put(MapSharedShreddingDefine.FIELD_DICT_ORIGINAL_SIZE, "2");
+ assertThatThrownBy(
+ () ->
MapSharedShreddingUtils.deserializeMetadata(missingFieldDict, "none"))
+ .hasMessageContaining(
+ "missing shredding metadata key:
paimon.map.shared-shredding.field-dict");
+ }
+
+ @Test
+ void testHasShreddingMetadata() {
+
assertThat(MapSharedShreddingUtils.hasShreddingMetadata(null)).isFalse();
+
+ Map<String, String> metadata = new HashMap<>();
+ metadata.put(
+ MapShreddingDefine.STORAGE_LAYOUT,
+ MapShreddingDefine.STORAGE_LAYOUT_SHARED_SHREDDING);
+
assertThat(MapSharedShreddingUtils.hasShreddingMetadata(metadata)).isTrue();
+
+ metadata.put(MapShreddingDefine.STORAGE_LAYOUT, "default");
+
assertThat(MapSharedShreddingUtils.hasShreddingMetadata(metadata)).isFalse();
+
+ assertThat(MapSharedShreddingUtils.hasShreddingMetadata(new
HashMap<>())).isFalse();
+ }
+
+ @Test
+ void testPhysicalColumnName() {
+
assertThat(MapSharedShreddingDefine.physicalColumnName(0)).isEqualTo("__col_0");
+
assertThat(MapSharedShreddingDefine.physicalColumnName(1)).isEqualTo("__col_1");
+
assertThat(MapSharedShreddingDefine.physicalColumnName(99)).isEqualTo("__col_99");
+ }
+}
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 7733b6a080..fedfecaafb 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
@@ -20,8 +20,10 @@ package org.apache.paimon.schema;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.CoreOptions.ChangelogProducer;
+import org.apache.paimon.CoreOptions.MapStorageLayout;
import org.apache.paimon.CoreOptions.MergeEngine;
import org.apache.paimon.TableType;
+import org.apache.paimon.data.shredding.MapSharedShreddingUtils;
import org.apache.paimon.factories.FactoryUtil;
import org.apache.paimon.fileindex.FileIndexOptions;
import org.apache.paimon.fileindex.FileIndexerFactory;
@@ -69,6 +71,7 @@ import static
org.apache.paimon.CoreOptions.FULL_COMPACTION_DELTA_COMMITS;
import static org.apache.paimon.CoreOptions.INCREMENTAL_BETWEEN;
import static org.apache.paimon.CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP;
import static org.apache.paimon.CoreOptions.INCREMENTAL_TO_AUTO_TAG;
+import static org.apache.paimon.CoreOptions.MAP_STORAGE_LAYOUT;
import static org.apache.paimon.CoreOptions.PRIMARY_KEY;
import static org.apache.paimon.CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS;
import static org.apache.paimon.CoreOptions.SCAN_MODE;
@@ -355,6 +358,8 @@ public class SchemaValidation {
validatePkClusteringOverride(options);
validateManifestSort(schema, options);
+
+ validateMapStorageLayout(schema, options);
}
public static void validateFallbackBranch(SchemaManager schemaManager,
TableSchema schema) {
@@ -604,6 +609,52 @@ public class SchemaValidation {
createMergeFunctionFactory(schema);
}
+ private static void validateMapStorageLayout(TableSchema schema,
CoreOptions options) {
+ String layoutSuffix = "." + MAP_STORAGE_LAYOUT;
+ Map<String, DataField> fieldMap = new HashMap<>();
+ for (DataField field : schema.fields()) {
+ fieldMap.put(field.name(), field);
+ }
+
+ for (String key : options.toMap().keySet()) {
+ if (!key.startsWith(FIELDS_PREFIX + ".") ||
!key.endsWith(layoutSuffix)) {
+ continue;
+ }
+
+ String fieldName =
+ key.substring(
+ (FIELDS_PREFIX + ".").length(), key.length() -
layoutSuffix.length());
+ DataField field = fieldMap.get(fieldName);
+ if (field == null) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Column '%s' is configured with
map.storage-layout but does not exist in table schema.",
+ fieldName));
+ }
+
+ DataType fieldType = field.type();
+ if (!(fieldType instanceof MapType)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Column '%s' is configured with
map.storage-layout but its type is not MAP.",
+ fieldName));
+ }
+
+ MapStorageLayout layout = options.mapStorageLayout(fieldName);
+ if (layout != MapStorageLayout.SHARED_SHREDDING) {
+ continue;
+ }
+
+ if (!MapSharedShreddingUtils.isShreddingKeyMap(fieldType)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Column '%s' is configured with
map.storage-layout=shared-shredding but its type is not MAP<STRING, T>.",
+ fieldName));
+ }
+ options.mapSharedShreddingMaxColumns(fieldName);
+ }
+ }
+
private static void validateFileIndex(TableSchema schema) {
CoreOptions options = new CoreOptions(schema.options());
FileIndexOptions fileIndexOptions = options.indexColumnsOptions();
diff --git a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java
b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java
index 359b9f25e8..3399990a8a 100644
--- a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java
@@ -23,6 +23,7 @@ import org.apache.paimon.options.Options;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link org.apache.paimon.CoreOptions}. */
public class CoreOptionsTest {
@@ -117,4 +118,43 @@ public class CoreOptionsTest {
options = new CoreOptions(conf);
assertThat(options.blobSplitByFileSize()).isTrue();
}
+
+ @Test
+ public void testMapStorageLayout() {
+ Options conf = new Options();
+ CoreOptions options = new CoreOptions(conf);
+ assertThat(options.mapStorageLayout("metrics"))
+ .isEqualTo(CoreOptions.MapStorageLayout.DEFAULT);
+
assertThat(options.mapSharedShreddingMaxColumns("metrics")).isEqualTo(256);
+
+ conf.setString("fields.metrics.map.storage-layout",
"shared-shredding");
+ conf.setString("fields.metrics.map.shared-shredding.max-columns",
"32");
+ options = new CoreOptions(conf);
+ assertThat(options.mapStorageLayout("metrics"))
+ .isEqualTo(CoreOptions.MapStorageLayout.SHARED_SHREDDING);
+
assertThat(options.mapSharedShreddingMaxColumns("metrics")).isEqualTo(32);
+
+ conf = new Options();
+ conf.setString("fields.metrics.map.storage-layout",
"Shared-Shredding");
+ options = new CoreOptions(conf);
+ assertThat(options.mapStorageLayout("metrics"))
+ .isEqualTo(CoreOptions.MapStorageLayout.SHARED_SHREDDING);
+
+ conf = new Options();
+ conf.setString("fields.metrics.map.storage-layout", "invalid");
+ final CoreOptions invalidLayoutOptions = new CoreOptions(conf);
+ assertThatThrownBy(() ->
invalidLayoutOptions.mapStorageLayout("metrics"))
+ .hasMessageContaining("invalid");
+
+ conf = new Options();
+ conf.setString("fields.metrics.map.shared-shredding.max-columns", "0");
+ final CoreOptions zeroMaxColumnsOptions = new CoreOptions(conf);
+ assertThatThrownBy(() ->
zeroMaxColumnsOptions.mapSharedShreddingMaxColumns("metrics"))
+ .hasMessageContaining("options
map.shared-shredding.max-columns must > 0");
+
+ conf.setString("fields.metrics.map.shared-shredding.max-columns",
"-1");
+ final CoreOptions negativeMaxColumnsOptions = new CoreOptions(conf);
+ assertThatThrownBy(() ->
negativeMaxColumnsOptions.mapSharedShreddingMaxColumns("metrics"))
+ .hasMessageContaining("options
map.shared-shredding.max-columns must > 0");
+ }
}
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 a306adc670..ee3e25846a 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
@@ -210,6 +210,109 @@ class SchemaValidationTest {
.hasMessageContaining("fields.data.num.aggregate-function");
}
+ @Test
+ public void testMapStorageLayout() {
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "id", DataTypes.INT()),
+ new DataField(
+ 1, "metrics",
DataTypes.MAP(DataTypes.STRING(), DataTypes.INT())),
+ new DataField(
+ 2, "codes", DataTypes.MAP(DataTypes.INT(),
DataTypes.STRING())));
+
+ Map<String, String> options = new HashMap<>();
+ options.put(BUCKET.key(), "-1");
+ options.put("fields.metrics.map.storage-layout", "shared-shredding");
+ assertThatNoException()
+ .isThrownBy(
+ () ->
+ validateTableSchema(
+ new TableSchema(
+ 1,
+ fields,
+ 10,
+ emptyList(),
+ emptyList(),
+ options,
+ "")));
+
+ options.put("fields.metrics.map.storage-layout", "default");
+ assertThatNoException()
+ .isThrownBy(
+ () ->
+ validateTableSchema(
+ new TableSchema(
+ 1,
+ fields,
+ 10,
+ emptyList(),
+ emptyList(),
+ options,
+ "")));
+
+ options.put("fields.nonexist.map.storage-layout", "shared-shredding");
+ assertThatThrownBy(
+ () ->
+ validateTableSchema(
+ new TableSchema(
+ 1,
+ fields,
+ 10,
+ emptyList(),
+ emptyList(),
+ options,
+ "")))
+ .hasMessageContaining("Field nonexist can not be found in
table schema.");
+
+ options.remove("fields.nonexist.map.storage-layout");
+ options.put("fields.id.map.storage-layout", "default");
+ assertThatThrownBy(
+ () ->
+ validateTableSchema(
+ new TableSchema(
+ 1,
+ fields,
+ 10,
+ emptyList(),
+ emptyList(),
+ options,
+ "")))
+ .hasMessageContaining(
+ "Column 'id' is configured with map.storage-layout but
its type is not MAP.");
+
+ options.remove("fields.id.map.storage-layout");
+ options.put("fields.codes.map.storage-layout", "shared-shredding");
+ assertThatThrownBy(
+ () ->
+ validateTableSchema(
+ new TableSchema(
+ 1,
+ fields,
+ 10,
+ emptyList(),
+ emptyList(),
+ options,
+ "")))
+ .hasMessageContaining(
+ "Column 'codes' is configured with
map.storage-layout=shared-shredding but its type is not MAP<STRING, T>.");
+
+ options.remove("fields.codes.map.storage-layout");
+ options.put("fields.metrics.map.storage-layout", "shared-shredding");
+ options.put("fields.metrics.map.shared-shredding.max-columns", "0");
+ assertThatThrownBy(
+ () ->
+ validateTableSchema(
+ new TableSchema(
+ 1,
+ fields,
+ 10,
+ emptyList(),
+ emptyList(),
+ options,
+ "")))
+ .hasMessageContaining("options
map.shared-shredding.max-columns must > 0");
+ }
+
@Test
public void testChainTableAllowsNonDeduplicateMergeEngine() {
Map<String, String> options = new HashMap<>();
