vbhanuchander-lang opened a new pull request, #28996:
URL: https://github.com/apache/flink/pull/28996
## What is the purpose of the change
Bytes-backed Avro decimal logical types were encoded incompatibly with
Flink's JVM serializer when a
`GenericRecord` crossed the JVM/Python boundary, silently corrupting the
record.
FLINK-37192 replaced `avro-python3` with `avro>=1.12.0`. That version routes
a bytes-backed decimal
through `BinaryEncoder.write_decimal_bytes` and
`BinaryDecoder.read_decimal_from_bytes`, both of
which size the payload with `write_long` / `read_long`. `FlinkAvroEncoder`
and `FlinkAvroDecoder`
override the primitive reads and writes for JVM compatibility but not those
two, and in this encoder
`write_long` is a fixed 8-byte long — so the length prefix was written as 8
bytes where the JVM
frames a bytes field with a 4-byte int.
For a record of `amount=Decimal("12.34"), tail=7`:
```
expected (JVM) : 0000000204d200000007
actual : 000000000000000204d200000007
```
A JVM `GenericDatumReader` consumes the oversized prefix as the whole bytes
field, so every following
field shifts: the decimal reads back empty and `tail` reads back as `2`,
leaving four bytes unread.
Nothing is raised, which is why this is silent corruption rather than a
decode failure. Flink's own
reader is wrong in the same direction, so a pure-Python round trip looks
correct and the defect only
appears across the boundary.
This does not affect standard Avro interoperability — `avro`, `fastavro` and
ordinary Kafka Avro
payloads are unchanged. It is specific to Flink's internal JVM-compatible
serializer.
## Brief change log
- `FlinkAvroEncoder.write_decimal_bytes` frames the unscaled payload as a
bytes field rather than
letting avro size it with an 8-byte long
- `FlinkAvroDecoder.read_decimal_from_bytes` reads that 4-byte size
symmetrically
- The payload itself is untouched: the same two's-complement big-endian
unscaled value avro already
produced, so only the length prefix moves
## Verifying this change
This change added tests and can be verified as follows:
- `pyflink/fn_execution/tests/test_avro_format.py`, four tests: the exact
encoded bytes for a
decimal followed by an int; that the record reads back field by field
with JVM framing; that the
unscaled payload is unchanged for `0`, `±12.34` and the `±1.28`/`-1.29`
byte-boundary cases; and
a round trip through `FlinkAvroDatumWriter`/`FlinkAvroDatumReader`
- Reverting the two overrides fails eight assertions across those tests
- Separately checked against avro 1.12.2 that the payload bytes stay
identical to
`avro.io.BinaryEncoder.write_decimal_bytes` across 23 values — positive,
negative, zero and the
±128, ±256, ±32768 boundaries — so only the framing differs
A note on how that was run: the module imports only `struct` and `avro`, and
I exercised the test
body with `PyFlinkTestCase` substituted by `unittest.TestCase` in a
virtualenv with avro 1.12.2. I
did not build a full PyFlink environment locally, so the assertions are
verified but the change has
not been through the project's own Python test harness on my machine. flake8
is clean under
`flink-python/tox.ini` (max-line-length 100).
## 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** — this is the JVM/Python boundary encoding for
Avro generic records.
It is a wire format used for in-flight data exchange rather than
persisted state, and the
previous encoding was already unreadable by the JVM, so the only case
that changes behaviour is
a decimal field that was silently corrupted before.
- The runtime per-record code paths (performance sensitive): **yes** — the
decimal path only. The
work is equivalent to what avro already did; one `int.to_bytes` replaces
a manual byte loop.
- 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? **no** — it corrects an
encoding defect, so there
is nothing to document.
--
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]
