mobuchowski commented on code in PR #75860:
URL: https://github.com/apache/airflow/pull/75860#discussion_r4401827383
##########
providers/openlineage/src/airflow/providers/openlineage/utils/utils.py:
##########
@@ -1522,6 +1526,7 @@ class TaskInfoComplete(TaskInfo):
"_BaseOperator__instantiated",
"_BaseOperator__init_kwargs", # Causes recursion error on AF3,
nothing useful there
"_dag",
+ "dag", # A stored operator's Dag, e.g. in the scheduler, which can't
be pickled
Review Comment:
With `include_full_task_info` on, `TaskInfoComplete` copies every attribute
in the operator's `__dict__` that isn't in `excludes`. On a scheduler-side
mapped operator, that includes `expand_input`, which `SerializedDAG`
deserialization sets with `setattr(task, k, kwargs_ref.deref(dag))`. When the
mapped task's input comes from an upstream task, `deref` makes it a
`SchedulerPlainXComArg` whose `.operator` is the live upstream operator. From
there pickling reaches the operator's Dag, and through the Dag the other
operators and their `methodtools` caches. `_cast_basic_types` leaves attrs
objects as they are, so all of this ends up in the dict item.
I reproduced it by changing one line of the new test to
`PythonOperator.partial(task_id="mapped",
python_callable=list).expand(op_kwargs=plain.output)`. It then fails on
`TaskInfoComplete(dag.get_task("mapped"))`:
```
_pickle.PicklingError: Can't pickle <functools._lru_cache_wrapper object at
0x…>: it's not the same object as
airflow.serialization.definitions.mappedoperator.SerializedMappedOperator.get_parse_time_mapped_ti_count
```
`expand_input` is the only item that fails. The `isidentifier()` filter
doesn't help here, because the cache is reached through a value rather than
being a key on the encoded operator. In production, task events for a mapped
task fed from an upstream task's output would still be dropped with `Failed to
submit method to executor` whenever full task info is on. Mapping over an
upstream task's output is probably the most common way dynamic mapping is used.
Adding `"expand_input"` and `"op_kwargs_expand_input"` to
`TaskInfoComplete.excludes` makes the changed test pass. Could you add that,
and test the upstream-output case next to the literal one? Longer term, storing
only JSON-safe values in `TaskInfoComplete` (for example
`json.loads(json.dumps(v, default=str))`) would stop the next live attribute
from causing the same failure.
--
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]
