feat(encoding): add range and delta block codecs

docs/src/format/file/encoding.md

@@ -585,6 +585,8 @@ on a per-value basis. We use ☑️ to mark a technique that is applied on a per
 | Flat            | ✅ (2.1)              | ✅ (2.1)                 | ✅ (2.1)                   |
 | Variable        | ✅ (2.1)              | ✅ (2.1)                 | ✅ (2.1)                   |
 | Constant        | ✅ (2.1)              | ❓                       | ❓                         |
+| Range           | ✅ (2.3)              | ❌                       | ❓                         |
+| Delta           | ✅ (2.3)              | ❌                       | ❓                         |
 | Bitpacking      | ✅ (2.1)              | ❓                       | ✅ (2.1)                   |
 | Fsst            | ❓                    | ✅ (2.1)                 | ✅ (2.1)                   |
 | Rle             | ✅ (2.2)              | ❌                       | ✅ (2.1)                   |
@@ -594,6 +596,28 @@ on a per-value basis. We use ☑️ to mark a technique that is applied on a per
 In the following sections we will describe each technique in a bit more detail and explain how it is utilized
 in various contexts.
 
+### Generic Block Sequences
+
+Starting in Lance 2.3, block compression can describe unsigned `u32` and `u64` sequences with a bounded,
+zero-or-one-payload codec tree. The containing layout supplies the value type and cardinality.
+
+`Range` is metadata-only. It stores the unsigned width, first value, and positive step. The value at index `i`
+is `start + step * i`; readers reject cardinalities below two, overflow, and widths other than 32 or 64 bits.
+
+```protobuf
+%%% proto.message.Range %%%
+```
+
+`Delta` stores the first value inline. Its child describes the `n - 1` non-negative adjacent differences and
+determines whether the Delta tree has a payload. Zero differences are valid. Readers reconstruct the sequence
+with checked prefix sums.
+
+```protobuf
+%%% proto.message.Delta %%%
+```
+
+Lance 2.0 through 2.2 writers do not emit either encoding.
+
 ### Flat
 
 Flat compression is the uncompressed representation of fixed-width data. There is a single buffer of data

protos/encodings_v2_1.proto

@@ -426,6 +426,39 @@ message Constant {
   optional bytes value = 1;
 }
 
+// A metadata-only arithmetic sequence of unsigned fixed-width values.
+//
+// The value at index i is `start + step * i`. The container supplies the
+// number of values. Writers use this encoding only when step is positive and
+// the final value fits in the declared bit width.
+//
+// The input is a u32 or u64 fixed-width data block.
+// There is no output buffer.
+message Range {
+  // The width of each decompressed value. Must be 32 or 64.
+  uint64 uncompressed_bits_per_value = 1;
+  // The first value in the sequence.
+  uint64 start = 2;
+  // The positive difference between adjacent values.
+  uint64 step = 3;
+}
+
+// A non-decreasing unsigned fixed-width sequence represented by its first
+// value and the differences between adjacent values.
+//
+// The child contains one fewer value than the input. The container supplies
+// the input cardinality. Delta has a payload exactly when its child has one.
+//
+// The input is a u32 or u64 fixed-width data block.
+message Delta {
+  // The width of each decompressed value. Must be 32 or 64.
+  uint64 uncompressed_bits_per_value = 1;
+  // The first uncompressed value.
+  uint64 base = 2;
+  // Compression applied to the adjacent differences.
+  CompressiveEncoding deltas = 3;
+}
+
 // A compression scheme in which a single fixed-width block is "packed" into
 // a smaller fixed-width block values where each value has fewer bits.
 //
@@ -631,5 +664,7 @@ message CompressiveEncoding {
         FixedSizeList fixed_size_list = 11;
         PackedStruct packed_struct = 12;
         VariablePackedStruct variable_packed_struct = 13;
+        Range range = 14;
+        Delta delta = 15;
     }
 }

rust/lance-encoding/src/compression.rs

@@ -1209,6 +1209,11 @@ impl DecompressionStrategy for DefaultDecompressionStrategy {
             Compression::OutOfLineBitpacking(_) => Err(Error::not_supported_source(
                 "this runtime was not built with bitpacking support".into(),
             )),
+            Compression::Range(_) | Compression::Delta(_) => {
+                let value_type = block::infer_block_value_type(description)?;
+                block::create_block_decompressor(description, value_type)
+                    .map(|(decompressor, _)| decompressor)
+            }
             Compression::Rle(rle) => Ok(Box::new(create_rle_decompressor(rle, self)?)),
             Compression::Variable(variable) => {
                 let offsets = variable.offsets.as_deref().ok_or_else(|| {
@@ -1443,6 +1448,8 @@ fn compression_name(compression: &Compression) -> &'static str {
         Compression::FixedSizeList(_) => "fixed-size list",
         Compression::VariablePackedStruct(_) => "variable packed struct",
         Compression::Rle(_) => "rle",
+        Compression::Range(_) => "range",
+        Compression::Delta(_) => "delta",
     }
 }
 

rust/lance-encoding/src/compression/block.rs

@@ -51,7 +51,6 @@ impl BlockValueType {
         (self.bits_per_value() / 8) as usize
     }
 
-    #[cfg(test)]
     pub(crate) fn max_value(self) -> u64 {
         match self {
             Self::UInt8 => u8::MAX as u64,
@@ -67,7 +66,9 @@ pub(crate) mod fixed;
 
 #[cfg(test)]
 pub(crate) use factory::encode_scalar;
-pub(crate) use factory::{create_block_decompressor, validate_fixed_payload_len};
+pub(crate) use factory::{
+    create_block_decompressor, infer_block_value_type, validate_fixed_payload_len,
+};
 #[cfg(feature = "bitpacking")]
 pub(crate) use factory::{validate_inline_bitpacking_payload, validate_out_of_line_payload};
 #[cfg(test)]

rust/lance-encoding/src/compression/block/factory.rs

@@ -13,7 +13,9 @@ use crate::{
     encodings::physical::{
         block::{CompressionConfig, CompressionScheme},
         constant::ConstantBlockDecompressor,
+        delta::DeltaDecompressor,
         general::GenericGeneralBlockDecompressor,
+        range::RangeDecompressor,
         rle::{BlockRleDecompressor, BlockRunCount, MetadataRunLengths},
         value::FixedWidthBlockDecompressor,
     },
@@ -198,6 +200,65 @@ fn create_inner(
                 false,
             ))
         }
+        Compression::Range(range) => {
+            validate_declared_bits(range.uncompressed_bits_per_value, expected_type, "Range")?;
+            if !matches!(
+                expected_type,
+                BlockValueType::UInt32 | BlockValueType::UInt64
+            ) {
+                return Err(Error::invalid_input(format!(
+                    "Range only supports 32 or 64-bit values, got {expected_bits}"
+                )));
+            }
+            if range.start > expected_type.max_value() {
+                return Err(Error::invalid_input(format!(
+                    "Range start {} exceeds the {expected_bits}-bit value range",
+                    range.start
+                )));
+            }
+            if range.step == 0 {
+                return Err(Error::invalid_input("Range step must be positive"));
+            }
+            Ok((
+                Box::new(RangeDecompressor::new(
+                    expected_bits,
+                    range.start,
+                    range.step,
+                )),
+                false,
+            ))
+        }
+        Compression::Delta(delta) => {
+            if position != Position::Root {
+                return Err(Error::invalid_input(
+                    "Delta is not supported as a block codec child",
+                ));
+            }
+            validate_declared_bits(delta.uncompressed_bits_per_value, expected_type, "Delta")?;
+            if !matches!(
+                expected_type,
+                BlockValueType::UInt32 | BlockValueType::UInt64
+            ) {
+                return Err(Error::invalid_input(format!(
+                    "Delta only supports 32 or 64-bit values, got {expected_bits}"
+                )));
+            }
+            if delta.base > expected_type.max_value() {
+                return Err(Error::invalid_input(format!(
+                    "Delta base {} exceeds the {expected_bits}-bit value range",
+                    delta.base
+                )));
+            }
+            let child = delta.deltas.as_deref().ok_or_else(|| {
+                Error::invalid_input("Delta is missing its deltas child encoding")
+            })?;
+            let (child, child_has_payload) =
+                create_inner(child, expected_type, Position::Child, false)?;
+            Ok((
+                Box::new(DeltaDecompressor::new(expected_bits, delta.base, child)),
+                child_has_payload,
+            ))
+        }
         Compression::InlineBitpacking(bitpacking) => {
             validate_declared_bits(
                 bitpacking.uncompressed_bits_per_value,
@@ -332,6 +393,17 @@ fn create_inner(
                     )?;
                     Some(MetadataRunLengths::Constant(value))
                 }
+                Some(Compression::Range(range)) => {
+                    validate_declared_bits(
+                        range.uncompressed_bits_per_value,
+                        run_length_type,
+                        "RLE run lengths Range",
+                    )?;
+                    Some(MetadataRunLengths::Range {
+                        start: range.start,
+                        step: range.step,
+                    })
+                }
                 _ => None,
             };
             let run_count = if let Some(metadata) = metadata_run_lengths {
@@ -444,6 +516,11 @@ fn validate_compression_config(
     Ok(CompressionConfig::new(scheme, compression.level))
 }
 
+/// Infers the unsigned fixed-width result of a bounded block descriptor.
+pub fn infer_block_value_type(encoding: &CompressiveEncoding) -> Result<BlockValueType> {
+    infer_inner(encoding, Position::Root)
+}
+
 fn infer_inner(encoding: &CompressiveEncoding, position: Position) -> Result<BlockValueType> {
     let compression = encoding
         .compression
@@ -465,6 +542,10 @@ fn infer_inner(encoding: &CompressiveEncoding, position: Position) -> Result<Blo
         Compression::OutOfLineBitpacking(bitpacking) => {
             BlockValueType::from_bits(bitpacking.uncompressed_bits_per_value)
         }
+        Compression::Range(range) => BlockValueType::from_bits(range.uncompressed_bits_per_value),
+        Compression::Delta(delta) if position == Position::Root => {
+            BlockValueType::from_bits(delta.uncompressed_bits_per_value)
+        }
         Compression::General(general) if position == Position::Root => infer_inner(
             general
                 .values
@@ -500,6 +581,8 @@ fn compression_name(compression: &Compression) -> &'static str {
         Compression::FixedSizeList(_) => "fixed-size list",
         Compression::PackedStruct(_) => "packed struct",
         Compression::VariablePackedStruct(_) => "variable packed struct",
+        Compression::Range(_) => "range",
+        Compression::Delta(_) => "delta",
     }
 }
 
@@ -530,4 +613,19 @@ mod tests {
         let error = create_block_decompressor(&flat, BlockValueType::UInt64).unwrap_err();
         assert!(error.to_string().contains("expected 64"));
     }
+
+    #[test]
+    fn validates_range_cardinality_when_decoding() {
+        let encoding = crate::format::ProtobufUtils21::range(32, u32::MAX as u64, 1);
+        let (decoder, has_payload) =
+            create_block_decompressor(&encoding, BlockValueType::UInt32).unwrap();
+        assert!(!has_payload);
+        let error = decoder.decompress(None, 2).unwrap_err();
+        assert!(error.to_string().contains("exceeds u32::MAX"));
+    }
+
+    #[test]
+    fn range_helper_rejects_overflow() {
+        assert!(crate::encodings::physical::range::checked_range_last(64, u64::MAX, 1, 2).is_err());
+    }
 }

rust/lance-encoding/src/compression/block/tests.rs

@@ -8,7 +8,7 @@ use crate::{
     compression::BlockCompressor,
     data::{BlockInfo, DataBlock, FixedWidthDataBlock},
     encodings::physical::{
-        constant::ConstantBlockCompressor, rle::BlockRleCompressor,
+        constant::ConstantBlockCompressor, range::RangeEncoder, rle::BlockRleCompressor,
         value::FixedWidthBlockCompressor,
     },
     format::{ProtobufUtils21, pb21::CompressiveEncoding},
@@ -30,6 +30,13 @@ fn decoded_u64(block: DataBlock) -> Vec<u64> {
     block.data.borrow_to_typed_slice::<u64>().to_vec()
 }
 
+fn decoded_u32(block: DataBlock) -> Vec<u32> {
+    let DataBlock::FixedWidth(block) = block else {
+        panic!("expected fixed-width output");
+    };
+    block.data.borrow_to_typed_slice::<u32>().to_vec()
+}
+
 #[test]
 fn scalar_encoding_is_little_endian_and_bounded() {
     assert_eq!(
@@ -73,6 +80,12 @@ fn metadata_compressors_validate_and_decode() {
         )),
         ProtobufUtils21::constant(Some(7_u64.to_le_bytes().to_vec().into())),
     );
+    let values = (0..32_u64).map(|value| 5 + value * 3).collect::<Vec<_>>();
+    round_trip_u64(
+        &values,
+        Box::new(RangeEncoder::new(64, 5, 3)),
+        ProtobufUtils21::range(64, 5, 3),
+    );
 }
 
 #[test]
@@ -217,6 +230,40 @@ fn constant_cardinality_contract_is_checked_at_decode() {
     );
 }
 
+#[test]
+fn range_checks_cardinality_and_overflow() {
+    let (decoder, has_payload) =
+        create_block_decompressor(&ProtobufUtils21::range(32, 3, 5), BlockValueType::UInt32)
+            .unwrap();
+    assert!(!has_payload);
+    assert_eq!(
+        decoded_u32(decoder.decompress(None, 4).unwrap()),
+        vec![3, 8, 13, 18]
+    );
+    assert!(decoder.decompress(None, 1).is_err());
+
+    let (overflowing, has_payload) = create_block_decompressor(
+        &ProtobufUtils21::range(32, u32::MAX as u64, 1),
+        BlockValueType::UInt32,
+    )
+    .unwrap();
+    assert!(!has_payload);
+    assert!(overflowing.decompress(None, 2).is_err());
+}
+
+#[test]
+fn delta_checks_prefix_sum_overflow() {
+    let encoding = ProtobufUtils21::delta(
+        64,
+        u64::MAX,
+        ProtobufUtils21::constant(Some(1_u64.to_le_bytes().to_vec().into())),
+    );
+    let (decoder, has_payload) =
+        create_block_decompressor(&encoding, BlockValueType::UInt64).unwrap();
+    assert!(!has_payload);
+    assert!(decoder.decompress(None, 2).is_err());
+}
+
 #[test]
 fn metadata_only_rle_round_trip() {
     let encoding = ProtobufUtils21::rle(
@@ -233,6 +280,22 @@ fn metadata_only_rle_round_trip() {
     assert!(decoder.decompress(None, 5).is_err());
 }
 
+#[test]
+fn metadata_range_rle_round_trip() {
+    let encoding = ProtobufUtils21::rle(
+        ProtobufUtils21::range(64, 10, 1),
+        ProtobufUtils21::range(32, 1, 1),
+    );
+    let (decoder, has_payload) =
+        create_block_decompressor(&encoding, BlockValueType::UInt64).unwrap();
+    assert!(!has_payload);
+    assert_eq!(
+        decoded_u64(decoder.decompress(None, 6).unwrap()),
+        vec![10, 11, 11, 12, 12, 12]
+    );
+    assert!(decoder.decompress(None, 7).is_err());
+}
+
 #[test]
 fn rle_framing_is_fallible() {
     let encoding = ProtobufUtils21::rle(

rust/lance-encoding/src/encodings/logical/primitive/sparse.rs

@@ -1386,8 +1386,10 @@ impl SparseStructuralScheduler {
                             .filter_map(|field| field.value.as_ref()),
                     );
                 }
+                Compression::Delta(delta) => stack.extend(delta.deltas.as_deref()),
                 Compression::Flat(_)
                 | Compression::Constant(_)
+                | Compression::Range(_)
                 | Compression::InlineBitpacking(_) => {}
             }
         }
@@ -1535,7 +1537,9 @@ impl SparseStructuralScheduler {
             Compression::Constant(_)
             | Compression::OutOfLineBitpacking(_)
             | Compression::Dictionary(_)
-            | Compression::VariablePackedStruct(_) => Err(Error::invalid_input_source(
+            | Compression::VariablePackedStruct(_)
+            | Compression::Range(_)
+            | Compression::Delta(_) => Err(Error::invalid_input_source(
                 "Sparse value compression uses an unsupported mini-block encoding".into(),
             )),
         }