Skip to content

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

Open
raminqaf wants to merge 1 commit into
apache:masterfrom
raminqaf:FLINK-40241-raw-format
Open

[FLINK-40241][table-runtime] Support VARIANT as column type in the 'raw' format#29010
raminqaf wants to merge 1 commit into
apache:masterfrom
raminqaf:FLINK-40241-raw-format

Conversation

@raminqaf

Copy link
Copy Markdown
Contributor

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

flinkbot commented Aug 24, 2026

Copy link
Copy Markdown
Collaborator

CI report:

Bot commands The @flinkbot bot supports the following commands:
  • @flinkbot run azure re-run the last Azure build

…aw' format

The 'raw' format now accepts a single VARIANT column, treating the bytes as a JSON document. On read the bytes are parsed like PARSE_JSON with duplicate keys rejected; on write the value is rendered with Variant#toJson. The round trip is value-lossless but not byte-lossless: insignificant whitespace is dropped and object keys are ordered.

Reads decode with 'raw.charset'. For the default UTF-8 the bytes are parsed directly through a new BinaryVariantInternalBuilder#parseJson(byte[]) overload, which lets Jackson decode straight from the byte array and avoids an intermediate String. Other charsets are decoded to a String first so 'raw.charset' stays honored and read remains symmetric with write. 'raw.endianness' does not apply to VARIANT.

Combined with 'raw.line-delimiter' this reads and writes newline-delimited JSON, one row per line.
@raminqaf
raminqaf force-pushed the FLINK-40241-raw-format branch from 31f70e1 to 3b54633 Compare August 24, 2026 12:35
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants