Use Serialization to support more objects in XCom

[kaxil]

closes #8703

This way we will support Set, Tuple (and others like DAG,
BaseOperator, relativedelta, datetime) in XCom in serialization and
de-serialization.

---

^ 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.

[potiuk]

LGTM. But I wonder whether it will support old serialized data with full compatibility? Maybe worth to add a test where we serialize complex-ish structure with json and de-serialize with BaseSerializer (unless it is already done?).

[github-actions]

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 main at your convenience, or amend the last commit of the PR, and push it with --force-with-lease.

[uranusjr]

Maybe we should add a BaseSerialization.to_json() method for symmetry

[ashb]

It already exists.

[ashb]

Oh, excetp:

    @classmethod
    def to_json(cls, var: Union[DAG, BaseOperator, dict, list, set, tuple]) -> str:
        """Stringifies DAGs and operators contained by var and returns a JSON string of var."""
        return json.dumps(cls.to_dict(var), ensure_ascii=True)

    @classmethod
    def to_dict(cls, var: Union[DAG, BaseOperator, dict, list, set, tuple]) -> dict:
        """Stringifies DAGs and operators contained by var and returns a dict of var."""
        # Don't call on this class directly - only SerializedDAG or
        # SerializedBaseOperator should be used as the "entrypoint"
        raise NotImplementedError()

[ashb]

Wouldn't this entirely break loading current XCom values?

And "BaseSerialization" has all sorts of behaviour for registering OperatorLinks etc that we don't want here.

So 🤔 not really sure this is right

[ashb]

Yeah, this toally breaks all existing JSON XComs:

In [1]: from airflow.serialization.serialized_objects import BaseSerialization

In [2]: BaseSerialization.from_dict({"x": "y"})
---------------------------------------------------------------------------
KeyError                                  Traceback (most recent call last)
<ipython-input-2-5a1376cdcc27> in <module>
----> 1 BaseSerialization.from_dict({"x": "y"})

~/code/airflow/airflow/airflow/serialization/serialized_objects.py in from_dict(cls, serialized_obj)
    136         reconstructs all DAGs and operators it contains.
    137         """
--> 138         return cls._deserialize(serialized_obj)
    139 
    140     @classmethod

~/code/airflow/airflow/airflow/serialization/serialized_objects.py in _deserialize(cls, encoded_var)
    269         if not isinstance(encoded_var, dict):
    270             raise ValueError(f"The encoded_var should be dict and is {type(encoded_var)}")
--> 271         var = encoded_var[Encoding.VAR]
    272         type_ = encoded_var[Encoding.TYPE]
    273 

KeyError: <Encoding.VAR: '__var'>

In [3]: BaseSerialization.from_dict({"x": "y"})

[ashb]

Maybe we need to start adding info against XCom rows, as I mentioned in #16823 (comment) -- that way this could be used for new XComs but old ones would stay as the were.

[github-actions]

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.