[
https://issues.apache.org/jira/browse/FLINK-40370?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18103851#comment-18103851
]
Joey Pereira commented on FLINK-40370:
--------------------------------------
cc [~shuyichen] (or [~suez1224] - not sure which handle of yours is current) as
we've identified this correctness risk in our deployment, although we have not
seen it occurring in the wild.
> 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)
