Allow passing a different tokio runtime into `DeltaOps` to execute CPU-bound tasks

[abhiaagarwal]

Description

Use Case

Per datafusion's recommendations, spawning a separate tokio runtime for CPU-bound tasks can lead to better tail latencies. Therefore, I propose that DeltaOps/Merge/Write/etcBuilder take an optional with_runtime parameter that allows passing an optional handle to another runtime. Then, delta-rs can spawn the task on a separate joinset, await it in there, and pass the result back to the io-bound handle via an oneshot channel.

This requires a bit more plumbing due to object_store calls inside of various executors, but that can be worked across with SpawnedReqwestConnector.

This could also potentially be benchmarked for the python impl to see if it substantially improves performance.

Related Issue(s)

Related to #3640.

[ion-elgreco]

@abhiaagarwal we already have the IO runtime stuff, so if you provide a runtime handle just for the store, then it's already separated.

delta-rs/crates/core/src/table/builder.rs

Lines 268 to 272 in 18f949e

	/// Provide a custom runtime handle or runtime config
	pub fn with_io_runtime(mut self, io_runtime: IORuntime) -> Self {
	self.table_config.io_runtime = Some(io_runtime);
	self
	}

and this is the default in Python:

delta-rs/python/src/lib.rs

Lines 195 to 199 in 18f949e

	let table_url = deltalake::table::builder::parse_table_uri(table_uri)
	.map_err(error::PythonError::from)?;
	let mut builder = deltalake::DeltaTableBuilder::from_uri(table_url)
	.map_err(error::PythonError::from)?
	.with_io_runtime(IORuntime::default());

[abhi-airspace-intelligence]

I can't believe I missed this! But I think I'm looking for the opposite solution, where instead of implicitly assuming the "other" runtime is the CPU bound one, I want to explicitly pass a separate runtime into DeltaOps (since it's more common to have a "cpu" runtime than an "io" runtime)

[ion-elgreco]

I can't believe I missed this! But I think I'm looking for the opposite solution, where instead of implicitly assuming the "other" runtime is the CPU bound one, I want to explicitly pass a separate runtime into DeltaOps (since it's more common to have a "cpu" runtime than an "io" runtime)

Is it though? The reason this exists in object-store was to make Datafusion and IO separation easier

[abhi-airspace-intelligence]

In production code, I've implemented something on all my operations that looks something like this:

let delete_builder = DeleteBuilder::new((*table).log_store(), table.snapshot()?.snapshot().clone())
            .with_predicate(delete_predicate)
            .with_writer_properties(config.into())
            .with_session_state(state);

let mut join_set = JoinSet::new();
join_set.spawn_on(delete_builder.into_future(), cpu_handle.handle());

let (deleted_table, metrics) = join_set.join_next().await.unwrap().unwrap()?;

Basically, under the proposed API, this would be handled by the various builders if the optional cpu_runtime parameter was set. If you think this is worthwhile addition to the public API @ion-elgreco?