<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.6.0</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">After upgrading to 2.6.0 version, pipelines with SparkKubernetesOperator -&gt; SparkKubernetesSensor stopped working correctly.</p>
<p dir="auto"><a href="https://github.com/apache/airflow/pull/29977" data-hovercard-type="pull_request" data-hovercard-url="/apache/airflow/pull/29977/hovercard">this PR</a> 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.<br>
On the other hand, using previous pipelines (Operator + Sensor) is impossible now, cause SparkKubernetesSensor fails with <code class="notranslate">jinja2.exceptions.UndefinedError: 'None' has no attribute 'metadata'</code> as SparkKubernetesOperator is no longer pushing info to xcom.</p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">Old pipelines should be compatible with Airflow 2.6.0, even though the log would be retrieved in two places - operator and sensor.<br>
OR remove the sensor completely and implement all the functionality in the operator (log + status)</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">Create a DAG with two operators</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="t1 = SparkKubernetesOperator(
        kubernetes_conn_id='common/kubernetes_default',
        task_id=f&quot;task-submit&quot;,
        namespace=&quot;namespace&quot;,
        application_file=&quot;spark-applications/app.yaml&quot;,
        do_xcom_push=True,
        dag=dag,
    )

    t2 = SparkKubernetesSensor(
        task_id=f&quot;task-sensor&quot;,
        namespace=&quot;namespace&quot;,
        application_name=f&quot;{{{{ task_instance.xcom_pull(task_ids='task-submit')['metadata']['name'] }}}}&quot;,
        dag=dag,
        attach_log=True,
    )
"><pre class="notranslate"><code class="notranslate">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,
    )

</code></pre></div>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Debian GNU/Linux 10 (buster)</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">apache-airflow-providers-amazon==8.0.0<br>
apache-airflow-providers-apache-spark==4.0.1<br>
apache-airflow-providers-celery==3.1.0<br>
apache-airflow-providers-cncf-kubernetes==6.1.0<br>
apache-airflow-providers-common-sql==1.4.0<br>
apache-airflow-providers-docker==3.6.0<br>
apache-airflow-providers-elasticsearch==4.4.0<br>
apache-airflow-providers-ftp==3.3.1<br>
apache-airflow-providers-google==10.0.0<br>
apache-airflow-providers-grpc==3.1.0<br>
apache-airflow-providers-hashicorp==3.3.1<br>
apache-airflow-providers-http==4.3.0<br>
apache-airflow-providers-imap==3.1.1<br>
apache-airflow-providers-microsoft-azure==6.0.0<br>
apache-airflow-providers-microsoft-mssql==3.3.2<br>
apache-airflow-providers-microsoft-psrp==2.2.0<br>
apache-airflow-providers-microsoft-winrm==3.1.1<br>
apache-airflow-providers-mysql==5.0.0<br>
apache-airflow-providers-odbc==3.2.1<br>
apache-airflow-providers-oracle==3.6.0<br>
apache-airflow-providers-postgres==5.4.0<br>
apache-airflow-providers-redis==3.1.0<br>
apache-airflow-providers-sendgrid==3.1.0<br>
apache-airflow-providers-sftp==4.2.4<br>
apache-airflow-providers-slack==7.2.0<br>
apache-airflow-providers-snowflake==4.0.5<br>
apache-airflow-providers-sqlite==3.3.2<br>
apache-airflow-providers-ssh==3.6.0<br>
apache-airflow-providers-telegram==4.0.0</p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Other 3rd-party Helm chart</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Anything else</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>
<h3 dir="auto">Body</h3>
<p dir="auto"><strong>TL;DR:</strong><br>
Multiple providers need to handle the logic of splitting SQL statements<br>
<a href="https://github.com/apache/airflow/blob/501a3c3fbefbcc0d6071a00eb101110fc4733e08/airflow/providers/snowflake/hooks/snowflake.py#L287-L347">Snowflake.run()</a> , <a href="https://github.com/apache/airflow/pull/22391/files">RedShift.run()</a> and it will be very similar should be eventually create <code class="notranslate">PrestoOperator</code> or <code class="notranslate">TrinoOperator</code>.</p>
<p dir="auto">I want to discuss the idea of extracting the split statement logic into <code class="notranslate">DbApiHook</code>.<br>
e.g create a <code class="notranslate">DbApiHook.run_many()</code> that will split the statements and use <code class="notranslate">DbApiHook.run()</code> under the hood.</p>
<p dir="auto">On one hand it will reduce duplicate code and allow to develop operators for these providers more easily<br>
On the other hand if we will make it in <code class="notranslate">dbapihook</code> it means to bump minimum supported Airflow version for these providers (once they will actually use the new function)</p>
<hr>
<p dir="auto"><strong>The long story:</strong><br>
Some DBs like MySQL, PostgreSql, MsSQL are able to run a batch of SQL statements:<br>
<code class="notranslate">SELECT 1; SELECT2;</code></p>
<p dir="auto">So you can do:<br>
<code class="notranslate">MySQLOperator(.., sql='SELECT 1; SELECT2;')</code><br>
it will invoke :<br>
</p><div class="Box Box--condensed my-2">
  <div class="Box-header f6">
    <p class="mb-0 text-bold">
      <a href="https://github.com/apache/airflow/blob/501a3c3fbefbcc0d6071a00eb101110fc4733e08/airflow/hooks/dbapi.py#L163">airflow/airflow/hooks/dbapi.py</a>
    </p>
    <p class="mb-0 color-fg-muted">
         Line 163
      in
      <a data-pjax="true" class="commit-tease-sha" href="/apache/airflow/commit/501a3c3fbefbcc0d6071a00eb101110fc4733e08">501a3c3</a>
    </p>
  </div>
  <div itemprop="text" class="Box-body p-0 blob-wrapper blob-wrapper-embedded data">
    <table class="highlight tab-size mb-0 js-file-line-container" data-tab-size="8" data-paste-markdown-skip="">

        <tbody><tr class="border-0">
          <td id="L163" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="163"></td>
          <td id="LC163" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line"> <span class="pl-k">def</span> <span class="pl-en">run</span>(<span class="pl-s1">self</span>, <span class="pl-s1">sql</span>, <span class="pl-s1">autocommit</span><span class="pl-c1">=</span><span class="pl-c1">False</span>, <span class="pl-s1">parameters</span><span class="pl-c1">=</span><span class="pl-c1">None</span>, <span class="pl-s1">handler</span><span class="pl-c1">=</span><span class="pl-c1">None</span>): </td>
        </tr>
    </tbody></table>
  </div>
</div>
<p></p>
<p dir="auto">which will execute the sql in a single query. (The run function will convert this sql to <code class="notranslate">['SELECT 1; SELECT2;']</code>  - Note this is a list with <strong>1 cell</strong>)</p>
<p dir="auto">But in other DBs like: Snowflake, Redshift, Trino, Presto</p>
<p dir="auto">You can not run <code class="notranslate">SELECT 1; SELECT2;</code> as is. You must split the statement.<br>
This makes the <code class="notranslate">run()</code> irrelevant for these DBs.</p>
<p dir="auto">For example in Snowflake when users pass <code class="notranslate">SELECT 1; SELECT2;</code> what we actually do is create:<br>
<code class="notranslate">['SELECT 1;', 'SELECT2;']</code> - Note this is a list with <strong>2 cells</strong><br>
</p><div class="Box Box--condensed my-2">
  <div class="Box-header f6">
    <p class="mb-0 text-bold">
      <a href="https://github.com/apache/airflow/blob/501a3c3fbefbcc0d6071a00eb101110fc4733e08/airflow/providers/snowflake/hooks/snowflake.py#L314-L316">airflow/airflow/providers/snowflake/hooks/snowflake.py</a>
    </p>
    <p class="mb-0 color-fg-muted">
        Lines 314 to 316
      in
      <a data-pjax="true" class="commit-tease-sha" href="/apache/airflow/commit/501a3c3fbefbcc0d6071a00eb101110fc4733e08">501a3c3</a>
    </p>
  </div>
  <div itemprop="text" class="Box-body p-0 blob-wrapper blob-wrapper-embedded data">
    <table class="highlight tab-size mb-0 js-file-line-container" data-tab-size="8" data-paste-markdown-skip="">

        <tbody><tr class="border-0">
          <td id="L314" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="314"></td>
          <td id="LC314" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line"> <span class="pl-k">if</span> <span class="pl-en">isinstance</span>(<span class="pl-s1">sql</span>, <span class="pl-s1">str</span>): </td>
        </tr>

        <tr class="border-0">
          <td id="L315" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="315"></td>
          <td id="LC315" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line">     <span class="pl-s1">split_statements_tuple</span> <span class="pl-c1">=</span> <span class="pl-en">split_statements</span>(<span class="pl-v">StringIO</span>(<span class="pl-s1">sql</span>)) </td>
        </tr>

        <tr class="border-0">
          <td id="L316" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="316"></td>
          <td id="LC316" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line">     <span class="pl-s1">sql</span> <span class="pl-c1">=</span> [<span class="pl-s1">sql_string</span> <span class="pl-k">for</span> <span class="pl-s1">sql_string</span>, <span class="pl-s1">_</span> <span class="pl-c1">in</span> <span class="pl-s1">split_statements_tuple</span> <span class="pl-k">if</span> <span class="pl-s1">sql_string</span>] </td>
        </tr>
    </tbody></table>
  </div>
</div>
<p></p>
<p dir="auto"><strong>The problem?</strong><br>
These providers (Snowflake, Redshift, Trino, Presto) <strong>are forced to override dbapi.run()</strong> but they all actually need the exact same functionality.</p>
<p dir="auto">If you will take a look at <a href="https://github.com/apache/airflow/blob/501a3c3fbefbcc0d6071a00eb101110fc4733e08/airflow/providers/snowflake/hooks/snowflake.py#L287-L347">Snowflake.run()</a> you will see that it's almost identical to <a href="https://github.com/apache/airflow/pull/22391/files">RedShift.run()</a> and it will be very similar should be eventually create <code class="notranslate">PrestoOperator</code> or <code class="notranslate">TrinoOperator</code>.</p>
<p dir="auto">This is something already brought up earlier in <a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="867848913" data-permission-text="Title is private" data-url="https://github.com/apache/airflow/issues/15533" data-hovercard-type="pull_request" data-hovercard-url="/apache/airflow/pull/15533/hovercard?comment_id=620481685&amp;comment_type=review_comment" href="https://github.com/apache/airflow/pull/15533#discussion_r620481685">#15533 (comment)</a></p>
<p dir="auto"><strong>So why not just set the sql param to accept list only?</strong><br>
Because we want to support using .sql file so we must handle the split of the statements.</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="SomeSqlOperator(
...,
sql=my_queries.sql
)"><pre class="notranslate"><code class="notranslate">SomeSqlOperator(
...,
sql=my_queries.sql
)
</code></pre></div>
<p dir="auto">We are reading the content of the file and loading it as a single statement.</p>
<hr>
<p dir="auto"><strong>What do others think?</strong><br>
Should we create <code class="notranslate">DbApiHook.run_many()</code> or leave it as it is today where every provider needs to handle the logic on it's own?</p>
<h3 dir="auto">Committer</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I acknowledge that I am a maintainer/committer of the Apache Airflow project.</li>
</ul>