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