This is an automated email from the ASF dual-hosted git repository.
amoghrajesh pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new f5bd40ed261 Make `durable` reach `default_args` and warn when set
below Airflow 3.3 (#71531)
f5bd40ed261 is described below
commit f5bd40ed261c127c7666c0422de648a7f519ad01
Author: Amogh Desai <[email protected]>
AuthorDate: Fri Aug 14 11:22:01 2026 +0530
Make `durable` reach `default_args` and warn when set below Airflow 3.3
(#71531)
---
.../docs/operators/redshift/redshift_data.rst | 4 +++
.../amazon/aws/operators/redshift_data.py | 24 +++++++++++++--
.../amazon/aws/operators/test_redshift_data.py | 34 +++++++++++++++++++++-
.../providers/apache/livy/operators/livy.py | 24 +++++++++++++--
.../tests/unit/apache/livy/operators/test_livy.py | 28 +++++++++++++++++-
.../apache/spark/operators/spark_submit.py | 5 ++++
.../apache/spark/operators/test_spark_submit.py | 4 +++
providers/databricks/docs/operators/run_now.rst | 4 +++
providers/databricks/docs/operators/submit_run.rst | 4 +++
.../providers/databricks/operators/databricks.py | 29 ++++++++++++++++--
.../unit/databricks/operators/test_databricks.py | 28 ++++++++++++++++++
providers/google/docs/operators/cloud/bigquery.rst | 8 ++---
.../providers/google/cloud/operators/bigquery.py | 23 +++++++++++++--
.../unit/google/cloud/operators/test_bigquery.py | 22 ++++++++++++++
providers/snowflake/docs/operators/snowflake.rst | 4 +++
.../providers/snowflake/operators/snowflake.py | 24 +++++++++++++--
.../unit/snowflake/operators/test_snowflake.py | 22 ++++++++++++++
17 files changed, 275 insertions(+), 16 deletions(-)
diff --git a/providers/amazon/docs/operators/redshift/redshift_data.rst
b/providers/amazon/docs/operators/redshift/redshift_data.rst
index 35d3bf19a07..f4e100e3bdd 100644
--- a/providers/amazon/docs/operators/redshift/redshift_data.rst
+++ b/providers/amazon/docs/operators/redshift/redshift_data.rst
@@ -125,6 +125,10 @@ Durable execution applies to the synchronous path. When
``deferrable=True`` is s
already tracks the statement across the wait, so deferrable mode takes
precedence and ``durable``
has no effect.
+Durable execution requires Airflow 3.3 or newer, since it relies on the task
state store. Below
+3.3, ``durable`` has no effect either way: setting it explicitly only emits a
warning, and the
+operator always submits fresh SQL on retry, exactly as before this feature
existed.
+
Reference
---------
diff --git
a/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_data.py
b/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_data.py
index b81403be214..f385fcc1761 100644
---
a/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_data.py
+++
b/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_data.py
@@ -17,6 +17,7 @@
# under the License.
from __future__ import annotations
+import warnings
from typing import TYPE_CHECKING, Any, cast
import botocore.exceptions
@@ -28,6 +29,20 @@ from airflow.providers.amazon.aws.utils import
validate_execute_complete_event
from airflow.providers.amazon.aws.utils.mixins import aws_template_fields
from airflow.providers.common.compat.sdk import AirflowException, conf
+_DURABLE_UNSET = object()
+
+
+def _warn_and_disable_durable_pre_3_3(durable: Any) -> bool:
+ """Shared by the <3.3 compat stub: durable has no effect below 3.3, warn
if it was set."""
+ if durable is not _DURABLE_UNSET:
+ warnings.warn(
+ "`durable` has no effect on Airflow versions below 3.3.",
+ UserWarning,
+ stacklevel=3,
+ )
+ return False
+
+
try:
from airflow.sdk import ResumableJobMixin
except ImportError:
@@ -37,9 +52,9 @@ except ImportError:
external_id_key: str = "redshift_statement_id"
- def __init__(self, *, durable: bool = True, **kwargs: Any) -> None:
+ def __init__(self, *, durable: Any = _DURABLE_UNSET, **kwargs: Any) ->
None:
super().__init__(**kwargs)
- self.durable = durable
+ self.durable = _warn_and_disable_durable_pre_3_3(durable)
def execute_resumable(self, context):
external_id = self.submit_job(context)
@@ -135,8 +150,13 @@ class RedshiftDataOperator(ResumableJobMixin,
AwsBaseOperator[RedshiftDataHook])
session_id: str | None = None,
session_keep_alive_seconds: int | None = None,
cancel_on_kill: bool = True,
+ durable: bool | None = None,
**kwargs,
) -> None:
+ # Named here (not left to **kwargs) so default_args reaches it on every
+ # supported Airflow version.
+ if durable is not None:
+ kwargs["durable"] = durable
super().__init__(**kwargs)
self.database = database
self.sql = sql
diff --git
a/providers/amazon/tests/unit/amazon/aws/operators/test_redshift_data.py
b/providers/amazon/tests/unit/amazon/aws/operators/test_redshift_data.py
index 1b33c7ab183..7831cf51021 100644
--- a/providers/amazon/tests/unit/amazon/aws/operators/test_redshift_data.py
+++ b/providers/amazon/tests/unit/amazon/aws/operators/test_redshift_data.py
@@ -17,13 +17,20 @@
# under the License.
from __future__ import annotations
+import warnings
+from datetime import datetime
from unittest import mock
import botocore.exceptions
import pytest
+from airflow.models.dag import DAG
from airflow.providers.amazon.aws.hooks.redshift_data import
QueryExecutionOutput
-from airflow.providers.amazon.aws.operators.redshift_data import
RedshiftDataOperator
+from airflow.providers.amazon.aws.operators.redshift_data import (
+ _DURABLE_UNSET,
+ RedshiftDataOperator,
+ _warn_and_disable_durable_pre_3_3,
+)
from airflow.providers.amazon.aws.triggers.redshift_data import
RedshiftDataTrigger
from airflow.providers.common.compat.sdk import AirflowException, TaskDeferred
@@ -721,3 +728,28 @@ class TestRedshiftDataOperatorDurable:
assert operator.is_job_succeeded("FINISHED") is True
assert operator.is_job_succeeded("STARTED") is False
+
+ def test_default_args_durable_reaches_operator(self):
+ with DAG(
+ dag_id="test_redshift_data_durable_default_args",
+ schedule=None,
+ start_date=datetime(2024, 1, 1),
+ default_args={"durable": False},
+ ):
+ operator = self._make_operator()
+ assert operator.durable is False
+
+
+class TestWarnAndDisableDurableAirflowPre3_3:
+ def test_no_warning_when_unset(self):
+ with warnings.catch_warnings(record=True) as caught:
+ warnings.simplefilter("always")
+ result = _warn_and_disable_durable_pre_3_3(_DURABLE_UNSET)
+ assert result is False
+ assert caught == []
+
+ @pytest.mark.parametrize("value", [True, False])
+ def test_warns_and_disables_when_explicitly_set(self, value):
+ with pytest.warns(UserWarning, match="durable.*no effect"):
+ result = _warn_and_disable_durable_pre_3_3(value)
+ assert result is False
diff --git
a/providers/apache/livy/src/airflow/providers/apache/livy/operators/livy.py
b/providers/apache/livy/src/airflow/providers/apache/livy/operators/livy.py
index 9f0a6b08506..b95b63cbd00 100644
--- a/providers/apache/livy/src/airflow/providers/apache/livy/operators/livy.py
+++ b/providers/apache/livy/src/airflow/providers/apache/livy/operators/livy.py
@@ -17,6 +17,7 @@
from __future__ import annotations
import time
+import warnings
from collections.abc import Sequence
from functools import cached_property
from typing import TYPE_CHECKING, Any, cast
@@ -29,6 +30,20 @@ from airflow.providers.common.compat.openlineage.utils.spark
import (
)
from airflow.providers.common.compat.sdk import AirflowException,
BaseOperator, conf
+_DURABLE_UNSET = object()
+
+
+def _warn_and_disable_durable_pre_3_3(durable: Any) -> bool:
+ """Shared by the <3.3 compat stub: durable has no effect below 3.3, warn
if it was set."""
+ if durable is not _DURABLE_UNSET:
+ warnings.warn(
+ "`durable` has no effect on Airflow versions below 3.3.",
+ UserWarning,
+ stacklevel=3,
+ )
+ return False
+
+
# ResumableJobMixin ships in airflow.sdk, which only exists on Airflow 3,
while this provider
# still targets apache-airflow>=2.11. Guard the import and fall back to a stub
on Airflow 2;
# drop the fallback once the provider's minimum Airflow version is >=3.0.
@@ -41,10 +56,10 @@ except ImportError:
external_id_key: str = "livy_batch_id"
- def __init__(self, *, durable: bool = True, **kwargs: Any) -> None:
+ def __init__(self, *, durable: Any = _DURABLE_UNSET, **kwargs: Any) ->
None:
# Swallow ``durable`` so it doesn't reach BaseOperator; crash
recovery is a no-op here.
super().__init__(**kwargs)
- self.durable = durable
+ self.durable = _warn_and_disable_durable_pre_3_3(durable)
def execute_resumable(self, context):
external_id = self.submit_job(context)
@@ -130,8 +145,13 @@ class LivyOperator(ResumableJobMixin, BaseOperator):
openlineage_inject_transport_info: bool = conf.getboolean(
"openlineage", "spark_inject_transport_info", fallback=False
),
+ durable: bool | None = None,
**kwargs: Any,
) -> None:
+ # Named here (not left to **kwargs) so default_args reaches it on every
+ # supported Airflow version.
+ if durable is not None:
+ kwargs["durable"] = durable
super().__init__(**kwargs)
if conf is None:
diff --git
a/providers/apache/livy/tests/unit/apache/livy/operators/test_livy.py
b/providers/apache/livy/tests/unit/apache/livy/operators/test_livy.py
index 2914b85166f..7fb6d22b459 100644
--- a/providers/apache/livy/tests/unit/apache/livy/operators/test_livy.py
+++ b/providers/apache/livy/tests/unit/apache/livy/operators/test_livy.py
@@ -17,6 +17,7 @@
from __future__ import annotations
import logging
+import warnings
from typing import Any
from unittest.mock import MagicMock, patch
@@ -25,7 +26,11 @@ import pytest
from airflow.models import Connection
from airflow.models.dag import DAG
from airflow.providers.apache.livy.hooks.livy import BatchState, LivyHook
-from airflow.providers.apache.livy.operators.livy import LivyOperator
+from airflow.providers.apache.livy.operators.livy import (
+ _DURABLE_UNSET,
+ LivyOperator,
+ _warn_and_disable_durable_pre_3_3,
+)
from airflow.providers.common.compat.sdk import AirflowException, timezone
from tests_common.test_utils.version_compat import AIRFLOW_V_3_3_PLUS
@@ -714,6 +719,12 @@ class TestLivyOperatorResumable:
assert operator.is_job_succeeded("success") is True
assert operator.is_job_succeeded("dead") is False
+ def test_default_args_durable_reaches_operator(self):
+ operator = LivyOperator(
+ task_id="livy_default_args", file="sparkapp.jar",
default_args={"durable": False}
+ )
+ assert operator.durable is False
+
def test_get_job_status_reads_batch_state_value(self):
operator = self._make_operator()
hook = self._make_hook()
@@ -731,3 +742,18 @@ class TestLivyOperatorResumable:
operator.poll_until_complete(55, {})
assert operator._batch_id == 55
+
+
+class TestWarnAndDisableDurableAirflowPre3_3:
+ def test_no_warning_when_unset(self):
+ with warnings.catch_warnings(record=True) as caught:
+ warnings.simplefilter("always")
+ result = _warn_and_disable_durable_pre_3_3(_DURABLE_UNSET)
+ assert result is False
+ assert caught == []
+
+ @pytest.mark.parametrize("value", [True, False])
+ def test_warns_and_disables_when_explicitly_set(self, value):
+ with pytest.warns(UserWarning, match="durable.*no effect"):
+ result = _warn_and_disable_durable_pre_3_3(value)
+ assert result is False
diff --git
a/providers/apache/spark/src/airflow/providers/apache/spark/operators/spark_submit.py
b/providers/apache/spark/src/airflow/providers/apache/spark/operators/spark_submit.py
index d70729cb27f..48d70155326 100644
---
a/providers/apache/spark/src/airflow/providers/apache/spark/operators/spark_submit.py
+++
b/providers/apache/spark/src/airflow/providers/apache/spark/operators/spark_submit.py
@@ -400,6 +400,7 @@ class SparkSubmitOperator(ResumableJobMixin, BaseOperator):
"openlineage", "spark_inject_transport_info", fallback=False
),
reconnect_on_retry: bool | None = None,
+ durable: bool | None = None,
**kwargs: Any,
) -> None:
if reconnect_on_retry is not None:
@@ -409,6 +410,10 @@ class SparkSubmitOperator(ResumableJobMixin, BaseOperator):
stacklevel=2,
)
kwargs.setdefault("durable", reconnect_on_retry)
+ # Named here (not left to **kwargs) so default_args={"durable": ...}
reaches it on every
+ # supported Airflow version; applied after reconnect_on_retry so an
explicit durable wins.
+ if durable is not None:
+ kwargs["durable"] = durable
super().__init__(**kwargs)
self.application = application
self.conf = conf
diff --git
a/providers/apache/spark/tests/unit/apache/spark/operators/test_spark_submit.py
b/providers/apache/spark/tests/unit/apache/spark/operators/test_spark_submit.py
index 953af3408d0..3b525582ed7 100644
---
a/providers/apache/spark/tests/unit/apache/spark/operators/test_spark_submit.py
+++
b/providers/apache/spark/tests/unit/apache/spark/operators/test_spark_submit.py
@@ -601,6 +601,10 @@ class TestSparkSubmitOperatorResumable:
assert "reconnect_on_retry" in str(w[0].message)
assert operator.durable is False
+ def test_default_args_durable_reaches_operator(self):
+ operator = self._make_operator(default_args={"durable": False})
+ assert operator.durable is False
+
def test_durable_false_submits_fresh_and_polls(self):
operator = self._make_operator(durable=False)
operator._hook = self._make_hook(should_track=True)
diff --git a/providers/databricks/docs/operators/run_now.rst
b/providers/databricks/docs/operators/run_now.rst
index ff34ccf6242..0b53f659b8a 100644
--- a/providers/databricks/docs/operators/run_now.rst
+++ b/providers/databricks/docs/operators/run_now.rst
@@ -136,6 +136,10 @@ Durable execution applies to the synchronous path. When
``deferrable=True`` is s
already tracks the run across the wait, so deferrable mode takes precedence
and ``durable`` has no
effect.
+Durable execution requires Airflow 3.3 or newer, since it relies on the task
state store. Below
+3.3, ``durable`` has no effect either way: setting it explicitly only emits a
warning, and the
+operator always triggers a fresh run on retry, exactly as before this feature
existed.
+
DatabricksRunNowDeferrableOperator
==================================
diff --git a/providers/databricks/docs/operators/submit_run.rst
b/providers/databricks/docs/operators/submit_run.rst
index 76e1e15fcb8..82919d982c2 100644
--- a/providers/databricks/docs/operators/submit_run.rst
+++ b/providers/databricks/docs/operators/submit_run.rst
@@ -215,6 +215,10 @@ Durable execution applies to the synchronous path. When
``deferrable=True`` is s
Triggerer already tracks the run across the wait, so deferrable mode takes
precedence and
``durable`` has no effect.
+Durable execution requires Airflow 3.3 or newer, since it relies on the task
state store. Below
+3.3, ``durable`` has no effect either way: setting it explicitly only emits a
warning, and the
+operator always submits a fresh run on retry, exactly as before this feature
existed.
+
DatabricksSubmitRunDeferrableOperator
=====================================
diff --git
a/providers/databricks/src/airflow/providers/databricks/operators/databricks.py
b/providers/databricks/src/airflow/providers/databricks/operators/databricks.py
index d820b099c8d..34a0c47e511 100644
---
a/providers/databricks/src/airflow/providers/databricks/operators/databricks.py
+++
b/providers/databricks/src/airflow/providers/databricks/operators/databricks.py
@@ -24,6 +24,7 @@ import copy
import hashlib
import json as json_utils
import time
+import warnings
from abc import ABC, abstractmethod
from collections.abc import Mapping, Sequence
from functools import cached_property
@@ -68,6 +69,20 @@ if TYPE_CHECKING:
from airflow.sdk import TaskGroup
from airflow.sdk.types import Context, Logger
+_DURABLE_UNSET = object()
+
+
+def _warn_and_disable_durable_pre_3_3(durable: Any) -> bool:
+ """Shared by the <3.3 compat stub: durable has no effect below 3.3, warn
if it was set."""
+ if durable is not _DURABLE_UNSET:
+ warnings.warn(
+ "`durable` has no effect on Airflow versions below 3.3.",
+ UserWarning,
+ stacklevel=3,
+ )
+ return False
+
+
try:
from airflow.sdk import ResumableJobMixin
except ImportError:
@@ -77,9 +92,9 @@ except ImportError:
external_id_key: str = "databricks_run_id"
- def __init__(self, *, durable: bool = True, **kwargs: Any) -> None:
+ def __init__(self, *, durable: Any = _DURABLE_UNSET, **kwargs: Any) ->
None:
super().__init__(**kwargs)
- self.durable = durable
+ self.durable = _warn_and_disable_durable_pre_3_3(durable)
def execute_resumable(self, context):
external_id = self.submit_job(context)
@@ -776,9 +791,14 @@ class DatabricksSubmitRunOperator(ResumableJobMixin,
BaseOperator):
openlineage_inject_transport_info: bool = conf.getboolean(
"openlineage", "spark_inject_transport_info", fallback=False
),
+ durable: bool | None = None,
**kwargs,
) -> None:
"""Create a new ``DatabricksSubmitRunOperator``."""
+ # Named here (not left to **kwargs) so default_args reaches it on every
+ # supported Airflow version.
+ if durable is not None:
+ kwargs["durable"] = durable
super().__init__(**kwargs)
self.json = json
self.tasks = tasks
@@ -1246,9 +1266,14 @@ class DatabricksRunNowOperator(ResumableJobMixin,
BaseOperator):
databricks_repair_reason_new_settings: dict[str, Any] | None = None,
cancel_previous_runs: bool = False,
forward_dag_params: bool = True,
+ durable: bool | None = None,
**kwargs,
) -> None:
"""Create a new ``DatabricksRunNowOperator``."""
+ # Named here (not left to **kwargs) so default_args reaches it on every
+ # supported Airflow version.
+ if durable is not None:
+ kwargs["durable"] = durable
super().__init__(**kwargs)
self.json = json
self.job_id = job_id
diff --git
a/providers/databricks/tests/unit/databricks/operators/test_databricks.py
b/providers/databricks/tests/unit/databricks/operators/test_databricks.py
index 0f190da3661..a581bc117b3 100644
--- a/providers/databricks/tests/unit/databricks/operators/test_databricks.py
+++ b/providers/databricks/tests/unit/databricks/operators/test_databricks.py
@@ -19,6 +19,7 @@ from __future__ import annotations
import copy
import hashlib
+import warnings
from datetime import datetime, timedelta
from typing import Any
from unittest import mock
@@ -40,6 +41,7 @@ from airflow.providers.common.compat.sdk import
AirflowException, BaseOperator,
from airflow.providers.databricks.exceptions import DatabricksApiError
from airflow.providers.databricks.hooks.databricks import RunState,
SQLStatementState
from airflow.providers.databricks.operators.databricks import (
+ _DURABLE_UNSET,
DatabricksCreateJobsOperator,
DatabricksNotebookOperator,
DatabricksRunNowOperator,
@@ -47,6 +49,7 @@ from airflow.providers.databricks.operators.databricks import
(
DatabricksSubmitRunOperator,
DatabricksTaskBaseOperator,
DatabricksTaskOperator,
+ _warn_and_disable_durable_pre_3_3,
)
from airflow.providers.databricks.triggers.databricks import (
DatabricksExecutionTrigger,
@@ -1889,6 +1892,12 @@ class TestDatabricksSubmitRunOperatorDurable:
op = DatabricksSubmitRunOperator(task_id=TASK_ID,
json={"notebook_task": NOTEBOOK_TASK})
assert op.is_job_succeeded(status) is expected
+ def test_default_args_durable_reaches_operator(self):
+ op = DatabricksSubmitRunOperator(
+ task_id=TASK_ID, json={"notebook_task": NOTEBOOK_TASK},
default_args={"durable": False}
+ )
+ assert op.durable is False
+
class TestDatabricksRunNowOperator:
def test_init_with_named_parameters(self):
@@ -3202,6 +3211,10 @@ class TestDatabricksRunNowOperatorDurable:
op = DatabricksRunNowOperator(task_id=TASK_ID, job_id=JOB_ID)
assert op.is_job_succeeded(status) is expected
+ def test_default_args_durable_reaches_operator(self):
+ op = DatabricksRunNowOperator(task_id=TASK_ID, job_id=JOB_ID,
default_args={"durable": False})
+ assert op.durable is False
+
class TestDatabricksSQLStatementsOperator:
def test_init(self):
@@ -4800,3 +4813,18 @@ class TestDatabricksTaskOperator:
operator.monitor_databricks_job()
assert mock_databricks_hook.return_value.get_run.call_count == 3
+
+
+class TestWarnAndDisableDurableAirflowPre3_3:
+ def test_no_warning_when_unset(self):
+ with warnings.catch_warnings(record=True) as caught:
+ warnings.simplefilter("always")
+ result = _warn_and_disable_durable_pre_3_3(_DURABLE_UNSET)
+ assert result is False
+ assert caught == []
+
+ @pytest.mark.parametrize("value", [True, False])
+ def test_warns_and_disables_when_explicitly_set(self, value):
+ with pytest.warns(UserWarning, match="durable.*no effect"):
+ result = _warn_and_disable_durable_pre_3_3(value)
+ assert result is False
diff --git a/providers/google/docs/operators/cloud/bigquery.rst
b/providers/google/docs/operators/cloud/bigquery.rst
index 6f42b4e841a..fa9b71a7337 100644
--- a/providers/google/docs/operators/cloud/bigquery.rst
+++ b/providers/google/docs/operators/cloud/bigquery.rst
@@ -403,10 +403,10 @@ the same logical run doesn't submit a second job, that
guarantee is unrelated to
and keeping ``force_rerun=False`` is still the right choice for it.
Durable execution requires Airflow 3.3 or newer, since it relies on the task
state store. On
-earlier Airflow versions the flag is a no-op and the operator always submits a
fresh job on retry,
-exactly as before -- including the pre-existing
``reattach_states``/``Conflict`` behavior, which is
-unchanged. If the task state store is unavailable at runtime, the operator
logs that crash
-recovery is disabled and behaves the same way.
+earlier Airflow versions the flag is a no-op -- setting it explicitly only
emits a warning -- and
+the operator always submits a fresh job on retry, exactly as before --
including the pre-existing
+``reattach_states``/``Conflict`` behavior, which is unchanged. If the task
state store is
+unavailable at runtime, the operator logs that crash recovery is disabled and
behaves the same way.
Like the persisted state itself, the stored job id isn't deleted
automatically, that only happens
when someone runs ``airflow state-store clean``. If a task's ``retry_delay``
is longer than
diff --git
a/providers/google/src/airflow/providers/google/cloud/operators/bigquery.py
b/providers/google/src/airflow/providers/google/cloud/operators/bigquery.py
index afc279e6548..872ff30fbff 100644
--- a/providers/google/src/airflow/providers/google/cloud/operators/bigquery.py
+++ b/providers/google/src/airflow/providers/google/cloud/operators/bigquery.py
@@ -73,6 +73,20 @@ except ImportError:
return value is not NOTSET
+_DURABLE_UNSET = object()
+
+
+def _warn_and_disable_durable_pre_3_3(durable: Any) -> bool:
+ """Shared by the <3.3 compat stub: durable has no effect below 3.3, warn
if it was set."""
+ if durable is not _DURABLE_UNSET:
+ warnings.warn(
+ "`durable` has no effect on Airflow versions below 3.3.",
+ UserWarning,
+ stacklevel=3,
+ )
+ return False
+
+
try:
from airflow.sdk import ResumableJobMixin
except ImportError:
@@ -82,9 +96,9 @@ except ImportError:
external_id_key: str = "bigquery_job_id"
- def __init__(self, *, durable: bool = True, **kwargs: Any) -> None:
+ def __init__(self, *, durable: Any = _DURABLE_UNSET, **kwargs: Any) ->
None:
super().__init__(**kwargs)
- self.durable = durable
+ self.durable = _warn_and_disable_durable_pre_3_3(durable)
def execute_resumable(self, context):
external_id = self.submit_job(context)
@@ -2373,8 +2387,13 @@ class BigQueryInsertJobOperator(
result_timeout: float | None = None,
deferrable: bool = conf.getboolean("operators", "default_deferrable",
fallback=False),
poll_interval: float = 4.0,
+ durable: bool | None = None,
**kwargs,
) -> None:
+ # Named here (not left to **kwargs) so default_args reaches it on every
+ # supported Airflow version.
+ if durable is not None:
+ kwargs["durable"] = durable
super().__init__(**kwargs)
self.configuration = configuration
self.location = location
diff --git
a/providers/google/tests/unit/google/cloud/operators/test_bigquery.py
b/providers/google/tests/unit/google/cloud/operators/test_bigquery.py
index 5d4e6718bf8..2c0d163373f 100644
--- a/providers/google/tests/unit/google/cloud/operators/test_bigquery.py
+++ b/providers/google/tests/unit/google/cloud/operators/test_bigquery.py
@@ -20,6 +20,7 @@ from __future__ import annotations
import json
import logging
import os
+import warnings
from contextlib import suppress
from unittest import mock
from unittest.mock import ANY, MagicMock
@@ -51,6 +52,7 @@ from airflow.providers.common.compat.sdk import (
)
from airflow.providers.google.cloud.openlineage.utils import BIGQUERY_NAMESPACE
from airflow.providers.google.cloud.operators.bigquery import (
+ _DURABLE_UNSET,
BigQueryCheckOperator,
BigQueryColumnCheckOperator,
BigQueryCreateEmptyDatasetOperator,
@@ -73,6 +75,7 @@ from airflow.providers.google.cloud.operators.bigquery import
(
BigQueryUpdateTableSchemaOperator,
BigQueryUpsertTableOperator,
BigQueryValueCheckOperator,
+ _warn_and_disable_durable_pre_3_3,
)
from airflow.providers.google.cloud.triggers.bigquery import (
BigQueryCheckTrigger,
@@ -2560,6 +2563,10 @@ class TestBigQueryInsertJobOperatorDurable:
assert op.is_job_succeeded("success") is True
assert op.is_job_succeeded("RUNNING") is False
+ def test_default_args_durable_reaches_operator(self):
+ op = self._make_operator(default_args={"durable": False})
+ assert op.durable is False
+
class TestBigQueryIntervalCheckOperator:
def test_bigquery_interval_check_operator_execute_complete(self):
@@ -3439,3 +3446,18 @@ class TestBigQueryListRoutinesOperator:
call_kwargs = mock_hook.return_value.list_routines.call_args.kwargs
assert call_kwargs["dataset_id"] == TEST_DATASET
assert call_kwargs["max_results"] == 10
+
+
+class TestWarnAndDisableDurableAirflowPre3_3:
+ def test_no_warning_when_unset(self):
+ with warnings.catch_warnings(record=True) as caught:
+ warnings.simplefilter("always")
+ result = _warn_and_disable_durable_pre_3_3(_DURABLE_UNSET)
+ assert result is False
+ assert caught == []
+
+ @pytest.mark.parametrize("value", [True, False])
+ def test_warns_and_disables_when_explicitly_set(self, value):
+ with pytest.warns(UserWarning, match="durable.*no effect"):
+ result = _warn_and_disable_durable_pre_3_3(value)
+ assert result is False
diff --git a/providers/snowflake/docs/operators/snowflake.rst
b/providers/snowflake/docs/operators/snowflake.rst
index 113b1832b5b..4cb3059718c 100644
--- a/providers/snowflake/docs/operators/snowflake.rst
+++ b/providers/snowflake/docs/operators/snowflake.rst
@@ -214,3 +214,7 @@ To opt out and always submit fresh SQL on retry, set
``durable=False``:
Durable execution applies to the synchronous path. When ``deferrable=True`` is
set, the Triggerer
already tracks the statement handles across the wait, so deferrable mode takes
precedence and
``durable`` has no effect.
+
+Durable execution requires Airflow 3.3 or newer, since it relies on the task
state store. Below
+3.3, ``durable`` has no effect either way: setting it explicitly only emits a
warning, and the
+operator always submits fresh SQL on retry, exactly as before this feature
existed.
diff --git
a/providers/snowflake/src/airflow/providers/snowflake/operators/snowflake.py
b/providers/snowflake/src/airflow/providers/snowflake/operators/snowflake.py
index 8e3cda63ef3..fa8b6348ca7 100644
--- a/providers/snowflake/src/airflow/providers/snowflake/operators/snowflake.py
+++ b/providers/snowflake/src/airflow/providers/snowflake/operators/snowflake.py
@@ -18,6 +18,7 @@
from __future__ import annotations
import time
+import warnings
from collections.abc import Iterable, Mapping, Sequence
from datetime import timedelta
from functools import cached_property
@@ -35,6 +36,20 @@ from airflow.providers.common.sql.operators.sql import (
from airflow.providers.snowflake.hooks.snowflake_sql_api import
SnowflakeSqlApiHook
from airflow.providers.snowflake.triggers.snowflake_trigger import
SnowflakeSqlApiTrigger
+_DURABLE_UNSET = object()
+
+
+def _warn_and_disable_durable_pre_3_3(durable: Any) -> bool:
+ """Shared by the <3.3 compat stub: durable has no effect below 3.3, warn
if it was set."""
+ if durable is not _DURABLE_UNSET:
+ warnings.warn(
+ "`durable` has no effect on Airflow versions below 3.3.",
+ UserWarning,
+ stacklevel=3,
+ )
+ return False
+
+
try:
from airflow.sdk import ResumableJobMixin
except ImportError:
@@ -44,9 +59,9 @@ except ImportError:
external_id_key: str = "snowflake_query_ids"
- def __init__(self, *, durable: bool = True, **kwargs: Any) -> None:
+ def __init__(self, *, durable: Any = _DURABLE_UNSET, **kwargs: Any) ->
None:
super().__init__(**kwargs)
- self.durable = durable
+ self.durable = _warn_and_disable_durable_pre_3_3(durable)
def execute_resumable(self, context):
external_id = self.submit_job(context)
@@ -413,8 +428,13 @@ class SnowflakeSqlApiOperator(ResumableJobMixin,
SQLExecuteQueryOperator):
timeout: int | None = None,
deferrable: bool = conf.getboolean("operators", "default_deferrable",
fallback=False),
snowflake_api_retry_args: dict[str, Any] | None = None,
+ durable: bool | None = None,
**kwargs: Any,
) -> None:
+ # Named here (not left to **kwargs) so default_args reaches it on every
+ # supported Airflow version.
+ if durable is not None:
+ kwargs["durable"] = durable
self.snowflake_conn_id = snowflake_conn_id
self.poll_interval = poll_interval
self.statement_count = statement_count
diff --git
a/providers/snowflake/tests/unit/snowflake/operators/test_snowflake.py
b/providers/snowflake/tests/unit/snowflake/operators/test_snowflake.py
index d37f420f2a6..7cf8713ab7d 100644
--- a/providers/snowflake/tests/unit/snowflake/operators/test_snowflake.py
+++ b/providers/snowflake/tests/unit/snowflake/operators/test_snowflake.py
@@ -17,6 +17,7 @@
# under the License.
from __future__ import annotations
+import warnings
from unittest import mock
from unittest.mock import MagicMock, call
@@ -31,10 +32,12 @@ from airflow.models.taskinstance import TaskInstance
from airflow.providers.common.compat.sdk import TaskDeferred
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
from airflow.providers.snowflake.operators.snowflake import (
+ _DURABLE_UNSET,
SnowflakeCheckOperator,
SnowflakeIntervalCheckOperator,
SnowflakeSqlApiOperator,
SnowflakeValueCheckOperator,
+ _warn_and_disable_durable_pre_3_3,
)
from airflow.providers.snowflake.triggers.snowflake_trigger import
SnowflakeSqlApiTrigger
from airflow.utils.types import DagRunType
@@ -967,3 +970,22 @@ class TestSnowflakeSqlApiOperatorDurable:
assert operator.is_job_active("success") is False
assert operator.is_job_succeeded("success") is True
assert operator.is_job_succeeded("running") is False
+
+ def test_default_args_durable_reaches_operator(self):
+ operator = self._make_operator(default_args={"durable": False})
+ assert operator.durable is False
+
+
+class TestWarnAndDisableDurableAirflowPre3_3:
+ def test_no_warning_when_unset(self):
+ with warnings.catch_warnings(record=True) as caught:
+ warnings.simplefilter("always")
+ result = _warn_and_disable_durable_pre_3_3(_DURABLE_UNSET)
+ assert result is False
+ assert caught == []
+
+ @pytest.mark.parametrize("value", [True, False])
+ def test_warns_and_disables_when_explicitly_set(self, value):
+ with pytest.warns(UserWarning, match="durable.*no effect"):
+ result = _warn_and_disable_durable_pre_3_3(value)
+ assert result is False
