peach12345 opened a new issue, #17615:
URL: https://github.com/apache/iceberg/issues/17615
### Apache Iceberg version
1.11.0 (latest release)
### Query engine
Flink
### Please describe the bug 🐞
### Environment
- Apache Iceberg: 1.11.0
- Module: `iceberg-flink-runtime-2.1`
- Flink: 2.2.1
### Description
When using `DynamicIcebergSink` to write to an Iceberg table that contains a
`VARIANT` column, the following exception is thrown at runtime:
```
Caused by: java.lang.UnsupportedOperationException: Unsupported type: variant
at
org.apache.iceberg.schema.SchemaWithPartnerVisitor.variant(SchemaWithPartnerVi
sitor.java:167)
at
org.apache.iceberg.schema.SchemaWithPartnerVisitor.visit(SchemaWithPartnerVisi
tor.java:111)
at
org.apache.iceberg.schema.SchemaWithPartnerVisitor.visit(SchemaWithPartnerVisi
tor.java:62)
at
org.apache.iceberg.schema.SchemaWithPartnerVisitor.visit(SchemaWithPartnerVisi
tor.java:45)
at
org.apache.iceberg.flink.sink.dynamic.CompareSchemasVisitor.visit(CompareSchem
asVisitor.java:60)
at
org.apache.iceberg.flink.sink.dynamic.TableMetadataCache.schema(TableMetadataC
ache.java:161)
at
org.apache.iceberg.flink.sink.dynamic.TableMetadataCache.schema(TableMetadataC
ache.java:112)
at
org.apache.iceberg.flink.sink.dynamic.DynamicRecordProcessor.collect(DynamicRe
cordProcessor.java:138)
```
### Root Cause
`SchemaWithPartnerVisitor` correctly dispatches `VARIANT` types via its
`variant()` method,
but the default implementation throws `UnsupportedOperationException`:
```java
// SchemaWithPartnerVisitor.java
public R variant(Types.VariantType variant, P partner) {
throw new UnsupportedOperationException("Unsupported type: variant");
}
```
The following two classes in the Flink Dynamic Sink inherit from
SchemaWithPartnerVisitor but do not override variant():
1. CompareSchemasVisitor — compares the input schema against the table
schema on every record. Called from TableMetadataCache.schema().
2. EvolveSchemaVisitor — performs schema evolution when
CompareSchemasVisitor returns SCHEMA_UPDATE_NEEDED. Called from
TableUpdater.findOrCreateSchema().
### Proposed Fix:
CompareSchemasVisitor needs a variant() override that compares the input
VARIANT against the table schema field:
```java
@Override
public Result variant(Types.VariantType variant, Integer tableSchemaId) {
if (tableSchemaId == null) {
return Result.SCHEMA_UPDATE_NEEDED;
}
Type tableSchemaType = tableSchema.findField(tableSchemaId).type();
if (tableSchemaType.isVariantType()) {
return Result.SAME;
}
return Result.SCHEMA_UPDATE_NEEDED;
}
```
EvolveSchemaVisitor needs a variant() override that is a no-op when the
VARIANT field already exists in the table schema (no type evolution needed):
```java
@Override
public Boolean variant(Types.VariantType variant, Integer partnerId) {
return partnerId == null;
}
```
Additionally, DataConverter.get() is missing a case VARIANT: branch, which
would cause a secondary UnsupportedOperationException if DATA_CONVERSION_NEEDED
were ever reached for a VARIANT field.
Files to change
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSc
hemasVisitor.java
-
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSch
emaVisitor.java
-
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DataConve
rter.java
-
### Willingness to contribute
- [x] I can contribute a fix for this bug independently
- [ ] I would be willing to contribute a fix for this bug with guidance from
the Iceberg community
- [ ] I cannot contribute a fix for this bug at this time
--
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]
