Fixed default XCom deserialization. #14827
Conversation
suiting-young
requested review from
ashb,
kaxil,
turbaszek and
XD-DENG
as code owners
| @@ -62,13 +62,7 @@ def init_on_load(self): | |||
| Called by the ORM after the instance has been loaded from the DB or otherwise reconstituted | |||
| i.e automatically deserialize Xcom value when loading from DB. | |||
| """ | |||
| try: | |||
| self.value = self.orm_deserialize_value() | |||
| except (UnicodeEncodeError, ValueError): | |||
There was a problem hiding this comment.
Those exceptions came from Python2 json.loads, have been replaced by JSONDecodeError (as below) in Python3.
| log.error( | ||
| "Could not deserialize the XCom value from JSON. " | ||
| "If you are using pickles instead of JSON " | ||
| "for XCom, then you need to enable pickle " | ||
| "support for XCom in your airflow config." | ||
| ) |
There was a problem hiding this comment.
This message is meaningless here, as user can do nothing with values already existing in database.
airflow/models/xcom.py
Outdated
| # As 'enable_xcom_pickling' changed default from True to False, | ||
| # pickled (old) and JSON (new) are both existing. | ||
| pass | ||
| return pickle.loads(result.value) |
There was a problem hiding this comment.
So what happens here where some values in DB are JSON Serialized and enable_xcom_pickling is True.
The previous code accounted for that, this code does not
There was a problem hiding this comment.
Isn't it pickle deprecated (as in 1.10.x) any more?
I'm trying to fix the default path that user haven't specify enable_xcom_pickling in airflow.cfg.
That the value changed from True (implicit, and then old) to False (new) silently.
If pickle is still the first class option, I'd like to implement a bi-direction fallback.
if conf.getboolean('core', 'enable_xcom_pickling'):
try:
return pickle.loads(result.value)
except pickle.UnpicklingError:
return json.loads(result.value.decode('UTF-8'))
else:
try:
return json.loads(result.value.decode('UTF-8'))
except JSONDecodeError:
return pickle.loads(result.value)
|
The PR most likely needs to run full matrix of tests because it modifies parts of the core of Airflow. However, committers might decide to merge it quickly and take the risk. If they don't merge it quickly - please rebase it to the latest master at your convenience, or amend the last commit of the PR, and push it with --force-with-lease. |
github-actions
bot
added
the
full tests needed
suiting-young
force-pushed
the
xcom-pickle
branch
from
5bc5ff3
to
962bafd
Compare
suiting-young
force-pushed
the
xcom-pickle
branch
from
962bafd
to
6a0fe3b
Compare
|
Awesome work, congrats on your first merged pull request! |
This is a fix for migrating with the default
enable_xcom_pickling(before & after).I.e.
pickle-d (old) andJSON(new) data both existing in database.^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code change, Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in UPDATING.md.