Flink: DynamicIcebergSink throws UnsupportedOperationException for VARIANT columns

### 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

Start at DynamicRecordProcessor.collect and TableMetadataCache.schema, then inspect CompareSchemasVisitor.java, EvolveSchemaVisitor.java, and DataConverter.java in flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/. Reproduce the reported Flink VARIANT-column failure and verify that DynamicIcebergSink handles existing VARIANT fields without UnsupportedOperationException or incorrect schema evolution.

Written by the indexing model from the issue text.