srpconfluent opened a new issue, #17298:
URL: https://github.com/apache/iceberg/issues/17298
### Apache Iceberg version
1.10.0 (latest release) — also reproduces on `main`
### Query engine
Flink
### Please describe the bug 🐞
**Summary:** Iceberg's Flink integration converts a Flink `MULTISET<T>` to
an Iceberg `map<T, int>` at the *schema* level, but the Flink *data writers*
reject `MultisetType`. As a result a table with a multiset column can be
created, but writing to it always fails.
**Root cause**
`FlinkSchemaUtil.convert()` → `FlinkTypeToType.visit(MultisetType)` maps
`MULTISET<T>` to `Types.MapType.ofRequired(..., IntegerType)` (element →
occurrence count), so schema conversion and table creation succeed.
On the write path, `ParquetWithFlinkSchemaVisitor` (and
`AvroWithFlinkSchemaVisitor`) walk the file schema's map node while pairing it
with the source Flink `LogicalType`, and assert:
```java
Preconditions.checkArgument(sType instanceof MapType, "Invalid map: %s is
not a map", sType);
MapType map = (MapType) sType;
```
For a multiset column the paired Flink type is `MultisetType`, which is a
`final` class extending `LogicalType` directly — it is **not** a `MapType`. So
the check throws:
```
java.lang.IllegalArgumentException: Invalid map: MULTISET<INT NOT NULL> is
not a map
at
org.apache.iceberg.flink.data.ParquetWithFlinkSchemaVisitor.visit(ParquetWithFlinkSchemaVisitor.java)
at
org.apache.iceberg.flink.data.FlinkParquetWriters.buildWriter(FlinkParquetWriters.java)
```
The Avro writer path fails the same way via
`AvroWithFlinkSchemaVisitor.isMapType(...)`.
**Minimal reproduction**
```java
import org.apache.flink.table.types.logical.IntType;
import org.apache.flink.table.types.logical.MultisetType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.iceberg.Schema;
import org.apache.iceberg.flink.data.FlinkParquetWriters;
import org.apache.iceberg.parquet.ParquetSchemaUtil;
import org.apache.iceberg.types.Types;
import org.apache.parquet.schema.MessageType;
// Iceberg schema as produced by FlinkTypeToType for a MULTISET<INT> column:
map<int, int>
Schema iceberg =
new Schema(
Types.NestedField.required(
1,
"f0",
Types.MapType.ofRequired(2, 3, Types.IntegerType.get(),
Types.IntegerType.get())));
// The canonical Flink schema still carries the MULTISET type
RowType flinkSchema = RowType.of(new MultisetType(false, new
IntType(false)));
MessageType parquetType = ParquetSchemaUtil.convert(iceberg, "table");
FlinkParquetWriters.buildWriter(flinkSchema, parquetType);
// => java.lang.IllegalArgumentException: Invalid map: MULTISET<INT NOT
NULL> is not a map
```
**Expected behavior**
Multiset columns should be writable using their `map<element, int>`
representation, consistent with the schema converter. Flink already backs
`MULTISET` with the same `MapData` runtime representation as `MAP` (element →
occurrence count) and exposes it via `RowData#getMap`, so the existing map
write path can serialize it once the visitor accepts `MultisetType`
(normalizing it to `MAP<element, INT NOT NULL>`).
**Affected code** (all maintained Flink versions — `flink/v1.20`,
`flink/v2.0`, `flink/v2.1`):
- `org.apache.iceberg.flink.data.ParquetWithFlinkSchemaVisitor`
- `org.apache.iceberg.flink.data.AvroWithFlinkSchemaVisitor`
### Willingness to contribute
- [x] I can contribute a fix for this bug independently
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
