Fix: SnowflakeSqlApiOperator(without deferable flag) incorrectly marks the task as successful even when jobs are still active

[subbota19]

---

Fix #46648.

Problem Description
In deferable mode, Airflow uses triggers to check the job status, so every time the deferable interval is reached, it checks the status and yields an event on this line.

However, when not using deferable logic, the execution follows the else statement, which does not include a while loop to continuously monitor execution. Instead, on this line, the task iterates through active Snowflake jobs, checks their status and then sleeps for poll_interval.

This results in the following behavior:

✅ If a job finishes in less than 5 seconds ( time.sleep using the default poll_interval), the logic works correctly.
❌ If a job takes longer than 5 seconds, there is no mechanism to keep monitoring it, and the task incorrectly finishes after the loop ends - even if the Snowflake job is still running.

Suggested Fix
I propose adding a running status check inside poll_on_queries and modifying the else statement with a loop to ensure Airflow waits until all queries finish, the logic is quick similar to what we use in deferable trigger while loop

With this change, the Airflow task would periodically call poll_on_queries and continue waiting until all jobs are complete.

Tests
Simulates a scenario where the Snowflake job is running for two checks (10 seconds apart, assuming default poll_interval): ffter the second check, the job is completed successfully and task is marked as successful.

Simulates a scenario where the Snowflake job failed and ensures AirflowException is raised.

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide (https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our pre-commits will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[potiuk]

You need to resolve conflicts now

[subbota19]

Hi @potiuk!
I've resolved the conflict on my end, could you please double-check it? Thanks!

[subbota19]

Hi @rawwar !
Could you please review this PR again and resolve the remaining conversations? Thanks!

[rawwar]

Hi @rawwar ! Could you please review this PR again and resolve the remaining conversations? Thanks!

I think you should be able to mark the conversations as resolved. If not, a committer will be able to help. I don't have permission to mark the conversation as resolved.

[subbota19]

Hi @potiuk!
I've resolved the conflict on my end and added suggestions, could you please double-check it? Thanks!

[rawwar]

Hi @potiuk! I've resolved the conflict on my end and added suggestions, could you please double-check it? Thanks!

@subbota19 , take changes from main. test failures are already fixed.

[subbota19]

Hi @potiuk! I've resolved the conflict on my end and added suggestions, could you please double-check it? Thanks!

@subbota19 , take changes from main. test failures are already fixed.

Hi @rawwar, thx, I've rebased to the main branch

[subbota19]

Hi @eladkal @potiuk !
I believe we can go ahead and merge this PR

[boring-cyborg]

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.