fmorillo7694 opened a new pull request, #17900:
URL: https://github.com/apache/iceberg/pull/17900
Fixes #17615
## Problem
`DynamicIcebergSink` fails on any `DynamicRecord` whose schema contains a
`variant` field:
```
java.lang.UnsupportedOperationException: Unsupported type: variant
at
org.apache.iceberg.schema.SchemaWithPartnerVisitor.variant(SchemaWithPartnerVisitor.java:167)
at
org.apache.iceberg.schema.SchemaWithPartnerVisitor.visit(SchemaWithPartnerVisitor.java:111)
at
org.apache.iceberg.schema.SchemaWithPartnerVisitor.visit(SchemaWithPartnerVisitor.java:62)
at
org.apache.iceberg.schema.SchemaWithPartnerVisitor.visit(SchemaWithPartnerVisitor.java:45)
at
org.apache.iceberg.flink.sink.dynamic.CompareSchemasVisitor.visit(CompareSchemasVisitor.java:60)
at
org.apache.iceberg.flink.sink.dynamic.TableMetadataCache.schema(TableMetadataCache.java:161)
at
org.apache.iceberg.flink.sink.dynamic.TableMetadataCache.schema(TableMetadataCache.java:112)
at
org.apache.iceberg.flink.sink.dynamic.DynamicRecordProcessor.collect(DynamicRecordProcessor.java:138)
```
`CompareSchemasVisitor` and `EvolveSchemaVisitor` extend
`SchemaWithPartnerVisitor` but do not override `variant()`, so they inherit the
base implementation which throws unconditionally. Since
`TableMetadataCache.schema()` runs `CompareSchemasVisitor` on every incoming
record's schema, the failure occurs during schema resolution — before any
writer is created, and **even when the target table already exists with an
exactly matching schema**. `DataConverter.get()` also lacks a `VARIANT` case on
the conversion path.
The rest of the stack already supports variant: `FlinkSchemaUtil` maps
`VariantType` in both directions, `FlinkParquetWriters`/`FlinkParquetReaders`
read and write it, and the static `IcebergSink` writes variant fields
end-to-end (verified: identical schema + row data succeeds through
`IcebergSink.forRowData` against the same table where the dynamic sink throws).
Notably, `VariantAvroDynamicTableRecordGenerator` (#16450) already produces
variant-bearing `DynamicRecord`s that the pipeline is currently unable to
process.
## Changes
- `CompareSchemasVisitor`: add `variant()` override — `SAME` when the table
field is also variant, `SCHEMA_UPDATE_NEEDED` otherwise (variant has no
widening/conversion semantics).
- `EvolveSchemaVisitor`: add `variant()` override, consistent with
`primitive()`.
- `DataConverter`: add identity `case VARIANT`.
- Applied to the `v2.1`, `v2.2`, and `v2.3` modules. `v1.20` is
intentionally excluded: Flink 1.20 has no `VariantType` /
`LogicalTypeRoot.VARIANT`, so the dynamic sink cannot receive variant data
there and a `case VARIANT:` in `DataConverter` would not compile against Flink
1.20.
## Tests
Per module (v2.1/v2.2/v2.3):
- `TestCompareSchemasVisitor`: same variant schema → `SAME`; variant vs
non-variant table field → `SCHEMA_UPDATE_NEEDED`; variant field missing from
table → `SCHEMA_UPDATE_NEEDED`.
- `TestEvolveSchemaVisitor`: adding a top-level variant field via schema
evolution; identical variant schema is a no-op.
- `TestRowDataConverter`: variant value passes through `DataConverter`
unchanged.
`:iceberg-flink:iceberg-flink-{2.1,2.2,2.3}:test` for the three test
classes: 79 tests per module, 0 failures. Spotless clean.
## Relationship to #17631
This supersedes #17631 (thank you @waterWang for the original fix — credited
via `Co-authored-by` on the commit). That PR targets the `flink/v2.0` module
which no longer exists on `main`, and its `v1.20` `DataConverter` change cannot
compile against Flink 1.20; this PR retargets the same fix to the current
module layout (v2.1/v2.2/v2.3) and adds test coverage.
--
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]
