fix: RepeatedScan::execute is CPU-work not I/O

I also renamed `Handle::spawn_blocking` to `Handle::spawn_blocking_io` which better reflects what it
does.

Signed-off-by: Daniel King <dan@spiraldb.com>

Author: danking

vortex-cuda/src/pooled_read_at.rs

@@ -106,7 +106,7 @@ impl VortexReadAt for PooledFileReadAt {
         async move {
             let mut target = pool.get(length)?;
             let target = handle
-                .spawn_blocking(move || {
+                .spawn_blocking_io(move || {
                     read_exact_at(&file, target.as_mut_slice(), offset)?;
                     Ok::<_, io::Error>(target)
                 })
@@ -234,7 +234,7 @@ impl VortexReadAt for PooledObjectStoreReadAt {
                 #[cfg(not(target_arch = "wasm32"))]
                 GetResultPayload::File(file, _) => {
                     target = handle
-                        .spawn_blocking(move || {
+                        .spawn_blocking_io(move || {
                             read_exact_at(&file, target.as_mut_slice(), range.start)?;
                             Ok::<_, io::Error>(target)
                         })

vortex-duckdb/src/duckdb/file_system.rs

@@ -118,7 +118,7 @@ impl VortexWrite for DuckDbFsWriter {
 
         let runtime = RUNTIME.handle();
         let buffer = runtime
-            .spawn_blocking(move || {
+            .spawn_blocking_io(move || {
                 let mut err: cpp::duckdb_vx_error = ptr::null_mut();
                 let mut out_len: cpp::idx_t = 0;
                 let status = unsafe {
@@ -149,7 +149,7 @@ impl VortexWrite for DuckDbFsWriter {
 
         let runtime = RUNTIME.handle();
         runtime
-            .spawn_blocking(move || {
+            .spawn_blocking_io(move || {
                 let mut err: cpp::duckdb_vx_error = ptr::null_mut();
                 let status = unsafe { cpp::duckdb_vx_fs_sync(handle.as_ptr(), &raw mut err) };
                 if status != cpp::duckdb_state::DuckDBSuccess {

vortex-duckdb/src/filesystem.rs

@@ -133,7 +133,7 @@ impl FileSystem for DuckDbFileSystem {
         stream::once(async move {
             RUNTIME
                 .handle()
-                .spawn_blocking(move || list_recursive(ctx, &directory_url, &base_url))
+                .spawn_blocking_io(move || list_recursive(ctx, &directory_url, &base_url))
                 .await
         })
         .flat_map(|result| match result {
@@ -271,7 +271,7 @@ impl VortexReadAt for DuckDbFsReader {
 
             let runtime = RUNTIME.handle();
             let size = runtime
-                .spawn_blocking(move || {
+                .spawn_blocking_io(move || {
                     let mut err: cpp::duckdb_vx_error = ptr::null_mut();
                     let mut size_out: cpp::idx_t = 0;
                     let status = unsafe {
@@ -301,7 +301,7 @@ impl VortexReadAt for DuckDbFsReader {
         async move {
             let runtime = RUNTIME.handle();
             let result: VortexResult<BufferHandle> = runtime
-                .spawn_blocking(move || -> VortexResult<BufferHandle> {
+                .spawn_blocking_io(move || -> VortexResult<BufferHandle> {
                     let mut buffer = ByteBufferMut::with_capacity_aligned(length, alignment);
                     unsafe { buffer.set_len(length) };
 

vortex-io/src/object_store/read_at.rs

@@ -134,7 +134,7 @@ impl VortexReadAt for ObjectStoreReadAt {
                 #[cfg(not(target_arch = "wasm32"))]
                 GetResultPayload::File(file, _) => {
                     handle
-                        .spawn_blocking(move || {
+                        .spawn_blocking_io(move || {
                             read_exact_at(&file, buffer.as_mut_slice(), range.start)?;
                             Ok::<_, io::Error>(buffer)
                         })

vortex-io/src/runtime/handle.rs

@@ -122,7 +122,7 @@ impl Handle {
     }
 
     /// Spawn a blocking I/O task for execution on the runtime.
-    pub fn spawn_blocking<F, R>(&self, f: F) -> Task<R>
+    pub fn spawn_blocking_io<F, R>(&self, f: F) -> Task<R>
     where
         F: FnOnce() -> R + Send + 'static,
         R: Send + 'static,

vortex-io/src/std_file/read_at.rs

@@ -126,7 +126,7 @@ impl VortexReadAt for FileReadAt {
         let allocator = Arc::clone(&self.allocator);
         async move {
             handle
-                .spawn_blocking(move || {
+                .spawn_blocking_io(move || {
                     let mut buffer = allocator.allocate(length, alignment)?;
                     read_exact_at(&file, buffer.as_mut_slice(), offset)?;
                     Ok(BufferHandle::new_host(buffer.freeze()))

vortex-layout/src/scan/scan_builder.rs

@@ -372,9 +372,8 @@ impl<A: 'static + Send> Stream for LazyScanStream<A> {
                         .unwrap_or(1);
                     let concurrency = builder.concurrency * num_workers;
                     let handle = builder.session.handle();
-                    let task = handle.spawn_blocking(move || {
-                        builder.prepare().and_then(|scan| scan.execute(None))
-                    });
+                    let task = handle
+                        .spawn_cpu(move || builder.prepare().and_then(|scan| scan.execute(None)));
                     self.state = LazyScanState::Preparing(PreparingScan {
                         ordered,
                         concurrency,