Add deferrable mode to InfluxDB 3 Operator

[subhramit]

Closes #67107
Follow-up to #58929

Add deferrable support to InfluxDB3Operator

When deferrable=True, the operator now defers to a new InfluxDB3QueryTrigger so the worker slot is released while the query runs. The trigger awaits InfluxDBClient3.query_async() and resumes the task with the same JSON-serializable record shape returned by the synchronous path.
Also update provider metadata and documentation to surface the trigger, add a deferrable example, and bump the influxdb3-python minimum version to >=0.12.0 (the release that introduced query_async()).

Note that this deviates from the issue's proposed API as I did not add a "poll_interval"/polling loop. InfluxDB 3's query_async streams the full result over one Arrow Flight call, which means there's no server-side job to poll a status on, unlike Snowflake/BigQuery/Redshift. The trigger awaits the query once and emits a single event.
This is more similar in shape to SQLExecuteQueryTrigger (used by GenericTransfer's deferrable path) - which also has a single await yielding one TriggerEvent with no poll_interval for the same reason:

airflow/providers/common/sql/src/airflow/providers/common/sql/triggers/sql.py

Lines 92 to 102 in 4e4d060

	async def run(self) -> AsyncIterator[TriggerEvent]:
	try:
	self.log.info("Extracting data from %s", self.conn_id)
	self.log.info("Executing: \n %s", self.sql)
	self.log.info("Reading records from %s", self.conn_id)
	results = await self._get_records()
	self.log.info("Reading records from %s done!", self.conn_id)
	self.log.debug("results: %s", results)
	yield TriggerEvent({"status": "success", "results": results})

Have added tests for the new deferrable operator path, trigger serialization/execution, and async hook behavior.

MWE

Tested on a setup of influxdb:3-core with a file object store, two rows written to a home table over the v3 line-protocol endpoint, read back through the influxdb3 CLI:

The async client on its own against the same server, to isolate it from Airflow:

c = InfluxDBClient3(host="http://localhost:8181", token=TOKEN, database="airflow_demo")
print(asyncio.run(c.query_async("SELECT * FROM home")).to_pydict())

Then through the operator - the deferrable task alongside the sync one running the same query, so the two paths could be compared:

blocking = InfluxDB3Operator(task_id="query_blocking", sql=SQL, influxdb3_conn_id=CONN)
deferred = InfluxDB3Operator(task_id="query_deferred", sql=SQL, influxdb3_conn_id=CONN, deferrable=True)
blocking >> deferred

query_deferred deferred to the triggerer, the trigger fired a success event, and the task resumed from QUEUED with the same two records the sync task returned:

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Assisted-by: GPT-5.4 following the guidelines
All code changes done as a result (and also this description) were manually driven, edited & reviewed by me.

[eladkal]

cc @arpitrathore can you take a look?

[eladkal]

left few comments.
Can you please confirm if this was tested against a real influx environment?

[subhramit]

Can you please confirm if this was tested against a real influx environment?

Yeah.
Can give an MWE as well.

Edit - added to the PR description.

[subhramit]

@eladkal I noticed a separate dependency issue while working on this. InfluxDB3Hook.query() and query_async() import pandas at runtime, but the provider doesn't declare a it as an extra or convert missing-pandas imports into an optional-feature error, which means users would hit a ModuleNotFoundError when using those query paths if they don't have pandas installed.

I took the opportunity and covered that in 1b7294b by adding the pandas extra and raising AirflowOptionalProviderFeatureException from those paths when pandas isn’t installed, like how it’s done in

airflow/providers/apache/hive/src/airflow/providers/apache/hive/hooks/hive.py

Lines 1083 to 1087 in ee15456

	import pandas as pd
	except ImportError as e:
	from airflow.providers.common.compat.sdk import AirflowOptionalProviderFeatureException
	raise AirflowOptionalProviderFeatureException(e)

and also in transfers/sql_to_s3.py

Update - this has now been made a hard dependency after #71976 (review) and going over some existing patterns.

[ashb]

This pr also needs to add async get_conn support please(as fetching a connection on a worker requires network traffic so can block)

(Non exhaustive review)

[potiuk]

Re-reviewed after your latest push. I went through every open thread and checked the change against the current code rather than going by GitHub's "outdated" marker, and I've resolved the ten that landed:

• _convert_dataframe_to_records is private, and the operator and trigger now share it instead of duplicating the conversion.
• query_async returns a pd.DataFrame like the sync path, with the record conversion moved into the trigger behind asyncio.to_thread.
• The hasattr(client, "query_async") guard and InfluxDB3AsyncQueryNotAvailableError are both gone.
• The hook uses an async connection accessor rather than thread-offloading get_conn.
• The docs note carries upstream links and no longer discusses poll intervals, and the operator docstring explains the XCom sizing trade-off instead of restating the client version.
• The hook and operator tests pin their mocks with spec_set / autospec.

That is a thorough round of follow-up, thank you. I've left the thread about deferrable=True in the system-test example open, since that one is a question for its author to close rather than me.

One issue remains, and it is in the async connection accessor that came out of that feedback.

aget_connection is Airflow 3 only, but this provider supports apache-airflow>=2.11.0

Details inline on hooks/influxdb3.py:161. In short: common.compat.sdk resolves BaseHook to airflow.sdk on Airflow 3 and to airflow.hooks.base on Airflow 2, and aget_connection was added to the Task SDK BaseHook in #53831 — Airflow 2.11's BaseHook does not have it. So deferrable=True raises AttributeError in the triggerer on the oldest Airflow this provider claims to support.

airflow.providers.common.compat.connection.get_async_connection exists for exactly this, you already depend on it via common-compat>=1.8.0, and sftp/hooks/sftp.py:829 uses it the same way. It should be a one-line change.

Smaller observations

• The pandas import guard is duplicated verbatim between query() and query_async(). A small module-level helper would keep the two messages from drifting.
• _create_client(self, connection) is the only unannotated signature in the file — connection: Connection would match the rest.
• Declaring pandas as an extra is a real improvement over the undeclared hard dependency it replaced, but InfluxDB3Operator's only code path needs pandas, so a plain pip install apache-airflow-providers-influxdb now yields an operator that raises at runtime. Worth a conscious call on whether it should be a hard dependency instead — unless influxdb3-python already pulls pandas in for mode="pandas", in which case the extra is belt-and-braces and this is moot. I could not verify that offline; you'll know.
• Minor note on the shape rather than a request: _convert_dataframe_to_records is a private name now imported by two other modules, which is the compromise between "make it private" and "don't duplicate it". It works, and I would not hold the PR for it — but if you revisit this area, a small internal module would express the intent better than a leading underscore that three files reach across.

For what it's worth, I checked AirflowOptionalProviderFeatureException being raised at task runtime and decided it is fine — presto/hooks/presto.py:160 does the same thing for a missing pandas, so it matches established practice in the repo.

---

This review was drafted by an AI-assisted tool and
confirmed by an Airflow maintainer. The findings
below are observations, not blockers; an Airflow
maintainer — a real person — will take the next look at the
PR. If you think a finding is mis-applied, please reply on
the PR and a maintainer will weigh in.

More on how Airflow handles maintainer review:
contributing-docs/05_pull_requests.rst.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting