Prune scan conjuncts concurrently and extract length-only scan

vortex-layout/src/scan/layout.rs

@@ -126,23 +126,15 @@ impl DataSource for LayoutReaderDataSource {
             .transpose()?;
         let dtype = projection.dtype().clone();
 
-        // If the dtype is an empty struct, and there is no filter, we can return a special
-        // length-only scan.
-        if let DType::Struct(fields, Nullability::NonNullable) = &dtype
-            && fields.nfields() == 0
-            && filter.is_none()
+        if filter.is_none()
+            && let Some(empty) = length_only_scan(
+                &dtype,
+                &scan_request.selection,
+                &row_range,
+                scan_request.limit,
+            )
         {
-            // FIXME(ngates): extract out maybe?
-            let row_count = scan_request.selection.row_count_in_range(&row_range);
-
-            // Apply the limit.
-            let row_count = if let Some(limit) = scan_request.limit {
-                row_count.min(limit)
-            } else {
-                row_count
-            };
-
-            return Ok(Box::new(Empty { dtype, row_count }));
+            return Ok(Box::new(empty));
         }
 
         // Check file-level pruning: if the filter can be proven false for the entire row range
@@ -364,6 +356,31 @@ impl Partition for LayoutReaderSplit {
     }
 }
 
+/// Returns a length-only scan when the projection is an empty struct, since its output then
+/// depends only on how many rows are selected.
+///
+/// Callers must only use this for unfiltered scans, where the selected row count is exact.
+fn length_only_scan(
+    dtype: &DType,
+    selection: &Selection,
+    row_range: &Range<u64>,
+    limit: Option<u64>,
+) -> Option<Empty> {
+    let DType::Struct(fields, Nullability::NonNullable) = dtype else {
+        return None;
+    };
+    if fields.nfields() != 0 {
+        return None;
+    }
+
+    let row_count = selection.row_count_in_range(row_range);
+    let row_count = limit.map_or(row_count, |limit| row_count.min(limit));
+    Some(Empty {
+        dtype: dtype.clone(),
+        row_count,
+    })
+}
+
 /// A scan that produces no data, only empty arrays with the correct row count.
 struct Empty {
     dtype: DType,

vortex-layout/src/scan/tasks.rs

@@ -9,7 +9,9 @@ use std::sync::Arc;
 
 use bit_vec::BitVec;
 use futures::FutureExt;
+use futures::TryStreamExt;
 use futures::future::BoxFuture;
+use futures::stream::FuturesUnordered;
 use vortex_array::ArrayRef;
 use vortex_array::MaskFuture;
 use vortex_array::expr::BoundExpression;
@@ -90,22 +92,29 @@ fn build_filter_mask(
     let filter_row_range = row_range.clone();
     MaskFuture::new(row_mask.len(), async move {
         let mut mask = row_mask;
-        let mut dynamic_versions = vec![None; filter.conjuncts().len()];
+        if mask.all_false() {
+            return Ok(mask);
+        }
 
-        // TODO(ngates): we could use FuturedUnordered to intersect the masks in parallel.
-        for (idx, conjunct) in filter.conjuncts().iter().enumerate() {
+        // Store the latest version of each dynamic expression prior to pruning.
+        // We will re-run the pruning later if the version has changed in the meantime.
+        let mut dynamic_versions: Vec<_> = (0..filter.conjuncts().len())
+            .map(|idx| filter.dynamic_updates(idx).map(|du| du.version()))
+            .collect();
+
+        // Prune with every conjunct concurrently, intersecting the masks as they resolve. Each
+        // pruning sees only the input mask, and the remaining evaluations are dropped once the
+        // intersection rules out every row.
+        let mut pruning: FuturesUnordered<_> = filter
+            .conjuncts()
+            .iter()
+            .map(|conjunct| reader.pruning_evaluation(&filter_row_range, conjunct, mask.clone()))
+            .collect::<VortexResult<_>>()?;
+        while let Some(conjunct_mask) = pruning.try_next().await? {
+            mask = mask.bitand(&conjunct_mask);
             if mask.all_false() {
                 return Ok(mask);
             }
-
-            // Store the latest version of the dynamic expression prior to pruning.
-            // We will re-run the pruning later if the version has changed in the meantime.
-            dynamic_versions[idx] = filter.dynamic_updates(idx).map(|du| du.version());
-
-            let conjunct_mask = reader
-                .pruning_evaluation(&filter_row_range, conjunct, mask.clone())?
-                .await?;
-            mask = mask.bitand(&conjunct_mask);
         }
 
         // Now we loop through the conjuncts in the preferred order and evaluate them.