Add defer support for SQLExecuteQueryOperator

[eladkal]

Body

The SQLExecuteQueryOperator serves several kind of DBs: Sqlite, PostgresSQL, Snowflake etc..
We want to explore the option to add defer capabilities to this operator which will be very helpful when running long duration queries.

Committer

• I acknowledge that I am a maintainer/committer of the Apache Airflow project.

[notatallshaw-gts]

Astronomer provides as Snowflake sensor, you should take a look what is possible with it: https://github.com/astronomer/astronomer-providers#extras

FYI I believe Astronomer have also investigated other databases and it was non-trivial as getting the async to be performant and provide enough information, it also relied on both the Python library and the underlying drivers so could be tricky to confirm the environment is correctly setup.

I'm sure PRs would be welcome by the Airflow team.

[bruno-uy]

This would be awesome. We have some long-running queries that would be great if we could defer.

[raphaelauv]

first layers to make it possible -> #43944 , #49392

[uranusjr]

I’ve made #49392 ready for review.

[raphaelauv]

works in progress for SQLExecuteQueryTrigger #55094

[faridnsh]

I'm thinking of giving this a try to make a smallest MVP PR with least amount of feature to have defer support with one of the Postgres drivers, maybe picking up from where @dabla left it. I'm generally a pretty confident person who goes head into problems and then realize that it's bigger than I can handle. 😄 I don't have a lot of experience with writing operators for Airflow, so I would appreciate it, if anyone has any suggestions or tips on how I should tackle and implement the change.

[dabla]

works in progress for SQLExecuteQueryTrigger #55094

I’ve closed this PR as it was an old one and will be fixed in another one as now its easy to fix with common compat.

[dabla]

I'm thinking of giving this a try to make a smallest MVP PR with least amount of feature to have defer support with one of the Postgres drivers, maybe picking up from where @dabla left it. I'm generally a pretty confident person who goes head into problems and then realize that it's bigger than I can handle. 😄 I don't have a lot of experience with writing operators for Airflow, so I would appreciate it, if anyone has any suggestions or tips on how I should tackle and implement the change.

If I was you I would experiment with implementing a PostgresAsyncHook first.