[FLINK-37925][table] Define VARIANT cast rules so a cast never alters the stored value

[raminqaf]

What is the purpose of the change

Follow-up to the initial VARIANT to primitive cast support in FLINK-37925. It settles the cast semantics under one rule and turns the string cast into a real value cast.

A cast from a VARIANT succeeds only when the target holds the stored value without altering it. A value is never wrapped, rounded, truncated, or padded to make it fit; anything else fails CAST and returns NULL for TRY_CAST.

Stored kind	Succeeds for
integer kinds	any numeric target that holds the value, DECIMAL, FLOAT, DOUBLE
FLOAT, DOUBLE	FLOAT, DOUBLE
DECIMAL	DECIMAL without rounding, FLOAT, DOUBLE
BOOLEAN, DATE	the same type
TIMESTAMP	TIMESTAMP(p) that keeps the fractional seconds
TIMESTAMP_LTZ	TIMESTAMP_LTZ(p) that keeps the fractional seconds
BYTES	BINARY(n), VARBINARY(n)
any scalar	CHAR(n), VARCHAR(n), STRING
NULL	SQL NULL for any nullable target

The conditions in that table are:

• An integer target needs the value in range and without a fractional part. So PARSE_JSON('7.0') reaches INT as 7 while PARSE_JSON('7.2') does not, and CAST(PARSE_JSON('1000') AS TINYINT) fails instead of wrapping.
• A DECIMAL target has to fit the precision and the scale. Trailing zeros may be appended, so 42 reaches DECIMAL(5, 2) as 42.00, but a scale that would have to round is rejected.
• FLOAT and DOUBLE are the one exception to exactness. They are approximate by definition, so they accept every numeric kind and drop decimal digits, rejecting only a magnitude they cannot represent such as 1e40 to a FLOAT.
• CHAR(n) requires the exact length and VARCHAR(n) at most n. Nothing is padded or truncated.
  Reading one kind as another is never implicit, so a DECIMAL is not read as an integer and a TIMESTAMP is not read as a TIMESTAMP_LTZ. To reach such a type the inner cast names the stored kind and a regular cast around it does the conversion, which then applies the usual rules and may round, truncate, or overflow:

CAST(CAST(PARSE_JSON('1000') AS SMALLINT) AS TINYINT)     -- returns -24 (after overflow)                                                                                                                           
CAST(CAST(PARSE_JSON('3.9') AS DECIMAL(2, 1)) AS INT)     -- returns 3 (truncated)                                                                                                                                  

Casting to a character string

CAST(VARIANT AS CHAR/VARCHAR) now renders the value exactly as a regular SQL cast of the stored kind would, instead of reusing the JSON serialization. So a boolean becomes TRUE rather than true, a timestamp uses the SQL format 2021-09-24 12:34:56.123456 rather than the ISO form with a T, and a TIMESTAMP_LTZ is shifted into the session time zone rather than always printed as UTC. This is implemented by calling the same runtime formatters the native to-string rules call, so the two cannot drift apart, and the tests assert against the very constants the native cases use.

JSON_STRING remains the way to obtain the JSON text, where a string stays quoted as "foo" and an object or array is serialized. A variant holding an object, array, or binary value has no scalar rendering, so that cast fails and the error points to JSON_STRING. A variant storing a JSON null casts to SQL NULL rather than to the text null.

Kinds that PARSE_JSON cannot produce

The same rule applies to the kinds a variant can hold but PARSE_JSON never creates. A timestamp reaches a TIMESTAMP whose precision keeps its fractional seconds, since a variant always stores microseconds, and a binary value reaches BINARY(n) of the exact length or VARBINARY(n) at most n. TIMESTAMP and TIMESTAMP_LTZ are separate kinds and are not read as one another. These paths are only reachable through the programmatic VariantBuilder today.

TIME has no counterpart in the variant type model and is rejected during validation.

Brief change log

• Add VariantCastUtils in flink-table-runtime holding the per-target checks: integer range, DECIMAL precision and scale, floating-point magnitude, timestamp precision, and character or binary length.
• VariantToPrimitiveCastRule: route every target through those checks. This also fixes two cases that silently altered the value, namely a timestamp keeping more precision than its declared type and a binary
  value being padded or truncated to the target length.
• VariantToStringCastRule: extract the scalar value instead of serializing to JSON, handle CHAR and VARCHAR directly to enforce the length, and map a null-valued variant to SQL NULL for a nullable target.
• LogicalTypeCasts: express VARIANT cast validation in the per-target rules and drop the JSON_STRING cast hint.

Verifying this change

This change added and updated tests:

• CastRulesTest: per-kind coverage of every numeric target including range rejection, DECIMAL precision and scale fitting, the FLOAT and DOUBLE acceptance of any numeric kind with overflow rejection, the timestamp precision rule, and the TIMESTAMP versus TIMESTAMP_LTZ split.
• CastFunctionITCase: the same through PARSE_JSON, plus string value extraction, strict CHAR and VARCHAR length, rejection of object and array values, a JSON null casting to SQL NULL, and the two nested casts above. Each failing case also asserts that TRY_CAST returns NULL.
• LogicalTypeCastsTest: the VARIANT cast support matrix.

Trade-offs

Rejecting a cast that would change the value is stricter than Snowflake and Spark, which coerce. For the kind-preserving part it matches Snowflake's AS_INTEGER and AS_DOUBLE accessor family, where a mismatch yields NULL. The reasoning is that a VARIANT cast is an extraction step, so any conversion decision belongs in an explicit second cast, and TRY_CAST provides the non-failing form.

Because the stored kind is only known per row, these are runtime failures and cannot be reported during validation.

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): yes, javadoc-only change to the @PublicEvolving Variant interface, no signature change
• The serializers: no
• The runtime per-record code paths (performance sensitive): yes, only on the VARIANT cast path
• Anything that affects deployment or recovery: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no, it refines existing FLINK-37925 behavior
• If yes, how is the feature documented? docs and JavaDocs

[flinkbot]

CI report:

• f9bd3a0 UNKNOWN
• c110934 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[twalthr]

Thank you @raminqaf. I left some feedback.

[twalthr]

Another round of comments.

[twalthr]

Thank you @raminqaf. We are almost there. But I had to post some more critical comments.

[twalthr]

LGTM. Thank you @raminqaf for addressing all the feedback comments!

[twalthr]

LGTM