Flink: DynamicIcebergSink throws UnsupportedOperationException for VARIANT columns

[peach12345]

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:

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

@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):

@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

• 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