feat: Spawn layout evaluation

vortex-datafusion/Cargo.toml

@@ -47,7 +47,7 @@ vortex-error = { workspace = true, features = ["datafusion"] }
 vortex-expr = { workspace = true, features = ["datafusion"] }
 vortex-file = { workspace = true, features = ["object_store", "tokio"] }
 vortex-io = { workspace = true, features = ["object_store", "tokio"] }
-vortex-layout = { workspace = true }
+vortex-layout = { workspace = true, features = ["tokio"] }
 [features]
 tracing = ["dep:tracing", "dep:tracing-futures", "vortex-io/tracing"]
 

vortex-datafusion/src/persistent/opener.rs

@@ -5,9 +5,11 @@ use datafusion::datasource::physical_plan::{FileMeta, FileOpenFuture, FileOpener
 use datafusion_common::Result as DFResult;
 use futures::{FutureExt as _, StreamExt};
 use object_store::{ObjectStore, ObjectStoreScheme};
+use tokio::runtime::Handle;
 use vortex_array::{ContextRef, IntoArrayVariant};
 use vortex_error::VortexResult;
 use vortex_expr::{ExprRef, VortexExpr};
+use vortex_file::executor::{TaskExecutor, TokioExecutor};
 use vortex_file::{SplitBy, VortexOpenOptions};
 use vortex_io::ObjectStoreReadAt;
 
@@ -65,6 +67,7 @@ impl FileOpener for VortexFileOpener {
         let object_store = self.object_store.clone();
         let projected_arrow_schema = self.projected_arrow_schema.clone();
         let batch_size = self.batch_size;
+        let executor = TaskExecutor::Tokio(TokioExecutor::new(Handle::current()));
 
         Ok(async move {
             let vxf = VortexOpenOptions::file(read_at)
@@ -86,6 +89,7 @@ impl FileOpener for VortexFileOpener {
                 // but at the moment our scanner has too much overhead to process small
                 // batches efficiently.
                 .with_split_by(SplitBy::RowCount(8 * batch_size))
+                .with_task_executor(executor)
                 .into_array_stream()?
                 .map(move |array| {
                     let st = array?.into_struct()?;

vortex-file/src/lib.rs

@@ -86,7 +86,6 @@
 //! buffers, and [cloud storage](vortex_io::ObjectStoreReadAt), can be used as the "linear and
 //! contiguous memory".
 
-pub mod exec;
 mod file;
 mod footer;
 mod generic;

vortex-layout/Cargo.toml

@@ -21,18 +21,20 @@ bit-vec = { workspace = true }
 bytes = { workspace = true }
 exponential-decay-histogram = { workspace = true }
 flatbuffers = { workspace = true }
-futures = { workspace = true, features = ["alloc"] }
+flume = { workspace = true }
+futures = { workspace = true, features = ["alloc", "executor"] }
 itertools = { workspace = true }
 log = { workspace = true }
 pin-project-lite = { workspace = true }
 vortex-array = { workspace = true }
 vortex-buffer = { workspace = true }
 vortex-dtype = { workspace = true }
-vortex-error = { workspace = true }
+vortex-error = { workspace = true, features = ["tokio"] }
 vortex-expr = { workspace = true }
 vortex-flatbuffers = { workspace = true, features = ["layout"] }
 vortex-mask = { workspace = true }
 vortex-scalar = { workspace = true }
+tokio = { workspace = true, optional = true }
 
 [dev-dependencies]
 futures = { workspace = true, features = ["executor"] }

vortex-layout/src/layouts/chunked/eval_expr.rs

@@ -175,7 +175,7 @@ mod test {
             ],
         )
         .unwrap();
-        (ScanExecutor::inline(Arc::new(segments)), layout)
+        (ScanExecutor::new(Arc::new(segments)), layout)
     }
 
     #[test]

vortex-layout/src/layouts/flat/eval_expr.rs

@@ -35,7 +35,7 @@ impl ExprEvaluator for FlatReader {
             tasks.push(ScanTask::Expr(expr));
         }
 
-        self.executor().evaluate(&array, &tasks).await
+        self.executor().evaluate(&array, &tasks)
     }
 
     async fn prune_mask(&self, row_mask: RowMask, _expr: ExprRef) -> VortexResult<RowMask> {
@@ -72,7 +72,7 @@ mod test {
                 .unwrap();
 
             let result = layout
-                .reader(ScanExecutor::inline(Arc::new(segments)), Default::default())
+                .reader(ScanExecutor::new(Arc::new(segments)), Default::default())
                 .unwrap()
                 .evaluate_expr(
                     RowMask::new_valid_between(0, layout.row_count()),
@@ -98,7 +98,7 @@ mod test {
 
             let expr = gt(Identity::new_expr(), lit(3i32));
             let result = layout
-                .reader(ScanExecutor::inline(Arc::new(segments)), Default::default())
+                .reader(ScanExecutor::new(Arc::new(segments)), Default::default())
                 .unwrap()
                 .evaluate_expr(RowMask::new_valid_between(0, layout.row_count()), expr)
                 .await
@@ -123,7 +123,7 @@ mod test {
                 .unwrap();
 
             let result = layout
-                .reader(ScanExecutor::inline(Arc::new(segments)), Default::default())
+                .reader(ScanExecutor::new(Arc::new(segments)), Default::default())
                 .unwrap()
                 .evaluate_expr(RowMask::new_valid_between(2, 4), ident())
                 .await

vortex-layout/src/layouts/flat/writer.rs

@@ -119,7 +119,7 @@ mod tests {
                 .unwrap();
 
             let result = layout
-                .reader(ScanExecutor::inline(Arc::new(segments)), Default::default())
+                .reader(ScanExecutor::new(Arc::new(segments)), Default::default())
                 .unwrap()
                 .evaluate_expr(RowMask::new_valid_between(0, layout.row_count()), ident())
                 .await

vortex-layout/src/layouts/struct_/eval_expr.rs

@@ -54,7 +54,6 @@ impl ExprEvaluator for StructReader {
 
         self.executor()
             .evaluate(&root_scope, &[ScanTask::Expr(partitioned.root.clone())])
-            .await
     }
 
     async fn prune_mask(&self, row_mask: RowMask, expr: ExprRef) -> VortexResult<RowMask> {
@@ -128,7 +127,7 @@ mod tests {
             .map(IntoArray::into_array)],
         )
         .unwrap();
-        (ScanExecutor::inline(Arc::new(segments)), layout)
+        (ScanExecutor::new(Arc::new(segments)), layout)
     }
 
     #[test]

vortex-layout/src/scan/executor/mod.rs

@@ -0,0 +1,49 @@
+#[cfg(feature = "tokio")]
+mod tokio;
+
+mod threads;
+
+use std::future::Future;
+
+use futures::future::BoxFuture;
+pub use threads::*;
+#[cfg(feature = "tokio")]
+pub use tokio::*;
+use vortex_error::VortexResult;
+
+pub trait Executor {
+    /// Spawns a future to run on a different runtime. The shouldn't be polled to make progress.
+    fn spawn<F>(&self, f: F) -> VortexResult<BoxFuture<'static, VortexResult<F::Output>>>
+    where
+        F: Future + Send + 'static,
+        <F as Future>::Output: Send + 'static;
+}
+
+/// Generic wrapper around different async runtimes. Used to spawn async tasks to run in the background, concurrently with other tasks.
+#[derive(Clone)]
+pub enum TaskExecutor {
+    Threads(ThreadsExecutor),
+    #[cfg(feature = "tokio")]
+    Tokio(TokioExecutor),
+}
+
+pub enum TaskExecutorMode {
+    Threads,
+    #[cfg(feature = "tokio")]
+    Tokio,
+}
+
+#[async_trait::async_trait]
+impl Executor for TaskExecutor {
+    fn spawn<F>(&self, f: F) -> VortexResult<BoxFuture<'static, VortexResult<F::Output>>>
+    where
+        F: Future + Send + 'static,
+        <F as Future>::Output: Send + 'static,
+    {
+        match self {
+            TaskExecutor::Threads(threads_executor) => threads_executor.spawn(f),
+            #[cfg(feature = "tokio")]
+            TaskExecutor::Tokio(tokio_executor) => tokio_executor.spawn(f),
+        }
+    }
+}