Improve error messages for Snowflake operators to include query and context

[nagabharathnr]

Description

Error messages raised by Snowflake operators can be generic and may not always provide enough context (such as the SQL query or operator details), making debugging more difficult for users.

Use case/motivation

Error messages should include additional context such as:

• the SQL query being executed (or a truncated version)
• operator/task context
• clearer indication of failure type (connection vs execution)

This would improve debugging and developer experience when working with Snowflake in Airflow.

Related issues

No

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[srini-koth]

Proposed Fix: Enhance _process_response in snowflake_sql_api Hook

To address the lack of error context in Snowflake operator failures, I'd suggest the following targeted changes in airflow/providers/snowflake/hooks/snowflake_sql_api.py.

---

1. Modify _process_response signature to accept SQL query context

Currently, _process_response handles 422 errors with structured fields (code, sqlState, message), but all other error responses are raised generically without this context.

Current signature:

def _process_response(self, response: dict) -> None:

Proposed signature:

def _process_response(self, response: dict, sql: str | None = None) -> None:

Passing sql allows the method to include the originating query in error output — critical for debugging multi-statement or dynamic SQL workflows.

---

2. Extend structured error handling to ALL error responses (not just 422)

Currently only 422 responses extract code, sqlState, and message. All other HTTP error statuses fall through to a generic raise.

Proposed change:

def _process_response(self, response: dict, sql: str | None = None) -> None:
    status = response.get("status")
    message = response.get("message", "Unknown error")
    code = response.get("code", "N/A")
    sql_state = response.get("sqlState", "N/A")
    sql_preview = (sql[:300] + "...") if sql and len(sql) > 300 else sql or "N/A"

    if status == "ERROR":
        raise AirflowException(
            f"Snowflake query failed.\n"
            f"Code: {code} | SQL State: {sql_state}\n"
            f"Message: {message}\n"
            f"SQL: {sql_preview}"
        )

This ensures every error path surfaces code, sqlState, and the SQL query — not just 422s.

3. Update all callers to pass SQL

Anywhere _process_response is called (e.g., in execute_query, get_query_status), pass the relevant sql argument:

# Before
self._process_response(response)

# After
self._process_response(response, sql=self.sql)

[jroachgolf84]

@nagabharathnr - which Snowflake Operators specifically?