[python] Support writing shared-shredding MAPs in Parquet

paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java

@@ -1620,6 +1620,67 @@ public void testJavaWriteSharedShreddingMapTable() throws Exception {
         }
     }
 
+    /** Java reads shared-shredding MAP columns written by Python. */
+    @Test
+    @EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
+    public void testJavaReadSharedShreddingMapTable() throws Exception {
+        FileStoreTable table =
+                (FileStoreTable)
+                        catalog.getTable(identifier("shared_shredding_map_python_test_parquet"));
+        Map<Integer, Map<String, Long>> rows = new HashMap<>();
+        List<Split> splits = new ArrayList<>(table.newSnapshotReader().read().dataSplits());
+        try (org.apache.paimon.reader.RecordReader<InternalRow> reader =
+                table.newRead().createReader(splits)) {
+            reader.forEachRemaining(
+                    row -> {
+                        int id = row.getInt(0);
+                        for (int column = 2; column <= 3; column++) {
+                            if (id == 3) {
+                                assertThat(row.isNullAt(column)).isTrue();
+                            } else {
+                                InternalMap required = row.getMap(column);
+                                assertThat(required.size()).isEqualTo(id == 2 ? 0 : 1);
+                                if (id != 2) {
+                                    InternalArray values = required.valueArray();
+                                    long value =
+                                            column == 2
+                                                    ? values.getLong(0)
+                                                    : values.getRow(0, 1).getLong(0);
+                                    assertThat(value).isEqualTo(id == 1 ? 1L : 2L);
+                                }
+                            }
+                        }
+                        if (row.isNullAt(1)) {
+                            rows.put(id, null);
+                            return;
+                        }
+                        InternalMap map = row.getMap(1);
+                        InternalArray keys = map.keyArray();
+                        InternalArray values = map.valueArray();
+                        Map<String, Long> converted = new LinkedHashMap<>();
+                        for (int i = 0; i < map.size(); i++) {
+                            converted.put(
+                                    keys.getString(i).toString(),
+                                    values.isNullAt(i) ? null : values.getLong(i));
+                        }
+                        rows.put(id, converted);
+                    });
+        }
+
+        assertThat(rows).containsOnlyKeys(1, 2, 3, 4);
+        assertThat(rows.get(1))
+                .containsOnlyKeys("hot", "warm", "overflow")
+                .containsEntry("hot", 10L)
+                .containsEntry("warm", 20L)
+                .containsEntry("overflow", 30L);
+        assertThat(rows.get(2))
+                .containsOnlyKeys("hot", "new")
+                .containsEntry("hot", null)
+                .containsEntry("new", 40L);
+        assertThat(rows.get(3)).isEmpty();
+        assertThat(rows.get(4)).isNull();
+    }
+
     private Map<Integer, Map<Integer, byte[]>> readMapBlobRows(FileStoreTable table)
             throws Exception {
         Map<Integer, Map<Integer, byte[]>> rows = new HashMap<>();

paimon-python/dev/run_mixed_tests.sh

@@ -1033,6 +1033,21 @@ run_shared_shredding_map_test() {
         return 1
     fi
     echo -e "${GREEN}✓ Python shared-shredding MAP read test completed successfully${NC}"
+
+    echo "Running Python shared-shredding MAP write test..."
+    if ! python -m pytest java_py_read_write_test.py::JavaPyReadWriteTest::test_write_shared_shredding_map_for_java -v; then
+        echo -e "${RED}✗ Python shared-shredding MAP write test failed${NC}"
+        return 1
+    fi
+    echo -e "${GREEN}✓ Python shared-shredding MAP write test completed successfully${NC}"
+
+    cd "$PROJECT_ROOT"
+    echo "Running Maven test for JavaPyE2ETest.testJavaReadSharedShreddingMapTable..."
+    if ! mvn test -Dtest=org.apache.paimon.JavaPyE2ETest#testJavaReadSharedShreddingMapTable -pl paimon-core -q -Drun.e2e.tests=true; then
+        echo -e "${RED}✗ Java shared-shredding MAP read test failed${NC}"
+        return 1
+    fi
+    echo -e "${GREEN}✓ Java shared-shredding MAP read test completed successfully${NC}"
 }
 
 # Function to run VARIANT test (Java write, Python read)

paimon-python/pypaimon/common/options/core_options.py

@@ -158,6 +158,10 @@ class CoreOptions:
     NESTED_SEQUENCE_FIELD = "nested-sequence-field"
     COUNT_LIMIT = "count-limit"
     MERGE_MAP_TS_FIELD = "ts-field"
+    MAP_STORAGE_LAYOUT = "map.storage-layout"
+    MAP_SHARED_SHREDDING_MAX_COLUMNS = "map.shared-shredding.max-columns"
+    MAP_SHARED_SHREDDING_COLUMN_PLACEMENT_POLICY = \
+        "map.shared-shredding.column-placement-policy"
 
     # Basic options
     AUTO_CREATE: ConfigOption[bool] = (
@@ -1873,6 +1877,46 @@ def field_merge_map_ts_field(self, field_name: str) -> str:
             .no_default_value()
         )
 
+    def map_storage_layout(self, field_name: str) -> str:
+        return self.options.get(
+            ConfigOptions.key(
+                f'{CoreOptions.FIELDS_PREFIX}.{field_name}.{CoreOptions.MAP_STORAGE_LAYOUT}'
+            )
+            .string_type()
+            .default_value('default')
+        ).lower()
+
+    def map_shared_shredding_max_columns(self, field_name: str) -> int:
+        value = self.options.get(
+            ConfigOptions.key(
+                f'{CoreOptions.FIELDS_PREFIX}.{field_name}.'
+                f'{CoreOptions.MAP_SHARED_SHREDDING_MAX_COLUMNS}'
+            )
+            .int_type()
+            .default_value(256)
+        )
+        if value <= 0:
+            raise ValueError(
+                '{} must be greater than 0'.format(
+                    CoreOptions.MAP_SHARED_SHREDDING_MAX_COLUMNS))
+        return value
+
+    def map_shared_shredding_column_placement_policy(
+            self, field_name: str) -> str:
+        value = self.options.get(
+            ConfigOptions.key(
+                f'{CoreOptions.FIELDS_PREFIX}.{field_name}.'
+                f'{CoreOptions.MAP_SHARED_SHREDDING_COLUMN_PLACEMENT_POLICY}'
+            )
+            .string_type()
+            .default_value('lru')
+        ).lower()
+        if value not in ('plain', 'sequential', 'lru'):
+            raise ValueError(
+                "Unsupported shared-shredding column placement policy: {}".format(
+                    value))
+        return value
+
     @property
     def query_auth_enabled(self) -> bool:
         return self.options.get(CoreOptions.QUERY_AUTH_ENABLED)

paimon-python/pypaimon/data/map_shared_shredding.py

@@ -14,7 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""Read support for Paimon's shared-shredding MAP storage layout."""
+"""Metadata and read support for the shared-shredding MAP layout."""
 
 import json
 import struct
@@ -33,6 +33,7 @@
 _NUM_COLUMNS = b"paimon.map.shared-shredding.num-columns"
 _FIELD_COLUMNS = b"paimon.map.shared-shredding.field-columns"
 _OVERFLOW_SET = b"paimon.map.shared-shredding.overflow-set"
+_MAX_ROW_WIDTH = b"paimon.map.shared-shredding.max-row-width"
 _FIELD_MAPPING = "__field_mapping"
 _OVERFLOW = "__overflow"
 _PHYSICAL_COLUMN_PREFIX = "__col_"
@@ -104,6 +105,40 @@ def parse_shared_shredding_selection_metadata(field: pa.Field):
     return name_by_id, field_to_columns, set(overflow_json), num_columns
 
 
+def shared_shredding_metadata(
+        name_to_id, field_to_columns, overflow_set, num_columns,
+        max_row_width, compression):
+    """Build Java-compatible Arrow field metadata for one data file."""
+    compression = compression.lower()
+    if compression not in ("none", "lz4", "zstd"):
+        raise ValueError(
+            "MAP shared-shredding only supports none/lz4/zstd compression, "
+            "but is {}.".format(compression))
+    field_dict = json.dumps(
+        dict(sorted(name_to_id.items())),
+        separators=(",", ":"),
+        sort_keys=True,
+    ).encode("utf-8")
+    encoded_dict = _compress(field_dict, compression)
+    columns = {
+        str(field_id): sorted(column_ids)
+        for field_id, column_ids in sorted(field_to_columns.items())
+    }
+    return {
+        _STORAGE_LAYOUT: b"shared-shredding",
+        _VERSION: b"1",
+        _FIELD_DICT: encoded_dict.decode("latin-1").encode("utf-8"),
+        _FIELD_DICT_COMPRESSION: compression.encode("utf-8"),
+        _FIELD_DICT_ORIGINAL_SIZE: str(len(field_dict)).encode("utf-8"),
+        _FIELD_COLUMNS: json.dumps(
+            columns, separators=(",", ":"), sort_keys=True).encode("utf-8"),
+        _OVERFLOW_SET: json.dumps(
+            sorted(overflow_set), separators=(",", ":")).encode("utf-8"),
+        _NUM_COLUMNS: str(num_columns).encode("utf-8"),
+        _MAX_ROW_WIDTH: str(max_row_width).encode("utf-8"),
+    }
+
+
 def map_selected_keys(description: str) -> List[str]:
     if not description or not description.startswith(_SELECTED_KEYS_PREFIX):
         raise ValueError("Invalid selected-key MAP metadata: {}".format(
@@ -589,6 +624,20 @@ def _decompress(data: bytes, original_size: int, compression: str) -> bytes:
     return result
 
 
+def _compress(data: bytes, compression: str) -> bytes:
+    if compression == "none":
+        return data
+    if compression == "zstd":
+        import zstandard as zstd
+        return zstd.ZstdCompressor(level=1).compress(data)
+    if compression == "lz4":
+        payload = bytes(pa.Codec("lz4_raw").compress(data))
+        return struct.pack("<ii", len(payload), len(data)) + payload
+    raise ValueError(
+        "Unsupported shared-shredding dictionary compression: {}".format(
+            compression))
+
+
 def _required(metadata, key):
     try:
         return metadata[key]

paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py

@@ -1636,6 +1636,52 @@ def test_read_selected_shared_shredding_keys_written_by_java(self):
                 self.assertEqual([None] * 5,
                                  result.column('metrics_missing').to_pylist())
 
+    def test_write_shared_shredding_map_for_java(self):
+        schema = pa.schema([
+            pa.field('id', pa.int32()),
+            pa.field('metrics', pa.map_(pa.string(), pa.int64())),
+            pa.field('required_values', pa.map_(pa.string(), pa.field(
+                'value', pa.int64(), nullable=False))),
+            pa.field('required_rows', pa.map_(pa.string(), pa.field(
+                'value', pa.struct([pa.field('score', pa.int64(), nullable=False)]),
+                nullable=False))),
+        ])
+        rows = pa.Table.from_pydict({
+            'id': [1, 2, 3, 4],
+            'metrics': [
+                [('hot', 10), ('warm', 20), ('overflow', 30)],
+                [('hot', None), ('new', 40)],
+                [],
+                None,
+            ],
+            'required_values': [[('a', 1)], [], None, [('b', 2)]],
+            'required_rows': [[('a', {'score': 1})], [], None, [('b', {'score': 2})]],
+        }, schema=schema)
+
+        table_name = 'default.shared_shredding_map_python_test_parquet'
+        self.catalog.drop_table(table_name, True)
+        self.catalog.create_table(
+            table_name,
+            Schema.from_pyarrow_schema(schema, options={
+                'bucket': '-1',
+                'file.format': 'parquet',
+                'write-only': 'true',
+                'fields.metrics.map.storage-layout': 'shared-shredding',
+                'fields.metrics.map.shared-shredding.max-columns': '2',
+                'fields.required_values.map.storage-layout': 'shared-shredding',
+                'fields.required_values.map.shared-shredding.max-columns': '2',
+                'fields.required_rows.map.storage-layout': 'shared-shredding',
+                'fields.required_rows.map.shared-shredding.max-columns': '2',
+            }),
+            False,
+        )
+        table = self.catalog.get_table(table_name)
+        builder = table.new_batch_write_builder()
+        writer = builder.new_write()
+        writer.write_arrow(rows)
+        builder.new_commit().commit(writer.prepare_commit())
+        writer.close()
+
     def test_write_map_blob_for_java(self):
         map_blob_type = pa.map_(pa.int32(), pa.large_binary())
         boolean_map_blob_type = pa.map_(pa.bool_(), pa.large_binary())