lance_encoding/encodings/physical/
delta.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: Copyright The Lance Authors
3
4//! Delta encoding for non-decreasing unsigned block sequences.
5
6use super::{checked_fixed_values, try_vec_with_capacity};
7use crate::{
8    buffer::LanceBuffer,
9    compression::{BlockCompressor, BlockDecompressor, validate_delta_child_encoding},
10    data::{BlockInfo, DataBlock, FixedWidthDataBlock},
11    format::{EncodingUtils, metadata::CompressiveEncoding},
12};
13use lance_core::{Error, Result};
14
15#[derive(Debug)]
16/// Encodes adjacent differences with a concrete child block codec.
17pub struct DeltaEncoder {
18    bits_per_value: u64,
19    base: u64,
20    child: Box<dyn BlockCompressor>,
21}
22
23impl DeltaEncoder {
24    /// Creates a delta codec with the first value and child difference codec.
25    pub fn new(bits_per_value: u64, base: u64, child: Box<dyn BlockCompressor>) -> Self {
26        Self {
27            bits_per_value,
28            base,
29            child,
30        }
31    }
32}
33
34impl BlockCompressor for DeltaEncoder {
35    fn compress(&self, data: DataBlock) -> Result<(Option<LanceBuffer>, CompressiveEncoding)> {
36        let DataBlock::FixedWidth(data) = data else {
37            return Err(Error::invalid_input(
38                "Delta encoding requires a fixed-width data block",
39            ));
40        };
41        if data.bits_per_value != self.bits_per_value {
42            return Err(Error::invalid_input(format!(
43                "Delta codec expects {}-bit values, got {}",
44                self.bits_per_value, data.bits_per_value
45            )));
46        }
47        let deltas = encode_deltas(data, self.base)?;
48        let (payload, child_encoding) = self.child.compress(DataBlock::FixedWidth(deltas))?;
49        validate_delta_child_encoding(&child_encoding, self.bits_per_value)?;
50        Ok((
51            payload,
52            EncodingUtils::delta(self.bits_per_value, self.base, child_encoding),
53        ))
54    }
55}
56
57fn encode_deltas(data: FixedWidthDataBlock, expected_base: u64) -> Result<FixedWidthDataBlock> {
58    if !matches!(data.bits_per_value, 32 | 64) {
59        return Err(Error::invalid_input(format!(
60            "Delta only supports 32 or 64-bit values, got {}",
61            data.bits_per_value
62        )));
63    }
64    if data.num_values < 2 {
65        return Err(Error::invalid_input(format!(
66            "Delta requires at least 2 values, got {}",
67            data.num_values
68        )));
69    }
70
71    let deltas = match data.bits_per_value {
72        32 => {
73            let values = checked_fixed_values::<u32>(&data, "Delta input")?;
74            if u64::from(values[0]) != expected_base {
75                return Err(Error::invalid_input(format!(
76                    "Delta base mismatch: codec expects {expected_base}, input starts with {}",
77                    values[0]
78                )));
79            }
80            let mut deltas = try_vec_with_capacity::<u32>(data.num_values - 1, "Delta output")?;
81            for (index, pair) in values.windows(2).enumerate() {
82                deltas.push(pair[1].checked_sub(pair[0]).ok_or_else(|| {
83                    Error::invalid_input(format!(
84                        "Delta input decreases at index {}: {} -> {}",
85                        index + 1,
86                        pair[0],
87                        pair[1]
88                    ))
89                })?);
90            }
91            LanceBuffer::reinterpret_vec(deltas)
92        }
93        64 => {
94            let values = checked_fixed_values::<u64>(&data, "Delta input")?;
95            if values[0] != expected_base {
96                return Err(Error::invalid_input(format!(
97                    "Delta base mismatch: codec expects {expected_base}, input starts with {}",
98                    values[0]
99                )));
100            }
101            let mut deltas = try_vec_with_capacity::<u64>(data.num_values - 1, "Delta output")?;
102            for (index, pair) in values.windows(2).enumerate() {
103                deltas.push(pair[1].checked_sub(pair[0]).ok_or_else(|| {
104                    Error::invalid_input(format!(
105                        "Delta input decreases at index {}: {} -> {}",
106                        index + 1,
107                        pair[0],
108                        pair[1]
109                    ))
110                })?);
111            }
112            LanceBuffer::reinterpret_vec(deltas)
113        }
114        _ => unreachable!("delta width was validated above"),
115    };
116    Ok(FixedWidthDataBlock {
117        bits_per_value: data.bits_per_value,
118        data: deltas,
119        num_values: data.num_values - 1,
120        block_info: BlockInfo::default(),
121    })
122}
123
124#[derive(Debug)]
125pub(crate) struct DeltaDecompressor {
126    bits_per_value: u64,
127    base: u64,
128    child: Box<dyn BlockDecompressor>,
129}
130
131impl DeltaDecompressor {
132    pub(crate) fn new(bits_per_value: u64, base: u64, child: Box<dyn BlockDecompressor>) -> Self {
133        Self {
134            bits_per_value,
135            base,
136            child,
137        }
138    }
139}
140
141impl BlockDecompressor for DeltaDecompressor {
142    fn decompress(&self, data: Option<LanceBuffer>, num_values: u64) -> Result<DataBlock> {
143        validate_delta_header(self.bits_per_value, self.base, num_values)?;
144        let child = self.child.decompress(data, num_values - 1)?;
145        reconstruct_deltas(child, self.bits_per_value, self.base, num_values)
146    }
147
148    fn requires_payload(&self) -> bool {
149        self.child.requires_payload()
150    }
151}
152
153pub(crate) fn validate_delta_header(bits_per_value: u64, base: u64, num_values: u64) -> Result<()> {
154    if !matches!(bits_per_value, 32 | 64) {
155        return Err(Error::invalid_input(format!(
156            "Delta only supports 32 or 64-bit values, got {bits_per_value}"
157        )));
158    }
159    if bits_per_value == 32 && base > u32::MAX as u64 {
160        return Err(Error::invalid_input(format!(
161            "Delta base {base} exceeds u32::MAX"
162        )));
163    }
164    if num_values < 2 {
165        return Err(Error::invalid_input(format!(
166            "Delta requires at least 2 values, got {num_values}"
167        )));
168    }
169    Ok(())
170}
171
172fn reconstruct_deltas(
173    child: DataBlock,
174    bits_per_value: u64,
175    base: u64,
176    num_values: u64,
177) -> Result<DataBlock> {
178    let DataBlock::FixedWidth(child) = child else {
179        return Err(Error::invalid_input(
180            "Delta child decoded to a non fixed-width block",
181        ));
182    };
183    if child.bits_per_value != bits_per_value || child.num_values != num_values - 1 {
184        return Err(Error::invalid_input(format!(
185            "Delta child decoded {} {}-bit values, expected {} {}-bit values",
186            child.num_values,
187            child.bits_per_value,
188            num_values - 1,
189            bits_per_value
190        )));
191    }
192
193    let data = match bits_per_value {
194        32 => {
195            let deltas = checked_fixed_values::<u32>(&child, "Delta child")?;
196            let mut values = try_vec_with_capacity::<u32>(num_values, "Delta output")?;
197            let mut current = u32::try_from(base)
198                .map_err(|_| Error::invalid_input(format!("Delta base {base} exceeds u32::MAX")))?;
199            values.push(current);
200            for (index, delta) in deltas.iter().enumerate() {
201                current = current.checked_add(*delta).ok_or_else(|| {
202                    Error::invalid_input(format!(
203                        "Delta prefix sum overflows u32 at index {}",
204                        index + 1
205                    ))
206                })?;
207                values.push(current);
208            }
209            LanceBuffer::reinterpret_vec(values)
210        }
211        64 => {
212            let deltas = checked_fixed_values::<u64>(&child, "Delta child")?;
213            let mut values = try_vec_with_capacity::<u64>(num_values, "Delta output")?;
214            let mut current = base;
215            values.push(current);
216            for (index, delta) in deltas.iter().enumerate() {
217                current = current.checked_add(*delta).ok_or_else(|| {
218                    Error::invalid_input(format!(
219                        "Delta prefix sum overflows u64 at index {}",
220                        index + 1
221                    ))
222                })?;
223                values.push(current);
224            }
225            LanceBuffer::reinterpret_vec(values)
226        }
227        _ => unreachable!("delta header was validated before reconstruction"),
228    };
229
230    Ok(DataBlock::FixedWidth(FixedWidthDataBlock {
231        bits_per_value,
232        data,
233        num_values,
234        block_info: BlockInfo::default(),
235    }))
236}
237
238#[cfg(test)]
239mod tests {
240    use crate::compression::{DecompressionStrategy, DefaultDecompressionStrategy};
241    use crate::encodings::physical::{
242        constant::ConstantEncoder, range::RangeEncoder, rle::RleEncoder, value::ValueEncoder,
243    };
244
245    use super::*;
246
247    #[test]
248    fn delta_round_trip_with_payload() {
249        let input = DataBlock::FixedWidth(FixedWidthDataBlock {
250            bits_per_value: 64,
251            data: LanceBuffer::reinterpret_vec(vec![5_u64, 5, 9, 12]),
252            num_values: 4,
253            block_info: BlockInfo::default(),
254        });
255        let encoder = DeltaEncoder::new(64, 5, Box::new(ValueEncoder::default()));
256        let (payload, encoding) = encoder.compress(input).unwrap();
257        let crate::format::metadata::compressive_encoding::Compression::Delta(delta) =
258            encoding.compression.as_ref().unwrap()
259        else {
260            panic!("expected Delta encoding");
261        };
262        let child = crate::compression::DefaultDecompressionStrategy::default()
263            .create_block_decompressor(delta.deltas.as_deref().unwrap())
264            .unwrap();
265        let decoded = DeltaDecompressor::new(64, 5, child)
266            .decompress(payload, 4)
267            .unwrap()
268            .as_fixed_width()
269            .unwrap();
270        assert_eq!(
271            decoded.data.borrow_to_typed_slice::<u64>().as_ref(),
272            &[5, 5, 9, 12]
273        );
274    }
275
276    #[rstest::rstest]
277    #[case::range(vec![0_u32, 1, 3, 6], Box::new(RangeEncoder::new(32, 1, 1)))]
278    #[case::constant(vec![0_u32, 3, 6, 9], Box::new(ConstantEncoder::new(32, 3)))]
279    fn delta_propagates_metadata_only_child(
280        #[case] values: Vec<u32>,
281        #[case] child: Box<dyn BlockCompressor>,
282    ) {
283        let input = DataBlock::FixedWidth(FixedWidthDataBlock {
284            bits_per_value: 32,
285            data: LanceBuffer::reinterpret_vec(values.clone()),
286            num_values: 4,
287            block_info: BlockInfo::default(),
288        });
289        let encoder = DeltaEncoder::new(32, 0, child);
290        let (payload, encoding) = encoder.compress(input).unwrap();
291        assert!(payload.is_none());
292
293        let decoder = crate::compression::DefaultDecompressionStrategy::default()
294            .create_block_decompressor(&encoding)
295            .unwrap();
296        assert!(!decoder.requires_payload());
297        let decoded = decoder
298            .decompress(None, 4)
299            .unwrap()
300            .as_fixed_width()
301            .unwrap();
302        assert_eq!(
303            decoded.data.borrow_to_typed_slice::<u32>().as_ref(),
304            values.as_slice()
305        );
306    }
307
308    #[test]
309    fn delta_rejects_decreasing_input() {
310        let input = FixedWidthDataBlock {
311            bits_per_value: 32,
312            data: LanceBuffer::reinterpret_vec(vec![2_u32, 1]),
313            num_values: 2,
314            block_info: BlockInfo::default(),
315        };
316        let error = encode_deltas(input, 2).unwrap_err();
317        assert!(error.to_string().contains("decreases"));
318    }
319
320    #[test]
321    fn delta_rejects_unsupported_rle_child_before_emitting_encoding() {
322        let input = DataBlock::FixedWidth(FixedWidthDataBlock {
323            bits_per_value: 32,
324            data: LanceBuffer::reinterpret_vec(vec![10_u32, 10, 12, 12, 15]),
325            num_values: 5,
326            block_info: BlockInfo::default(),
327        });
328        let encoder = DeltaEncoder::new(32, 10, Box::new(RleEncoder::new()));
329
330        let error = encoder.compress(input).unwrap_err();
331
332        assert!(matches!(&error, Error::InvalidInput { .. }));
333        assert!(error.to_string().contains("does not support a rle child"));
334    }
335
336    #[test]
337    fn range_and_delta_return_errors_in_miniblock_positions() {
338        let range = EncodingUtils::range(32, 0, 1);
339        let delta = EncodingUtils::delta(32, 0, range.clone());
340        let strategy = DefaultDecompressionStrategy::default();
341
342        for encoding in [&range, &delta] {
343            let Err(error) = strategy.create_miniblock_decompressor(encoding, &strategy) else {
344                panic!("expected block-only encoding to be rejected in a mini-block position");
345            };
346            assert!(matches!(&error, Error::NotSupported { .. }));
347            assert!(
348                error
349                    .to_string()
350                    .contains("only supported in block positions")
351            );
352        }
353    }
354}
lance_encoding/encodings/physical/delta.rs

lance_encoding/encodings/physical/
delta.rs
