Snowflake provider (SnowflakeSqlApiOperator) - Not exposing query results with multiple queries

[set92]

Apache Airflow version

Other Airflow 2/3 version (please specify below)

If "Other Airflow 2/3 version" selected, which one?

2.10.3

What happened?

When calling the .execute, and retrieve the query_ids I get the parent ID, but if I'm running multiple queries I want to know all the queries IDs to retrieve results for any of them.

On the logs, I can see the API response, but I don't see a method to get it.

While checking the code I see it calls to the Hook, and on the hook I can see

    def _process_response(self, status_code, resp):
        self.log.info("Snowflake SQL GET statements status API response: %s", resp)
        if status_code == 202:
            return {"status": "running", "message": "Query statements are still running"}
        elif status_code == 422:
            return {"status": "error", "message": resp["message"]}
        elif status_code == 200:
            if resp_statement_handles := resp.get("statementHandles"):
                statement_handles = resp_statement_handles
            elif resp_statement_handle := resp.get("statementHandle"):
                statement_handles = [resp_statement_handle]
            else:
                statement_handles = []
            return {
                "status": "success",
                "message": resp["message"],
                "statement_handles": statement_handles,
            }
        else:
            return {"status": "error", "message": resp["message"]}

So, seeing this I would say in the Hook is correct, because if it has statementHandles it returns it. Although I would say that is not totally right, because you may want/need both. The parent id that is the statementHandle, and the IDs of the children (statementHandles).

But the Operator only returns the parent ID, even so when in the resp I can see everything. I think is because it calls to check_query_output() which only logs the json response, but doesn't return it, and the hook uses get_sql_api_query_status that parses it correctly?

I also think that the poll_queries would be a nice method to have exposed to the hook, since now the hook I think it runs the query and forget about it.

resp:
[TIMESTAMP] {snowflake_sql_api.py:287} INFO - Snowflake SQL GET statements status API response: {'resultSetMetaData': {'numRows': 1, 'format': 'jsonv2', 'partitionInfo': [{'rowCount': 1, 'uncompressedSize': 57}], 'rowType': [{'name': 'multiple statement execution', 'database': '', 'schema': '', 'table': '', 'scale': None, 'nullable': False, 'byteLength': 16777216, 'precision': None, 'collation': None, 'length': 16777216, 'type': 'text'}]}, 'data': [['Multiple statements executed successfully.']], 'code': '090001', 'statementHandles': ['XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX', 'XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX', 'XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX', 'XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX', 'XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX', 'XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX', 'XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX', 'XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX', 'XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX'], 'statementStatusUrl': '/api/v2/statements/XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX?requestId=XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX', 'requestId': 'XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX', 'sqlState': '00000', 'statementHandle': 'XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX', 'message': 'Statement executed successfully.', 'createdOn': 1763984429070}

What you think should happen instead?

No response

How to reproduce

I don't have a basic example since I'm debugging production code, but it should be

    op = SnowflakeSqlApiOperator(
        task_id="run_sql",
        snowflake_conn_id=snflk_conn_id,
        sql=sql_statement,
        database=sql_params["database"],
        schema=sql_params["schema_db"],
        deferrable=False,
        statement_count=stmt_count,
        warehouse=kwargs.get("warehouse", "").upper(),
    )

sql_statement = """
ALTER SESSION SET TRANSACTION_ABORT_ON_ERROR = TRUE;
BEGIN TRANSACTION;

SELECT 4 as "number of rows inserted";
SELECT 2 as "number of rows updated";

COMMIT;
"""

Operating System

AWS MWAA

Versions of Apache Airflow Providers

Providers

Package Name	Version	Description
apache-airflow-providers-amazon	9.0.0	Amazon integration (including Amazon Web Services (AWS)).
apache-airflow-providers-celery	3.8.3	Celery
apache-airflow-providers-common-compat	1.7.3	Common Compatibility Provider - providing compatibility code for previous Airflow versions
apache-airflow-providers-common-io	1.4.2	Common IO Provider
apache-airflow-providers-common-sql	1.28.0	Common SQL Provider
apache-airflow-providers-fab	1.5.0	Flask App Builder
apache-airflow-providers-ftp	3.11.1	File Transfer Protocol (FTP)
apache-airflow-providers-http	4.13.2	Hypertext Transfer Protocol (HTTP)
apache-airflow-providers-imap	3.7.0	Internet Message Access Protocol (IMAP)
apache-airflow-providers-microsoft-mssql	3.9.1	Microsoft SQL Server (MSSQL)
apache-airflow-providers-pagerduty	5.0.2	Pagerduty
apache-airflow-providers-postgres	5.13.1	PostgreSQL
apache-airflow-providers-slack	9.2.0	Slack services integration including: - Slack API - Slack Incoming Webhook
apache-airflow-providers-smtp	1.8.0	Simple Mail Transfer Protocol (SMTP)
apache-airflow-providers-snowflake	6.5.3	Snowflake
apache-airflow-providers-sqlite	3.9.0	SQLite

Deployment

Amazon (AWS) MWAA

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

[jroachgolf84]

Thanks for creating this issue! Can you tell me a bit more about your desired functionality? Are you looking for each statement ID to be written to XComs? Added to logs? Tell me more.

[set92]

In general, I would like to have a better integration between Snowflake <-> Airflow. Most of all for debugging purposes. For example, in Snowflake you can't see the query results of queries that are not your own. So, it would be nice to have it in Airflow.

I feel the UX for that would be a plugin that can connects to Snowsight, or have a panel like Astro showed for their guides in 3.1. But I'm not going to daydream with it, and since I don't want to use >AF3, I was thinking I could get the query ids to at least do some counts on them, having at the end of all atomic blocks some select $variables outputting how many rows I have modified on the DML queries.

From my POV, those queries ids should be available as a parameter for anyone to use them as they like. Having them in the XCOM? I suppose it makes sense since right now is outputting the parent_id, but with it, you would need to call to Snowflake again to know the childs IDs, so I'm trying to avoid this query.

[potiuk]

nce I don't want to use >AF3

Just to make it clear - I doubt any effort from the Airflow dev community will be put in adding new things in Airflow 2 since we are only fixing critical and security issues in Airflow 2.

Also the support of future releases of Snowflake provider will end in April 2026, so maybe a good idea to switch and to develop it for Airflow 3 (just saying) . PRs as usual are most welcome.

[set92]

Yep, but as I was talking with the Astro people while doing the demo, there were too many things missing. Like that I couldn't opt out of the dag versioning, and I couldn't rerun a dagrun with different code was crazy. After a while seemed it was possible, but only see/change the version using the top right panel on the Graph view, but I normally use the accordion menu (where you can expand and collapse task groups).

In general, felt it was an unfinished project. Also, the button to clear all the tasks inside a task group was removed, they told me that would be added in 3.1, but my trust is already broken. I prefer to wait until it has been polish a bit more. I have too much these months. And I mostly only use it for Snowflake, so I can update the providers version and keep using 2.11 or whatever is not >=3.0.

[jroachgolf84]

I'd echo what @potiuk is saying. @set92, please feel free to open up a PR if this is functionality that is absolutely essential for your workflow. Or, you could author your own custom Operators if needed.

[potiuk]

In general, felt it was an unfinished project. Also, the button to clear all the tasks inside a task group was removed, they told me that would be added in 3.1, but my trust is already broken. I prefer to wait until it has been polish a bit more. I have too much these months. And I mostly only use it for Snowflake, so I can update the providers version and keep using 2.11 or whatever is not >=3.0.

Sure, you can wait with upgrade - but this also mean that unless you implement it yourself in your copy with your plugin (And we won't really even need to accept it community), this is highly unlikely somoene will ever do it for Airflow 2.

But you still can of course - for yourself if you want to use the old plugin mechanisms. They were designed for extensions and there is absolutely nothing wrong with you implementing that plugin for yourself.

Converting it into a discussion - it's hhighly unlikely this turns into something that the community would do

[set92]

@potiuk From my point of view, what we are talking is a bug, since not all the mechanisms work in the same way. The conversation if I'm going to use it or not is another thing. The problem at hand is:

1. Running a single query -> Perfect, you only have a query, so the parent ID is enough. ✅
2. Multiple queries -> In _process_response in SnowflakeHook we can see it is prepared for it, since it evaluates, and returns the parent id, or the child ids.

            if resp_statement_handles := resp.get("statementHandles"):
                statement_handles = resp_statement_handles
            elif resp_statement_handle := resp.get("statementHandle"):
                statement_handles = [resp_statement_handle]
            else:
                statement_handles = []
            return {
                "status": "success",
                "message": resp["message"],
                "statement_handles": statement_handles,

2.1. What I'm requesting is that the SnowflakeSqlApiOperator works in the same way. It doesn't need to be for Airflow 2, only improve the product that is Airflow for anyone to use it.
2.2. I think that implementation has a bug, because Snowflake doesn't add the parent ID to the child IDs. I think as a user/developer you are going to expect to have the parent ID AND the child IDs.

[potiuk]

Sure. It does not matter if it is a bug or feature . It needs someone to implement it. Whether it's a bug or feature makes absolutely no difference. And the best (fastest and most certain) is that you implement it. If you do it for Airflow 3 you can even contribute it. If not - likely something for yourself.

Or you can wait until someone volunteers and fixes it. That's how open-source works.

You can certainly speed up things by contributing them in the way community accept it - but other than that you just have to hope someone will do it - but there are no guarantees whatsoever - if only you are interested you might wait forever.

All I can do is to mark it as good first issue

[set92]

I thought that since it's a bug, it would be given priority. Although it's also true that it's not a critical bug.

I was surprised that it got turned into a conversation because it's like dismissing the bug.
But okay, I'll look into making a superclass for the Snowflake operator to keep using airflow 2,X, and I'll mention it to Astro in case they want to resolve the issue in future versions (v3.X).

[potiuk]

I thought that since it's a bug, it would be given priority. Although it's also true that it's not a critical bug.

By whom?

We have > 3500 contributors - people like you as well or others. There are no "priorities" given by some central priority something - each of people contributing prioritises accordng to what they think is important for them to fix. that's how open-source works.

This was converted into a dicusssion because the original request was about much more sophisticated idea about adding a UI. Converting to a discssion means that ... there is a discussion that you raised yourself:

You could have started a discussion related to the issue that you originally raised, but by commenting on a more general thing you made it a discussion about way more than that.

And that was far beyond a "bug" - that was a discussion about adding new UI for the provider that would be Airflow 2 compatible and that would be new feature (for Airflow 3) but since you wanted to discuss it for Airflow 2 additions and dismissed Airflow 3 , it's been converted to a discussion - in case you wanted to continue it there. That was no longer bug - you turned into a discussion about adding new feature to Airflow 2 (which is not going to happen)

But if you want to raise a bug to solve without diving into a discussion. fill free open a separate issue, i will mark it as a good firs issue if it will be descibed well and hopefuly someone will fix it (or maybe even you).

[potiuk]

Learning - good idea is to not mixed "digression" discussions as comments to existing issues. I was literally convinced you want to propose to approach the problem you described by creating a plugin, the way you commented it.