You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Joey Pereira (Jira)" <ji...@apache.org> on 2026/08/11 22:04:00 UTC
[jira] [Updated] (FLINK-40370) Bytes-backed decimal logical types corrupt PyFlink GenericRecordAvroTypeInfo JVM/Python serialization
[ https://issues.apache.org/jira/browse/FLINK-40370?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Joey Pereira updated FLINK-40370:
---------------------------------
Description:
After FLINK-37192 replaced {{avro-python3}} with {{avro>=1.12.0}}, PyFlink began encoding bytes-backed Avro decimal logical types incompatibly with Flink’s internal Java serializer when generic records cross the JVM/Python boundary.
This does *not* affect standard Avro interoperability. Standard Python Avro, Java Avro, {{fastavro}}, and ordinary Kafka Avro payloads remain compatible. The regression affects Flink’s separate internal serializer for {{GenericRecordAvroTypeInfo}}.
For a record containing {{amount = Decimal("12.34")}} followed by {{tail = 7}}:
||Serialization boundary||Encoded bytes||
|Standard Python Avro and {{fastavro}}|{{0404d20e}}|
|Flink internal Java serializer|{{0000000204d200000007}}|
|Flink internal Python serializer with modern Avro|{{000000000000000204d200000007}}|
The actual Java {{GenericDatumReader}} successfully decodes the Python-generated bytes as a record with an empty decimal payload and {{tail = 2}}, leaving unread bytes. This is *silent data corruption*, not merely a decoding failure.
h3. Reproduction
{code:python}
from decimal import Decimal
from io import BytesIO
import avro.io
import avro.schema
import fastavro
from pyflink.fn_execution.formats.avro import (
FlinkAvroDatumReader,
FlinkAvroDatumWriter,
FlinkAvroDecoder,
FlinkAvroEncoder,
)
schema_json = """
{
"type": "record",
"name": "DecimalRecord",
"fields": [
{
"name": "amount",
"type": {
"type": "bytes",
"logicalType": "decimal",
"precision": 8,
"scale": 2
}
},
{"name": "tail", "type": "int"}
]
}
"""
schema = avro.schema.parse(schema_json)
record = {"amount": Decimal("12.34"), "tail": 7}
standard = BytesIO()
avro.io.DatumWriter(schema).write(
record,
avro.io.BinaryEncoder(standard),
)
fast = BytesIO()
fastavro.schemaless_writer(fast, schema.to_json(), record)
python_flink = BytesIO()
FlinkAvroDatumWriter(schema).write(
record,
FlinkAvroEncoder(python_flink),
)
java_flink = BytesIO()
java_compatible_encoder = FlinkAvroEncoder(java_flink)
java_compatible_encoder.write_bytes(bytes.fromhex("04d2"))
java_compatible_encoder.write_int(7)
print("Standard Python Avro:", standard.getvalue().hex())
print("Standard fastavro:", fast.getvalue().hex())
print("Expected Flink JVM:", java_flink.getvalue().hex())
print("Actual PyFlink:", python_flink.getvalue().hex())
try:
result = FlinkAvroDatumReader(schema, schema).read(
FlinkAvroDecoder(BytesIO(java_flink.getvalue()))
)
print("Python reads JVM:", result)
except Exception as exc:
print("Python reads JVM:", type(exc).__name__, str(exc))
stream = BytesIO(python_flink.getvalue())
decoder = FlinkAvroDecoder(stream)
amount = decoder.read_bytes()
tail = decoder.read_int()
print(
"JVM-compatible read:",
{
"amount_length": len(amount),
"tail": tail,
"unread": stream.read().hex(),
},
)
{code}
Observed using Python {{avro==1.12.1}} and {{fastavro==1.10.0}}:
{code}
Standard Python Avro: 0404d20e
Standard fastavro: 0404d20e
Expected Flink JVM: 0000000204d200000007
Actual PyFlink: 000000000000000204d200000007
Python reads JVM: InvalidAvroBinaryEncoding Read 2 bytes, expected 8670806016 bytes
JVM-compatible read: {
"amount_length": 0,
"tail": 2,
"unread": "04d200000007"
}
{code}
Additionally, compiling and running Flink’s actual Java {{DataOutputEncoder}}, {{DataInputDecoder}}, and Avro {{GenericDatumReader}} produced:
{code}
actual Java DataOutputEncoder:
0000000204d200000007
actual Java DataInputDecoder:
amount_length=0, tail=2, unread=04d200000007
actual Java GenericDatumReader:
amount_length=0, tail=2, unread=04d200000007
{code}
h3. Expected behavior
Flink’s Python and Java internal generic-record serializers agree on the four-byte byte-array length prefix. Decimal-containing records round-trip without corrupting subsequent fields.
h3. Root cause
Modern Python Avro’s decimal-specialized writer calls {{write_long}}, and its decimal-specialized reader calls {{read_long}}.
PyFlink overrides those methods using *eight-byte* fixed-width operations. However, Flink’s Java {{DataOutputEncoder.writeBytes}} and {{DataInputDecoder.readBytes}} use *four-byte* lengths.
Consequently:
* Python to Java can silently corrupt the decimal and subsequent fields.
* Java to Python fails decoding.
* Python to Python succeeds, so symmetric Python-only tests miss the incompatibility.
h3. Affected workflows
* {{GenericRecordAvroTypeInfo}} with bytes-backed decimal fields.
* {{AvroBulkWriters.for_generic_record(...)}}.
* {{AvroParquetWriters.for_generic_record(...)}}.
* Generic Avro readers crossing from Java into Python.
* Explicitly Avro-typed Python state.
Ordinary Kafka Avro payloads, {{fastavro}}, standard Java Avro, plain bytes fields, and unrelated SQL decimal coders are unaffected.
h3. Suggested fix
Ensure PyFlink’s decimal-specialized encoder and decoder use Flink’s four-byte internal byte-array framing.
Because affected Flink versions may already have checkpointed the eight-byte representation in explicitly Avro-typed Python state, the decoder should consider accepting both historical eight-byte and corrected four-byte representations, while new writes always use four-byte framing.
Add coverage for bidirectional JVM/Python compatibility, adjacent-field corruption, negative and high-precision decimals, nested and nullable records, Avro/Parquet sinks, both Python/Cython coder implementations, and previously persisted Python state.
h3. References
* [FLINK-37192: Dependency change introducing the regression|https://issues.apache.org/jira/browse/FLINK-37192]
* [PyFlink custom decimal-related encoder|https://github.com/apache/flink/blob/release-2.3.0/flink-python/pyflink/fn_execution/formats/avro.py#L174-L199]
* [PyFlink custom decoder|https://github.com/apache/flink/blob/release-2.3.0/flink-python/pyflink/fn_execution/formats/avro.py#L50-L88]
* [Java internal encoder|https://github.com/apache/flink/blob/release-2.3.0/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/utils/DataOutputEncoder.java#L86-L101]
* [Java internal decoder|https://github.com/apache/flink/blob/release-2.3.0/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/utils/DataInputDecoder.java#L85-L97]
* [Apache Avro decimal writer|https://github.com/apache/avro/blob/release-1.12.1/lang/py/avro/io.py#L468-L493]
* [Avro generic-record file sink|https://github.com/apache/flink/blob/release-2.3.0/flink-python/pyflink/datastream/formats/avro.py#L132-L166]
* [Parquet generic-record file sink|https://github.com/apache/flink/blob/release-2.3.0/flink-python/pyflink/datastream/formats/parquet.py#L76-L110]
was:
After FLINK-37192 replaced `avro-python3` with `avro>=1.12.0`, PyFlink began encoding bytes-backed Avro decimal logical types incompatibly with Flink’s internal Java serializer when generic records cross the JVM/Python boundary.
This does **not** affect standard Avro interoperability. Standard Python Avro, Java Avro, `fastavro`, and ordinary Kafka Avro payloads remain compatible. The regression affects Flink’s separate internal serializer for `GenericRecordAvroTypeInfo`.
For a record containing `amount = Decimal("12.34")` followed by `tail = 7`:
| Serialization boundary | Encoded bytes |
|---|---|
| Standard Python Avro and `fastavro` | `0404d20e` |
| Flink internal Java serializer | `0000000204d200000007` |
| Flink internal Python serializer with modern Avro | `000000000000000204d200000007` |
The actual Java `GenericDatumReader` successfully decodes the Python-generated bytes as a record with an empty decimal payload and `tail = 2`, leaving unread bytes. This is **silent data corruption**, not merely a decoding failure.
### Reproduction
```python
from decimal import Decimal
from io import BytesIO
import avro.io
import avro.schema
import fastavro
from pyflink.fn_execution.formats.avro import (
FlinkAvroDatumReader,
FlinkAvroDatumWriter,
FlinkAvroDecoder,
FlinkAvroEncoder,
)
schema_json = """
{
"type": "record",
"name": "DecimalRecord",
"fields": [
{
"name": "amount",
"type": {
"type": "bytes",
"logicalType": "decimal",
"precision": 8,
"scale": 2
}
},
{"name": "tail", "type": "int"}
]
}
"""
schema = avro.schema.parse(schema_json)
record = {"amount": Decimal("12.34"), "tail": 7}
standard = BytesIO()
avro.io.DatumWriter(schema).write(
record,
avro.io.BinaryEncoder(standard),
)
fast = BytesIO()
fastavro.schemaless_writer(fast, schema.to_json(), record)
python_flink = BytesIO()
FlinkAvroDatumWriter(schema).write(
record,
FlinkAvroEncoder(python_flink),
)
java_flink = BytesIO()
java_compatible_encoder = FlinkAvroEncoder(java_flink)
java_compatible_encoder.write_bytes(bytes.fromhex("04d2"))
java_compatible_encoder.write_int(7)
print("Standard Python Avro:", standard.getvalue().hex())
print("Standard fastavro:", fast.getvalue().hex())
print("Expected Flink JVM:", java_flink.getvalue().hex())
print("Actual PyFlink:", python_flink.getvalue().hex())
try:
result = FlinkAvroDatumReader(schema, schema).read(
FlinkAvroDecoder(BytesIO(java_flink.getvalue()))
)
print("Python reads JVM:", result)
except Exception as exc:
print("Python reads JVM:", type(exc).__name__, str(exc))
stream = BytesIO(python_flink.getvalue())
decoder = FlinkAvroDecoder(stream)
amount = decoder.read_bytes()
tail = decoder.read_int()
print(
"JVM-compatible read:",
{
"amount_length": len(amount),
"tail": tail,
"unread": stream.read().hex(),
},
)
```
Observed using Python `avro==1.12.1` and `fastavro==1.10.0`:
```text
Standard Python Avro: 0404d20e
Standard fastavro: 0404d20e
Expected Flink JVM: 0000000204d200000007
Actual PyFlink: 000000000000000204d200000007
Python reads JVM: InvalidAvroBinaryEncoding Read 2 bytes, expected 8670806016 bytes
JVM-compatible read: {
"amount_length": 0,
"tail": 2,
"unread": "04d200000007"
}
```
Additionally, compiling and running Flink’s actual Java `DataOutputEncoder`, `DataInputDecoder`, and Avro `GenericDatumReader` produced:
```text
actual Java DataOutputEncoder:
0000000204d200000007
actual Java DataInputDecoder:
amount_length=0, tail=2, unread=04d200000007
actual Java GenericDatumReader:
amount_length=0, tail=2, unread=04d200000007
```
### Expected behavior
Flink’s Python and Java internal generic-record serializers agree on the four-byte byte-array length prefix. Decimal-containing records round-trip without corrupting subsequent fields.
### Root cause
Modern Python Avro’s decimal-specialized writer calls `write_long`, and its decimal-specialized reader calls `read_long`.
PyFlink overrides those methods using **eight-byte** fixed-width operations. However, Flink’s Java `DataOutputEncoder.writeBytes` and `DataInputDecoder.readBytes` use **four-byte** lengths.
Consequently:
- Python to Java can silently corrupt the decimal and subsequent fields.
- Java to Python fails decoding.
- Python to Python succeeds, so symmetric Python-only tests miss the incompatibility.
### Affected workflows
- `GenericRecordAvroTypeInfo` with bytes-backed decimal fields.
- `AvroBulkWriters.for_generic_record(...)`.
- `AvroParquetWriters.for_generic_record(...)`.
- Generic Avro readers crossing from Java into Python.
- Explicitly Avro-typed Python state.
Ordinary Kafka Avro payloads, `fastavro`, standard Java Avro, plain bytes fields, and unrelated SQL decimal coders are unaffected.
### Suggested fix
Ensure PyFlink’s decimal-specialized encoder and decoder use Flink’s four-byte internal byte-array framing.
Because affected Flink versions may already have checkpointed the eight-byte representation in explicitly Avro-typed Python state, the decoder should consider accepting both historical eight-byte and corrected four-byte representations, while new writes always use four-byte framing.
Add coverage for bidirectional JVM/Python compatibility, adjacent-field corruption, negative and high-precision decimals, nested and nullable records, Avro/Parquet sinks, both Python/Cython coder implementations, and previously persisted Python state.
### References
- [FLINK-37192: Dependency change introducing the regression](https://issues.apache.org/jira/browse/FLINK-37192)
- [PyFlink custom decimal-related encoder](https://github.com/apache/flink/blob/release-2.3.0/flink-python/pyflink/fn_execution/formats/avro.py#L174-L199)
- [PyFlink custom decoder](https://github.com/apache/flink/blob/release-2.3.0/flink-python/pyflink/fn_execution/formats/avro.py#L50-L88)
- [Java internal encoder](https://github.com/apache/flink/blob/release-2.3.0/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/utils/DataOutputEncoder.java#L86-L101)
- [Java internal decoder](https://github.com/apache/flink/blob/release-2.3.0/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/utils/DataInputDecoder.java#L85-L97)
- [Apache Avro decimal writer](https://github.com/apache/avro/blob/release-1.12.1/lang/py/avro/io.py#L468-L493)
- [Avro generic-record file sink](https://github.com/apache/flink/blob/release-2.3.0/flink-python/pyflink/datastream/formats/avro.py#L132-L166)
- [Parquet generic-record file sink](https://github.com/apache/flink/blob/release-2.3.0/flink-python/pyflink/datastream/formats/parquet.py#L76-L110)
> Bytes-backed decimal logical types corrupt PyFlink GenericRecordAvroTypeInfo JVM/Python serialization
> -----------------------------------------------------------------------------------------------------
>
> Key: FLINK-40370
> URL: https://issues.apache.org/jira/browse/FLINK-40370
> Project: Flink
> Issue Type: Bug
> Components: API / Python, Formats (JSON, Avro, Parquet, ORC, SequenceFile)
> Affects Versions: 2.0.2, 2.3.0, 2.2.1, 2.1.3
> Reporter: Joey Pereira
> Priority: Major
>
> After FLINK-37192 replaced {{avro-python3}} with {{avro>=1.12.0}}, PyFlink began encoding bytes-backed Avro decimal logical types incompatibly with Flink’s internal Java serializer when generic records cross the JVM/Python boundary.
> This does *not* affect standard Avro interoperability. Standard Python Avro, Java Avro, {{fastavro}}, and ordinary Kafka Avro payloads remain compatible. The regression affects Flink’s separate internal serializer for {{GenericRecordAvroTypeInfo}}.
> For a record containing {{amount = Decimal("12.34")}} followed by {{tail = 7}}:
> ||Serialization boundary||Encoded bytes||
> |Standard Python Avro and {{fastavro}}|{{0404d20e}}|
> |Flink internal Java serializer|{{0000000204d200000007}}|
> |Flink internal Python serializer with modern Avro|{{000000000000000204d200000007}}|
> The actual Java {{GenericDatumReader}} successfully decodes the Python-generated bytes as a record with an empty decimal payload and {{tail = 2}}, leaving unread bytes. This is *silent data corruption*, not merely a decoding failure.
> h3. Reproduction
> {code:python}
> from decimal import Decimal
> from io import BytesIO
> import avro.io
> import avro.schema
> import fastavro
> from pyflink.fn_execution.formats.avro import (
> FlinkAvroDatumReader,
> FlinkAvroDatumWriter,
> FlinkAvroDecoder,
> FlinkAvroEncoder,
> )
> schema_json = """
> {
> "type": "record",
> "name": "DecimalRecord",
> "fields": [
> {
> "name": "amount",
> "type": {
> "type": "bytes",
> "logicalType": "decimal",
> "precision": 8,
> "scale": 2
> }
> },
> {"name": "tail", "type": "int"}
> ]
> }
> """
> schema = avro.schema.parse(schema_json)
> record = {"amount": Decimal("12.34"), "tail": 7}
> standard = BytesIO()
> avro.io.DatumWriter(schema).write(
> record,
> avro.io.BinaryEncoder(standard),
> )
> fast = BytesIO()
> fastavro.schemaless_writer(fast, schema.to_json(), record)
> python_flink = BytesIO()
> FlinkAvroDatumWriter(schema).write(
> record,
> FlinkAvroEncoder(python_flink),
> )
> java_flink = BytesIO()
> java_compatible_encoder = FlinkAvroEncoder(java_flink)
> java_compatible_encoder.write_bytes(bytes.fromhex("04d2"))
> java_compatible_encoder.write_int(7)
> print("Standard Python Avro:", standard.getvalue().hex())
> print("Standard fastavro:", fast.getvalue().hex())
> print("Expected Flink JVM:", java_flink.getvalue().hex())
> print("Actual PyFlink:", python_flink.getvalue().hex())
> try:
> result = FlinkAvroDatumReader(schema, schema).read(
> FlinkAvroDecoder(BytesIO(java_flink.getvalue()))
> )
> print("Python reads JVM:", result)
> except Exception as exc:
> print("Python reads JVM:", type(exc).__name__, str(exc))
> stream = BytesIO(python_flink.getvalue())
> decoder = FlinkAvroDecoder(stream)
> amount = decoder.read_bytes()
> tail = decoder.read_int()
> print(
> "JVM-compatible read:",
> {
> "amount_length": len(amount),
> "tail": tail,
> "unread": stream.read().hex(),
> },
> )
> {code}
> Observed using Python {{avro==1.12.1}} and {{fastavro==1.10.0}}:
> {code}
> Standard Python Avro: 0404d20e
> Standard fastavro: 0404d20e
> Expected Flink JVM: 0000000204d200000007
> Actual PyFlink: 000000000000000204d200000007
> Python reads JVM: InvalidAvroBinaryEncoding Read 2 bytes, expected 8670806016 bytes
> JVM-compatible read: {
> "amount_length": 0,
> "tail": 2,
> "unread": "04d200000007"
> }
> {code}
> Additionally, compiling and running Flink’s actual Java {{DataOutputEncoder}}, {{DataInputDecoder}}, and Avro {{GenericDatumReader}} produced:
> {code}
> actual Java DataOutputEncoder:
> 0000000204d200000007
> actual Java DataInputDecoder:
> amount_length=0, tail=2, unread=04d200000007
> actual Java GenericDatumReader:
> amount_length=0, tail=2, unread=04d200000007
> {code}
> h3. Expected behavior
> Flink’s Python and Java internal generic-record serializers agree on the four-byte byte-array length prefix. Decimal-containing records round-trip without corrupting subsequent fields.
> h3. Root cause
> Modern Python Avro’s decimal-specialized writer calls {{write_long}}, and its decimal-specialized reader calls {{read_long}}.
> PyFlink overrides those methods using *eight-byte* fixed-width operations. However, Flink’s Java {{DataOutputEncoder.writeBytes}} and {{DataInputDecoder.readBytes}} use *four-byte* lengths.
> Consequently:
> * Python to Java can silently corrupt the decimal and subsequent fields.
> * Java to Python fails decoding.
> * Python to Python succeeds, so symmetric Python-only tests miss the incompatibility.
> h3. Affected workflows
> * {{GenericRecordAvroTypeInfo}} with bytes-backed decimal fields.
> * {{AvroBulkWriters.for_generic_record(...)}}.
> * {{AvroParquetWriters.for_generic_record(...)}}.
> * Generic Avro readers crossing from Java into Python.
> * Explicitly Avro-typed Python state.
> Ordinary Kafka Avro payloads, {{fastavro}}, standard Java Avro, plain bytes fields, and unrelated SQL decimal coders are unaffected.
> h3. Suggested fix
> Ensure PyFlink’s decimal-specialized encoder and decoder use Flink’s four-byte internal byte-array framing.
> Because affected Flink versions may already have checkpointed the eight-byte representation in explicitly Avro-typed Python state, the decoder should consider accepting both historical eight-byte and corrected four-byte representations, while new writes always use four-byte framing.
> Add coverage for bidirectional JVM/Python compatibility, adjacent-field corruption, negative and high-precision decimals, nested and nullable records, Avro/Parquet sinks, both Python/Cython coder implementations, and previously persisted Python state.
> h3. References
> * [FLINK-37192: Dependency change introducing the regression|https://issues.apache.org/jira/browse/FLINK-37192]
> * [PyFlink custom decimal-related encoder|https://github.com/apache/flink/blob/release-2.3.0/flink-python/pyflink/fn_execution/formats/avro.py#L174-L199]
> * [PyFlink custom decoder|https://github.com/apache/flink/blob/release-2.3.0/flink-python/pyflink/fn_execution/formats/avro.py#L50-L88]
> * [Java internal encoder|https://github.com/apache/flink/blob/release-2.3.0/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/utils/DataOutputEncoder.java#L86-L101]
> * [Java internal decoder|https://github.com/apache/flink/blob/release-2.3.0/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/utils/DataInputDecoder.java#L85-L97]
> * [Apache Avro decimal writer|https://github.com/apache/avro/blob/release-1.12.1/lang/py/avro/io.py#L468-L493]
> * [Avro generic-record file sink|https://github.com/apache/flink/blob/release-2.3.0/flink-python/pyflink/datastream/formats/avro.py#L132-L166]
> * [Parquet generic-record file sink|https://github.com/apache/flink/blob/release-2.3.0/flink-python/pyflink/datastream/formats/parquet.py#L76-L110]
--
This message was sent by Atlassian Jira
(v8.20.10#820010)