h74zhou opened a new pull request, #28764:
URL: https://github.com/apache/flink/pull/28764
> **⚠️ Stacked on #28108** (`khaledh:handle-recursive-proto-fields`). #28108
adds the recursive-message **deserialize** support this builds on; it is not
yet merged, so this PR's diff includes that commit as its base. **Please review
only the top commit** (`[FLINK-34620][formats][protobuf] Add serialize-side
handling...`) and merge this **after** #28108.
## What is the purpose of the change
Completes recursive-protobuf support for the `flink-protobuf` format
(FLINK-34620). #28108 added the **deserialize** side — a message field whose
type is (transitively) recursive is resolved to `VARBINARY` via cycle
detection, and read back with `message.toByteArray()`. This PR adds the missing
**serialize** side so that a `RowToProtoConverter` can be constructed for any
proto whose descriptor transitively contains a recursive message, and fixes an
absent-field round-trip bug that only surfaces once both paths exist.
Without this, building a serializer for such a proto fails at construction
(job start), before any data:
```
PbCodegenException -> InvalidProgramException -> CompileException:
Assignment conversion not possible from "com.google.protobuf.ByteString" to
"<recursive message type>"
```
Root cause: `PbToRowTypeUtil` resolves a recursive field to `VARBINARY`. On
read, `PbCodegenDeserializeFactory` routes it to `PbCodegenBytesDeserializer`.
On write, `PbCodegenSerializeFactory` had no matching branch, so `VARBINARY`
fell through to `PbCodegenSimpleSerializer`, which emits
`ByteString.copyFrom(...)` into a message-typed variable — which does not
compile.
## Brief change log
- `PbCodegenBytesSerializer` (new) — parses the `VARBINARY` bytes back
into the message (`<MessageType>.parseFrom(bytes)`), the inverse of
`PbCodegenBytesDeserializer`.
- `PbCodegenSerializeFactory` — dispatches the `MESSAGE` + `VARBINARY`
case to the new serializer, mirroring the branch already used in
`PbCodegenDeserializeFactory`.
- `PbCodegenRowDeserializer` — a recursive field is `VARBINARY` (a "simple
type") but is still a `MESSAGE`; guard the proto3 primitive-default path with
`getJavaType() != MESSAGE` so an absent sub-message stays absent instead of
round-tripping back as a present (empty) one.
- Docs — note that a proto2 recursive message declaring `required` fields
cannot be serialized when an absent recursive field is read with
`read-default-values=true`.
## Verifying this change
This change added tests and can be verified as follows:
- `RecursiveMessageRowToProtoTest` (new) — serialize + round trips: proto3
nested recursive, null-field omission, proto2 round-trip, repeated recursive
fields (`ARRAY<VARBINARY>`), absent-field round trips asserting absence is
preserved.
- `RecursiveMessageProtoToRowTest` (extended) — proto3 unset recursive
field asserts `null` (`read-default-values=false`) / default bytes (`=true`),
mirroring the proto2 pair.
- `mvn -pl flink-formats/flink-protobuf test` → `Tests run: 61, Failures:
0, Errors: 0, Skipped: 0`; `checkstyle:check` and `spotless:check` clean.
## 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 serializers: **yes** (protobuf format codegen; serialize path for
recursive message fields)
- The runtime per-record code path (performance sensitive): **no**
- Anything that affects deployment or recovery: **no**
- The S3 file system connector: **no**
## Documentation
- Does this pull request introduce a new feature? **no** (completes an
existing in-progress feature)
- If yes, how is the feature documented? **docs** (protobuf format options
note updated)
--
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]
