Deferred SnowflakeSqlApiOperator + Airflow 3.0.3 failure #55747
Description
Apache Airflow Provider(s)
snowflake
Versions of Apache Airflow Providers
apache-airflow-providers-snowflake==6.4.0
Apache Airflow version
Airflow 3.0.6
Operating System
Linux
Deployment
Astronomer
Deployment details
No response
What happened
Running the SnowflakeSqlApiOperator async + Airflow 3.0.6 gives this failure:
Task failed with exception: source="task"
AirflowException: error: You cannot use AsyncToSync in the same thread as an async event loop - just await the async function directly.
File "/usr/local/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 920 in run
File "/usr/local/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 1215 in _execute_task
File "/usr/local/lib/python3.12/site-packages/airflow/sdk/bases/operator.py", line 1606 in resume_execution
File "/usr/local/lib/python3.12/site-packages/airflow/providers/snowflake/operators/snowflake.py", line 510 in execute_complete
The operator runs fine on Airflow 2.
What you think should happen instead
The operator should not fail
How to reproduce
Run SnowflakeSqlApiOperator async + Airflow 3.0.6
Sample code
import pendulum
from airflow.decorators import dag
from airflow.providers.snowflake.operators.snowflake import (
SnowflakeSqlApiOperator,
)
@dag(
dag_id="test_dbt_multi_repo_v1",
default_args={
"owner": "data-engineering",
"depends_on_past": False,
"retries": 0,
},
start_date=pendulum.today("UTC").add(days=-1),
schedule=None,
max_active_runs=1,
tags=["data-engineering", "test"],
)
def create_dag():
snow = SnowflakeSqlApiOperator(
deferrable=True,
task_id="snow",
snowflake_conn_id="snowflake_test",
poll_interval=60,
sql="SELECT * from snowflake_sample_data.tpcds_sf100tcl.call_center",
# sql="select * from HG_ANALYTICS_PROD.OPEN.V_GA4_ALL where event_date >'20250101'",
)
(
snow
)
dag = create_dag()
Anything else
No response
Are you willing to submit PR?
- Yes I am willing to submit a PR!
Code of Conduct
- I agree to follow this project's Code of Conduct