S3 requests after the hyper s3 timeout connection duration fail over large table operations

[liamphmurphy]

Environment

Delta-rs version: python v0.16.0

Binding: Python

Environment:

• Cloud provider: AWS

---

Bug

What happened:

On three of our large Delta tables (out of 500, most of the others are a lot smaller) I've repeatedly been seeing errors related to an S3 timeout. I get two kinds of errors; one indicates an error reading data from a parquet file, the other is just a much more generic error. Here they are:

Failed to parse parquet: Parquet error: AsyncChunkReader::get_bytes error: Generic S3 error: Error after 10 retries in 174.881515313s, max_retries:10, retry_timeout:180s, source:error sending request for url (https://s3.us-east-1.amazonaws.com/...redacted...some-data-file.parquet): operation timed out

AND

ERROR - Error deleting events from s3a://path-to-delta-table: Failed to parse parquet: Parquet error: AsyncChunkReader::get_bytes error: Generic S3 error: request or response body error: operation timed out

This can happen in various contexts, but the prominent example in my case is needing to scan the entire Delta table for records matching a value and deleting those matches. I've logged out right before this delete op runs, and the time delta between that log and the error is usually around 5 minutes. This time delta is relevant.

The second one in particular comes up a lot. This is on a table that gets around 250,000 rows added every hour and is partitioned by date 2024-03-18 and hour 15 as example.

My initial thought was that this was due to a failure to download a larger file such as a checkpoint; but those are "only" 668 MB and it is certainly not unheard of, moving files using various AWS SDK's will gladly handle larger files.

Our current theory is that instead this timeout indicate that the hyper client makes a request, then the library does some in-mem operations (such as scanning a large delta table) that takes a long time, and then when delta-rs is ready to reach out to S3 again, it exceeds some kind of keep-alive timeout and fails. This is why that time delta of 5 minutes I mentioned earlier is important.

This seems related to this comment:

delta-rs/crates/aws/src/storage.rs

Lines 401 to 406 in 785667e

	/// The `pool_idle_timeout` option of aws http client. Has to be lower than 20 seconds, which is
	/// default S3 server timeout <https://aws.amazon.com/premiumsupport/knowledge-center/s3-socket-connection-timeout-error/>.
	/// However, since rusoto uses hyper as a client, its default timeout is 90 seconds
	/// <https://docs.rs/hyper/0.13.2/hyper/client/struct.Builder.html#method.keep_alive_timeout>.
	/// Hence, the `connection closed before message completed` could occur.
	/// To avoid that, the default value of this setting is 15 seconds if it's not set otherwise.

And finally I think this sort of timeout is expected from the hyper client, per this (admittedly old) ticket but it explains this exact situation in another context: hyperium/hyper#763

What you expected to happen:

Even on long running operations, such as a scan over the whole table for deletion, the operation would eventually succeed barring no other issues.

How to reproduce it:

IF I AM CORRECT: perform a long running Delta operation, such as a delete on a table with millions of rows and have it try to make a final reach out to S3

More details:

Note this is just a theory. I wanted to get this issue out there to see what others think, I'm going to dig in and see if I can figure out whether it's a issue with the library or something else.

[ion-elgreco]

@liamphmurphy I'm happy I wasn't the only one facing this. Was questioning our infra setup

[liamphmurphy]

@liamphmurphy I'm happy I wasn't the only one facing this. Was questioning our infra setup

Haha I'm also glad to know I'm not completely crazy!

This may be relevant: awslabs/aws-sdk-rust#1079 (comment)

If this fix is related, I've noticed the hyper version delta-rs uses is quite old so it'll be a big version increase.

[ion-elgreco]

@liamphmurphy just to double check, do you also see the behavior on 0.15.3? That's what we use at the moment

[ion-elgreco]

The linked code you shared is actually not out yet in a Python release. It was recently refactored by @mightyshazam to the AWS sdk's.

[liamphmurphy]

@ion-elgreco Tested with 0.15.3 and yep I get that second more generic error still.

And ahh okay understood, guess that explains why changing that env var wasn't doing anything haha. I'll bring that down locally and see if that fixes it

[ion-elgreco]

@liamphmurphy we actually see it happen in 1.5minute already:

The above exception was caused by the following exception:
_internal.DeltaError: Generic DeltaTable error: External error: External error: Arrow error: External error: Parquet error: ParquetObjectReader::get_byte_ranges error: Generic S3 error: request or response body error: operation timed out

[liamphmurphy]

I did some debugging and found where this error occurs (at least for me)

It happens here:

delta-rs/crates/core/src/kernel/snapshot/log_segment.rs

Line 316 in dadf264

let batch = batch?;

The checkpoint_stream.next() call hangs for some time, and then eventually returns an error;

Err(Parquet { source: General("ParquetObjectReader::get_byte_ranges error: Generic S3 error: request or response body error: operation timed out") })

So it appears that it finds the right checkpoint file (which for me is around 800 MB large) and detects any unaccounted for logs made after that checkpoint, but then fails on scanning this stream.

[liamphmurphy]

And for some additional logs in case it's helpful (the checkpoint files print line is my own), here's the last snippet of RUST_LOG=debug lines leading to the error.

checkpoint files: [ObjectMeta { location: Path { raw: "_delta_log/00000000000000108548.checkpoint.parquet" }, last_modified: 2024-03-19T13:01:23Z, size: 953533417, e_tag: Some("\"tag\""), version: None }]
[2024-03-19T14:23:36Z DEBUG hyper::client::pool] reuse idle connection for ("https", s3.us-east-1.amazonaws.com)
[2024-03-19T14:23:36Z DEBUG hyper::proto::h1::io] flushed 1575 bytes
[2024-03-19T14:23:36Z DEBUG hyper::proto::h1::io] parsed 12 headers
[2024-03-19T14:23:36Z DEBUG hyper::proto::h1::conn] incoming body is content-length (8 bytes)
[2024-03-19T14:23:36Z DEBUG hyper::proto::h1::conn] incoming body completed
[2024-03-19T14:23:36Z DEBUG hyper::client::pool] pooling idle connection for ("https", s3.us-east-1.amazonaws.com)
[2024-03-19T14:23:36Z DEBUG hyper::client::pool] reuse idle connection for ("https", s3.us-east-1.amazonaws.com)
[2024-03-19T14:23:36Z DEBUG hyper::proto::h1::io] flushed 1575 bytes
[2024-03-19T14:23:36Z DEBUG hyper::proto::h1::io] parsed 12 headers
[2024-03-19T14:23:36Z DEBUG hyper::proto::h1::conn] incoming body is content-length (86293 bytes)
[2024-03-19T14:23:36Z DEBUG hyper::proto::h1::conn] incoming body completed
[2024-03-19T14:23:36Z DEBUG hyper::client::pool] pooling idle connection for ("https", s3.us-east-1.amazonaws.com)
[2024-03-19T14:23:36Z DEBUG hyper::client::pool] reuse idle connection for ("https", s3.us-east-1.amazonaws.com)
[2024-03-19T14:23:36Z DEBUG hyper::proto::h1::io] flushed 1567 bytes
[2024-03-19T14:23:36Z DEBUG hyper::proto::h1::io] parsed 12 headers
[2024-03-19T14:23:36Z DEBUG hyper::proto::h1::conn] incoming body is content-length (953409770 bytes)
[2024-03-19T14:23:58Z DEBUG hyper::client::client] client connection error: connection closed before message completed

[ion-elgreco]

@liamphmurphy very useful! so for you it's reading these huge checkpoints then

I think in our case I figured whats happenning it's sometimes our incoming data has 0 rows, which was causing the merge to scan the whole table, will create something separate for that but there is crossover in the connection issues

[liamphmurphy]

@ion-elgreco Yeah appears to be. I presume the reason why a stream is used here was to accommodate large checkpoints like this and to not have to load it all in memory, but then that hyper client connection closes before it can finish. 🤔

[thomasfrederikhoeck]

@ion-elgreco we seing a similar issue using Azure. When we run the following code it fails with _internal.DeltaError: Failed to parse parquet: Parquet error: Z-order failed while scanning data: ArrowError(ExternalError(General("ParquetObjectReader::get_byte_ranges error: Generic MicrosoftAzure error: request or response body error: operation timed out")), None)

import os
os.environ["RUST_LOG"]="debug"

from deltalake import DeltaTable
 
blob_path = "az://<redacted path>"
storage_options = {"AZURE_STORAGE_ACCOUNT_NAME": "<redacted sa>", "AZURE_CONTAINER_NAME":'<redacted container>', 'use_azure_cli': 'true'}

dt = DeltaTable(blob_path, storage_options=storage_options)
dt.optimize.z_order(["StatusDateTime"])

The table consists of 12 parquet of roughly 400mb each.

You can see the full error log here:
https://gist.githubusercontent.com/thomasfrederikhoeck/8db799b3fe4fbc2414626a97da9249a9/raw/658e6cbcf090fba424e7eafae99eea30c4ae7231/txt