fix: wrap future streams also in io_runtime

[ion-elgreco]

@alamb pointed out that the get stream would be not executed on the IO runtime unless it was wrapped. So thanks to his example PR I've added this now for the other streams :) Thanks @alamb!

Also addressed some other missing areas where I should have enabled the IO runtime:

• DeltaFileSystemHandler
• Creating a table
• Is Delta Table check
• Writing a new table (used by pyarrow engine)
• Writing a new table (rust engine)

I've also made it possible to pass an optional runtime in DeltaOps::try_from_uri, since this was missing but it seems a common used entry point to open a delta table

[ion-elgreco]

@Tom-Newton can you give this a spin?

[codecov]

Codecov Report

Attention: Patch coverage is 12.57143% with 153 lines in your changes missing coverage. Please review.

Project coverage is 71.53%. Comparing base (f7745fd) to head (01da20b).

Files with missing lines	Patch %	Lines
crates/core/src/storage/mod.rs	0.00%	133 Missing ⚠️
crates/core/src/operations/mod.rs	40.00%	7 Missing and 2 partials ⚠️
python/src/lib.rs	0.00%	7 Missing ⚠️
crates/benchmarks/src/bin/merge.rs	0.00%	2 Missing ⚠️
crates/core/src/protocol/checkpoints.rs	66.66%	1 Missing ⚠️
python/src/filesystem.rs	0.00%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #3162      +/-   ##
==========================================
- Coverage   71.76%   71.53%   -0.23%     
==========================================
  Files         138      138              
  Lines       44362    44492     +130     
  Branches    44362    44492     +130     
==========================================
- Hits        31835    31829       -6     
- Misses      10495    10620     +125     
- Partials     2032     2043      +11     

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

[alamb]

I plan to give this a look later today

[Tom-Newton]

@Tom-Newton can you give this a spin?

I gave it a try, but I can definitely still reproduce _internal.DeltaError: Failed to parse parquet: External: Generic MicrosoftAzure error: error decoding response body when running at scale, cross region. I would guess that the cause is network or Azure flakiness, where object-store is not retrying, rather than a tokio runtime issue. I'm quite interested in https://github.com/apache/arrow-rs/pull/6519/files.

[ion-elgreco]

@Tom-Newton what is your timeout set to?

[Tom-Newton]

@Tom-Newton what is your timeout set to?

deltalake.DeltaTable(path, storage_options={"timeout": "120s"})

[ion-elgreco]

@Tom-Newton can you set it to a very large interval, 30 minutes or something?

[Tom-Newton]

@Tom-Newton can you set it to a very large interval, 30 minutes or something?

I gave it a try with 30 minutes. The success rate is not noticeably better but we do get a different error message.

      File "/redacted/pip-svc_deltalake/site-packages/deltalake/table.py", line 1270, in update_incremental
        self._table.update_incremental()
    OSError: Generic MicrosoftAzure error: Error after 0 retries in 993.500001046s, max_retries:10, retry_timeout:180s, source:error sending request for url (https://redacted.blob.core.windows.net/redacted/_delta_log/00000000000000146433.json)

Potentially this is useful information, but I have not yet had a chance to investigate further. It sounds like in this case Azure is returning an error response, which would likely be informative.

[ion-elgreco]

@Tom-Newton this is during reading I presume?