strawboat 0.2.6

A native storage format based on Apache Arrow.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

use std::io::BufRead;

use arrow::array::PrimitiveArray;
use bitpacking::{BitPacker, BitPacker4x};

use arrow::error::Result;
use byteorder::ReadBytesExt;

use crate::{
    compression::{Compression, SAMPLE_COUNT, SAMPLE_SIZE},
    write::WriteOptions,
};

use super::{compress_sample_ratio, IntegerCompression, IntegerStats, IntegerType};

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct DeltaBitpacking {}

impl<T: IntegerType> IntegerCompression<T> for DeltaBitpacking {
    fn compress(
        &self,
        array: &PrimitiveArray<T>,
        _stats: &IntegerStats<T>,
        _write_options: &WriteOptions,
        output: &mut Vec<u8>,
    ) -> Result<usize> {
        let start: usize = output.len();
        let bitpacker = BitPacker4x::new();
        let my_data = bytemuck::cast_slice(array.values().as_slice());

        let mut initial = 0;
        for chunk in my_data.chunks(BitPacker4x::BLOCK_LEN) {
            let num_bits: u8 = bitpacker.num_bits(chunk);
            output.push(num_bits);
            output.reserve(BitPacker4x::BLOCK_LEN * 4);

            let out_slice = unsafe {
                core::slice::from_raw_parts_mut(
                    output.as_mut_ptr().add(output.len()) as *mut u8,
                    BitPacker4x::BLOCK_LEN * 4,
                )
            };

            let size = bitpacker.compress_sorted(initial, chunk, out_slice, num_bits);
            initial = *chunk.last().unwrap();
            unsafe { output.set_len(output.len() + size) };
        }

        Ok(output.len() - start)
    }

    fn decompress(&self, mut input: &[u8], length: usize, output: &mut Vec<T>) -> Result<()> {
        log::debug!("DeltaBitpacking::decompress {}", input.len());
        let bitpacker = BitPacker4x::new();

        let mut initial = 0;

        output.reserve(BitPacker4x::BLOCK_LEN * length);
        for _ in (0..length).step_by(BitPacker4x::BLOCK_LEN) {
            let num_bits = input.read_u8()?;
            let out_slice = unsafe {
                core::slice::from_raw_parts_mut(
                    output.as_mut_ptr().add(output.len()) as *mut u32,
                    BitPacker4x::BLOCK_LEN,
                )
            };
            let size = bitpacker.decompress_sorted(initial, input, out_slice, num_bits);
            input.consume(size);

            initial = *out_slice.last().unwrap();
            unsafe { output.set_len(output.len() + BitPacker4x::BLOCK_LEN) };
        }
        Ok(())
    }

    fn to_compression(&self) -> Compression {
        Compression::DeltaBitpacking
    }

    fn compress_ratio(&self, stats: &IntegerStats<T>) -> f64 {
        if stats.min.as_i64() < 0
            || std::mem::size_of::<T>() != 4
            || stats.src.len() % BitPacker4x::BLOCK_LEN != 0
            || !stats.is_sorted
            || stats.null_count > 0
        {
            return 0.0f64;
        }

        let bpk = super::bp::Bitpacking {};
        compress_sample_ratio(&bpk, stats, SAMPLE_COUNT, SAMPLE_SIZE) * 1.50f64
    }
}