Add deferrable mode to InfluxDB3Operator

providers/influxdb/README.rst

@@ -56,7 +56,7 @@ PIP package                                 Version required
 ``apache-airflow``                          ``>=2.11.0``
 ``apache-airflow-providers-common-compat``  ``>=1.8.0``
 ``influxdb-client``                         ``>=1.19.0``
-``influxdb3-python``                        ``>=0.7.0``
+``influxdb3-python``                        ``>=0.12.0``
 ``requests``                                ``>=2.32.0,<3``
 ==========================================  ==================
 

providers/influxdb/docs/index.rst

@@ -104,7 +104,7 @@ PIP package                                 Version required
 ``apache-airflow``                          ``>=2.11.0``
 ``apache-airflow-providers-common-compat``  ``>=1.8.0``
 ``influxdb-client``                         ``>=1.19.0``
-``influxdb3-python``                        ``>=0.7.0``
+``influxdb3-python``                        ``>=0.12.0``
 ``requests``                                ``>=2.32.0,<3``
 ==========================================  ==================
 

providers/influxdb/docs/operators/index.rst

@@ -46,3 +46,24 @@ Example usage:
     :language: python
     :start-after: [START howto_operator_influxdb3]
     :end-before: [END howto_operator_influxdb3]
+
+Deferrable mode
+^^^^^^^^^^^^^^^
+
+Set ``deferrable=True`` to release the worker slot while the query runs. The task is resumed by the
+:class:`~airflow.providers.influxdb.triggers.influxdb3.InfluxDB3QueryTrigger` once results are ready.
+
+.. exampleinclude:: /../../influxdb/tests/system/influxdb/example_influxdb3.py
+    :language: python
+    :start-after: [START howto_operator_influxdb3_deferrable]
+    :end-before: [END howto_operator_influxdb3_deferrable]
+
+.. note::
+
+    InfluxDB 3 streams query results over a single Arrow Flight call rather than exposing a job that
+    can be polled, so the trigger awaits the query once instead of polling at an interval, and there
+    is no ``poll_interval`` parameter. Results still travel back through XCom, so deferring is most
+    useful for long-running queries with small-to-moderate result sets. For very large extracts,
+    keep using :class:`~airflow.providers.influxdb.hooks.influxdb3.InfluxDB3Hook` from a Python task.
+
+    Deferrable mode requires ``influxdb3-python>=0.12.0`` and a running ``triggerer``.

providers/influxdb/provider.yaml

@@ -90,6 +90,11 @@ operators:
     python-modules:
       - airflow.providers.influxdb.operators.influxdb3
 
+triggers:
+  - integration-name: InfluxDB 3
+    python-modules:
+      - airflow.providers.influxdb.triggers.influxdb3
+
 connection-types:
   - hook-class-name: airflow.providers.influxdb.hooks.influxdb.InfluxDBHook
     hook-name: "Influxdb"

providers/influxdb/pyproject.toml

@@ -62,7 +62,7 @@ dependencies = [
     "apache-airflow>=2.11.0",
     "apache-airflow-providers-common-compat>=1.8.0",
     "influxdb-client>=1.19.0",
-    "influxdb3-python>=0.7.0",
+    "influxdb3-python>=0.12.0",
     "requests>=2.32.0,<3",
 ]
 

providers/influxdb/src/airflow/providers/influxdb/get_provider_info.py

@@ -56,6 +56,12 @@ def get_provider_info():
                 "python-modules": ["airflow.providers.influxdb.operators.influxdb3"],
             },
         ],
+        "triggers": [
+            {
+                "integration-name": "InfluxDB 3",
+                "python-modules": ["airflow.providers.influxdb.triggers.influxdb3"],
+            }
+        ],
         "connection-types": [
             {
                 "hook-class-name": "airflow.providers.influxdb.hooks.influxdb.InfluxDBHook",

providers/influxdb/src/airflow/providers/influxdb/hooks/influxdb3.py

@@ -24,6 +24,7 @@
 
 from __future__ import annotations
 
+import json
 from typing import TYPE_CHECKING, Any
 
 try:
@@ -49,6 +50,15 @@
     from airflow.models import Connection
 
 
+class InfluxDB3AsyncQueryNotAvailableError(RuntimeError):
+    """Raised when the installed InfluxDB 3 client lacks async query support."""
+
+
+def convert_dataframe_to_records(dataframe: pd.DataFrame) -> list[dict[str, Any]]:
+    """Convert a query result DataFrame into a JSON-serializable list of dictionaries."""
+    return json.loads(dataframe.to_json(orient="records", date_format="iso"))
+
+
 class InfluxDB3Hook(BaseHook):
     """
     Interact with InfluxDB 3.x (Core/Enterprise/Cloud Dedicated).
@@ -205,6 +215,39 @@ def query(self, query: str) -> pd.DataFrame:
 
         return result
 
+    async def query_async(self, query: str) -> list[dict[str, Any]]:
+        """
+        Run a SQL query from the triggerer and return JSON-serializable records.
+
+        ``InfluxDBClient3.query_async`` runs the blocking Arrow Flight calls in the event
+        loop's default executor, so awaiting it does not block the triggerer. It is a plain
+        coroutine that resolves once the whole result stream has been read -- InfluxDB 3 has
+        no submit-then-poll query API, so there is nothing to poll in between.
+
+        Requires ``influxdb3-python>=0.12.0``, the release that introduced ``query_async``.
+
+        :param query: SQL query string
+        :return: List of dictionaries representing query results
+        """
+        client = self.get_conn()
+        if not hasattr(client, "query_async"):
+            raise InfluxDB3AsyncQueryNotAvailableError(
+                "Deferrable mode requires influxdb3-python>=0.12.0, which introduced "
+                "InfluxDBClient3.query_async(). Upgrade with: pip install 'influxdb3-python>=0.12.0'"
+            )
+
+        import pandas as pd
+
+        result = await client.query_async(query=query, language="sql", mode="pandas")
+
+        if not isinstance(result, pd.DataFrame):
+            raise ValueError(
+                f"Query did not return a DataFrame. "
+                f"Result type: {type(result).__module__}.{type(result).__name__}"
+            )
+
+        return convert_dataframe_to_records(result)
+
     def write(
         self,
         measurement: str,

providers/influxdb/src/airflow/providers/influxdb/operators/influxdb3.py

@@ -19,12 +19,12 @@
 
 from __future__ import annotations
 
-import json
 from collections.abc import Sequence
 from typing import TYPE_CHECKING, Any
 
-from airflow.providers.common.compat.sdk import BaseOperator
-from airflow.providers.influxdb.hooks.influxdb3 import InfluxDB3Hook
+from airflow.providers.common.compat.sdk import BaseOperator, conf
+from airflow.providers.influxdb.hooks.influxdb3 import InfluxDB3Hook, convert_dataframe_to_records
+from airflow.providers.influxdb.triggers.influxdb3 import InfluxDB3QueryTrigger
 
 if TYPE_CHECKING:
     from airflow.sdk.definitions.context import Context
@@ -40,6 +40,11 @@ class InfluxDB3Operator(BaseOperator):
 
     :param sql: The SQL query to be executed
     :param influxdb3_conn_id: Reference to :ref:`InfluxDB 3 connection id <howto/connection:influxdb3>`.
+    :param deferrable: Run the query from the triggerer instead of holding a worker slot for its
+        duration. Requires ``influxdb3-python>=0.12.0``. Note that InfluxDB 3 streams results over
+        Arrow Flight rather than exposing a job that can be polled, so the whole result set still flows back
+        through XCom -- deferring helps with long-running queries returning modest result sets
+        (aggregations, freshness probes), not with very large extracts.
     """
 
     template_fields: Sequence[str] = ("sql",)
@@ -49,24 +54,50 @@ def __init__(
         *,
         sql: str,
         influxdb3_conn_id: str = "influxdb3_default",
+        deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False),
         **kwargs,
     ) -> None:
         super().__init__(**kwargs)
         self.influxdb3_conn_id = influxdb3_conn_id
         self.sql = sql
+        self.deferrable = deferrable
 
-    def execute(self, context: Context) -> list[dict[str, Any]]:
+    def execute(self, context: Context) -> list[dict[str, Any]] | None:
         """
         Execute SQL query and return results as JSON-serializable list of dictionaries.
 
         :param context: Airflow context
-        :return: List of dictionaries representing query results
+        :return: List of dictionaries representing query results, or ``None`` when deferring
         """
         self.log.info("Executing SQL query: %s", self.sql)
+
+        if self.deferrable:
+            self.defer(
+                timeout=self.execution_timeout,
+                trigger=InfluxDB3QueryTrigger(
+                    sql=self.sql,
+                    influxdb3_conn_id=self.influxdb3_conn_id,
+                ),
+                method_name="execute_complete",
+            )
+
         hook = InfluxDB3Hook(conn_id=self.influxdb3_conn_id)
         result = hook.query(self.sql)
 
         self.log.info("Query executed successfully. Rows returned: %d", len(result))
+        return convert_dataframe_to_records(result)
+
+    def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> list[dict[str, Any]]:
+        """Return the query results produced by :class:`InfluxDB3QueryTrigger`."""
+        if event is None:
+            raise RuntimeError("InfluxDB 3 query did not return an event")
+
+        status = event.get("status")
+        if status == "error":
+            raise RuntimeError(event.get("message", "InfluxDB 3 query failed"))
+        if status != "success":
+            raise RuntimeError(f"InfluxDB 3 query returned unexpected status: {status!r}")
 
-        json_str = result.to_json(orient="records", date_format="iso")
-        return json.loads(json_str)
+        records = event["records"]
+        self.log.info("Query executed successfully. Rows returned: %d", len(records))
+        return records

providers/influxdb/src/airflow/providers/influxdb/triggers/__init__.py

@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.

providers/influxdb/src/airflow/providers/influxdb/triggers/influxdb3.py

@@ -0,0 +1,75 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Trigger for running InfluxDB 3.x SQL queries from the triggerer."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+from airflow.providers.influxdb.hooks.influxdb3 import InfluxDB3Hook
+from airflow.triggers.base import BaseTrigger, TriggerEvent
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncIterator
+
+
+class InfluxDB3QueryTrigger(BaseTrigger):
+    """
+    Run a SQL query against InfluxDB 3.x without occupying a worker slot.
+
+    InfluxDB 3 executes queries over a single Apache Arrow Flight stream: there is no
+    server-side job to submit and then poll for completion, and therefore no query state
+    and no poll interval. The trigger instead awaits the query coroutine once and emits a
+    single event when the stream has been fully read.
+
+    ``influxdb3-python`` implements ``query_async`` by running the blocking Flight calls in
+    the event loop's default executor, so concurrency in the triggerer is bounded by that
+    thread pool rather than by native async IO. This is a limitation of the upstream client,
+    not of this trigger.
+
+    :param sql: The SQL query to be executed.
+    :param influxdb3_conn_id: Reference to :ref:`InfluxDB 3 connection id <howto/connection:influxdb3>`.
+    """
+
+    def __init__(
+        self,
+        sql: str,
+        influxdb3_conn_id: str = "influxdb3_default",
+    ) -> None:
+        super().__init__()
+        self.sql = sql
+        self.influxdb3_conn_id = influxdb3_conn_id
+
+    def serialize(self) -> tuple[str, dict[str, Any]]:
+        return (
+            "airflow.providers.influxdb.triggers.influxdb3.InfluxDB3QueryTrigger",
+            {
+                "sql": self.sql,
+                "influxdb3_conn_id": self.influxdb3_conn_id,
+            },
+        )
+
+    async def run(self) -> AsyncIterator[TriggerEvent]:
+        hook = InfluxDB3Hook(conn_id=self.influxdb3_conn_id)
+        try:
+            records = await hook.query_async(self.sql)
+        except Exception as error:
+            self.log.exception("InfluxDB 3 query failed in trigger")
+            yield TriggerEvent({"status": "error", "message": str(error)})
+            return
+
+        yield TriggerEvent({"status": "success", "records": records})

providers/influxdb/tests/system/influxdb/example_influxdb3.py

@@ -57,6 +57,15 @@ def write_to_influxdb3():
 )
 # [END howto_operator_influxdb3]
 
+# [START howto_operator_influxdb3_deferrable]
+deferrable_query_task = InfluxDB3Operator(
+    task_id="query_data_deferrable",
+    sql="SELECT * FROM \"temperature\" WHERE time > now() - INTERVAL '1 hour'",
+    influxdb3_conn_id="influxdb3_default",
+    deferrable=True,
+)
+# [END howto_operator_influxdb3_deferrable]
+
 ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")
 DAG_ID = "influxdb3_example_dag"
 
@@ -68,7 +77,7 @@ def write_to_influxdb3():
     tags=["example", "influxdb3"],
 ) as dag:
     write_task = write_to_influxdb3()
-    write_task >> query_task
+    write_task >> [query_task, deferrable_query_task]
 
     from tests_common.test_utils.watcher import watcher
 

providers/influxdb/tests/unit/influxdb/hooks/test_influxdb3.py

@@ -21,7 +21,7 @@
 import pytest
 
 from airflow.models import Connection
-from airflow.providers.influxdb.hooks.influxdb3 import InfluxDB3Hook
+from airflow.providers.influxdb.hooks.influxdb3 import InfluxDB3AsyncQueryNotAvailableError, InfluxDB3Hook
 
 
 class TestInfluxDB3Hook:
@@ -85,6 +85,46 @@ def test_query(self):
         assert isinstance(result, pd.DataFrame)
         assert len(result) == 2
 
+    @pytest.mark.asyncio
+    async def test_query_async(self):
+        """Test async query with InfluxDB 3.x."""
+        pd = pytest.importorskip("pandas")
+
+        self.influxdb3_hook.client = mock.Mock()
+        mock_df = pd.DataFrame({"col1": [1, 2], "col2": [3, 4]})
+        self.influxdb3_hook.client.query_async = mock.AsyncMock(return_value=mock_df)
+        self.influxdb3_hook.get_conn = mock.Mock(return_value=self.influxdb3_hook.client)
+
+        influxdb_query = 'SELECT "duration" FROM "pyexample"'
+        result = await self.influxdb3_hook.query_async(influxdb_query)
+
+        self.influxdb3_hook.get_conn.assert_called()
+        self.influxdb3_hook.client.query_async.assert_awaited_once_with(
+            query=influxdb_query, language="sql", mode="pandas"
+        )
+        assert result == [{"col1": 1, "col2": 3}, {"col1": 2, "col2": 4}]
+
+    @pytest.mark.asyncio
+    async def test_query_async_requires_supported_client(self):
+        """Deferrable execution requires an InfluxDB client exposing query_async."""
+        self.influxdb3_hook.client = mock.Mock(spec=[])
+        self.influxdb3_hook.get_conn = mock.Mock(return_value=self.influxdb3_hook.client)
+
+        with pytest.raises(InfluxDB3AsyncQueryNotAvailableError, match="influxdb3-python>=0.12.0"):
+            await self.influxdb3_hook.query_async('SELECT "duration" FROM "pyexample"')
+
+    @pytest.mark.asyncio
+    async def test_query_async_requires_dataframe_result(self):
+        """Async query results must resolve to a pandas DataFrame."""
+        pytest.importorskip("pandas")
+
+        self.influxdb3_hook.client = mock.Mock()
+        self.influxdb3_hook.client.query_async = mock.AsyncMock(return_value=[{"col1": 1}])
+        self.influxdb3_hook.get_conn = mock.Mock(return_value=self.influxdb3_hook.client)
+
+        with pytest.raises(ValueError, match="did not return a DataFrame"):
+            await self.influxdb3_hook.query_async('SELECT "duration" FROM "pyexample"')
+
     def test_write(self):
         """Test write with InfluxDB 3.x."""
         self.influxdb3_hook.client = mock.Mock()