[FLINK-40241][table-runtime] Support VARIANT as column type in the 'raw' format

[raminqaf]

What is the purpose of the change

The raw format is for reading and writing schemaless topics and files, and VARIANT is Flink's type for semi-structured data, but the two could not be combined: CREATE TABLE t (payload VARIANT) WITH ('format' = 'raw', ...) failed at plan time. Users had to declare the column as STRING and call PARSE_JSON in every query on read, and flatten with JSON_STRING on write, which misrepresents the column type in catalogs and downstream consumers.

This change makes the raw format accept a single VARIANT column, treating the bytes as a JSON document. The change is purely additive; no existing type behavior changes.

Brief change log

• Add LogicalTypeRoot.VARIANT to RawFormatFactory's supported types.
• Deserialize by parsing the decoded bytes into a Variant, matching PARSE_JSON defaults (duplicate object keys rejected, malformed JSON fails the job). For the default UTF-8 the bytes are parsed directly via a new @Internal BinaryVariantInternalBuilder#parseJson(byte[]) overload, letting Jackson decode straight from the byte array with no intermediate String. Other charsets are decoded to a String with raw.charset first, so the option stays honored and read is symmetric with write.
• Serialize by rendering the Variant with Variant#toJson, encoded with raw.charset.
• Document the new type mapping on the raw format page (English and Chinese placeholder).

Verifying this change

This change added tests and can be verified as follows:

• RawFormatFactoryTest: a VARIANT column is accepted and produces the expected schemas.
• RawFormatSerDeSchemaTest: value-lossless round trips for objects, arrays, and JSON scalars; SQL NULL; a UTF-16 round trip (exercises the decode-then-parse branch); malformed JSON, duplicate keys, and empty message all raise DeserializationException.
• RawFormatLineDelimiterTest: newline-delimited JSON round trips one row per line when combined with raw.line-delimiter.

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 new builder method is @Internal)
• The serializers: no
• The runtime per-record code paths (performance sensitive): yes (adds a per-record converter for VARIANT columns on the raw read/write path)
• 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 (raw format connector page)

Notes for reviewers

• The round trip is value-lossless, not byte-lossless: Variant#toJson drops insignificant whitespace and orders object keys. raw.endianness does not apply to VARIANT.
• The write path is still Variant#toJson().getBytes(charset). A byte-direct emitter would avoid the intermediate String, but it needs a new flink-core Variant API with a byte-level JSON escaper, so it is deferred as a follow-up. The read-side byte path was taken here because createParser(byte[]) already exists and is algorithmically faster, not just one fewer allocation.
• FLINK-40218 (build a Variant from a token source) is complementary but targets callers that have already tokenized their JSON; it does not help this bytes-in path or the write side, so it is out of scope here.

---

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

• Yes (Opus 4.8)

[flinkbot]

CI report:

• 6ebb456 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

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

[dalelane]

I'm curious why you're not using the charset in both paths.

If the user configures anything that isn't UTF-8, we'd respect and enforce what they declared.

If the user configures it to use UTF-8, we give the data to Jackson which IIUC will do it's own pattern-based encoding detection. So if Jackson doesn't think the data looks like UTF-8, the UTF-8 config would be ignored.

Is that intentional? (Am I mis-reading the intent of the config?)

[raminqaf]

Good catch, and no, that wasn't intentional. You read it right.

The byte path handed raw bytes to Jackson, whose createParser(byte[]) runs its own RFC 4627 encoding auto-detection. I confirmed it parses UTF-16 bytes even when raw.charset is UTF-8, so the declared charset was not enforced on that path, while every non-UTF-8 charset was.

Fixed by always decoding with the declared charset before parsing. UTF-8 is now enforced exactly like every other charset, and read is symmetric with write.

I also dropped the now-unused parseJson(byte[]) overload, so this no longer touches flink-core.

[dalelane]

Do you think we need to update the catch here to cover VariantTypeException that would be possible now?

[raminqaf]

Good point. VariantTypeException extends RuntimeException, so it wasn't caught here. Added

[raminqaf]

@flinkbot run azure