vbhanuchander-lang opened a new pull request, #17644: URL: https://github.com/apache/iceberg/pull/17644
Closes #17298. ### The bug `FlinkTypeToType#visit(MultisetType)` converts a Flink `MULTISET<T>` to an Iceberg `map<T, int>` of element to occurrence count, so a table with a multiset column is created happily. The write path then pairs that Iceberg map with the Flink `MultisetType` — which is `final` and extends `LogicalType` directly, so it is *not* a `MapType` — and every data file writer rejected it. The column could never be written, in any format: | Format | Where | | --- | --- | | Parquet | `ParquetWithFlinkSchemaVisitor` — `checkArgument(sType instanceof MapType)` then a cast | | Avro | `AvroWithFlinkSchemaVisitor` — `isMapType` / `mapKeyType` / `mapValueType` | | ORC | `FlinkSchemaVisitor` — unchecked `(MapType) flinkType` cast | ### The fix Visit a multiset as its equivalent `map<element, int NOT NULL>`. This is not a new convention — it is the one this repository already uses. `RowData` represents `MULTISET` and `MAP` identically as `MapData`, which is why Iceberg's own `RowDataToAvroConverters` routes `case MAP: case MULTISET:` to the same converter, and `AvroSchemaConverter#extractValueTypeToAvroMap` performs exactly this element-as-key, int-as-value extraction. The value is **required**, matching `Types.MapType.ofRequired(...)` in `FlinkTypeToType`. Applied to Flink **1.20, 2.0 and 2.1**, since all three share the defect. ### Why no existing test caught it The writer tests derive the Flink type with `FlinkSchemaUtil.convert(icebergSchema)`, which turns `map<string, int>` back into a `MapType`. That path **cannot produce a `MultisetType`**, so the multiset write path was unreachable from the existing harness — which is why a column you can create but never write to went unnoticed. The new `TestFlinkMultisetWrite` therefore pairs an explicit `ROW<id INT NOT NULL, tags MULTISET<STRING NOT NULL>>` with the Iceberg schema, the way Flink actually does at runtime. ### Verification | | Result | | --- | --- | | Without the change | the three write cases FAIL on 1.20, 2.0 and 2.1 | | With the change | 4 tests pass on each of 1.20, 2.0, 2.1 | | Regression | `flink.data.*` on 2.1 — **669 tests, 0 failures** | | Style | `spotlessCheck` on all three, `checkstyleMain`/`checkstyleTest` on 2.1 — clean | ### Note Only the write path is changed. Reads already work, because reading produces the Iceberg `map<string, int>` and Flink accepts a map where a multiset is expected in that direction; this PR does not alter read behaviour or the schema conversion itself. 🤖 AI-assisted — generated with Claude Code (Opus 5) and reviewed by me before submitting. -- 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]
