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

[srpconfluent]

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 — in every file format (Parquet, Avro, and ORC).

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, 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:

  Preconditions.checkArgument(sType instanceof MapType, "Invalid map: %s is not a map", sType);
  MapType map = (MapType) sType;

  → IllegalArgumentException: Invalid map: MULTISET<INT NOT NULL> is not a map
• Avro — AvroWithFlinkSchemaVisitor.isMapType(...) performs the same instanceof MapType check → same IllegalArgumentException.
• ORC — FlinkSchemaVisitor casts directly in case MAP:

  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)

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

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

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 paths can serialize it once the visitors accept 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 (Parquet)
• org.apache.iceberg.flink.data.AvroWithFlinkSchemaVisitor (Avro)
• org.apache.iceberg.flink.data.FlinkSchemaVisitor (ORC)

Willingness to contribute

• I can contribute a fix for this bug independently