utilize map_index for deterministic generation of OpenLineage's run_id

[mobuchowski]

Use queued_dttm to differentiate between sensors with mode=reschedule runs.
Use map_index to differentiate between mapped tasks runs.

[kacpermuda]

Looks good, let's run all the compat tests to make sure it works for past Airflow versions as well.

[dstandish]

you might look into whether you can use TaskReschedule to know whether you are in a reschedule. if so, could limit unnec queries by checking whether you inherit from basesensor and have mode==reschedule. and then, on the exit side, you could check state == up-for-reschedule.
although, this could be an issue with deferrables as well. with deferrables though you could check whether next_method / next kwargs populated to infer coming out of deferral. i think we do that somewhere and selectively emit a log message. then on exit, you could look at state deferred perhaps.
but fundamentally, you are OL master so, i defer to you

[mobuchowski]

@dstandish at the end just

        session.query(
            exists().where(
                TaskReschedule.dag_id == ti.dag_id,
                TaskReschedule.task_id == ti.task_id,
                TaskReschedule.run_id == ti.run_id,
                TaskReschedule.map_index == ti.map_index,
                TaskReschedule.try_number == ti.try_number,
            )
        ).scalar()
        is True

works - the additional query isn't a big problem because I can perform those just for reschedulable sensors.

[dstandish]

@dstandish at the end just

        session.query(
            exists().where(
                TaskReschedule.dag_id == ti.dag_id,
                TaskReschedule.task_id == ti.task_id,
                TaskReschedule.run_id == ti.run_id,
                TaskReschedule.map_index == ti.map_index,
                TaskReschedule.try_number == ti.try_number,
            )
        ).scalar()
        is True

works - the additional query isn't a big problem because I can perform those just for reschedulable sensors.

cool @mobuchowski

[potiuk]

Is not that something that should be back-ported to 2.10.4 ? It certainly looks like

[potiuk]

I provisionally added 2.10.4 milestone now but @mobuchowski -> maybe you can use the new cherry-picker manual flow to back-port it (it's already merged, so we missed the opportunity to auto cherry-pick it) https://github.com/apache/airflow/blob/main/dev/README_AIRFLOW3_DEV.md#how-to-backport-pr-with-cherry-picker-cli

[potiuk]

Ah. STupid me. It's provider-only :). Forget it.