fix: pad delta remainder with last value

encodings/fastlanes/src/delta/array/delta_compress.rs

@@ -15,6 +15,7 @@ use vortex_array::arrays::primitive::PrimitiveArrayExt;
 use vortex_array::dtype::NativePType;
 use vortex_array::match_each_unsigned_integer_ptype;
 use vortex_array::validity::Validity;
+use vortex_buffer::BitBufferMut;
 use vortex_buffer::Buffer;
 use vortex_buffer::BufferMut;
 use vortex_error::VortexResult;
@@ -40,6 +41,20 @@ pub fn delta_compress(
         let validity = match validity {
             Validity::Array(mask) => {
                 let bits = mask.execute::<BoolArray>(ctx)?.into_bit_buffer();
+                let pad = bits.len().next_multiple_of(FL_CHUNK_SIZE) - bits.len();
+                // Pad remainder bits as valid to match last-value remainder padding.
+                // `transpose_bitbuffer` would otherwise zero-fill, and those nulls scatter
+                // onto real residual slots (bit-transpose ≠ integer-transpose), where
+                // bitpacking would then skip their patches.
+                let bits = if pad == 0 {
+                    bits
+                } else {
+                    // `sliced` first so the copy covers only the logical range, not whatever
+                    // wider buffer the mask was sliced out of.
+                    let mut padded = BitBufferMut::copy_from(&bits.sliced());
+                    padded.append_n(true, pad);
+                    padded.freeze()
+                };
                 Validity::Array(
                     BoolArray::new(transpose_bitbuffer(bits), Validity::NonNullable).into_array(),
                 )
@@ -94,8 +109,10 @@ where
     }
 
     // Pad the remainder to 1024 elements and process as a full chunk.
-    if !remainder.is_empty() {
-        let mut padded_chunk = [T::default(); FL_CHUNK_SIZE];
+    if let Some(&last) = remainder.last() {
+        // Repeat the last value for padding to prevent a value-to-zero step from producing
+        // huge wrapping deltas in the padded tail (same rationale as RLE compression).
+        let mut padded_chunk = [last; FL_CHUNK_SIZE];
         padded_chunk[..remainder.len()].copy_from_slice(remainder);
         process_chunk(&padded_chunk, &mut output_deltas[full_chunks.len()]);
     }
@@ -110,12 +127,14 @@ where
 
 #[cfg(test)]
 mod tests {
+    use std::iter;
     use std::sync::LazyLock;
 
     use rstest::rstest;
     use vortex_array::IntoArray;
     use vortex_array::VortexSessionExecute;
     use vortex_array::arrays::Bool;
+    use vortex_array::arrays::BoolArray;
     use vortex_array::arrays::PrimitiveArray;
     use vortex_array::assert_arrays_eq;
     use vortex_array::validity::Validity;
@@ -125,6 +144,8 @@ mod tests {
     use vortex_session::VortexSession;
 
     use crate::Delta;
+    use crate::FL_CHUNK_SIZE;
+    use crate::bit_transpose::untranspose_bitbuffer;
     use crate::bitpack_compress::bitpack_encode;
     use crate::delta::array::delta_decompress::delta_decompress;
     use crate::delta_compress;
@@ -169,6 +190,83 @@ mod tests {
         Ok(())
     }
 
+    /// Zero-padding the trailing chunk inflated delta span on unaligned monotone columns,
+    /// causing DeltaScheme to reject encoding. Pad positions must repeat the last value.
+    #[test]
+    fn remainder_pad_preserves_small_delta_span() -> VortexResult<()> {
+        let mut ctx = SESSION.create_execution_ctx();
+        for n in [1025usize, 2049] {
+            let array = PrimitiveArray::from_iter((0..n as u32).map(|i| 1000 + i));
+            let (_bases, deltas) = delta_compress(&array, &mut ctx)?;
+            let d = deltas.as_slice::<u32>();
+            let min = *d.iter().min().unwrap();
+            let max = *d.iter().max().unwrap();
+            assert!(
+                max - min <= 1,
+                "n={n}: delta span should stay O(1), got min={min} max={max}",
+            );
+            assert!(
+                !d.iter().any(|&v| v > u32::MAX / 2),
+                "n={n}: padding must not produce wrapping deltas",
+            );
+        }
+        Ok(())
+    }
+
+    /// Padding remainder validity with `true` must not change logical nulls, including leading
+    /// and trailing nulls in the unaligned tail. After untranspose, pad bits are valid and are
+    /// sliced off by `logical_len`.
+    ///
+    /// The bit transpose is not the integer transpose, so which physical slots the pad bits land
+    /// on varies with the remainder length; cover several, plus an aligned length that pads
+    /// nothing at all.
+    #[rstest]
+    #[case::one_row_remainder(1025)]
+    #[case::mid_chunk_remainder(1500)]
+    #[case::two_chunks_plus_one(2049)]
+    #[case::one_row_short_of_aligned(3071)]
+    #[case::already_aligned(2048)]
+    fn remainder_validity_pad_does_not_clobber_logical_nulls(
+        #[case] len: usize,
+    ) -> VortexResult<()> {
+        let mut ctx = SESSION.create_execution_ctx();
+        // Nulls at both ends, so a leading null and a null inside the padded tail are covered.
+        let array = PrimitiveArray::from_option_iter(
+            iter::once(None)
+                .chain((1..len as i32 - 1).map(Some))
+                .chain(iter::once(None)),
+        );
+        assert_eq!(array.len(), len);
+
+        let (bases, deltas) = delta_compress(&array, &mut ctx)?;
+        let padded_len = len.next_multiple_of(FL_CHUNK_SIZE);
+        assert_eq!(deltas.len(), padded_len);
+
+        let Validity::Array(storage) = deltas.validity()? else {
+            vortex_bail!("expected array-backed storage validity")
+        };
+        let sequential =
+            untranspose_bitbuffer(storage.execute::<BoolArray>(&mut ctx)?.into_bit_buffer());
+        assert_eq!(sequential.len(), padded_len);
+        for i in 0..len {
+            assert_eq!(
+                sequential.value(i),
+                array.is_valid(i, &mut ctx)?,
+                "logical validity changed at {i}"
+            );
+        }
+        for i in len..padded_len {
+            assert!(sequential.value(i), "pad bit {i} should be valid");
+        }
+
+        let delta = Delta::try_new(bases.into_array(), deltas.into_array(), 0, len)?;
+        assert_eq!(delta.len(), len);
+        assert!(!delta.is_valid(0, &mut ctx)?);
+        assert!(!delta.is_valid(len - 1, &mut ctx)?);
+        assert_arrays_eq!(delta, array, &mut ctx);
+        Ok(())
+    }
+
     /// Regression test: delta + bitpacked encoding must correctly round-trip nullable arrays
     /// where null positions contain arbitrary values. Without fill-forward, the delta cumulative
     /// sum propagates corrupted values from null positions.

encodings/fastlanes/src/delta/array/mod.rs

@@ -32,7 +32,7 @@ pub struct DeltaSlots {
 ///
 /// A DeltaArray comprises a sequence of _chunks_ each representing exactly 1,024
 /// delta-encoded values. If the input array length is not a multiple of 1,024, the last chunk
-/// is padded with zeros to fill a complete 1,024-element chunk.
+/// is padded with the last value to fill a complete 1,024-element chunk.
 ///
 /// # Examples
 ///

vortex-btrblocks/src/schemes/integer/scheme_selection_tests.rs

@@ -195,6 +195,70 @@ fn test_delta_compressed() -> VortexResult<()> {
     Ok(())
 }
 
+/// Same as [`test_delta_compressed`], but with a length that is not a multiple of 1024.
+/// Zero-padding the trailing chunk used to inflate the delta span and cause DeltaScheme to skip.
+#[cfg(feature = "unstable_encodings")]
+#[test]
+fn test_delta_compressed_unaligned_length() -> VortexResult<()> {
+    let mut ctx = SESSION.create_execution_ctx();
+    use vortex_array::assert_arrays_eq;
+    use vortex_fastlanes::Delta;
+
+    let mut rng = StdRng::seed_from_u64(7u64);
+    let mut value = 500_000i32;
+    let values: Vec<i32> = (0..1025)
+        .map(|_| {
+            value += 1 + (rng.next_u32() % 6) as i32;
+            value
+        })
+        .collect();
+    let array = PrimitiveArray::new(Buffer::copy_from(&values), Validity::NonNullable);
+
+    let btr = BtrBlocksCompressor::default();
+    let compressed = btr.compress(
+        &array.clone().into_array(),
+        &mut SESSION.create_execution_ctx(),
+    )?;
+    assert!(
+        compressed.is::<Delta>(),
+        "expected Delta for unaligned near-monotone column, got tree:\n{}",
+        compressed.display_tree()
+    );
+    assert_arrays_eq!(compressed, array.into_array(), &mut ctx);
+    Ok(())
+}
+
+/// Nullable unaligned monotone must round-trip through Delta (and a cascaded residual).
+///
+/// Mirrors `duckdb/aggregate_pushdown.slt`: `NULL` then `1..=100000` (length 100001).
+#[cfg(feature = "unstable_encodings")]
+#[test]
+fn test_delta_nullable_unaligned_sum() -> VortexResult<()> {
+    use vortex_array::aggregate_fn::fns::sum::sum;
+    use vortex_array::assert_arrays_eq;
+    use vortex_fastlanes::Delta;
+
+    let mut ctx = SESSION.create_execution_ctx();
+    let array =
+        PrimitiveArray::from_option_iter(iter::once(None).chain((1i32..=100_000).map(Some)));
+
+    let btr = BtrBlocksCompressor::default();
+    let compressed = btr.compress(&array.clone().into_array(), &mut ctx)?;
+    assert!(
+        compressed.is::<Delta>(),
+        "expected Delta, got tree:\n{}",
+        compressed.display_tree()
+    );
+    assert_arrays_eq!(compressed, array.into_array(), &mut ctx);
+
+    let expected_sum: i64 = (1i64..=100_000).sum();
+    assert_eq!(
+        sum(&compressed, &mut ctx)?.as_primitive().as_::<i64>(),
+        Some(expected_sum),
+    );
+    Ok(())
+}
+
 /// Returns true if any `Delta` array appears below an ancestor `Delta` in the tree.
 #[cfg(feature = "unstable_encodings")]
 fn has_nested_delta(array: &vortex_array::ArrayRef, under_delta: bool) -> bool {

vortex-btrblocks/src/trace_tests.rs

@@ -333,7 +333,8 @@ fn trace_scan_compare_on_compressed_shipmode() -> VortexResult<()> {
 
 /// Q13-style predicate over the comment column: `l_comment LIKE '%special%'`.
 ///
-/// The column compresses to `fsst -> bitpacked lengths/offsets`.
+/// The column compresses to `fsst -> bitpacked lengths/offsets`, or to `fsst -> delta offsets`
+/// (with bitpacked residuals) when `unstable_encodings` makes Delta available.
 fn comment_predicate(column: ArrayRef, len: usize) -> VortexResult<ArrayRef> {
     Like::try_new(
         column,
@@ -358,13 +359,34 @@ fn trace_scan_like_on_compressed_comment() -> VortexResult<()> {
     // No reduce rule rewrites a like over FSST; the FSST like kernel compiles the pattern and
     // matches in compressed space at execution time.
     insta::assert_snapshot!(optimized.trace.to_string(), @"");
+    // Delta is only registered under `unstable_encodings`. Without it the offsets stay bitpacked
+    // and canonicalize inside the FSST kernel, so the scan has no extra children to execute.
+    #[cfg(not(feature = "unstable_encodings"))]
     insta::assert_snapshot!(executed.trace.to_string(), @"
     execute_until target=AnyCanonical root=vortex.like(bool, len=4096)
       iter 0 current=vortex.like(bool, len=4096) builder_active=false
         child_execute_parent session[0]:execute_parent_fn slot=0 parent=vortex.like(bool, len=4096) child=vortex.fsst(utf8, len=4096) -> vortex.bool(bool, len=4096)
       iter 1 current=vortex.bool(bool, len=4096) builder_active=false
       return output=vortex.bool(bool, len=4096)
     ");
+    #[cfg(feature = "unstable_encodings")]
+    insta::assert_snapshot!(executed.trace.to_string(), @"
+    execute_until target=AnyCanonical root=vortex.like(bool, len=4096)
+      iter 0 current=vortex.like(bool, len=4096) builder_active=false
+    execute_until target=AnyCanonical root=fastlanes.delta(u16, len=4097)
+      iter 0 current=fastlanes.delta(u16, len=4097) builder_active=false
+    execute_until target=AnyCanonical root=fastlanes.bitpacked(u16, len=5120)
+      iter 0 current=fastlanes.bitpacked(u16, len=5120) builder_active=false
+        Done array=vortex.primitive(u16, len=5120)
+      iter 1 current=vortex.primitive(u16, len=5120) builder_active=false
+      return output=vortex.primitive(u16, len=5120)
+        Done array=vortex.primitive(u16, len=4097)
+      iter 1 current=vortex.primitive(u16, len=4097) builder_active=false
+      return output=vortex.primitive(u16, len=4097)
+        child_execute_parent session[0]:execute_parent_fn slot=0 parent=vortex.like(bool, len=4096) child=vortex.fsst(utf8, len=4096) -> vortex.bool(bool, len=4096)
+      iter 1 current=vortex.bool(bool, len=4096) builder_active=false
+      return output=vortex.bool(bool, len=4096)
+    ");
 
     Ok(())
 }

vortex-btrblocks/tests/snapshots/golden__compact__list_of_int_runs.snap

@@ -3,11 +3,15 @@ source: vortex-btrblocks/tests/golden.rs
 expression: rendered
 ---
 input: list(i32), len=4066, nbytes=81804
-root: vortex.list(list(i32), len=4066) nbytes=4434
+root: vortex.list(list(i32), len=4066) nbytes=4748
   metadata: 
   elements: vortex.zigzag(i32, len=16384) nbytes=2969
     metadata: 
     encoded: vortex.pco(u32, len=16384) nbytes=2969
       metadata: ptype: u32, nrows: 16384, slice: 0..16384
-  offsets: vortex.pco(u16, len=4067) nbytes=1465
-    metadata: ptype: u16, nrows: 4067, slice: 0..4067
+  offsets: fastlanes.delta(u16, len=4067) nbytes=1779
+    metadata: offset: 0
+    bases: vortex.pco(u16, len=256) nbytes=243
+      metadata: ptype: u16, nrows: 256, slice: 0..256
+    deltas: fastlanes.bitpacked(u16, len=4096) nbytes=1536
+      metadata: bit_width: 3, offset: 0

vortex-btrblocks/tests/snapshots/golden__unstable__list_of_int_runs.snap

@@ -3,7 +3,7 @@ source: vortex-btrblocks/tests/golden.rs
 expression: rendered
 ---
 input: list(i32), len=4066, nbytes=81804
-root: vortex.list(list(i32), len=4066) nbytes=11146
+root: vortex.list(list(i32), len=4066) nbytes=6016
   metadata: 
   elements: vortex.runend(i32, len=16384) nbytes=3968
     metadata: offset: 0
@@ -15,11 +15,9 @@ root: vortex.list(list(i32), len=4066) nbytes=11146
       metadata: reference: -49931i32
       encoded: fastlanes.bitpacked(i32, len=1020) nbytes=2176
         metadata: bit_width: 17, offset: 0
-  offsets: fastlanes.bitpacked(u16, len=4067) nbytes=7178
-    metadata: bit_width: 14, offset: 0
-    patch_indices: vortex.primitive(u16, len=1) nbytes=2
+  offsets: fastlanes.delta(u16, len=4067) nbytes=2048
+    metadata: offset: 0
+    bases: vortex.primitive(u16, len=256) nbytes=512
       metadata: ptype: u16
-    patch_values: vortex.constant(u16, len=1) nbytes=4
-      metadata: scalar: 16384u16
-    patch_chunk_offsets: vortex.primitive(u8, len=4) nbytes=4
-      metadata: ptype: u8
+    deltas: fastlanes.bitpacked(u16, len=4096) nbytes=1536
+      metadata: bit_width: 3, offset: 0

vortex-btrblocks/tests/snapshots/golden__unstable__string_fsst_structured.snap

@@ -3,13 +3,31 @@ source: vortex-btrblocks/tests/golden.rs
 expression: rendered
 ---
 input: utf8, len=16384, nbytes=653785
-root: vortex.fsst(utf8, len=16384) nbytes=151382
+root: vortex.fsst(utf8, len=16384) nbytes=121766
   metadata: len: 16384, nsymbols: 223
-  uncompressed_lengths: vortex.sparse(u8, len=16384) nbytes=3154
+  uncompressed_lengths: vortex.sparse(u8, len=16384) nbytes=1802
     metadata: fill_value: 24u8
-    patch_indices: vortex.primitive(u16, len=1575) nbytes=3150
-      metadata: ptype: u16
+    patch_indices: fastlanes.delta(u16, len=1575) nbytes=1798
+      metadata: offset: 0
+      bases: vortex.primitive(u16, len=128) nbytes=256
+        metadata: ptype: u16
+      deltas: fastlanes.bitpacked(u16, len=2048) nbytes=1542
+        metadata: bit_width: 6, offset: 0
+        patch_indices: vortex.primitive(u16, len=1) nbytes=2
+          metadata: ptype: u16
+        patch_values: vortex.constant(u16, len=1) nbytes=2
+          metadata: scalar: 67u16
+        patch_chunk_offsets: vortex.primitive(u8, len=2) nbytes=2
+          metadata: ptype: u8
     patch_values: vortex.constant(u8, len=1575) nbytes=2
       metadata: scalar: 23u8
-  codes_offsets: fastlanes.bitpacked(u32, len=16385) nbytes=36992
-    metadata: bit_width: 17, offset: 0
+  codes_offsets: fastlanes.delta(u32, len=16385) nbytes=8728
+    metadata: offset: 0
+    bases: vortex.primitive(u32, len=544) nbytes=2176
+      metadata: ptype: u32
+    deltas: vortex.dict(u32, len=17408) nbytes=6552
+      metadata: all_values_referenced: true
+      codes: fastlanes.bitpacked(u8, len=17408) nbytes=6528
+        metadata: bit_width: 3, offset: 0
+      values: vortex.primitive(u32, len=6) nbytes=24
+        metadata: ptype: u32