fix: auto-size multipart parts for Dir.from_local_sync uploads

[mjurkus]

Bug: Dir.from_local_sync skips the multipart auto-sizer → large directory uploads fail on S3/Ceph with 416 InvalidRange

Summary

Dir.from_local_sync uploads via raw fsspec (fs.put(..., recursive=True)), which pins every multipart part to obstore's 5 MiB default. The async Dir.from_local instead routes through storage.put, which auto-sizes the part size. Any single file larger than ~48.8 GiB (10,000 parts × 5 MiB) therefore fails from a sync task but succeeds from an async task.

Impact

• A sync @env.task returning a Dir that contains a large file (observed: a single 160 GiB file) fails at upload.
• Real error from Ceph RadosGW (S3): POST .../file?uploadId=... → 416 Range Not Satisfiable, <Code>InvalidRange</Code>, rejected in ~1 ms (the CompleteMultipartUpload manifest exceeds the 10,000-part limit).
• 160 GiB ÷ 5 MiB = 32,768 parts → over the 10,000 hard cap. The same output from an async task produces ~9000 parts (~18 MiB each) and succeeds.

Root cause

src/flyte/io/_dir.py, from_local_sync:

fs = storage.get_underlying_filesystem(path=resolved_remote_path)
fs.put(local_path_str, resolved_remote_path, recursive=True)
return cls(path=resolved_remote_path, name=dirname, hash=dir_cache_key)

fsspec's obstore backend calls store.put_async without chunk_size, leaving it at obstore's 5 MiB default — the exact scenario flyte/storage/_storage.py documents (its _put_obstore_bypass/_compute_upload_chunk_size machinery exists specifically to avoid this, noting the "~48.8 GiB hard ceiling (10,000 parts)").

The async sibling from_local does it correctly:

output_path = await storage.put(
    from_path=local_path_str, to_path=resolved_remote_path, recursive=True, batch_size=batch_size
)

So this is a straight asymmetry: from_local_sync never got the obstore-aware path. Notably, Dir.download_sync already fixed the download side of this same class of bug using syncify:

# Route through the obstore-aware storage.get (via syncify) rather than the raw fsspec ...
return syncify(storage.get)(self.path, local_dest, recursive=True)

related: #1320

[kumare3]

this looks like a great fix. I am looking if file/dataframe also need this fix

[kumare3]

can you fix the lint and then +1. looks right. found similar issue in file and dataframe.

[kumare3]

@mjurkus / @danielsalvador fixed in file too - #1480 (not really a fix, but improvement) thank you

[mjurkus]

I think clippy version was bumpped and all .github/workflows/lint.yml runs should fail. On other PRs too.