Add map_index to XCom model and interface #22112
Conversation
|
Wow this conflicts very fast. |
4224c93 to
fe9bd73
Compare
33153e1 to
e494600
Compare
airflow/migrations/versions/c306b5b5ae4a_switch_xcom_table_to_use_run_id.py
Outdated
Show resolved
Hide resolved
| # separately, and enforce uniqueness with DagRun.id instead. | ||
| Index("idx_xcom_key", key), | ||
| Index("idx_xcom_ti_id", dag_id, task_id, run_id), | ||
| Index("idx_xcom_ti_id", dag_id, task_id, run_id, map_index), |
There was a problem hiding this comment.
The column order here doesn't match the migrations -- this matters from one of our DBs (sqlite? mssql?)
There was a problem hiding this comment.
Shouldn't we have an FK to TaskInstance?
There was a problem hiding this comment.
There wasn’t a foreign key before. I think previously (when XCom was based on execution date) it was possible to push an XCom to a future date, so a foreign key did not make sense back then; now it’s based on run ID perhaps it makes sense to have a ti-xcom relation, but that should be a separate discussion regardless.
There was a problem hiding this comment.
it was possible to push an XCom to a future date
Already done https://lists.apache.org/thread/gofj3g6m6vvksy6n0cmgq1qxd309bbbl (I don't think it ever actually worked.)
bdee31a to
3f1c4b6
Compare
|
heads up @uranusjr that migration has been renamed to |
This is not actually stored correctly yet. We still need to fix the XCom interface.
This adds an additional (optional) map_index argument to XCom's get/set/clear interface so mapped task instances can push to the correct entries, and have them pulled correctly by a downstream. To make the XCom interface easier to use for common scenarios, a convenience method get_value is added to take a TaskInstanceKey that automatically performs argument unpacking and call get_one underneath. This is not done as a get_one overload to simplify the implementation and typing.
3f1c4b6 to
cb60690
Compare
|
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. |
|
Hello @uranusjr @ashb - while preparaing the "Throve classifier" release for Providers i noticed that backwards-incompatible change sneaked in for a number of providers here. The |
|
Those |
Ah ok. Just thinking how to automate the check in this case so that it is not used accidentally. Let me take a look maybe I will come up with something. I do not want to spot it myself, our backwards compatibility checks should do it automatically for us I think. Same way as we check if "imports" are working. |
|
(BTW. there were no mis-uses, but I standardized the use of |
This adds an additional (optional) map_index argument to XCom's get/set/clear interface so mapped task instances can push to the correct entries, and have them pulled correctly by a downstream.
To make the XCom interface easier to use for common scenarios, a convenience method get_value is added to take a TaskInstanceKey that automatically performs argument unpacking and call get_one underneath. This is not done as a get_one overload to simplify the implementation and typing.