Cap the bytes of string and binary data blocks; skip chunks a known mask excludes

SPICE_PATCHES.md

@@ -60,6 +60,8 @@ exists: a grep proves presence, a test proves behaviour.
 | 12 | Restore lint checks on forks | `bb80c537b` | Fork CI not running lints | — (CI config) | No |
 | 13 | `set_available_parallelism` | this change | Scan and writer fan-out sized from the machine's core count rather than what the host process is entitled to (spiceai/spiceai#12328) | `cargo test -p vortex-utils --test parallelism_declared --test parallelism_declared_too_late` | Proposed — additive, detection unchanged |
 | 14 | Arrow `Map` alias | `1a6dc54f1` (`lukim/map`) | `Map` columns unwritable: `Array encoding not implemented for Arrow data type Map(...)` on every flush (spiceai/spiceai#13524) | `cargo test -p vortex-arrow --lib executor::map` | Yes — 0.86.1 native `Map` (`#9111`) |
+| 17 | Byte-capped data blocks for string/binary columns, with encodings chosen once per row block | this change | A string or binary column is written in blocks of 8,192 rows whatever its row width, because `effective_block_len` sizes blocks only for fixed-width types; a ~3 KB/row JSON column gets ~12.5 MiB blocks, and a one-row lookup reads and copies a whole block (spiceai/spiceai#15323). `WriteStrategyBuilder::with_data_block_max_bytes` splits such a block into power-of-two row counts under the cap. Searching encodings on each small block costs several full compressions per block (a block of at most 1,024 rows is its own sample), so `CompressingStrategy::with_group_rows` and `CascadingCompressor::plan_for`/`compress_with_plan` choose encodings once per row block, on the sample the compressor would draw from the unsplit block, and apply them to its split blocks (a choice made on the first split block alone fits only it: incompressible rows ahead of JSON picked FSST for the JSON, +15% size and 2.4x `LIKE` CPU) | `cargo test -p vortex-layout --lib repartition::tests::a_block_size_maximum_splits_wide_strings_into_power_of_two_blocks` **and** `cargo test -p vortex-btrblocks --lib canonical_compressor::tests::a_replayed_plan_roundtrips_other_data_of_the_same_dtype` **and** `cargo test -p vortex-layout --lib compressed::tests::a_group_is_encoded_as_its_unsplit_data_would_be` **and** `cargo test -p vortex-file --test test_write_table test_data_block_max_bytes_roundtrip` | Proposed — additive; off unless the cap is set |
+| 18 | Chunked reads skip chunks a known mask excludes | this change | A read of a row range given an already-resolved mask (`MaskFuture::ready`, as a point read passes) requested every chunk in the range, even those the mask selects no row of, because the flat reader fetches its segment before it looks at the mask. Once string blocks are split under a byte cap (row 17), a point read of one row of an 8,192-row range fetched all of its split blocks, so a lookup got slower rather than faster (spiceai/spiceai#15323). `MaskFuture::known` exposes a mask known up front, and `ChunkedReader` skips chunks it excludes | `cargo test -p vortex-layout --lib chunked::reader::test::a_known_mask_reads_only_the_chunks_it_selects` | Proposed — upstream defect, not Spice behaviour |
 
 **Confidence:** rows 1–4, 13 and 14 are verified against the branches. Rows 5–12 are seeded from an
 audit of non-merge commits on `spiceai-54` authored by Spice engineers; their descriptions

vortex-array/src/mask_future.rs

@@ -20,6 +20,8 @@ use vortex_mask::Mask;
 pub struct MaskFuture {
     inner: Shared<BoxFuture<'static, SharedVortexResult<Mask>>>,
     len: usize,
+    /// The mask, when it was known before the future was created; see [`Self::known`].
+    known: Option<Mask>,
 }
 
 impl MaskFuture {
@@ -40,9 +42,19 @@ impl MaskFuture {
                 .boxed()
                 .shared(),
             len,
+            known: None,
         }
     }
 
+    /// The mask, if it is known without awaiting: this future was created with [`Self::ready`]
+    /// or [`Self::new_true`], or sliced from one that was (and not given an [`Self::inspect`]
+    /// callback, which must run).
+    ///
+    /// A reader can use it to skip work for rows the mask excludes before requesting any data.
+    pub fn known(&self) -> Option<&Mask> {
+        self.known.as_ref()
+    }
+
     /// Returns the length of the mask.
     pub fn len(&self) -> usize {
         self.len
@@ -55,7 +67,10 @@ impl MaskFuture {
 
     /// Create a MaskFuture from a ready mask.
     pub fn ready(mask: Mask) -> Self {
-        Self::new(mask.len(), async move { Ok(mask) })
+        let known = mask.clone();
+        let mut future = Self::new(mask.len(), async move { Ok(mask) });
+        future.known = Some(known);
+        future
     }
 
     /// Create a MaskFuture that resolves to a mask with all values set to true.
@@ -71,6 +86,10 @@ impl MaskFuture {
             return self.clone();
         }
 
+        if let Some(known) = &self.known {
+            return Self::ready(known.slice(range));
+        }
+
         let inner = self.inner.clone();
         Self::new(range.len(), async move { Ok(inner.await?.slice(range)) })
     }
@@ -81,9 +100,12 @@ impl MaskFuture {
     ) -> Self {
         let len = self.len;
 
+        // Not `known`: a slice of a known mask resolves without awaiting this future, and so
+        // without running `f`.
         Self {
             inner: self.inner.inspect(f).boxed().shared(),
             len,
+            known: None,
         }
     }
 }
@@ -124,3 +146,47 @@ mod tests {
         })
     }
 }
+
+#[cfg(test)]
+mod known_tests {
+    use std::sync::Arc;
+    use std::sync::atomic::AtomicBool;
+    use std::sync::atomic::Ordering;
+
+    use vortex_error::VortexResult;
+    use vortex_mask::Mask;
+
+    use super::MaskFuture;
+
+    #[test]
+    fn a_ready_mask_and_its_slices_are_known() {
+        let ready = MaskFuture::ready(Mask::from_indices(10, vec![2, 7]));
+        assert_eq!(ready.known(), Some(&Mask::from_indices(10, vec![2, 7])));
+        assert_eq!(
+            ready.slice(5..10).known(),
+            Some(&Mask::from_indices(5, vec![2]))
+        );
+        assert_eq!(MaskFuture::new_true(4).known(), Some(&Mask::new_true(4)));
+    }
+
+    #[test]
+    fn a_pending_mask_is_not_known_until_awaited() {
+        let pending = MaskFuture::new(4, async { Ok(Mask::new_true(4)) });
+        assert!(pending.known().is_none());
+        assert!(pending.slice(1..3).known().is_none());
+    }
+
+    /// A slice of an inspected future still awaits it, so the callback runs.
+    #[test]
+    fn a_slice_of_an_inspected_future_runs_the_callback() -> VortexResult<()> {
+        futures::executor::block_on(async {
+            let ran = Arc::new(AtomicBool::new(false));
+            let flag = Arc::clone(&ran);
+            let fut = MaskFuture::ready(Mask::new_true(4))
+                .inspect(move |_| flag.store(true, Ordering::SeqCst));
+            assert_eq!(fut.slice(1..3).await?, Mask::new_true(2));
+            assert!(ran.load(Ordering::SeqCst));
+            Ok(())
+        })
+    }
+}

vortex-btrblocks/Cargo.toml

@@ -40,6 +40,10 @@ vortex-zstd = { workspace = true, optional = true }
 
 [dev-dependencies]
 arrow-array = { workspace = true }
+libc = "0.2"
+onpair = { workspace = true }
+tracing = { workspace = true, features = ["std"] }
+tracing-subscriber = { workspace = true, features = ["registry"] }
 divan = { workspace = true }
 insta = { workspace = true }
 rstest = { workspace = true }
@@ -64,6 +68,11 @@ name = "compress"
 harness = false
 test = false
 
+[[bench]]
+name = "small_chunk_compress"
+harness = false
+test = false
+
 [[bench]]
 name = "compress_listview"
 harness = false