Async/data persistence

[wild-endeavor]

Why are the changes needed?

A previous PR added async versions of the type engine. This change leverages that by converting flytekit's main I/O layer to async as well.

What changes were proposed in this pull request?

• Added a new function to get the async version of an fsspec filesystem. This takes care of calling fsspec with the current loop.
• Fix to get_filesystem to take into account passed in kwargs even if no special if condition handling is present (this affected using https filesystem).
• Added async versions of the popular functions.
• Updated some type transformers to async.

How was this patch tested?

Tested locally and then tested on internal clusters. Performance benefit is mainly noticeable across large number of small files. A task that produces a list of 1000 files, 2MB each, decreases from around 135s to about 40s on a t3a.xlarge. No improvement is seen for folders because those were already async.

Setup process

Screenshots

Check all the applicable boxes

• I updated the documentation accordingly.
• All new and existing tests passed.
• All commits are signed-off.

Related PRs

Docs link

[codecov]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 82.79%. Comparing base (1f8a273) to head (4a8c936).
Report is 18 commits behind head on master.

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #2829      +/-   ##
==========================================
- Coverage   87.23%   82.79%   -4.45%     
==========================================
  Files          26        3      -23     
  Lines        1426      186    -1240     
==========================================
- Hits         1244      154    -1090     
+ Misses        182       32     -150     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[kumare3]

Suggested change

	def get_async_filesystem_for_path(self, path: str = "", anonymous: bool = False, **kwargs) -> AsyncFileSystem:
	async def get_async_filesystem_for_path(self, path: str = "", anonymous: bool = False, **kwargs) -> AsyncFileSystem:

[thomasjpfan]

Can you share the code that you used for benchmarking?

[wild-endeavor]

Can you share the code that you used for benchmarking?

https://github.com/unionai/debugyt/blob/master/user/ytong/src/yt_dbg/data_perf/measure_wfs.py
@thomasjpfan

[eapolinario]

kf-pytorch tests are hanging (notice how they ran for 5h+): https://github.com/flyteorg/flytekit/actions/runs/11451023024/job/31860491868?pr=2829#step:7:1

I kicked off a re-run but it seems like it's also hanging.

[eapolinario]

This is amazing.