Flink: Support Variant type in DynamicIcebergSink

[fmorillo7694]

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 DynamicRecords 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.