Custom tokio runtime/thread pool when reading from Icechunk

[maximedion2]

Hey there!

I wasn't sure where to start looking, but would there be a way to read from an icechunk store with a user specified runtime? For example, the equivalent of this from object_store, https://docs.rs/object_store/0.12.2/object_store/client/struct.SpawnedReqwestConnector.html.

The use case I have in mind is exactly what they are describing in the SpawnedReqwestConnector docs, i.e. I want to avoid having compute heavy tasks in the thread pool that handles the tasks reading the data chunks. If the "main" task that ends up triggering reads from icechunk is launched on a tokio runtime with compute heavy stuff running in it, then when read tasks are spawned (presumably using tokio::spawn) those can end up being blocked by the blocking tasks because they are in the same thread pool, which I want to avoid (that's the short version).

Does that functionality exist somewhere, or does icechunk at least expose enough "machinery" for me to do that with what I can get out-of-the-box?

[mpiannucci]

Hi @maximedion2, thanks for pointing this out. Are you specifically looking to specify the connector for ObjectStore storage backends or do you want general control over the tokio thread pool? Icechunk does indeed use tokio::spawn in various places vs standard structured concurrency with tokio.

We don't currently have a built in way to control this, but i'm interested in exactly where you want control because there is thread stuff happening in Icechunks context, but then also downstream libraries (object_store, the native s3 api) that wont necessarily automatically share config/threading.

[maximedion2]

Thanks for your reply!

Okay so here's more context. I'm trying to write a library built around datafusion that will allow to read data from zarr (including icechunk) and apply all the SQL operations supported by datafusion to it (and if that works well, I want to add more operations, but that's for later).

datafusion is "pull" based, so you have a sequence of operations described in a plan, and from what I understand (I'm just starting to look into the details of how it works), operation N would need the results from operation N-1, so it will pull results from it, which triggers operation N-1 pulling from operation N-2, etc...all the way down to a stream of data that is read from a data source (in my case, say an icechunk store). By default, datafusion spawns everything on a tokio thread pool (not the i/o blocking one, it uses the non blocking thread pool, see here).

This means that all the way down into icechunk, when tokio::spawn is called (it would be the same if futures were "naively" awaited on the current thread), non blocking tasks, e.g. reading from a store on s3, gcp, etc... run in the same thread pool as compute heavy tasks (a JOIN condition, taking the average of the data, etc...), which is not great, especially since reading from icechunk will involve many async operations (because there are typically a lot of chunks). I don't want the i/o to constantly have to wait for compute tasks to finish.

Here's an example from the datafusion docs where they show how to handle this issue, but for reading data directly from object store. To answer your question, I was asking about general control over the runtime, like for example when you open an icechunk session, you would pass in a runtime, and every async task from that point on is spun up in that runtime (and not in the calling runtime, which is the issue here). That is, unless icechunk spins up its own runtime internally and spawns tasks in it, is that what you meant by "but then also downstream libraries (object_store, the native s3 api) that wont necessarily automatically share config/threading."?

To be clear I don't think this is the only way this could work, I think I could possibly make the calls to icechunk async functions from a separate runtime, instead of passing that separate runtime as an argument to icechunk (the hypothetical approach I'm asking about), but since it seems to be the way it works for object store, I thought I would ask if there was a similar mechanism for icechunk.

[rabernat]

Hi @maximedion2! Would love to learn more about your project as it intersects quite directly with some things on our internal roadmap at Earthmover. Feel free to send me an email at ryan@earthmover.io if you're interested in having a chat.

[mpiannucci]

Wow thanks for the write up, that is an awesome use case.

Based on what you've described, here is my understanding of how things stand:

• icechunk does not explicitly have a way to control the runtime. Wherever you run its functions by default will run in the current context.
• icechunks storage backends are configurable. So the way I would support this short term is to allow for configuring the object store runtime like in the data fusion example OR you can just create your own icechunk Storage implementation using whatever runtime you want.

From rust all of this is very configurable, and I'm happy to help figure out what works best.

Note that in Python layer everything runs in a configured tokio runtime.

[maximedion2]

@mpiannucci sounds good!

Tbh I haven't looked super deep into icechunk storage options, I'm using zarrs and the corresponding libraries for what I'm doing, and so far my code only relies on this trait being implemented, which it is for this which itself depends on an icechunk session. I'm just getting to the point where the actual implementations matter (i.e. it's not just about the trait existing, but also about what runtimes are used for the async stuff, etc...), so I will look in to this more.

Regarding configuring the icechunk backends, is this also possible on the custom s3 library (that, iirc, doesn't depend on object store)? Another option, which I was clumsily trying to explain in my first reply, is to use a custom runtime to spawn any and all calls to the icechunk backend (through the zarrs defined trait methods), such that the custom runtime is used to spawn any downstream tasks, I think that could possibly work.

In any case, thanks for your replies! I will look into this more closely and see what the best option is.

[maximedion2]

@rabernat so right now I'm working on this project on my own time, can't quite justify doing that for the company I work for, at least not yet, so progress is a bit slow, but it's moving forward. I initially started this to learn more about query engines, zarr, rust, etc... and I also think it could be a powerful tool. I started this a couple of years ago but I recently rewrote everything to use the zarrs library. Initially I was reading from zarr storage with code I wrote 100% myself, that was fun and a good learning experience but with a proper canonical zarr implementation in rust it's now pointless and just not a good idea, so I switched over to using that.

Are you guys internally working on a query engine on top of icechunk? in any case, yes we can definitely chat, I'll reach out soon.

[maximedion2]

Hey there!

Follow up on this. So it wasn't clear to me if right now, with no changes to the crate, it's possible to create a session with an ObjectStore? like the ObjectStorage (private) constructor takes in a Arc<dyn ObjectStoreBackend>, but it's not directly accessible? In theory one could implement ObjectStoreBackend for an ObjectStore, it's a bit silly but that would work, if it was possible to instantiate an ObjectStorage with anything that implements the trait (as a public method)? That might already exist, but I couldn't find it.

Also, I vaguely remember @rabernat mentioning in a conversation that icechunk relies on a custom implementation, i.e. not ObjectStore, for s3 i/o, and that that custom implementation was faster than with ObjectStore? Is that the case (@mpiannucci I think that's what you mentioned in your initial comment)? Are there plans to do that for other remote stores? I ask because assuming we can easily create a session from ObjectStore, and build stuff on top of that, would I be locking myself out of potential performance improvements if I basically commit to relying on ObjectStore instead of whatever native implementation is available in icechunk?

[maximedion2]

Looking at the code for local repositories, I see this, I'm a little confused, what does "prefer using object stores" mean here? What is the "production" ready way to set up a local icechunk repo?

[paraseba]

Basically, today there is no production-ready way to setup a local Icechunk repo on the filesystem. It's only safe if you don't have concurrent commits. "Prefer using object stores" means in this case something like "either use some supported cloud provider or install something like MinIO if you want a local solution". On the other hand, if you don't expect concurrent commits, you can use the current filesystem implementation. In the presence of two (or more) concurrent commits, one of them may report success but actually be overwritten by other.

This is a limitation of the object_store library we are using to implement the local filesystem Storage instance. We definitely want to remove this limitation, but we haven't had time to do it yet.

[kylebarron]

This is a limitation of the object_store library we are using to implement the local filesystem Storage instance.

Specifically, you mean that object_store doesn't guarantee stable list ordering for local filesystems?

[paraseba]

That's minor, larger issue is it doesn't support conditional PUT on local filesystems.

[rabernat]

larger issue is it doesn't support conditional PUT on local filesystems.

...which is in theory trivial to implement with file locking. It's ironic to me that all this work has gone into making object stores provide the sort of consistency guarantees we expect from filesystems...but now somehow filesystems have been left behind! 😆