Widen VarBinBuilder offets during FSST compress (#7853)

## Summary

Fixes: #7833

I don't think that a more complicated solution here makes a lot of
sense. We narrow these offsets is the compressor anyways, so there is no
reason to do something more complicated than this.

Additionally adds the logic for comparing with arrow large binary arrays
that was missing before.

To be honest, I feel like it doesn't really make any sense to allow the
user to specify what the offsets and sizes types should be at all, I
think we can have the builder _always_ use the 64-bit types because we
know that they will be compressed by the compressor anyways.

## Testing

Adds some regression tests. I've ignored the big test because it
allocates too much memory.

You can run the ignored test with:

```
cargo t --release -p vortex-fsst varbin_compress_offsets_overflow_i32 -- --ignored --no-capture
```

Signed-off-by: Connor Tsui <connor.tsui20@gmail.com>

Author: connortsui20

Cargo.lock

@@ -33,6 +33,7 @@ _test-harness = ["dep:rand", "vortex-array/_test-harness"]
 divan = { workspace = true }
 rand = { workspace = true }
 rstest = { workspace = true }
+test-with = { workspace = true }
 vortex-array = { workspace = true, features = ["_test-harness"] }
 
 [[bench]]

encodings/fsst/Cargo.toml

@@ -69,8 +69,12 @@ where
     I: Iterator<Item = Option<&'a [u8]>>,
 {
     let mut buffer = Vec::with_capacity(DEFAULT_BUFFER_LEN);
-    let mut builder = VarBinBuilder::<i32>::with_capacity(len);
+
+    // Offsets are widened to i64 because the cumulative compressed bytes can exceed i32::MAX for
+    // large inputs (see issue #7833). Per-string sizes still fit in i32.
+    let mut builder = VarBinBuilder::<i64>::with_capacity(len);
     let mut uncompressed_lengths: BufferMut<i32> = BufferMut::with_capacity(len);
+
     for string in iter {
         match string {
             None => {

encodings/fsst/src/compress.rs

@@ -1,6 +1,9 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
+use rand::SeedableRng;
+use rand::rngs::StdRng;
+use rand::seq::IndexedRandom;
 use vortex_array::ArrayRef;
 use vortex_array::IntoArray;
 use vortex_array::LEGACY_SESSION;
@@ -107,3 +110,61 @@ fn test_fsst_array_ops() {
 
     assert_arrays_eq!(fsst_array, canonical_array);
 }
+
+// TODO(someone): ideally CI would run this in release mode as well since debug builds make the
+// allocation and compression loop substantially slower.
+/// Regression for #7833: [`fsst_compress`] must accept inputs whose cumulative compressed
+/// bytes exceed [`i32::MAX`]. Before the fix, [`fsst_compress_iter`] hardcoded
+/// [`VarBinBuilder<i32>`] for the FSST output and panicked in
+/// [`VarBinBuilder::append_value`] once cumulative compressed bytes crossed the boundary.
+///
+/// The input is built with [`VarBinBuilder<i64>`] so the input itself does not panic, which
+/// confirms the overflow is on the FSST output side. After the fix the test must succeed
+/// with the row count preserved.
+///
+/// Allocates ~2.5 GiB for the input and ~2.5 GiB for the FSST output (~5 GiB total), so it
+/// is gated to CI runs and skipped when `VORTEX_SKIP_SLOW_TESTS` is set. To run it locally:
+///
+/// ```text
+/// CI=1 cargo test --release -p vortex-fsst fsst_compress_offsets
+/// ```
+///
+/// [`fsst_compress_iter`]: crate::compress::fsst_compress_iter
+#[test_with::env(CI)]
+#[test_with::no_env(VORTEX_SKIP_SLOW_TESTS)]
+fn fsst_compress_offsets_overflow_i32() {
+    // High-entropy ASCII strings sliced from a random pool. FSST is a symbol-table
+    // compressor; pseudo-random data with no recurring byte sequences resists compression,
+    // so the compressed output stays close to input size and crosses the i32 boundary.
+    const STRING_LEN: usize = 64 * 1024;
+    const TOTAL_BYTES: usize = (1usize << 31) + (512 << 20); // ~2.5 GiB
+    const N: usize = TOTAL_BYTES / STRING_LEN;
+    const POOL_LEN: usize = 64 * 1024 * 1024;
+
+    // Printable ASCII alphabet so the result is valid UTF-8.
+    const ALPHABET: &[u8; 95] =
+        b" !\"#$%&'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijklmnopqrstuvwxyz{|}~";
+
+    let mut rng = StdRng::seed_from_u64(0xC0DE_C011_B711);
+    let pool: Vec<u8> = (0..POOL_LEN)
+        .map(|_| *ALPHABET.choose(&mut rng).unwrap())
+        .collect();
+
+    println!("building large VarBinArray");
+    let mut builder = VarBinBuilder::<i64>::with_capacity(N);
+    for i in 0..N {
+        let off = i.wrapping_mul(31337) % (POOL_LEN - STRING_LEN);
+        builder.append_value(&pool[off..off + STRING_LEN]);
+    }
+    let array = builder.finish(DType::Utf8(Nullability::NonNullable));
+
+    println!("training FSST compressor");
+    let compressor = fsst_train_compressor(&array);
+    let len = array.len();
+    let dtype = array.dtype().clone();
+    let mut ctx = LEGACY_SESSION.create_execution_ctx();
+
+    println!("compressing to FSST");
+    let compressed = fsst_compress(array, len, &dtype, &compressor, &mut ctx);
+    assert_eq!(compressed.len(), len);
+}

encodings/fsst/src/tests.rs

@@ -2,8 +2,11 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use arrow_array::BinaryArray;
+use arrow_array::LargeBinaryArray;
+use arrow_array::LargeStringArray;
 use arrow_array::StringArray;
 use arrow_ord::cmp;
+use arrow_schema::DataType;
 use vortex_buffer::BitBuffer;
 use vortex_error::VortexExpect as _;
 use vortex_error::VortexResult;
@@ -82,15 +85,26 @@ impl CompareKernel for VarBin {
 
             let lhs = Datum::try_new(lhs.array(), ctx)?;
 
-            // Use StringViewArray/BinaryViewArray to match the Utf8View/BinaryView types
-            // produced by Datum::try_new (which uses execute_arrow(None, ctx))
-            let arrow_rhs: &dyn arrow_array::Datum = match rhs_const.dtype() {
-                DType::Utf8(_) => &rhs_const
+            // The RHS scalar must match the LHS Arrow data type. VarBin with i64 offsets is
+            // converted to LargeBinary/LargeUtf8 (see `preferred_arrow_type`), and Arrow refuses to
+            // compare LargeBinary with Binary (or LargeUtf8 with Utf8).
+            let arrow_rhs: &dyn arrow_array::Datum = match (rhs_const.dtype(), lhs.data_type()) {
+                (DType::Utf8(_), DataType::LargeUtf8) => &rhs_const
+                    .as_utf8()
+                    .value()
+                    .map(LargeStringArray::new_scalar)
+                    .unwrap_or_else(|| arrow_array::Scalar::new(LargeStringArray::new_null(1))),
+                (DType::Utf8(_), _) => &rhs_const
                     .as_utf8()
                     .value()
                     .map(StringArray::new_scalar)
                     .unwrap_or_else(|| arrow_array::Scalar::new(StringArray::new_null(1))),
-                DType::Binary(_) => &rhs_const
+                (DType::Binary(_), DataType::LargeBinary) => &rhs_const
+                    .as_binary()
+                    .value()
+                    .map(LargeBinaryArray::new_scalar)
+                    .unwrap_or_else(|| arrow_array::Scalar::new(LargeBinaryArray::new_null(1))),
+                (DType::Binary(_), _) => &rhs_const
                     .as_binary()
                     .value()
                     .map(BinaryArray::new_scalar)
@@ -237,9 +251,14 @@ mod test {
 
 #[cfg(test)]
 mod tests {
+    use vortex_buffer::ByteBuffer;
+
     use crate::IntoArray;
+    use crate::arrays::BoolArray;
     use crate::arrays::ConstantArray;
     use crate::arrays::VarBinArray;
+    use crate::arrays::varbin::builder::VarBinBuilder;
+    use crate::assert_arrays_eq;
     use crate::builtins::ArrayBuiltins;
     use crate::dtype::DType;
     use crate::dtype::Nullability;
@@ -260,4 +279,55 @@ mod tests {
             &DType::Bool(Nullability::Nullable)
         );
     }
+
+    /// Regression: a [`VarBinArray`] built with `i64` offsets is canonicalised to
+    /// Arrow `LargeUtf8` / `LargeBinary` by `preferred_arrow_type`. Without an explicit
+    /// branch in [`CompareKernel`], the constant RHS is wrapped in a `StringArray` /
+    /// `BinaryArray` and Arrow rejects the `LargeUtf8 == Utf8` mismatch. Triggering
+    /// this only requires `i64` offsets, not large data.
+    ///
+    /// [`CompareKernel`]: super::CompareKernel
+    #[test]
+    fn varbin_i64_offsets_compare_constant() {
+        let mut builder = VarBinBuilder::<i64>::with_capacity(3);
+        builder.append_value(b"abc");
+        builder.append_value(b"xyz");
+        builder.append_value(b"abc");
+        let array = builder.finish(DType::Utf8(Nullability::NonNullable));
+
+        let result = array
+            .into_array()
+            .binary(
+                ConstantArray::new(Scalar::utf8("abc", Nullability::NonNullable), 3).into_array(),
+                Operator::Eq,
+            )
+            .unwrap();
+
+        let expected = BoolArray::from_iter([true, false, true]);
+        assert_arrays_eq!(result, expected);
+    }
+
+    #[test]
+    fn varbin_i64_offsets_compare_constant_binary() {
+        let mut builder = VarBinBuilder::<i64>::with_capacity(3);
+        builder.append_value(b"abc");
+        builder.append_value(b"xyz");
+        builder.append_value(b"abc");
+        let array = builder.finish(DType::Binary(Nullability::NonNullable));
+
+        let result = array
+            .into_array()
+            .binary(
+                ConstantArray::new(
+                    Scalar::binary(ByteBuffer::copy_from(b"abc"), Nullability::NonNullable),
+                    3,
+                )
+                .into_array(),
+                Operator::Eq,
+            )
+            .unwrap();
+
+        let expected = BoolArray::from_iter([true, false, true]);
+        assert_arrays_eq!(result, expected);
+    }
 }