Add deferrable mode to SnowparkContainerJobOperator

[justinpakzad]

---

Description

Follow-up to #68259 adding deferrable mode to SnowparkContainerJobOperator. Snowpark Container jobs are often long-running (ML training, batch processing, other containerized workloads that run for hours), which makes them a good fit for deferrable mode.

The SnowflakeHook is synchronous, so this follows a similar approach the Anthropic provider took. The trigger wraps each blocking hook call in asyncio.to_thread so a single poll doesn't stall the shared triggerer event loop. A timeout parameter is introduced for both modes. Cleanup is consistent across both modes, on timeout the service is dropped to stop billing on a job past its deadline. On poll errors and terminal failures it's left in place for inspection.

The status Enum and terminal/non-terminal frozensets were moved to the trigger file so the operator can import them without a circular dependency.

Testing

• Unit tests for the trigger (serialization, terminal/unexpected/timeout/error events, cleanup) and the operator's defer and execute_complete paths.
• Manually tested end-to-end against a real Snowflake account: deferred success (job completes, logs retrieved, service dropped), terminal failure, both timeout types, on_kill, and the drop toggles.

Was generative AI tooling used to co-author this PR?

• Yes (Claude Opus 4.6)

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[justinpakzad]

Would appreciate a review on this. Thanks.

[SameerMesiah97]

This needs some work (especially the trigger). I have left some comments.

[justinpakzad]

@SameerMesiah97 I appreciate the review. I've addressed and replied to all of your comments. Let me know your thoughts. Thanks.

[SameerMesiah97]

Just one nit. Please check my response to the enum comment too.

[justinpakzad]

Just one nit. Please check my response to the enum comment too.

Addressed both of the comments.

[justinpakzad]

Would appreciate a review on this. Thanks.

[potiuk]

Approach looks solid — the trigger runs the blocking SnowflakeHook through asyncio.to_thread rather than stalling the triggerer loop, on_kill is wired to the real BaseTrigger hook, and the tests are unusually thorough (parametrized, autospec, time_machine, deadline-precedence and boundary cases). Four observations inline, one of which I'd like resolved before this merges; none are blockers.

Deadline measured from defer time (operators/snowpark_containers.py:260)

The deferred path starts the execution_timeout clock when the task defers rather than when it started, so total task runtime can exceed execution_timeout by however long submission took. Details and a suggested fix inline.

Smaller observations

• operators/snowpark_containers.py:30 — TERMINAL_STATUSES / NON_TERMINAL_STATUSES were public names in a released module; the rename drops them without an alias.
• operators/snowpark_containers.py:197 — the timeout path drops the service before fetching container logs, and ignores drop_on_completion=False.
• hooks/snowflake.py:68 — status enum placement in the SQL hook module; the trigger module would break the import cycle just as well.

Worth noting the __init__ warning you added is in the right place — combination-of-arguments checks belong in __init__ per #70296, not in execute().

---

This review was drafted by an AI-assisted tool and confirmed by an Airflow maintainer. The findings below are observations, not blockers; an Airflow maintainer — a real person — will take the next look at the PR. If you think a finding is mis-applied, please reply on the PR and a maintainer will weigh in.

More on how Airflow handles maintainer review: contributing-docs/05_pull_requests.rst.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting

[justinpakzad]

Thanks for the review. I've addressed all of your comments. Please see my response in regards to the placement of the statuses as I have made some changes there. I also re-tested this e2e against my Snowflake account, specifically the new changes which log the container outputs on timeout. It's working as expected.

[potiuk]

LGTM — all four points from my last pass are addressed: the deadline now derives from ti.start_date, TERMINAL_STATUSES / NON_TERMINAL_STATUSES keep their original names and stay importable from the operator module, both timeout paths log container output before dropping and respect drop_on_completion, and the statuses moved out of the SQL hook module into utils/snowpark_containers.py. CI is green and every thread is resolved. Four smaller observations inline, none blocking.

Smaller observations

• triggers/snowpark_containers.py:159 — BaseTrigger.on_kill() only exists from Airflow 3.3.0; across the 2.11–3.2 range this provider still supports, nothing calls it, so a killed deferred task leaves the service running. Worth a line in the docs.
• operators/snowpark_containers.py:243 — the success-path drop used to be unguarded; routing it through _drop_service makes a failed drop silent.
• operators/snowpark_containers.py:194 — time.monotonic() rather than time.time() for the in-process poll window (AGENTS.md); the trigger's epoch deadlines correctly stay on time.time().
• operators/snowpark_containers.py:121 — the new 24h default also caps the pre-existing synchronous path; worth a changelog line, since providers' changelogs are generated from git log.

---

This review was drafted by an AI-assisted tool and confirmed by an Airflow maintainer. The maintainer approving this PR has read the findings and signed off. If something feels off, please reply on the PR and a maintainer will follow up.

More on how Airflow handles maintainer review: contributing-docs/05_pull_requests.rst.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting

[justinpakzad]

LGTM — all four points from my last pass are addressed: the deadline now derives from ti.start_date, TERMINAL_STATUSES / NON_TERMINAL_STATUSES keep their original names and stay importable from the operator module, both timeout paths log container output before dropping and respect drop_on_completion, and the statuses moved out of the SQL hook module into utils/snowpark_containers.py. CI is green and every thread is resolved. Four smaller observations inline, none blocking.

Appreciate the second round of review. All good callouts and they've been addressed.

[potiuk]

LGTM — all four points from my last pass are addressed:

• _poll_for_status is on time.monotonic() for both the deadline and the check, while the trigger's serialized epoch deadlines correctly stay on time.time().
• The deferrable on_kill version gap is documented in snowpark_containers.rst.
• The success-path drop keeps its best-effort behaviour with a comment giving the reason — a transient drop error would otherwise fail an already-successful job and retry it. That is the right call, and it is now legible to the next reader.
• The 24h default is called out as a breaking change directly below the Changelog header, which is exactly where that file's own instructions put breaking-change notes.

CI is green (56 passed, 28 skipped), no threads are open, and the branch merges clean.

One small observation, not blocking.

Brittle clock mock (providers/snowflake/tests/unit/snowflake/operators/test_snowpark_containers.py:176)

Details inline. Dropping time_machine here was correct — it does not patch the monotonic clock, which is precisely why the test needed changing.

---

This review was drafted by an AI-assisted tool and confirmed by an Airflow maintainer. The maintainer approving this PR has read the findings and signed off. If something feels off, please reply on the PR and a maintainer will follow up.

More on how Airflow handles maintainer review: contributing-docs/05_pull_requests.rst.