[FLINK-40491][core] Extend BinaryVariant with TIME, TIMESTAMP_NS, TIMESTAMP_LTZ_NS primitives

[manner]

What is the purpose of the change

The variant binary encoding spec (see Variant Encoding) defines primitive type codes 17-20 for TIME, TIMESTAMP_LTZ_NS, TIMESTAMP_NS, and UUID that are currently missing in Flink.
This PR adds support for codes 17-19 (TIME, TIMESTAMP_LTZ_NS, TIMESTAMP_NS) to BinaryVariant, so that a LocalTime and a nanosecond-precision Instant/LocalDateTime can be represented in a Variant without lossy truncation to microseconds.
UUID (code 20) is intentionally out of scope here and will be added in a separate ticket.

Brief change log

• Added BinaryVariantUtil constants for primitive codes 17-19, extended getType()/valueSize()/getLong() to handle them, and added a TIME_FORMATTER for JSON rendering
• Added Variant.Type.TIME/TIMESTAMP_NS/TIMESTAMP_LTZ_NS and the corresponding getTime()/getDateTimeNanos()/getInstantNanos() accessors to the Variant interface
• Added VariantBuilder.of(LocalTime); made of(Instant)/of(LocalDateTime) precision-aware so a value with no sub-microsecond component keeps using the existing compact micros encoding, and only switches to the new nanosecond encoding when the value actually needs it
• Implemented appendTime/appendTimestampNanos/appendTimestampLtzNanos in BinaryVariantInternalBuilder, and the matching read/get()/toJson() support in BinaryVariant

Verifying this change

This change added tests and can be verified as follows:

• Round-trip and generic get() dispatch tests for the new types (BinaryVariantTest#testScalarVariant)
• Dedicated precision-dispatch tests verifying Instant/LocalDateTime pick the existing TIMESTAMP_LTZ/TIMESTAMP encoding for microsecond-aligned values and the new TIMESTAMP_LTZ_NS/TIMESTAMP_NS encoding otherwise, including that the mismatched accessor throws VariantTypeException (BinaryVariantTest#testNanosecondPrecisionVariant)
• A test verifying that LocalTime silently truncates below microsecond precision, since TIME has no nanosecond-precision counterpart in the variant spec (BinaryVariantTest#testTimeSubMicrosecondTruncation)
• JSON rendering assertions for TIME/TIMESTAMP_NS/TIMESTAMP_LTZ_NS (BinaryVariantTest#testToJsonScalar)

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 (Variant and VariantBuilder are @PublicEvolving; this adds new enum constants and new interface methods)
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• 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? not documented (documentation is tracked separately in FLINK-40494)

---

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

• Yes (please specify the tool below)

Generated-by: Claude Sonnet 5 (Claude Code)

[flinkbot]

CI report:

• c7dc74a Azure: PENDING
• 5eea899 Azure: PENDING

Bot commands

The @flinkbot bot supports the following commands:

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

[raminqaf]

Thanks for the PR @manner! Left two comments

[raminqaf]

Try this case

        LocalDateTime nanoLocalDateTime2 = LocalDateTime.of(2300, 1, 1, 0, 0, 0, 1);
        Variant dateTimeVariant2 = builder.of(nanoLocalDateTime2);

This will overflow. We should handle the error better

[manner]

Good catch! I'll add proper handling

[manner]

Actually the same error will currently happen when using a microsecond timestamp such as
LocalDateTime nanoLocalDateTime = LocalDateTime.of(2300, 1, 1, 0, 0, 0, 0);
Instant.until()s function for calculating microseconds between two instants just uses the function for calculating nanoseconds between two instants and then divides the result by 1000. This then also results in an overflow.

So this was a pre-existing bug, that only surfaced now.

[snuyanzin]

is it TIME with micro/nanos or not?

[manner]

TIME is using micros:

https://parquet.apache.org/docs/file-format/types/variantencoding/

[snuyanzin]

why do we think 8 bytes is enough here?

[manner]

The Variant specification states 8 bytes for nanosecond timestamps as well. The downside for using the same 8 bytes for the higher precision values is the smaller range of timestamps that can be used (+/- 292 years around unix epoch)

https://parquet.apache.org/docs/file-format/types/variantencoding/

[snuyanzin]

ok, thanks for clarification
I tend to think we need to explicitly mention such limitation in docs

[snuyanzin]

and by the way what will happen with timestamp_NS -300 years?
will it fail (user friendly message?)
or produce some wrong result?

[manner]

You're right, this should be properly documented. I think documentation will mostly happen in this ticket:
https://issues.apache.org/jira/browse/FLINK-40494

When converting goes wrong it will now throw a VariantTypeException with a helpful error message:
https://github.com/apache/flink/pull/29050/changes#diff-2a1f60d1bf5ca4585c1c2d08c51866702d17d70306ee16ce096b238377b82770R137-R148

There's a test for this here:
https://github.com/apache/flink/pull/29050/changes#diff-662023392a0949caa3a0814536613caf4fb2a17c1e6f3d7217e829bee0d50390R142-R154

[raminqaf]

LGTM @manner! Thanks for the contribution! 🚀

[snuyanzin]

after looking one more time the thing I didn't get
why don't we update VariantCastUtils with newly added types?

I guess we should, or?

flink/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java

Lines 294 to 312 in 878844f

	case TIMESTAMP:
	// A wall-clock value needs no zone shift, which is what UTC_ZONE achieves here. A
	// variant keeps microseconds, so the precision is always 6.
	value =
	DateTimeUtils.formatTimestamp(
	TimestampData.fromLocalDateTime(variant.getDateTime()),
	DateTimeUtils.UTC_ZONE,
	TIMESTAMP_PRECISION);
	break;
	case TIMESTAMP_LTZ:
	value =
	DateTimeUtils.formatTimestamp(
	TimestampData.fromInstant(variant.getInstant()),
	sessionZone,
	TIMESTAMP_PRECISION);
	break;
	case NULL:
	// Only reachable for a NOT NULL target. A nullable target maps a null-valued
	// variant to SQL NULL before this method is called.