Lee-W commented on code in PR #58289:
URL: https://github.com/apache/airflow/pull/58289#discussion_r2581529254
##########
airflow-core/src/airflow/assets/manager.py:
##########
@@ -229,12 +242,44 @@ def notify_asset_alias_created(asset_assets: AssetAlias):
def notify_asset_changed(asset: Asset):
"""Run applicable notification actions when an asset is changed."""
try:
+ # TODO: AIP-76 this will have to change. needs to know *what*
happened to the asset (e.g. partition key)
Review Comment:
nit: let's add the link or issue number in the comment here so that we can
find where to modify easier
##########
airflow-core/src/airflow/assets/manager.py:
##########
@@ -243,15 +288,91 @@ def _queue_dagruns(cls, asset_id: int, dags_to_queue:
set[DagModel], session: Se
# "fallback" to running this in a nested transaction. This is needed
# so that the adding of these rows happens in the same transaction
# where `ti.state` is changed.
- if not dags_to_queue:
+ if get_dialect_name(session) == "postgresql":
+ return cls._queue_dagruns_nonpartitioned_postgres(asset_id,
non_partitioned_dags, session)
+ return cls._queue_dagruns_nonpartitioned_slow_path(asset_id,
non_partitioned_dags, session)
+
+ @classmethod
+ def _queue_partitioned_dags(
+ cls,
+ asset_id: int,
+ partition_dags: Iterable[DagModel],
+ event: AssetEvent,
+ partition_key: str | None,
+ session: Session,
+ ) -> None:
+ if partition_dags and not partition_key:
+ # TODO: AIP-76 how to best ensure users can see this? Probably add
Log record.
+ log.warning(
+ "Listening dags are partition-aware but run has no partition
key",
+ listening_dags=[x.dag_id for x in partition_dags],
+ asset_id=asset_id,
+ run_id=event.source_run_id,
+ dag_id=event.source_dag_id,
+ task_id=event.source_task_id,
+ )
return
Review Comment:
I feel this logic should exist in line 269. but non-blocking
##########
airflow-core/src/airflow/assets/manager.py:
##########
@@ -243,15 +288,91 @@ def _queue_dagruns(cls, asset_id: int, dags_to_queue:
set[DagModel], session: Se
# "fallback" to running this in a nested transaction. This is needed
# so that the adding of these rows happens in the same transaction
# where `ti.state` is changed.
- if not dags_to_queue:
+ if get_dialect_name(session) == "postgresql":
+ return cls._queue_dagruns_nonpartitioned_postgres(asset_id,
non_partitioned_dags, session)
+ return cls._queue_dagruns_nonpartitioned_slow_path(asset_id,
non_partitioned_dags, session)
+
+ @classmethod
+ def _queue_partitioned_dags(
+ cls,
+ asset_id: int,
+ partition_dags: Iterable[DagModel],
+ event: AssetEvent,
+ partition_key: str | None,
+ session: Session,
+ ) -> None:
+ if partition_dags and not partition_key:
+ # TODO: AIP-76 how to best ensure users can see this? Probably add
Log record.
+ log.warning(
+ "Listening dags are partition-aware but run has no partition
key",
+ listening_dags=[x.dag_id for x in partition_dags],
+ asset_id=asset_id,
+ run_id=event.source_run_id,
+ dag_id=event.source_dag_id,
+ task_id=event.source_task_id,
+ )
return
- if get_dialect_name(session) == "postgresql":
- return cls._postgres_queue_dagruns(asset_id, dags_to_queue,
session)
- return cls._slow_path_queue_dagruns(asset_id, dags_to_queue, session)
+ for target_dag in partition_dags:
+ if TYPE_CHECKING:
+ assert partition_key is not None
+ from airflow.models.serialized_dag import SerializedDagModel
+
+ serdag = SerializedDagModel.get(dag_id=target_dag.dag_id,
session=session)
+ if not serdag:
Review Comment:
```suggestion
if not (serdag :=
SerializedDagModel.get(dag_id=target_dag.dag_id, session=session)):
```
##########
airflow-core/src/airflow/example_dags/example_outlet_event_extra.py:
##########
@@ -68,6 +68,9 @@ def asset_with_extra_by_context(*, outlet_events=None):
def _asset_with_extra_from_classic_operator_post_execute(context, result):
context["outlet_events"][asset].extra = {"hi": "bye"}
+ # TODO: AIP-76 probably we want to make it so this could be
Review Comment:
I feel like it add some complexity I'm not sure we what to handle 🤔 Is it
needed for partition? Or would it be better to make it the next improvement?
##########
airflow-core/src/airflow/models/taskinstance.py:
##########
@@ -1315,6 +1315,9 @@ def register_asset_changes_in_db(
) -> None:
from airflow.sdk.definitions.asset import Asset, AssetAlias,
AssetNameRef, AssetUniqueKey, AssetUriRef
+ # TODO: AIP-76 should we provide an interface to override this, so
that the task can
+ # tell the truth if for some reason it touches a different partition?
Review Comment:
Why would the "for some reason" be?
--
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]
