AIP-83 run_id logic when no logical date #46398
Conversation
prabhusneha
requested review from
uranusjr,
ryanahamilton,
ashb,
bbovenzi,
pierrejeambrun,
jscheffl,
ephraimbuddy and
XD-DENG
as code owners
boring-cyborg
bot
added
area:API
| run_type=DagRunType.MANUAL, | ||
| logical_date=coerced_logical_date, | ||
| data_interval=data_interval, | ||
| run_after=data_interval.end, |
There was a problem hiding this comment.
At some point (before 3.0) I want to reduce the arguments here to just take a DagRunInfo, but that can be done separately instead.
There was a problem hiding this comment.
according to the doc, when logical date is null, then data interval end should be null. so it would not make sense to use data interval end as the run_after date. thoughts @uranusjr
There was a problem hiding this comment.
Oh yeah you’re right. This should do coerced_logical_date or timezone.utcnow() (as currently implemented, coerced_logical_date can never be None, but it will be when everything is finished).
There was a problem hiding this comment.
Should we include this change as a part of this PR or have a separate PR?
| @@ -1391,6 +1392,7 @@ def _create_dag_runs_asset_triggered( | |||
| run_type=DagRunType.ASSET_TRIGGERED, | |||
| logical_date=logical_date, | |||
| data_interval=data_interval, | |||
| run_after=max(logical_dates.values()), | |||
There was a problem hiding this comment.
There’s a task to make asset-triggered runs have None logical_date instead, so we’ll need to change this again soon. This is good enough for now.
| data["dag_run_id"] = DagRun.generate_run_id( | ||
| DagRunType.MANUAL, timezone.parse(data["logical_date"]) | ||
| DagRunType.MANUAL, | ||
| timezone.parse(data["logical_date"]), | ||
| ) |
There was a problem hiding this comment.
Do we want to pass run_after here?
There was a problem hiding this comment.
looks like there is no change here, just a new line?
There was a problem hiding this comment.
Yes and I think that’s wrong, we should change the logic here.
There was a problem hiding this comment.
Is this required here? as run_after was not added in this as part of the PR that added the new field run_after ( #46195 ).
I assumed this was going to be deprecated so wasn't added here.
| def generate_run_id(self, logical_date: datetime) -> str: | ||
| def generate_run_id(self, logical_date: datetime | None, run_after: datetime | None) -> str: | ||
| if logical_date is None: | ||
| if run_after is None: | ||
| raise ValueError("run_after cannot be None") | ||
| return run_after + get_random_string() | ||
| return f"{self}__{logical_date.isoformat()}" |
There was a problem hiding this comment.
Maybe if this function should continue to accept one single datetime value, and we do the if-else check outside instead.
There was a problem hiding this comment.
Maybe if this function should continue to accept one single datetime value, and we do the if-else check outside instead.
Here we need to know if logical_date is None or not before generating a random string that gets appended to run_after.
If we take just one argument, it wouldn't know when to append the random string. Are you suggesting to move this logic into the callers(Timetable.generate_run_id and DagRun.generate_run_id)?
I went with the current implementation to avoid duplicate code/multiple function calls.
jedcunningham
added
the
AIP-83
| @@ -371,6 +371,7 @@ def trigger_dag_run( | |||
| run_type=DagRunType.MANUAL, | |||
| logical_date=logical_date, | |||
| data_interval=data_interval, | |||
| run_after=data_interval.end, | |||
There was a problem hiding this comment.
again, if logical date is null, then we won't generally have a data interval.... correct @uranusjr ?
There was a problem hiding this comment.
data_interval would be null as well.
Closes: #46199
Construct run_id, when logical date is null, using run_after + random string.