Issue 1:
Apache Airflow version

2.6.0

What happened

After upgrading to 2.6.0 version, pipelines with SparkKubernetesOperator -> SparkKubernetesSensor stopped working correctly.

this PR introduces some enhancement into Spark Kubernetes logic, now SparkKubernetesOperator receives the log from spark pods (which is great), but it doesn't monitor the status of a pod, which means if spark application fails - a task in Airflow finishes successfully.

On the other hand, using previous pipelines (Operator + Sensor) is impossible now, cause SparkKubernetesSensor fails with jinja2.exceptions.UndefinedError: 'None' has no attribute 'metadata' as SparkKubernetesOperator is no longer pushing info to xcom.

What you think should happen instead

Old pipelines should be compatible with Airflow 2.6.0, even though the log would be retrieved in two places - operator and sensor.

OR remove the sensor completely and implement all the functionality in the operator (log + status)

How to reproduce

Create a DAG with two operators

t1 = SparkKubernetesOperator(
kubernetes_conn_id='common/kubernetes_default',
task_id=f"task-submit",
namespace="namespace",
application_file="spark-applications/app.yaml",
do_xcom_push=True,
dag=dag,
)

t2 = SparkKubernetesSensor(
task_id=f"task-sensor",
namespace="namespace",
application_name=f"{{{{ task_instance.xcom_pull(task_ids='task-submit')['metadata']['name'] }}}}",
dag=dag,
attach_log=True,
)

Operating System

Debian GNU/Linux 10 (buster)

Versions of Apache Airflow Providers

apache-airflow-providers-amazon==8.0.0

apache-airflow-providers-apache-spark==4.0.1

apache-airflow-providers-celery==3.1.0

apache-airflow-providers-cncf-kubernetes==6.1.0

apache-airflow-providers-common-sql==1.4.0

apache-airflow-providers-docker==3.6.0

apache-airflow-providers-elasticsearch==4.4.0

apache-airflow-providers-ftp==3.3.1

apache-airflow-providers-google==10.0.0

apache-airflow-providers-grpc==3.1.0

apache-airflow-providers-hashicorp==3.3.1

apache-airflow-providers-http==4.3.0

apache-airflow-providers-imap==3.1.1

apache-airflow-providers-microsoft-azure==6.0.0

apache-airflow-providers-microsoft-mssql==3.3.2

apache-airflow-providers-microsoft-psrp==2.2.0

apache-airflow-providers-microsoft-winrm==3.1.1

apache-airflow-providers-mysql==5.0.0

apache-airflow-providers-odbc==3.2.1

apache-airflow-providers-oracle==3.6.0

apache-airflow-providers-postgres==5.4.0

apache-airflow-providers-redis==3.1.0

apache-airflow-providers-sendgrid==3.1.0

apache-airflow-providers-sftp==4.2.4

apache-airflow-providers-slack==7.2.0

apache-airflow-providers-snowflake==4.0.5

apache-airflow-providers-sqlite==3.3.2

apache-airflow-providers-ssh==3.6.0

apache-airflow-providers-telegram==4.0.0

Deployment

Other 3rd-party Helm chart

Deployment details

No response

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

Issue 2:
Body

TL;DR:

Multiple providers need to handle the logic of splitting SQL statements

Snowflake.run() , RedShift.run() and it will be very similar should be eventually create PrestoOperator or TrinoOperator.

I want to discuss the idea of extracting the split statement logic into DbApiHook.

e.g create a DbApiHook.run_many() that will split the statements and use DbApiHook.run() under the hood.

On one hand it will reduce duplicate code and allow to develop operators for these providers more easily

On the other hand if we will make it in dbapihook it means to bump minimum supported Airflow version for these providers (once they will actually use the new function)

The long story:

Some DBs like MySQL, PostgreSql, MsSQL are able to run a batch of SQL statements:

SELECT 1; SELECT2;

So you can do:

MySQLOperator(.., sql='SELECT 1; SELECT2;')

it will invoke :

airflow/airflow/hooks/dbapi.py

Line 163
in
501a3c3

def run(self, sql, autocommit=False, parameters=None, handler=None):

which will execute the sql in a single query. (The run function will convert this sql to ['SELECT 1; SELECT2;'] - Note this is a list with 1 cell)

But in other DBs like: Snowflake, Redshift, Trino, Presto

You can not run SELECT 1; SELECT2; as is. You must split the statement.

This makes the run() irrelevant for these DBs.

For example in Snowflake when users pass SELECT 1; SELECT2; what we actually do is create:

['SELECT 1;', 'SELECT2;'] - Note this is a list with 2 cells

airflow/airflow/providers/snowflake/hooks/snowflake.py

Lines 314 to 316
in
501a3c3

if isinstance(sql, str):

split_statements_tuple = split_statements(StringIO(sql))

sql = [sql_string for sql_string, _ in split_statements_tuple if sql_string]

The problem?

These providers (Snowflake, Redshift, Trino, Presto) are forced to override dbapi.run() but they all actually need the exact same functionality.

If you will take a look at Snowflake.run() you will see that it's almost identical to RedShift.run() and it will be very similar should be eventually create PrestoOperator or TrinoOperator.

This is something already brought up earlier in #15533 (comment)

So why not just set the sql param to accept list only?

Because we want to support using .sql file so we must handle the split of the statements.

SomeSqlOperator(
...,
sql=my_queries.sql
)

We are reading the content of the file and loading it as a single statement.

What do others think?

Should we create DbApiHook.run_many() or leave it as it is today where every provider needs to handle the logic on it's own?

Committer

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