You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "kandharvishnu (via GitHub)" <gi...@apache.org> on 2025/06/25 05:49:23 UTC
[I] Sensor task is not skipped when used after BranchSQLOperator in Airflow 3.x [airflow]
kandharvishnu opened a new issue, #52219:
URL: https://github.com/apache/airflow/issues/52219
### Apache Airflow version
3.0.2
### If "Other Airflow 2 version" selected, which one?
_No response_
### What happened?
In Airflow 3.x, when a **SqlSensor** is used as one of the downstream tasks of a `BranchSQLOperator`, it is **not** being skipped as expected based on the branch condition. Even when the sensor task is not selected by the branching logic, it still runs instead of being skipped.
This behavior breaks the expected semantics of branching, where only one downstream path should execute and others should be marked as skipped.
### What you think should happen instead?
Only the task(s) that are selected by the `BranchSQLOperator` should run. All other downstream tasks, including sensors, should be skipped automatically. This was the observed and expected behavior in Airflow 2.x.
### How to reproduce
Use the following minimal DAG code:
```
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.common.sql.operators.sql import BranchSQLOperator
from airflow.providers.common.sql.sensors.sql import SqlSensor
from datetime import datetime
CONN_ID = "postgres_conn"
def task_2_func():
print("Branch: TASK - 2 – Executing this task.")
with DAG(
dag_id="example_branch_sql_with_python",
start_date=datetime(2024, 1, 1),
schedule=None,
catchup=False,
default_args={"owner": "airflow"},
) as dag:
branch_task = BranchSQLOperator(
task_id="branch_sql",
sql="SELECT CASE WHEN 1=1 THEN 0 ELSE 1 END",
conn_id=CONN_ID,
follow_task_ids_if_true=["task_1"],
follow_task_ids_if_false=["task_2"],
)
task_1 = SqlSensor(
task_id="task_1",
sql="SELECT CASE WHEN 1=1 THEN 1 ELSE 0 END",
conn_id=CONN_ID,
)
task_2 = PythonOperator(
task_id="task_2",
python_callable=task_2_func,
)
branch_task >> [task_1, task_2]
```
In this example, **task_1** should be **skipped** since the SQL in `BranchSQLOperator` returns 0, selecting task_2. But in Airflow 3.x, both tasks are executed.
### Operating System
Debian
### Versions of Apache Airflow Providers
_No response_
### Deployment
Astronomer
### Deployment details
_No response_
### Anything else?
_No response_
### Are you willing to submit PR?
- [ ] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
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.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Sensor task is not skipped when used after BranchSQLOperator in Airflow 3.x [airflow]
Posted by "fweilun (via GitHub)" <gi...@apache.org>.
fweilun commented on issue #52219:
URL: https://github.com/apache/airflow/issues/52219#issuecomment-3003994502
Hi, I'd like to work on this issue and submit a PR. Let me know if it's already taken. Thanks!
--
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
Re: [I] Sensor task is not skipped when used after BranchSQLOperator in Airflow 3.x [airflow]
Posted by "fweilun (via GitHub)" <gi...@apache.org>.
fweilun commented on issue #52219:
URL: https://github.com/apache/airflow/issues/52219#issuecomment-3018794579
Hi, I'm currently working on this issue.
I found that Skipping tasks [] is empty because the line below in SkipMixin::skip_all_except filters out all operators except MappedOperator and BaseOperator:
`
downstream_tasks = _ensure_tasks(task.downstream_list)
`
However, BaseSensorOperator from airflow.sdk.bases.sensor.BaseSensorOperator is not recognized as a BaseOperator due to differing module origins, even though it nominally inherits from BaseOperator. As a result, tasks like SqlSensor (which subclass BaseSensorOperator) are excluded from downstream_tasks, and Skipping tasks [], which is derived from downstream_tasks, ends up empty.
Below is the log. task_1 is filtered out in line 11-12.
<img width="829" alt="Image" src="https://github.com/user-attachments/assets/334e9838-ef1f-4016-b73a-900d05b9fdc0" />
I temporarily added BaseSensorOperator to _ensure_tasks() as a workaround, and it resolved the issue:
adjusted code:
```python
def _ensure_tasks(nodes: Iterable[DAGNode]) -> Sequence[Operator]:
from airflow.models.baseoperator import BaseOperator
from airflow.models.mappedoperator import MappedOperator
from airflow.sensors.base import BaseSensorOperator
return [n for n in nodes if isinstance(n, (BaseOperator, MappedOperator, BaseSensorOperator))]
```
Does this seem like the right direction?
Would appreciate any guidance on whether _ensure_tasks() should be updated to support SDK-based operators, or if there’s a better approach moving forward.
Thanks!
--
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
Re: [I] Sensor task is not skipped when used after BranchSQLOperator in Airflow 3.x [airflow]
Posted by "fweilun (via GitHub)" <gi...@apache.org>.
fweilun commented on issue #52219:
URL: https://github.com/apache/airflow/issues/52219#issuecomment-3018696381
Hi, I'm currently working on this issue.
`
def skip_all_except(
self,
ti: RuntimeTaskInstanceProtocol,
branch_task_ids: None | str | Iterable[str],
):
// some code in skip_all_except
task = ti.task
dag = ti.task.dag
valid_task_ids = set(dag.task_ids)
invalid_task_ids = branch_task_id_set - valid_task_ids
if invalid_task_ids:
raise AirflowException(
"'branch_task_ids' must contain only valid task_ids. "
f"Invalid tasks found: {invalid_task_ids}."
)
log.info(f"[DEBUG] {task.downstream_list}")
downstream_tasks = _ensure_tasks(task.downstream_list)
if downstream_tasks:
# For a branching workflow that looks like this, when "branch" does skip_all_except("task1"),
# we intuitively expect both "task1" and "join" to execute even though strictly speaking,
# "join" is also immediately downstream of "branch" and should have been skipped. Therefore,
# we need a special case here for such empty branches: Check downstream tasks of branch_task_ids.
# In case the task to skip is also downstream of branch_task_ids, we add it to branch_task_ids and
# exclude it from skipping.
#
# branch -----> join
# \ ^
# v /
# task1
#
for branch_task_id in list(branch_task_id_set):
branch_task_id_set.update(dag.get_task(branch_task_id).get_flat_relative_ids(upstream=False))
log.info(f"[DEBUG] downstream_tasks: {downstream_tasks}")
log.info(f"[DEBUG] branch_task_id_set: {branch_task_id_set}")
skip_tasks = [
(t.task_id, ti.map_index) for t in downstream_tasks if t.task_id not in branch_task_id_set
]
follow_task_ids = [t.task_id for t in downstream_tasks if t.task_id in branch_task_id_set]
log.info("Skipping tasks %s", skip_tasks)
ti.xcom_push(
key=XCOM_SKIPMIXIN_KEY,
value={XCOM_SKIPMIXIN_FOLLOWED: follow_task_ids},
)
# The following could be applied only for non-mapped tasks,
# as future mapped tasks have not been expanded yet. Such tasks
# have to be handled by NotPreviouslySkippedDep.
self._set_state_to_skipped(skip_tasks, ti.map_index) # type: ignore[arg-type]
`
--
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
Re: [I] Sensor task is not skipped when used after BranchSQLOperator in Airflow 3.x [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil closed issue #52219: Sensor task is not skipped when used after BranchSQLOperator in Airflow 3.x
URL: https://github.com/apache/airflow/issues/52219
--
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
Re: [I] Sensor task is not skipped when used after BranchSQLOperator in Airflow 3.x [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on issue #52219:
URL: https://github.com/apache/airflow/issues/52219#issuecomment-3084124905
On it, will be fixed asap
--
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
Re: [I] Sensor task is not skipped when used after BranchSQLOperator in Airflow 3.x [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on issue #52219:
URL: https://github.com/apache/airflow/issues/52219#issuecomment-3084193593
https://github.com/apache/airflow/pull/53455
--
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