kaxil opened a new pull request, #76270:
URL: https://github.com/apache/airflow/pull/76270
`SqlSensor` has no deferrable mode. Waiting for a table to fill either holds
a worker slot (`mode="poke"`) or starts the task on a worker again for every
check (`mode="reschedule"`). This adds `deferrable=True`: the triggerer runs
the query every `poke_interval`, and the task holds no worker slot while it
waits. It works with any `DbApiHook` connection.
```python
SqlSensor(
task_id="wait_for_orders",
conn_id="shop_db",
sql="SELECT COUNT(*) >= %(min_orders)s FROM orders WHERE order_date =
CURRENT_DATE",
parameters={"min_orders": 3},
poke_interval=60,
deferrable=True,
)
```
Like other provider operators, `deferrable` defaults to `[operators]
default_deferrable`. Prior art: `SnowflakeSensorAsync` in astronomer-providers
is a deferrable `SqlSensor` for Snowflake only. A generic version in common.sql
covers every `DbApiHook` backend.
## Design rationale
**The triggerer checks the condition, not the worker.** The sensor checks
once on the worker and returns straight away if the condition already holds.
Otherwise it defers to a new `SqlSensorTrigger`, which sleeps `poke_interval`,
runs the query, and fires once the first cell of the first row is truthy. With
`fail_on_empty=True` it fails on an empty result instead. The task defers
exactly once, and `timeout` is passed straight to `defer(timeout=...)`.
**With `success`, `failure` or `selector`, the sensor does not defer.**
These are Python callables from the Dag file, so the triggerer cannot run them.
The alternative was a trigger that hands each row back to the worker to
evaluate and then re-defers. For the usual shape, `SELECT COUNT(*)` with
`success=lambda n: n >= 100`, that starts the task on a worker for every check,
which costs as much as `mode="reschedule"` plus a trip through the triggerer.
So when any of them is set, the sensor logs a warning and runs in its usual
`mode`. `HttpSensor` does the same when `response_check` is set. The docs tell
users to write the condition in SQL instead.
The sensor also logs a warning and stays in its usual `mode` when:
- **`silent_fail`, `never_fail` or `exponential_backoff` is set.** They only
take effect inside the poke loop of `BaseSensorOperator.execute`.
- **A subclass overrides `poke`.** The triggerer only checks the first cell,
so it would skip the subclass's condition and succeed too early.
- **It runs on Airflow 3 before 3.2.2.** Each poll, connection lookup
included, runs in a thread (`sync_to_async(thread_sensitive=False)`) so a slow
query never blocks other triggers. Before 3.2.2, sync SDK calls from several
trigger threads at once can make the TriggerRunner read supervisor responses
out of order and crash. #66412 fixed that, and it was backported to 3.2.2.
Airflow 2 has no comms channel, so 2.11 defers too.
**`SQLExecuteQueryTrigger` gains an optional `parameters` argument.** It is
passed to `get_records` and serialized. `SqlSensorTrigger` subclasses it to
reuse `get_hook` and `serialize`. Apart from the new argument,
`SQLExecuteQueryTrigger` and `GenericTransfer`, which uses it, are unchanged.
**The trigger builds the hook and calls `get_records` on every poll.** That
is the same call the worker-side `poke()` makes. It picks up credentials
rotated during a long wait, and runs the identical query on hooks that override
only `get_records` (Hive passes `parameters` as `hive_conf` there).
## Gotchas
- On Airflow 2.11, a deferral that hits `timeout` fails with a retryable
`TaskDeferralError`, not `AirflowSensorTimeout`, so retries restart the wait.
That is core 2.x behaviour for every deferrable sensor.
- `soft_fail=True` turns any trigger failure into a skip, including
`fail_on_empty`, because `BaseSensorOperator.resume_execution` converts every
`AirflowException`. An error on the first worker-side check follows poke-mode
rules instead.
## Verified end to end
I ran this on `airflow standalone` from this branch (main, Postgres
backend), with a real scheduler, triggerer and worker, and the sensor reading a
Postgres `shop` database:
| Dag | What happened |
|---|---|
| `wait_for_orders` | Task went to `deferred`; the triggerer polled every 10
s ("Criteria not met"). I inserted three orders, the next poll fired, the
worker logged "Success criteria met", and `process_orders` ran. |
| `wait_for_orders_callable` (`success=lambda count: count >= 3`,
`mode="reschedule"`) | Logged the fallback warning and went to
`up_for_reschedule` without deferring. |
| `wait_for_orders_timeout` (`timeout=30`, condition never met) | The
triggerer polled every 5 s and logged "Trigger cancelled due to timeout". The
task failed with `AirflowSensorTimeout` on try 1, 33 s after it started. |
On Airflow 2.11.1, with wheels built from this branch and a real sqlite
connection, the trigger survived a `serialize()` round trip, polled in a thread
until the condition held, and the sensor succeeded on resume. The callable
sensor poked on the worker. A `fail_on_empty` event raised, or skipped under
`soft_fail`.
--
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]
