[FLINK-40825][table] Support casting primitive types to VARIANT

[raminqaf]

What is the purpose of the change

FLIP-521 lists the types that cast to and from VARIANT. The direction VARIANT to SQL type exists, but SqlCastFunction rejected every cast to VARIANT. This PR adds CAST and TRY_CAST from primitive types to VARIANT. Casting a whole ARRAY, MAP, or ROW into one VARIANT follows in FLINK-40826.

A type casts to VARIANT only if a VARIANT kind holds its value without loss. Any other type is rejected at validation.

Expression	Result
CAST(CAST(1 AS BIGINT) AS VARIANT)	1, stored as BIGINT
CAST(42 AS VARIANT)	42, stored as INT
CAST('{"a": 1}' AS VARIANT)	the string '{"a": 1}', not an object
CAST(CAST(NULL AS INT) AS VARIANT)	SQL NULL, not a variant null
CAST(CAST('NaN' AS DOUBLE) AS VARIANT)	NaN, stored as DOUBLE
CAST(ts AS VARIANT) for ts TIMESTAMP(9) in the year 3000	fails, outside the nanosecond range
TRY_CAST(REPEAT('x', 17000000) AS VARIANT)	NULL, a VARIANT holds at most 16 MiB
CAST(ARRAY[1, NULL] AS ARRAY<VARIANT>)	[1, NULL], each element a VARIANT, the NULL stays SQL NULL
CAST(INTERVAL '2' DAY AS VARIANT)	fails at validation
CAST(ARRAY[1, 2] AS VARIANT)	fails at validation, FLINK-40826

Brief change log

• Casting a VARIANT to FLOAT or DOUBLE keeps a stored NaN or infinity. The FLOAT cast rejects only a finite value that does not fit, such as 1e40. Before, a VARIANT holding NaN, for example from the Avro converter, could not be read back. This is the first commit.
• SqlCastFunction#canCastFrom routes a VARIANT target to LogicalTypeCasts instead of rejecting it, the same way it handles UUID.
• LogicalTypeCasts declares the supported sources explicitly: BOOLEAN, the numeric types, CHAR, VARCHAR, BINARY, VARBINARY, DATE, TIME, TIMESTAMP, TIMESTAMP_LTZ, and UUID. The cast is explicit only.
• New PrimitiveToVariantCastRule, backed by VariantCastUtils#fromXxx helpers. It fails only for a TIMESTAMP(p) or TIMESTAMP_LTZ(p) with p above 6 whose value is outside the nanosecond range, and for a string or binary value over 16 MiB. canFail reports exactly the types whose precision or declared length allows such a value, so TRY_CAST returns NULL for it.
• Docs: a "cast to VARIANT" section in data-types.md, the VARIANT column of the cast matrix, and TIME and UUID in the list of VARIANT kinds.
• A constructed cast checks each child with the same rules, so ARRAY<INT> to ARRAY<VARIANT>, and the same for ROW and MAP, works element by element at any depth. A NULL element stays a SQL NULL. A whole ARRAY, MAP, or ROW into one VARIANT is still rejected and follows in FLINK-40826. This also allows a MAP<VARIANT, ...> key and a MULTISET<VARIANT> element. Two VARIANT values are equal only when their bytes match, so a key cast from INT does not match the same number cast from BIGINT. The docs say so.

Notes for reviewers:

• An integer keeps the width of its SQL type. PARSE_JSON('1') picks the smallest kind because JSON text carries no width, but a cast keeps the declared type. Both are valid under the Parquet Variant spec, which treats all integer widths as one equivalence class.
• A string is wrapped as a STRING kind and never parsed. PARSE_JSON stays the way to parse JSON text.
• NaN and infinity are stored. The binary format holds IEEE numbers, and PARSE_JSON rejects them only because JSON has no literal for them. JSON_STRING still fails on such a value. Printing and CAST(v AS STRING) show NaN. The json and raw formats call toJson() as well, so writing such a value to those sinks fails the job. The docs say so.
• A timestamp keeps its declared precision, the same way CAST(TIMESTAMP(p) AS STRING) prints exactly p digits and the Avro format picks millis or micros from p. Up to a precision of 6 it is stored with microseconds, above with nanoseconds, even for a value without digits below a microsecond. Nanoseconds only cover 1677-09-21 to 2262-04-11, so for a precision above 6 a value outside that range fails, and TRY_CAST returns NULL. VariantBuilder picks the kind by value, so the helpers use BinaryVariantInternalBuilder directly.
• TIME is stored in microseconds, the only TIME precision of the VARIANT spec. Every Flink TIME fits, since the runtime keeps milliseconds.
• canFail trusts the declared length of a string or binary type. Flink does not enforce that length on values from a source, so TRY_CAST would still fail for a longer value in a VARCHAR(100). That is pathological, and the Javadoc says so.
• The explicit-only choice is deliberate. Making the cast implicit later is additive, but taking implicit back would break queries.
• SqlCastFunction is a copy of the Calcite class. The change replaces its existing TODO to support casts to VARIANT.
• A LogicalTypeCastsTest row asserted that UUID does not cast to VARIANT. It now asserts that it does.

Verifying this change

This change added tests and can be verified as follows:

• LogicalTypeCastsTest covers every supported source, and rejects INTERVAL, TIMESTAMP WITH TIME ZONE, MULTISET, ARRAY, MAP, and ROW. It allows ARRAY<INT> to ARRAY<VARIANT>, ROW and MAP alike, and rejects ARRAY<INTERVAL> to ARRAY<VARIANT>.
• VariantCastUtilsTest pins the size limit: 16,777,211 bytes cast, one more byte fails with a clear message.
• CastRuleProviderTest checks rule resolution, including the array, row, and map rules for constructed targets, that VARIANT to VARIANT stays the identity, and which types can fail: a timestamp with a precision above 6, and a string or binary type whose declared length allows more than 16 MiB, with the exact boundaries.
• CastRulesTest checks the stored kind for every source, including the kept integer width, TIME(0) and TIME(3) up to the last millisecond of the day, TIMESTAMP(3), TIMESTAMP(6), and a TIMESTAMP(9) without digits below a microsecond, NaN and infinity, and SQL NULL. It checks that a TIMESTAMP(9) and TIMESTAMP_LTZ(7) outside the nanosecond range fail, while a TIMESTAMP(6) in the year 3000 casts. It also covers NaN and infinity read back from VARIANT to FLOAT and DOUBLE. It covers a pre-epoch timestamp and element-wise casts into ARRAY<VARIANT>, ROW<.. VARIANT>, and MAP<STRING, VARIANT>, and into MAP<VARIANT, STRING> keys and MULTISET<VARIANT> elements.
• CastFunctionITCase round-trips each type through VARIANT in SQL and the Table API, for literals and for values computed at runtime, including TIME, NaN, and infinity. It checks that a late TIMESTAMP(9) fails with CAST, returns NULL with TRY_CAST, and that a late TIMESTAMP(6) round-trips. It also checks the validation errors. It round-trips TINYINT, SMALLINT, CHAR(n) with its padding, BINARY(n) with its zero padding, and the constructed casts. It checks that TRY_CAST returns NULL for a string over 16 MiB, and that a MAP<VARIANT, STRING> key cast from INT matches an INT VARIANT but not a BIGINT one.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): yes. The new cast runs per record and builds one VARIANT per value.
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? yes
• If yes, how is the feature documented? docs, in data-types.md (English and Chinese)

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Opus 5.5

[flinkbot]

CI report:

• 9e51eb8 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[AHeise]

Thanks, the direction looks good, and keeping the kind of the SQL type is the right call. Two things before merge: the size limit makes string and binary casts fail even though the rule says they can't, and ARRAY/ROW/MAP of VARIANT are now castable, which is untested and undocumented. The rest are questions and nits.

[AHeise]

Thanks for the quick turnaround, everything from the first round is addressed. One open question about VARIANT as a MAP key or MULTISET element; the rest are nits. Please squash the two follow-up commits into "Support casting primitive types to VARIANT" before merge, keeping the NaN commit separate. "Cover casts … already worked" describes the review history rather than the change.