[
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)
