raminqaf opened a new pull request, #29010:
URL: https://github.com/apache/flink/pull/29010
## 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?
- [X] Yes (Opus 4.8)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
