InfluxDB3Operator: add deferrable variant
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 9h
- Merged PRs (30d)
- 472
Description
### Description
The `InfluxDB3Operator` added in #58929 currently executes queries synchronously, which means each task holds an Airflow worker slot for the entire duration of the query. For long-running analytics or aggregation queries over InfluxDB 3 clusters, this prevents the worker from picking up other tasks while the query runs on the database.
This issue proposes adding a `deferrable: bool = False` parameter (and a separate trigger class) to `InfluxDB3Operator`, following the pattern already established in:
- `SnowflakeSqlApiOperator` (deferrable variant with `SnowflakeSqlApiTrigger`)
- `BigQueryInsertJobOperator` (deferrable mode with `BigQueryInsertJobTrigger`)
- `RedshiftDataOperator` (deferrable with `RedshiftDataTrigger`)
#### Proposed API
```python
from airflow.providers.influxdb.operators.influxdb3 import InfluxDB3Operator
InfluxDB3Operator(
task_id="aggregate_metrics",
sql="SELECT ... FROM measurement WHERE ...",
influxdb3_conn_id="influxdb3_default",
deferrable=True,
poll_interval=30,
)
```
When `deferrable=True`, `execute()` should:
1. Submit the query to InfluxDB 3 using `InfluxDBClient3.query_async()`, the asyncio coroutine exposed by `influxdb3-python`. (`query_async` was introduced in `influxdb3-python` 0.12.0; the influxdb provider currently declares `influxdb3-python>=0.7.0`, so the deferrable implementation will need to bump the lower bound to `>=0.12.0`.) For very large result sets, the synchronous `query(mode="reader")` Apache Arrow Flight stream is also available if a streaming result interface is preferred.
2. Call `self.defer(trigger=InfluxDB3QueryTrigger(...), method_name="execute_complete")`.
3. The trigger polls the query state on the InfluxDB cluster at `poll_interval` and yields a `TriggerEvent` when the query finishes or fails.
4. `execute_complete()` retrieves the result and pushes it to XCom in the same shape as the sync path.
#### Acceptance criteria
- [ ] New `deferrable: bool` parameter on `InfluxDB3Operator` (default `False` for backward compatibility).
- [ ] New `InfluxDB3QueryTrigger` (or similar) implementing the async polling loop using `influxdb3-python`'s async client.
- [ ] Unit tests covering both `deferrable=True` and `deferrable=False` code paths.
- [ ] Integration test exercising the trigger end-to-end against a local InfluxDB 3 instance (or mocked equivalent if a real one is impractical in CI).
- [ ] Provider documentation updated with a `deferrable=True` example.
#### Scope and tradeoffs
1. `InfluxDBClient3.query_async()` is a `loop.run_in_executor(None, ...)` wrapper around the blocking Apache Arrow Flight calls (see `influxdb3-python/influxdb_client_3/query/query_api.py`). The interface is a proper coroutine and is suitable for use in a Trigger. Concurrency in the Triggerer is bounded by the executor's thread pool rather than by native async IO. Native async over Arrow Flight is a future improvement on the InfluxData client side, not a blocker for this issue. The provider's current `influxdb3-python>=0.7.0` lower bound will need to be bumped to `>=0.12.0`, the release that introduced `query_async`.
2. The deferrable variant frees a worker slot while the query runs, but the result still flows back through the operator and into XCom. This is most useful for long-running queries with small-to-moderate result sets (aggregations, freshness probes, dashboard queries). For very large extracts, the existing `InfluxDB3Hook` + Python task pattern remains the right answer, and this issue does not aim to replace it.
### Use case/motivation
Analytics queries on InfluxDB Cloud Dedicated or InfluxDB 3 Enterprise routinely take minutes for windowed aggregations across high-cardinality measurements. In a constrained worker pool, holding a slot per long-running query limits the parallelism the rest of the DAG fleet can achieve. A deferrable variant releases the worker slot back to the pool the moment the query is submitted to InfluxDB. The Triggerer then polls the query state and resumes the task when the result is ready.
This matches established Airflow patterns and unblocks teams that want to standardise on the new `InfluxDB3Operator` without losing worker-slot efficiency.
### Related issues
- Original InfluxDB 3 support: #58929
- Snowflake deferrable reference: `airflow/providers/snowflake/operators/snowflake.py`, `airflow/providers/snowflake/triggers/snowflake_trigger.py`
- BigQuery deferrable reference: `airflow/providers/google/cloud/operators/bigquery.py`, `airflow/providers/google/cloud/triggers/bigquery.py`
### Are you willing to submit a PR?
- [x] 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)
Contributor guide
Assessment
This issue has not been assessed yet.