Skip to content

Implement async state store methods for object storage backend - #72131

Open
jason810496 wants to merge 1 commit into
apache:mainfrom
jason810496:fix/common-io/async-state-store-backend
Open

Implement async state store methods for object storage backend#72131
jason810496 wants to merge 1 commit into
apache:mainfrom
jason810496:fix/common-io/async-state-store-backend

Conversation

@jason810496

@jason810496 jason810496 commented Aug 27, 2026

Copy link
Copy Markdown
Member

Why

StateStoreObjectStorageBackend raised NotImplementedError from all four of its async methods, so the async state store accessors added in #68232 fail against the shipped object storage backend. adelete/aclear clear the DB reference before calling the backend, so the call raised only after the reference was gone, orphaning the stored object.

What

  • Implement aget, aset, adelete, aclear by offloading the existing sync methods with asyncio.to_thread.
  • Add tests for all four, including a parametrized check that the blocking work does not run on the loop thread.
  • Show the offload pattern in the custom backend guide, which requires the four async methods but only demonstrated sync get().

Why asyncio.to_thread and not fsspec's async API

fsspec does have one, but it is not reachable from where this backend sits:

  • The coroutines live on AsyncFileSystem and are all private (_cat_file, _pipe_file, _rm, _glob). open_async is the only public one.
  • They exist only on async implementations. The base path is user configured, and LocalFileSystem sets async_impl = False, so a native path still needs an isinstance branch and a sync fallback.
  • ObjectStoragePath exposes no coroutines at all, so going native means bypassing it and re-implementing the path building and compression that _write_to_object_storage/_read_from_object_storage already handle.
  • fsspec's sync methods already dispatch those coroutines onto fsspec's own IO loop via run_coroutine_threadsafe and block the caller. asyncio.to_thread only moves that block off Airflow's loop, so this follows fsspec's own concurrency model rather than working around it.

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

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant