dianfu commented on code in PR #28996:
URL: https://github.com/apache/flink/pull/28996#discussion_r3921412499
##########
flink-python/pyflink/fn_execution/formats/avro.py:
##########
@@ -87,6 +87,13 @@ def read_bytes(self):
assert (nbytes >= 0), nbytes
return self.read(nbytes)
+ def read_decimal_from_bytes(self, precision, scale):
+ # avro's implementation sizes the payload with read_long, which is an
Avro zig-zag long
+ # upstream but a fixed 8-byte long here. On the JVM a bytes-backed
decimal is framed like
+ # any other bytes field, so the size is a 4-byte int.
+ size = self.read_int()
Review Comment:
Would it make sense to keep backward compabitility in case
GenericRecordAvroTypeInfo is used for Python state?
We should be able to distinguish the two representations as following: a
positive first int is the corrected 4-byte length, while zero indicates the
high half of the historical fixed-width long.
For example:
```python
def read_decimal_from_bytes(self, precision, scale):
size = self.read_int()
if size == 0:
# avro>=1.12 previously used FlinkAvroDecoder.read_long()
# for this length, producing an 8-byte fixed-width prefix.
size = self.read_int()
return self.read_decimal_from_fixed(precision, scale, size)
```
--
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]
