Decode old-style nested Xcom value

[utkarsharma2]

Xcom values generated by airflow 2.5.2 was not getting decoded by airflow 2.6.
There were two issues

1. I think it was a typo, which was mentioned in the issue -

   airflow/airflow/serialization/serde.py

   Line 273 in 58fca5e

   return {CLASSNAME: old[OLD_TYPE], VERSION: DEFAULT_VERSION, DATA: old[OLD_DATA][OLD_DATA]}

2. When we have __type == dict in _convert() we don't need to wrap the data in dict again.

closes: #31769

[phanikumv]

Test docker-compose quick start is failing, other than that it lgtm

[jedcunningham]

Suggested change

	# Added to handle for handling xcom data from airflow 2.5.2
	# Added to handle for handling xcom data from airflow 2.5.2

This comment feels a little weird to me. Is this really just 2.5.2, or <2.6?

[hussein-awala]

same thought (comment)

[utkarsharma2]

I'm not sure what all airflow version generates this encoded value, but I'll check and update that here. Sorry, should have done this in 1st iteration.

[utkarsharma2]

I think the code comment should be applicable for 2.4>= to <2.6 because the dataset concept was introduced in 2.4. Will update accordingly. Will update the code comment accordingly.

[utkarsharma2]

Also, I was able to replicate the issue for airflow 2.4.3 as running example dag dataset_produces_1 produces data - [{“__type": "airflow.datasets.Dataset", "__source": "dataset_produces_1.producing_task_1", "__var": {"__var": {"uri": "s3://dag1/output_1.txt", "extra": {"__var": {"hi": "bye"}, "__type": "dict"}}, "__type": "dict"}}] and the issue was not caught by testcase - https://github.com/apache/airflow/blob/main/tests/serialization/test_serde.py#L259-L273 as it doesn't have more than two levels of nested __var

[hussein-awala]

I can't find any encoder which add __var in a nested object __var, and why it was moved from

    @staticmethod
    def _convert(old: dict) -> dict:
        """Converts an old style serialization to new style"""
        if OLD_TYPE in old and OLD_SOURCE in old:
            return {CLASSNAME: old[OLD_TYPE], VERSION: DEFAULT_VERSION, DATA: old[OLD_DATA]}

to

def _convert(old: dict) -> dict:
    """Converts an old style serialization to new style"""
    if OLD_TYPE in old and OLD_DATA in old:
        return {CLASSNAME: old[OLD_TYPE], VERSION: DEFAULT_VERSION, DATA: old[OLD_DATA][OLD_DATA]}

    return old

in #31866.
I wonder if it was a bug in that PR and we need just to remove it instead of adding a condition for 2.5.2.
BTW, this PR was released in 2.6.0, so we will have the same problem with all the previous version, maybe we need to test 2.5.3 and 2.4.3 before merging this one, in order to update the comments:

# Added to handle for handling xcom data from airflow 2.5.2

@bolkedebruin wdyt?

[utkarsharma2]

@hussein-awala In Airflow 2.4.3 if you run dag dataset_produces_1 it will produce xcom data as below, with nested __var -

{
        "__type": "airflow.datasets.Dataset",
        "__source": None,
        "__var": {
            "__var": {
                "uri": uri,
                "extra": {
                    "__var": {"hi": "bye"},
                    "__type": "dict"
                }
            },
            "__type": "dict",
        },
}

And I think old[OLD_DATA][OLD_DATA] was done to directly get to the second nested __var assuming there will be no more nested __var in serialized data but that is not the case as extra can have dicts too.

Also, in the old test case, we had only tests for the below data -
test case - https://github.com/apache/airflow/blob/main/tests/serialization/test_serde.py#L259-L273

{
    "__type": "airflow.datasets.Dataset",
    "__source": None,
    "__var": {
        "__var": {
            "uri": uri,
            "extra": None              # <-- change 
        },
        "__type": "dict",
    },
}

old[OLD_DATA][OLD_DATA] fails with extra = {"__var": {"hi": "bye"}, "__type": "dict"} and will works with extra=None

But if we explicitly handle the case for dict and do not assume the nested levels of __var below code will work in all cases.

def _convert(old: dict) -> dict:
    """Converts an old style serialization to the new style.""
    if OLD_TYPE in old and OLD_DATA in old:
        #Added to handle for handling xcom data from airflow < 2.6
        if old[OLD_TYPE] == "dict":
            return old[OLD_DATA]
        else:
            return {CLASSNAME: old[OLD_TYPE], VERSION: DEFAULT_VERSION, DATA: old[OLD_DATA]}
    return old

[bolkedebruin]

Looking

[bolkedebruin]

_convert was created to handle old (pre 2.6) XCom seralized data, but the issue is that the new deserializer cannot handle:

{'extra': {'__type': 'dict', '__var': {'hi': 'bye'}}, 'uri': 's3://does_not_exist'}

Which is the nested form of the old style serialization.

I think the solution is correct, but that the commit message and the inline comments could be improved. As mentioned this code was already handling pre 2.6 data, but it wasn't handling the nested version of that correctly. Also I think the test could use some improvement / addition. The above dict should be able to be deseralized directly. So I would expect an additional test.