Remove XCom pickling

[kaxil]

XCom pickling was disabled by default in Airflow 2.0.0: https://airflow.apache.org/docs/apache-airflow/1.10.15/configurations-ref.html#enable-xcom-pickling

Discussion

To avoid a time-consuming DB migration, I have not changed the column type of value; it is still LargeBinary/LONGBLOB (MySQL).

As part of Airflow 3, we should strongly recommend users to use the airflow db clean command to prune the DBs to the minimum required. If we assume, most users would do that, we can run the following migration:

Option 1: Try to migrate pickle to JSON

def upgrade():
    bind = op.get_bind()
    session = Session(bind=bind)

    for row in session.query(XCom).all():
        try:
            unpickled_data = pickle.loads(row.value)
            row.value = json.dumps(unpickled_data).encode('utf-8')
        except (pickle.UnpicklingError, json.JSONDecodeError):
            # If unpickling fails, assume it's already JSON and skip
            continue

    op.alter_column("xcom", "value", type_=sa.Text)

Option 2: Delete XCom rows with pickle

from airflow.models.xcom import BaseXCom

def upgrade():
    bind = op.get_bind()
    session = Session(bind=bind)

    pickled_xcoms = session.query(BaseXCom).filter(BaseXCom.value.isnot(None))
    
    deleted_count = 0
    for xcom in pickled_xcoms:
        if xcom.value.startswith(b'\x80'):  # Identify as pickled by protocol marker: https://github.com/python/cpython/blob/494360afd00dc8f6b549f160525c3e86ec14905d/Lib/pickletools.py#L2122-L2133
            session.delete(xcom)

Option 3: Keep the current column type

Not optimal, but we can keep the current column type to binary/long-blob

Any thoughts?

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an 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 a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[pgagnon]

@kaxil

Option 1: Try to migrate pickle to JSON

# If unpickling fails, assume it's already JSON and skip

If JSON decoding fails, maybe we could save the value in a backup blob field (to be deprecated and removed in a specified future version).

While XComs are generally not that important I feel it might be better to soft-delete wherever there's a risk of data loss as a general practice.

[jscheffl]

I like the drop of the pickle type - but I'd favor not keeping this as blob/binary in the DB. Then we can not use any DB feature to efficiently "use" the data other than a blob.

If not in this PR, can we have a follow-up that converts the data into json data type as described in https://www.postgresql.org/docs/17/datatype-json.html ?

(MySQL is: https://dev.mysql.com/doc/refman/8.4/en/json.html)

And totally forgot about the vote: I think we can also consider to provide an offline migration tool which we request to be executed by everybody manually prior upgrade such that we don't need to carry complex inline migration. Everybody who uses XCom pickling should know this from the config. So Option 1 or Option 1a (a=offline tool)

[hussein-awala]

+1

[kaxil]

I will create a separate PR to handle the migration so that can be reviewed independently -- will have a PR by EOD today

[kaxil]

PR created: #44166