Make Persister Download Snapshot Files from Object Store Concurrently or in Parallel

[mgattozzi]

See #24604 (comment) below

In #24588 I had created the initial implementation of the persister to read and write Segment Info Files, the Catalog, and Parquet data to the Object Store. However, one flaw is that we download the Segment Info Files one at a time as seen here:

influxdb/influxdb3_write/src/persister.rs

Lines 187 to 190 in 001a2a6

	for item in &list[0..end] {
	let bytes = self.object_store.get(&item.location).await?.bytes().await?;
	output.push(serde_json::from_slice(&bytes)?);
	}

If the network is slow this could really slow down any time we need to acquire the segment data.

In the PR I had tried to make it concurrent and while the code in theory should have been successful this was thwarted by the lifetimes created with the #[async_trait] proc macro. Getting it to use Rayon's par_iter() method also did not work. While async fn in traits is now stable it has it's own issues as well that were hard to untangle with the codebase at this time.

This is a known bottleneck currently and so to make sure we keep track of it I've opened up this issue to keep track of it and to keep track of any progress made addressing it.

[hiltontj]

Although things have changed a fair bit since this was opened - we now have snapshots instead of segments - we are still loading the snapshot files in sequence here:

influxdb/influxdb3_write/src/persister.rs

Lines 231 to 234 in 35e29d1

	for item in &list[0..end] {
	let bytes = self.object_store.get(&item.location).await?.bytes().await?;
	output.push(serde_json::from_slice(&bytes)?);
	}

[pauldix]

This is definitely something we'll want to do. We'll also want to fetch WAL files in parallel as well. Maybe we should create an epic around optimizing startup time?

[hiltontj]

I created an epic here: #25533