You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2022/03/09 10:14:30 UTC
[GitHub] [airflow] uranusjr opened a new pull request #22112: Add map_index to XCom model and interface
uranusjr opened a new pull request #22112:
URL: https://github.com/apache/airflow/pull/22112
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.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #22112: Add map_index to XCom model and interface
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #22112:
URL: https://github.com/apache/airflow/pull/22112#discussion_r823513413
##########
File path: airflow/migrations/versions/c306b5b5ae4a_switch_xcom_table_to_use_run_id.py
##########
@@ -132,6 +134,10 @@ def downgrade():
op.create_table("__airflow_tmp_xcom", *_get_old_xcom_columns())
xcom = Table("xcom", metadata, *_get_new_xcom_columns())
+
+ # Remoe XCom entries from mapped tis.
+ delete(xcom).where(xcom.c.map_index != -1)
Review comment:
`airflow db downgrade -s --revision a3bcd0914482` to get the SQL.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #22112: Add map_index to XCom model and interface
Posted by GitBox <gi...@apache.org>.
potiuk commented on pull request #22112:
URL: https://github.com/apache/airflow/pull/22112#issuecomment-1066859029
Added PR for that @uranusjr https://github.com/apache/airflow/pull/22244
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #22112: Add map_index to XCom model and interface
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #22112:
URL: https://github.com/apache/airflow/pull/22112#discussion_r823511148
##########
File path: airflow/providers/amazon/aws/operators/emr.py
##########
@@ -245,7 +245,7 @@ def get_link(
:return: url link
"""
if ti_key:
- flow_id = XCom.get_one(key="return_value", ti_key=ti_key)
+ flow_id = XCom.get_value(key="return_value", ti_key=ti_key)
Review comment:
Gah, no. This is the compat shim that _I_ wrote. Ingore me, this is fine.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #22112: Add map_index to XCom model and interface
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #22112:
URL: https://github.com/apache/airflow/pull/22112#discussion_r823554067
##########
File path: airflow/migrations/versions/c306b5b5ae4a_switch_xcom_table_to_use_run_id.py
##########
@@ -132,6 +134,10 @@ def downgrade():
op.create_table("__airflow_tmp_xcom", *_get_old_xcom_columns())
xcom = Table("xcom", metadata, *_get_new_xcom_columns())
+
+ # Remoe XCom entries from mapped tis.
+ delete(xcom).where(xcom.c.map_index != -1)
Review comment:
:lolsob:
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #22112: Add map_index to XCom model and interface
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #22112:
URL: https://github.com/apache/airflow/pull/22112#issuecomment-1065142292
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.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on pull request #22112: Add map_index to XCom model and interface
Posted by GitBox <gi...@apache.org>.
uranusjr commented on pull request #22112:
URL: https://github.com/apache/airflow/pull/22112#issuecomment-1062766686
Wow this conflicts very fast.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #22112: Add map_index to XCom model and interface
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #22112:
URL: https://github.com/apache/airflow/pull/22112#discussion_r823507342
##########
File path: airflow/models/xcom.py
##########
@@ -87,7 +88,7 @@ class BaseXCom(Base, LoggingMixin):
# but it goes over MySQL's index length limit. So we instead create indexes
# 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),
Review comment:
The column order here doesn't match the migrations -- this matters from one of our DBs (sqlite? mssql?)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #22112: Add map_index to XCom model and interface
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #22112:
URL: https://github.com/apache/airflow/pull/22112#discussion_r823510692
##########
File path: airflow/providers/amazon/aws/operators/emr.py
##########
@@ -245,7 +245,7 @@ def get_link(
:return: url link
"""
if ti_key:
- flow_id = XCom.get_one(key="return_value", ti_key=ti_key)
+ flow_id = XCom.get_value(key="return_value", ti_key=ti_key)
Review comment:
We can't use this new method :/ The providers need to continue to work with Airflow 2.1 :/
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a change in pull request #22112: Add map_index to XCom model and interface
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #22112:
URL: https://github.com/apache/airflow/pull/22112#discussion_r823728726
##########
File path: airflow/models/xcom.py
##########
@@ -87,7 +88,7 @@ class BaseXCom(Base, LoggingMixin):
# but it goes over MySQL's index length limit. So we instead create indexes
# 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),
Review 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.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #22112: Add map_index to XCom model and interface
Posted by GitBox <gi...@apache.org>.
potiuk commented on pull request #22112:
URL: https://github.com/apache/airflow/pull/22112#issuecomment-1066811849
> Those `get_value()` usages should only happen in `if ti_key is not None` blocks. The `ti_key` argument is only passed by Airflow 2.3+, so this check acts as feature detection to maintain backward compatibility. Please do tell me if you spot any usages that are not in such blocks.
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.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #22112: Add map_index to XCom model and interface
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #22112:
URL: https://github.com/apache/airflow/pull/22112#discussion_r823505404
##########
File path: airflow/migrations/versions/c306b5b5ae4a_switch_xcom_table_to_use_run_id.py
##########
@@ -132,6 +134,10 @@ def downgrade():
op.create_table("__airflow_tmp_xcom", *_get_old_xcom_columns())
xcom = Table("xcom", metadata, *_get_new_xcom_columns())
+
+ # Remoe XCom entries from mapped tis.
+ delete(xcom).where(xcom.c.map_index != -1)
Review comment:
This doesn't go via alembic's op which means it doesn't get captured in to `--sql` mode.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a change in pull request #22112: Add map_index to XCom model and interface
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #22112:
URL: https://github.com/apache/airflow/pull/22112#discussion_r823553328
##########
File path: airflow/migrations/versions/c306b5b5ae4a_switch_xcom_table_to_use_run_id.py
##########
@@ -132,6 +134,10 @@ def downgrade():
op.create_table("__airflow_tmp_xcom", *_get_old_xcom_columns())
xcom = Table("xcom", metadata, *_get_new_xcom_columns())
+
+ # Remoe XCom entries from mapped tis.
+ delete(xcom).where(xcom.c.map_index != -1)
Review comment:
Actually the way I wrote it the query won’t get executed at all 🤪
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #22112: Add map_index to XCom model and interface
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #22112:
URL: https://github.com/apache/airflow/pull/22112#discussion_r823926085
##########
File path: airflow/models/xcom.py
##########
@@ -87,7 +88,7 @@ class BaseXCom(Base, LoggingMixin):
# but it goes over MySQL's index length limit. So we instead create indexes
# 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),
Review 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.)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on pull request #22112: Add map_index to XCom model and interface
Posted by GitBox <gi...@apache.org>.
uranusjr commented on pull request #22112:
URL: https://github.com/apache/airflow/pull/22112#issuecomment-1066801141
Those `get_value()` usages should only happen in `if ti_key is not None` blocks. The `ti_key` argument is only passed by Airflow 2.3+, so this check acts as feature detection to maintain backward compatibility. Please do tell me if you spot any usages that are not in such blocks.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #22112: Add map_index to XCom model and interface
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #22112:
URL: https://github.com/apache/airflow/pull/22112#discussion_r823506315
##########
File path: airflow/migrations/versions/c306b5b5ae4a_switch_xcom_table_to_use_run_id.py
##########
@@ -132,6 +134,10 @@ def downgrade():
op.create_table("__airflow_tmp_xcom", *_get_old_xcom_columns())
xcom = Table("xcom", metadata, *_get_new_xcom_columns())
+
+ # Remoe XCom entries from mapped tis.
+ delete(xcom).where(xcom.c.map_index != -1)
Review comment:
```suggestion
op.execute(xcom.delete().where(xcom.c.map_index != -1))
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #22112: Add map_index to XCom model and interface
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #22112:
URL: https://github.com/apache/airflow/pull/22112#discussion_r823507947
##########
File path: airflow/models/xcom.py
##########
@@ -87,7 +88,7 @@ class BaseXCom(Base, LoggingMixin):
# but it goes over MySQL's index length limit. So we instead create indexes
# 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),
Review comment:
Shouldn't we have an FK to TaskInstance?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #22112: Add map_index to XCom model and interface
Posted by GitBox <gi...@apache.org>.
potiuk commented on pull request #22112:
URL: https://github.com/apache/airflow/pull/22112#issuecomment-1066860331
(BTW. there were no mis-uses, but I standardized the use of `if not None`
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #22112: Add map_index to XCom model and interface
Posted by GitBox <gi...@apache.org>.
potiuk commented on pull request #22112:
URL: https://github.com/apache/airflow/pull/22112#issuecomment-1066685670
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 `get_value` method of XCom is only added in 2.3.0, yet a number of providers have been switched to use it in some places in this change. I think we we should implement a backwards-compatibility here and add protection that this new XCom method iis not used accidentally by the providers.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dstandish commented on pull request #22112: Add map_index to XCom model and interface
Posted by GitBox <gi...@apache.org>.
dstandish commented on pull request #22112:
URL: https://github.com/apache/airflow/pull/22112#issuecomment-1064652037
heads up @uranusjr that migration has been renamed to `0102_c306b5b5ae4a_switch_xcom_table_to_use_run_id.py` in main
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb merged pull request #22112: Add map_index to XCom model and interface
Posted by GitBox <gi...@apache.org>.
ashb merged pull request #22112:
URL: https://github.com/apache/airflow/pull/22112
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #22112: Add map_index to XCom model and interface
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #22112:
URL: https://github.com/apache/airflow/pull/22112#discussion_r823509211
##########
File path: airflow/models/xcom.py
##########
@@ -203,27 +209,25 @@ def set(
session.add(new)
session.flush()
- @overload
@classmethod
- def get_one(
+ @provide_session
+ def get_value(
cls,
*,
- key: Optional[str] = None,
ti_key: "TaskInstanceKey",
+ key: Optional[str] = None,
session: Session = NEW_SESSION,
- ) -> Optional[Any]:
- """Retrieve an XCom value, optionally meeting certain criteria.
+ ) -> Any:
+ """Retrieve an XCom value for a task instance.
This method returns "full" XCom values (i.e. uses ``deserialize_value``
from the XCom backend). Use :meth:`get_many` if you want the "shortened"
value via ``orm_deserialize_value``.
- If there are no results, *None* is returned.
+ If there are no results, *None* is returned. If multiple XCom entries
+ match the criteria, an arbitrary one is returned.
- A deprecated form of this function accepts ``execution_date`` instead of
- ``run_id``. The two arguments are mutually exclusive.
-
- :param ti_key: The TaskInstanceKey to look up the XCom for
+ :param ti_key: The TaskInstanceKey to look up the XCom for.
:param key: A key for the XCom. If provided, only XCom with matching
keys will be returned. Pass *None* (default) to remove the filter.
:param include_prior_dates: If *False* (default), only XCom from the
Review comment:
This method doesn't accept include_prior_dates.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org