You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "arpitrathore (via GitHub)" <gi...@apache.org> on 2026/05/18 11:04:57 UTC
[I] InfluxDB3Operator: add deferrable variant [airflow]
arpitrathore opened a new issue, #67107:
URL: https://github.com/apache/airflow/issues/67107
### 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)
--
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] InfluxDB3Operator: add deferrable variant [airflow]
Posted by "subhramit (via GitHub)" <gi...@apache.org>.
subhramit commented on issue #67107:
URL: https://github.com/apache/airflow/issues/67107#issuecomment-5379697467
> Hi, I am new to this and I would like to take this
Hey, since the project's [policies](https://github.com/apache/airflow/blob/main/contributing-docs/04_how_to_contribute.rst#contribute-code-changes) state (and as also mentioned above) the convention to not expect assignment and directly work on issues, I had already started working on this and will be filing a PR soon.
Just so that your effort isn't duplicated, would you mind picking up something else or perhaps helping in review of this?
--
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] InfluxDB3Operator: add deferrable variant [airflow]
Posted by "codemouli (via GitHub)" <gi...@apache.org>.
codemouli commented on issue #67107:
URL: https://github.com/apache/airflow/issues/67107#issuecomment-5380881693
Sure No problem.
--
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] InfluxDB3Operator: add deferrable variant [airflow]
Posted by "arpitrathore (via GitHub)" <gi...@apache.org>.
arpitrathore commented on issue #67107:
URL: https://github.com/apache/airflow/issues/67107#issuecomment-4505014860
I'd like to take this. Plan: add a `deferrable` parameter on `InfluxDB3Operator` and a new `InfluxDB3QueryTrigger`, modelled on the BigQuery and Snowflake deferrable patterns referenced above. Implementation will use `InfluxDBClient3.query_async()` (introduced in `influxdb3-python` 0.12.0), which will also require bumping the provider's lower bound from `>=0.7.0` to `>=0.12.0`.
Follow-up to my earlier PR #58929. Companion proposal to #67109.
Happy to start once the issue is triaged. If a maintainer can take a quick look and assign it to me, I'll begin the PR.
--
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] InfluxDB3Operator: add deferrable variant [airflow]
Posted by "subhramit (via GitHub)" <gi...@apache.org>.
subhramit commented on issue #67107:
URL: https://github.com/apache/airflow/issues/67107#issuecomment-5381325366
Filed https://github.com/apache/airflow/pull/71976.
Did not follow the exact API shape proposed here (`poll_interval` doesn't apply, since InfluxDB 3's async query has no pollable job state to check). Have added more details in the PR description.
--
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] InfluxDB3Operator: add deferrable variant [airflow]
Posted by "shahar1 (via GitHub)" <gi...@apache.org>.
shahar1 closed issue #67107: InfluxDB3Operator: add deferrable variant
URL: https://github.com/apache/airflow/issues/67107
--
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] InfluxDB3Operator: add deferrable variant [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on issue #67107:
URL: https://github.com/apache/airflow/issues/67107#issuecomment-5365716538
> Happy to start once the issue is triaged. If a maintainer can take a quick look and assign it to me, I'll begin the PR.
Hey, we try to avoid assigning individuals as we found many times assigned people are not always delivering and with issue assigned there is less likelihood someone else will work on the issue.
If you have a fix/feature implementation just raise PR directly. It is much easier to discuss over an actual code proposal.
--
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] InfluxDB3Operator: add deferrable variant [airflow]
Posted by "codemouli (via GitHub)" <gi...@apache.org>.
codemouli commented on issue #67107:
URL: https://github.com/apache/airflow/issues/67107#issuecomment-5373782100
Hi, I am new to this and I would like to take this
--
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