apache / apache/iceberg

Flink: MULTISET columns cannot be written — Parquet/Avro/ORC data writers reject MultisetType

Open
#17,298 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
Java
Stars
9.2k
Forks
3.5k
Avg merge
2d 16h
Merged PRs (30d)
129

Description

### 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` to an Iceberg `map` 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 — in every file format (Parquet, Avro, and ORC).

**Root cause**

`FlinkSchemaUtil.convert()` → `FlinkTypeToType.visit(MultisetType)` maps `MULTISET` to `Types.MapType.ofRequired(..., IntegerType)` (element → occurrence count), so schema conversion and table creation succeed.

On the write path, the Flink write-schema visitors pair the map node of the file/Iceberg schema with the source Flink `LogicalType` and assume that type is a `MapType`. For a multiset column the paired Flink type is `MultisetType`, which is a `final` class extending `LogicalType` directly — it is **not** a `MapType`. All three data-file writers therefore fail:

- **Parquet** — `ParquetWithFlinkSchemaVisitor` asserts `sType instanceof MapType`:
```java
Preconditions.checkArgument(sType instanceof MapType, "Invalid map: %s is not a map", sType);
MapType map = (MapType) sType;
```
→ `IllegalArgumentException: Invalid map: MULTISET is not a map`
- **Avro** — `AvroWithFlinkSchemaVisitor.isMapType(...)` performs the same `instanceof MapType` check → same `IllegalArgumentException`.
- **ORC** — `FlinkSchemaVisitor` casts directly in `case MAP`:
```java
case MAP:
MapType mapType = (MapType) flinkType; // flinkType is a MultisetType
```
→ `ClassCastException: class org.apache.flink.table.types.logical.MultisetType cannot be cast to class org.apache.flink.table.types.logical.MapType`

**Minimal reproduction (Parquet)**

```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 column: map
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 is not a map
```

The Avro (`FlinkAvroWriter`) and ORC (`FlinkOrcWriter`) writers fail equivalently for the same `flinkSchema`.

**Expected behavior**

Multiset columns should be writable using their `map` 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 paths can serialize it once the visitors accept `MultisetType` (normalizing it to `MAP`).

**Affected code** (all maintained Flink versions — `flink/v1.20`, `flink/v2.0`, `flink/v2.1`):
- `org.apache.iceberg.flink.data.ParquetWithFlinkSchemaVisitor` (Parquet)
- `org.apache.iceberg.flink.data.AvroWithFlinkSchemaVisitor` (Avro)
- `org.apache.iceberg.flink.data.FlinkSchemaVisitor` (ORC)

### Willingness to contribute

- [x] I can contribute a fix for this bug independently

Contributor guide

Open the contributing guide

Research direction

Start by reading ParquetWithFlinkSchemaVisitor, AvroWithFlinkSchemaVisitor, and FlinkSchemaVisitor at their map-handling paths, then inspect the existing Flink writer tests or entry points. Done means MULTISET columns use the same map representation as MAP and can be written successfully in Parquet, Avro, and ORC across the affected Flink branches.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
data-engineering
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Quiet
Clarity
Clearly specified
Newbie friendliness
66/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.