Dynamic Task Mapping is Immutable within a Run

[fritz-astronomer]

Apache Airflow version

2.3.0 (latest released)

What happened

Looks like mapped tasks are immutable, even when the source XCOM that created them changes.

This is a problem for things like Late Arriving Data and Data Reprocessing

What you think should happen instead

Mapped tasks should change in response to a change of input

How to reproduce

Here is a writeup and MVP DAG demonstrating the issue
https://gist.github.com/fritz-astronomer/d159d0e29d57458af5b95c0f253a3361

Operating System

docker/debian

Versions of Apache Airflow Providers

No response

Deployment

Astronomer

Deployment details

No response

Anything else

Can look into a fix - but may not be able to submit a full PR

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[ephraimbuddy]

Just to mention that this works:

    xarg = pull_messages()

    op = DatabricksRunNowOperator.partial(
        task_id = "new_job",
        job_id=42,
        notebook_params={"dry-run": "true"},
        python_params=["douglas adams", "42"],
        spark_submit_params=["--class", "org.apache.spark.examples.SparkPi"],
    ).expand(jar_params= xarg)

[uranusjr]

That’s interesting… I would assume it does not. Both examples go through XCom, and both should go through the same code path. This paragraph is not relevant.

Regarding the bug, we currently have logic to mark existing expanded tasks as REMOVED if they are superfluous. Maybe that’s not working as intended.

[ephraimbuddy]

That’s interesting… I would assume it does not. Both examples go through XCom, and both should go through the same code path.

I commented on the wrong post...the comment was meant for #23642 and it's not working there as well...not thinking well today!

[uranusjr]

Ah OK, that makes sense. So I guess we “just” need to fix the REMOVED logic.

[ephraimbuddy]

Ah OK, that makes sense. So I guess we “just” need to fix the REMOVED logic.

I think fixing that would work if the dag file is edited. For a dynamically generated mapping args as in the issue, it won't get to the code because the dag_run.dag_hash would still match the latest serialized dag hash, right?

airflow/airflow/jobs/scheduler_job.py

Line 1188 in 7277122

if dag_run.dag_hash == latest_version: