diff --git a/parquet-testing b/parquet-testing index ffdcbb5e2282..e7d32dac5b4c 160000 --- a/parquet-testing +++ b/parquet-testing @@ -1 +1 @@ -Subproject commit ffdcbb5e22828186c7461e56dbd26a0fe3caee56 +Subproject commit e7d32dac5b4cbf017fefd598a03686f53370b292 diff --git a/parquet/Cargo.toml b/parquet/Cargo.toml index 181f5466aa5b..f4ce5aaf8550 100644 --- a/parquet/Cargo.toml +++ b/parquet/Cargo.toml @@ -89,7 +89,8 @@ zstd = { version = "0.13", default-features = false } serde_json = { version = "1.0", features = ["std"], default-features = false } arrow = { workspace = true, features = ["ipc", "test_utils", "prettyprint", "json"] } arrow-cast = { workspace = true } -tokio = { version = "1.0", default-features = false, features = ["macros", "rt-multi-thread", "io-util", "fs", "sync"] } +arrow-csv = { workspace = true } +tokio = { version = "1.0", default-features = false, features = ["macros", "rt-multi-thread", "io-util", "fs"] } rand = { version = "0.9", default-features = false, features = ["std", "std_rng", "thread_rng"] } object_store = { workspace = true, features = ["azure", "fs"] } sysinfo = { version = "0.39.6", default-features = false, features = ["system"] } diff --git a/parquet/src/arrow/arrow_writer/mod.rs b/parquet/src/arrow/arrow_writer/mod.rs index 207b4c3327ef..86b5a4b391f8 100644 --- a/parquet/src/arrow/arrow_writer/mod.rs +++ b/parquet/src/arrow/arrow_writer/mod.rs @@ -3236,7 +3236,7 @@ mod tests { Encoding::BYTE_STREAM_SPLIT, ], DataType::Float32 | DataType::Float64 => { - vec![Encoding::PLAIN, Encoding::BYTE_STREAM_SPLIT] + vec![Encoding::PLAIN, Encoding::BYTE_STREAM_SPLIT, Encoding::ALP] } _ => vec![Encoding::PLAIN], }; diff --git a/parquet/src/basic.rs b/parquet/src/basic.rs index b4f18f311723..e9ae1364c323 100644 --- a/parquet/src/basic.rs +++ b/parquet/src/basic.rs @@ -449,6 +449,10 @@ enum Encoding { /// afterwards. Note that the use of this encoding with FIXED_LEN_BYTE_ARRAY(N) data may /// perform poorly for large values of N. BYTE_STREAM_SPLIT = 9; + /// Adaptive Lossless floating-Point encoding (ALP). + /// + /// Currently specified for FLOAT and DOUBLE. + ALP = 10; } ); @@ -469,6 +473,7 @@ impl FromStr for Encoding { "DELTA_BYTE_ARRAY" | "delta_byte_array" => Ok(Encoding::DELTA_BYTE_ARRAY), "RLE_DICTIONARY" | "rle_dictionary" => Ok(Encoding::RLE_DICTIONARY), "BYTE_STREAM_SPLIT" | "byte_stream_split" => Ok(Encoding::BYTE_STREAM_SPLIT), + "ALP" | "alp" => Ok(Encoding::ALP), _ => Err(general_err!("unknown encoding: {}", s)), } } @@ -608,6 +613,7 @@ fn i32_to_encoding(val: i32) -> Encoding { 7 => Encoding::DELTA_BYTE_ARRAY, 8 => Encoding::RLE_DICTIONARY, 9 => Encoding::BYTE_STREAM_SPLIT, + 10 => Encoding::ALP, _ => panic!("Impossible encoding {val}"), } } @@ -1866,6 +1872,7 @@ mod tests { ); assert_eq!(Encoding::DELTA_BYTE_ARRAY.to_string(), "DELTA_BYTE_ARRAY"); assert_eq!(Encoding::RLE_DICTIONARY.to_string(), "RLE_DICTIONARY"); + assert_eq!(Encoding::ALP.to_string(), "ALP"); } #[test] @@ -2174,6 +2181,8 @@ mod tests { assert_eq!(encoding, Encoding::RLE_DICTIONARY); encoding = "BYTE_STREAM_SPLIT".parse().unwrap(); assert_eq!(encoding, Encoding::BYTE_STREAM_SPLIT); + encoding = "alp".parse().unwrap(); + assert_eq!(encoding, Encoding::ALP); // test lowercase encoding = "byte_stream_split".parse().unwrap(); @@ -2309,6 +2318,7 @@ mod tests { Encoding::PLAIN_DICTIONARY, Encoding::RLE_DICTIONARY, Encoding::BYTE_STREAM_SPLIT, + Encoding::ALP, ]; encodings_roundtrip(encodings.into()); } diff --git a/parquet/src/data_type.rs b/parquet/src/data_type.rs index d8c7b9201389..52e9c9e0d9df 100644 --- a/parquet/src/data_type.rs +++ b/parquet/src/data_type.rs @@ -702,6 +702,7 @@ pub(crate) mod private { + Send + HeapSize + crate::encodings::decoding::private::GetDecoder + + crate::encodings::encoding::private::GetEncoder + crate::file::statistics::private::MakeStatistics { const PHYSICAL_TYPE: Type; diff --git a/parquet/src/encodings/alp.rs b/parquet/src/encodings/alp.rs new file mode 100644 index 000000000000..c8ef5752f986 --- /dev/null +++ b/parquet/src/encodings/alp.rs @@ -0,0 +1,584 @@ +// 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. + +//! ALP (Adaptive Lossless floating-Point) Encoding +//! +//! Spec: +//! +//! # Page layout +//! +//! An ALP-encoded page consists of a fixed-size header, an offset array +//! locating each vector inside the body, and the vector data itself: +//! +//! ```text +//! +-------------+-----------------------------+--------------------------------------+ +//! | Header | Offset Array | Vector Data | +//! | (7 bytes) | (num_vectors * 4 bytes) | (variable) | +//! +-------------+------+------+-----+---------+----------+----------+-----+----------+ +//! | Page Header | off0 | off1 | ... | off N-1 | Vector 0 | Vector 1 | ... | Vec N-1 | +//! | (7 bytes) | (4B) | (4B) | | (4B) |(variable)|(variable)| |(variable)| +//! +-------------+------+------+-----+---------+----------+----------+-----+----------+ +//! ``` +//! +//! Each vector entry has the form +//! `[AlpInfo][ForInfo][PackedValues][ExceptionPositions][ExceptionValues]`. + +use crate::errors::{ParquetError, Result}; +use crate::util::bit_util::{FromBitpacked, FromBytes}; + +pub(crate) const ALP_HEADER_SIZE: usize = 7; +pub(crate) const ALP_COMPRESSION_MODE: u8 = 0; +pub(crate) const ALP_INTEGER_ENCODING_FOR_BIT_PACK: u8 = 0; +pub(crate) const ALP_MIN_LOG_VECTOR_SIZE: u8 = 3; +pub(crate) const ALP_MAX_LOG_VECTOR_SIZE: u8 = 15; +/// Spec-recommended default `log_vector_size`: 1024-value vectors. +pub(crate) const ALP_DEFAULT_LOG_VECTOR_SIZE: u8 = 10; +pub(crate) const ALP_MAX_EXPONENT_F32: u8 = 10; +pub(crate) const ALP_MAX_EXPONENT_F64: u8 = 18; + +/// Page-level ALP header (7 bytes). +/// +/// ```text +/// Byte: 0 1 2 3 4 5 6 +/// +----------------+---------------+--------------+----+----+----+----+ +/// | compression | integer | log_vector | num_elements | +/// | _mode | _encoding | _size | (int32 LE) | +/// +----------------+---------------+--------------+----+----+----+----+ +/// ``` +/// +/// Layout in bytes: +/// - `[0]` `compression_mode` +/// - `[1]` `integer_encoding` +/// - `[2]` `log_vector_size` +/// - `[3..7]` `num_elements` (little-endian `i32`) +/// +/// The fields hold the *decoded* values used throughout the decoder, not the +/// raw on-disk encoding: +/// - `num_elements` is stored on disk as an `i32`, kept in memory as a `usize`. +/// - vector size is stored on disk as a `u8` `log_vector_size`, kept in memory +/// as the actual `vector_size` (`1 << log_vector_size`) `usize`. +/// +/// Each conversion happens once, in [`AlpHeader::deserialize`] and +/// [`AlpHeader::serialize`], so the rest of the decoder computes offsets and +/// sizes in `usize`. Those methods reject only what the target type cannot +/// represent; spec-level validity, such as the allowed vector-size range, is +/// enforced by the page parser. +#[derive(Debug, Clone, Copy)] +pub(crate) struct AlpHeader { + pub(crate) compression_mode: u8, + pub(crate) integer_encoding: u8, + pub(crate) vector_size: usize, + pub(crate) num_elements: usize, +} + +impl AlpHeader { + /// Parse a 7-byte page header from its little-endian on-disk form, + /// converting each field to its in-memory type: + /// - `log_vector_size` (`u8`) is expanded to `vector_size` with an + /// overflow-checked shift. + /// - `num_elements` (`i32`) is checked for non-negativity. + pub(crate) fn deserialize(bytes: &[u8]) -> Result { + if bytes.len() < ALP_HEADER_SIZE { + return Err(general_err!( + "Invalid ALP page: expected at least {} bytes for header, got {}", + ALP_HEADER_SIZE, + bytes.len() + )); + } + + let log_vector_size = bytes[2]; + let vector_size = 1usize + .checked_shl(u32::from(log_vector_size)) + .ok_or_else(|| { + general_err!( + "Invalid ALP page: log_vector_size {} too large to represent a vector size", + log_vector_size + ) + })?; + + let num_elements_i32 = i32::from_le_bytes([bytes[3], bytes[4], bytes[5], bytes[6]]); + let num_elements = usize::try_from(num_elements_i32).map_err(|_| { + general_err!( + "Invalid ALP page: num_elements {} must be >= 0", + num_elements_i32 + ) + })?; + + Ok(Self { + compression_mode: bytes[0], + integer_encoding: bytes[1], + vector_size, + num_elements, + }) + } + + /// Serialize this header into its 7-byte little-endian on-disk form. + /// + /// Converts the in-memory values back to the on-disk encoding, rejecting + /// what cannot be represented: `vector_size` must be a power of two (its log + /// is the on-disk field), and `num_elements` must fit in an `i32`. + /// Counterpart to [`AlpHeader::deserialize`]; consumed by the ALP encoder. + pub(crate) fn serialize(&self) -> Result<[u8; ALP_HEADER_SIZE]> { + if !self.vector_size.is_power_of_two() { + return Err(general_err!( + "Invalid ALP page: vector_size {} is not a power of two", + self.vector_size + )); + } + let log_vector_size = self.vector_size.trailing_zeros() as u8; + + let num_elements = i32::try_from(self.num_elements).map_err(|_| { + general_err!( + "Invalid ALP page: num_elements {} exceeds i32::MAX", + self.num_elements + ) + })?; + + let mut out = [0u8; ALP_HEADER_SIZE]; + out[0] = self.compression_mode; + out[1] = self.integer_encoding; + out[2] = log_vector_size; + out[3..7].copy_from_slice(&num_elements.to_le_bytes()); + Ok(out) + } + + /// `vector_size` is always `1 << log_vector_size` (see [`AlpHeader::deserialize`]), + /// so the division a `div_ceil` would emit is a shift. `vector_size` is a + /// runtime value, so the compiler cannot see that on its own. + pub(crate) fn num_vectors(&self) -> usize { + debug_assert!(self.vector_size.is_power_of_two()); + (self.num_elements + self.vector_size - 1) >> self.vector_size.trailing_zeros() + } + + /// Number of elements in vector `vector_index`: a full vector, the short + /// trailing remainder, or zero past the end of the page. + pub(crate) fn vector_num_elements(&self, vector_index: usize) -> u16 { + let start = vector_index.saturating_mul(self.vector_size); + let remaining = self.num_elements.saturating_sub(start); + remaining.min(self.vector_size) as u16 + } +} + +/// Per-vector ALP metadata (4 bytes). +/// +/// ```text +/// Byte: 0 1 2 3 +/// +----------+----------+---------+---------+ +/// | exponent | factor | num_exceptions | +/// | (uint8) | (uint8) | (uint16 LE) | +/// +----------+----------+---------+---------+ +/// ``` +#[derive(Debug, Clone, Copy)] +pub(crate) struct AlpInfo { + pub(crate) exponent: u8, + pub(crate) factor: u8, + pub(crate) num_exceptions: u16, +} + +impl AlpInfo { + pub(crate) const STORED_SIZE: usize = 4; + + /// Append this vector's ALP metadata in its on-disk little-endian form. + pub(crate) fn extend_serialized(&self, out: &mut Vec) { + out.push(self.exponent); + out.push(self.factor); + out.extend_from_slice(&self.num_exceptions.to_le_bytes()); + } +} + +/// Per-vector FOR (frame of reference) metadata: 5 bytes for `f32`, 9 for `f64`. +/// +/// ```text +/// +--------------------+-----------+ +/// | frame_of_reference | bit_width | +/// | (Exact::WIDTH, LE) | (uint8) | +/// +--------------------+-----------+ +/// ``` +#[derive(Debug, Clone, Copy)] +pub(crate) struct ForInfo { + pub(crate) frame_of_reference: Exact, + pub(crate) bit_width: u8, +} + +impl ForInfo { + pub(crate) fn stored_size() -> usize { + Exact::WIDTH + 1 + } + + /// Append this vector's FOR metadata in its on-disk little-endian form. + pub(crate) fn extend_serialized(&self, out: &mut Vec) { + self.frame_of_reference.extend_le_bytes(out); + out.push(self.bit_width); + } + + pub(crate) fn get_bit_packed_size(&self, num_elements: u16) -> usize { + (self.bit_width as usize * num_elements as usize).div_ceil(8) + } + + pub(crate) fn get_data_stored_size(&self, num_elements: u16, num_exceptions: u16) -> usize { + let bit_packed_size = self.get_bit_packed_size(num_elements); + bit_packed_size + + num_exceptions as usize * std::mem::size_of::() + + num_exceptions as usize * Exact::WIDTH + } +} + +/// Exact integer type used by FOR reconstruction: `u32` for `f32`, `u64` for +/// `f64`. +/// +/// Why unsigned (not `i32`/`i64`)? The spec computes and stores deltas in +/// unsigned wrapping arithmetic: this avoids signed overflow when a vector's +/// range exceeds the signed maximum, and unpacking needs no sign extension. +/// Signed interpretation is applied later during decimal reconstruction. +pub(crate) trait AlpExact: + Copy + std::fmt::Debug + PartialEq + FromBitpacked + Default +{ + const WIDTH: usize; + type Signed: Copy + Ord + std::fmt::Debug + Send; + fn from_le_slice(slice: &[u8]) -> Self; + fn wrapping_add(self, rhs: Self) -> Self; + fn wrapping_sub(self, rhs: Self) -> Self; + fn reinterpret_as_signed(self) -> Self::Signed; + fn reinterpret_from_signed(signed: Self::Signed) -> Self; + /// Widen to `u64` for bit-packing, which is `u64`-oriented throughout. + fn to_u64(self) -> u64; + fn extend_le_bytes(self, out: &mut Vec); +} + +impl AlpExact for u32 { + const WIDTH: usize = 4; + type Signed = i32; + + fn from_le_slice(slice: &[u8]) -> Self { + u32::from_le_bytes([slice[0], slice[1], slice[2], slice[3]]) + } + + fn wrapping_add(self, rhs: Self) -> Self { + self.wrapping_add(rhs) + } + + fn wrapping_sub(self, rhs: Self) -> Self { + self.wrapping_sub(rhs) + } + + fn reinterpret_as_signed(self) -> Self::Signed { + i32::from_ne_bytes(self.to_ne_bytes()) + } + + fn reinterpret_from_signed(signed: Self::Signed) -> Self { + u32::from_ne_bytes(signed.to_ne_bytes()) + } + + fn to_u64(self) -> u64 { + u64::from(self) + } + + fn extend_le_bytes(self, out: &mut Vec) { + out.extend_from_slice(&self.to_le_bytes()); + } +} + +impl AlpExact for u64 { + const WIDTH: usize = 8; + type Signed = i64; + + fn from_le_slice(slice: &[u8]) -> Self { + u64::from_le_bytes([ + slice[0], slice[1], slice[2], slice[3], slice[4], slice[5], slice[6], slice[7], + ]) + } + + fn wrapping_add(self, rhs: Self) -> Self { + self.wrapping_add(rhs) + } + + fn wrapping_sub(self, rhs: Self) -> Self { + self.wrapping_sub(rhs) + } + + fn reinterpret_as_signed(self) -> Self::Signed { + i64::from_ne_bytes(self.to_ne_bytes()) + } + + fn reinterpret_from_signed(signed: Self::Signed) -> Self { + u64::from_ne_bytes(signed.to_ne_bytes()) + } + + fn to_u64(self) -> u64 { + self + } + + fn extend_le_bytes(self, out: &mut Vec) { + out.extend_from_slice(&self.to_le_bytes()); + } +} +pub(crate) const ALP_POW10_F32: [f32; 11] = [ + 1.0, + 10.0, + 100.0, + 1000.0, + 10000.0, + 100000.0, + 1000000.0, + 10000000.0, + 100000000.0, + 1000000000.0, + 10000000000.0, +]; + +pub(crate) const ALP_POW10_F64: [f64; 19] = [ + 1.0, + 10.0, + 100.0, + 1000.0, + 10000.0, + 100000.0, + 1000000.0, + 10000000.0, + 100000000.0, + 1000000000.0, + 10000000000.0, + 100000000000.0, + 1000000000000.0, + 10000000000000.0, + 100000000000000.0, + 1000000000000000.0, + 10000000000000000.0, + 100000000000000000.0, + 1000000000000000000.0, +]; + +pub(crate) const ALP_NEG_POW10_F32: [f32; 11] = [ + 1.0, + 0.1, + 0.01, + 0.001, + 0.0001, + 0.00001, + 0.000001, + 0.0000001, + 0.00000001, + 0.000000001, + 0.0000000001, +]; + +pub(crate) const ALP_NEG_POW10_F64: [f64; 19] = [ + 1.0, + 0.1, + 0.01, + 0.001, + 0.0001, + 0.00001, + 0.000001, + 0.0000001, + 0.00000001, + 0.000000001, + 0.0000000001, + 0.00000000001, + 0.000000000001, + 0.0000000000001, + 0.00000000000001, + 0.000000000000001, + 0.0000000000000001, + 0.00000000000000001, + 0.000000000000000001, +]; + +pub(crate) trait AlpFloat: + Copy + Default + PartialEq + std::ops::Mul +{ + type Exact: AlpExact + FromBytes; + type Scale: Copy + Send; + + /// Largest `exponent` this type admits: 10 for `f32`, 18 for `f64`. + const MAX_EXPONENT: u8; + + /// Rounding magic number: `2^22 + 2^23` (`f32`) or `2^51 + 2^52` (`f64`). + const MAGIC_NUMBER: Self; + + /// Bounds outside which the scaled value cannot reach the exact integer + /// type: `i32` for `f32`, `i64` for `f64`. + const ENCODING_UPPER_LIMIT: Self; + const ENCODING_LOWER_LIMIT: Self; + + /// [`AlpFloat::ENCODING_UPPER_LIMIT`] as the exact signed integer. Stands in + /// for values ALP cannot represent, so that the round-trip check that + /// follows fails and the value is recorded as an exception. + const ENCODING_SENTINEL: ::Signed; + + /// Precompute vector-level ALP decimal scale constants for: + /// `value = (encoded * 10^(factor)) * 10^(-exponent)`. + /// + /// Preconditions are validated during page parse. + fn decode_scale(exponent: u8, factor: u8) -> Self::Scale; + + /// Decode one signed exact integer using a precomputed two-step scale. + fn decode_value(signed_encoded: ::Signed, scale: Self::Scale) -> Self; + + fn from_exact_bits(bits: Self::Exact) -> Self; + + fn to_exact_bits(self) -> Self::Exact; + + /// Precompute vector-level ALP decimal scale constants for the encode + /// direction: `encoded = fast_round((value * 10^(exponent)) * 10^(-factor))`. + fn encode_scale(exponent: u8, factor: u8) -> Self::Scale; + + /// Apply a scale as the same two separate multiplications the decode side + /// uses. Two steps rather than one multiplication by a combined constant: + /// the spec requires this on the normative decode path, and encoding with + /// the same arithmetic maximizes the values that round-trip. + fn apply_scale(self, scale: Self::Scale) -> Self; + + /// True for values ALP cannot turn into an exact integer: NaN, the + /// infinities, anything scaled past the exact integer type, and `-0.0` + /// (which would come back as `+0.0` and lose its sign). + fn is_impossible_to_encode(self) -> bool; + + /// Round to nearest by the "magic number" technique. Adding `magic` pushes + /// `x` into the binade where floats are spaced exactly 1.0 apart, so the + /// add itself snaps to the nearest integer, and subtracting `magic` back + /// is exact. `magic = 1.5 * 2^mantissa_bits` keeps the sum in that binade + /// for negative `x` too. Values large enough to be mis-rounded just fail + /// the caller's round-trip check and become exceptions. The add/sub must + /// not be simplified away: it *is* the rounding. + fn fast_round(self) -> ::Signed; + + /// Encode one value with a precomputed [`AlpFloat::encode_scale`]. + /// + /// Values ALP cannot represent map to [`AlpFloat::ENCODING_SENTINEL`], whose + /// round trip is guaranteed to mismatch, so the caller's `decode == value` + /// check records them as exceptions without a separate test. + fn encode_value(self, scale: Self::Scale) -> ::Signed { + let scaled = self.apply_scale(scale); + if scaled.is_impossible_to_encode() { + return Self::ENCODING_SENTINEL; + } + scaled.fast_round() + } +} + +impl AlpFloat for f32 { + type Exact = u32; + type Scale = (f32, f32); + + const MAX_EXPONENT: u8 = ALP_MAX_EXPONENT_F32; + const MAGIC_NUMBER: Self = 12582912.0; // 2^22 + 2^23 + const ENCODING_UPPER_LIMIT: Self = 2147483520.0; + const ENCODING_LOWER_LIMIT: Self = -2147483520.0; + const ENCODING_SENTINEL: i32 = 2147483520; + + fn decode_scale(exponent: u8, factor: u8) -> Self::Scale { + debug_assert!(exponent <= ALP_MAX_EXPONENT_F32); + debug_assert!(factor <= exponent); + ( + ALP_POW10_F32[factor as usize], + ALP_NEG_POW10_F32[exponent as usize], + ) + } + + fn decode_value(signed_encoded: i32, scale: Self::Scale) -> Self { + ((signed_encoded as f32) * scale.0) * scale.1 + } + + fn from_exact_bits(bits: Self::Exact) -> Self { + f32::from_bits(bits) + } + + fn to_exact_bits(self) -> Self::Exact { + self.to_bits() + } + + fn encode_scale(exponent: u8, factor: u8) -> Self::Scale { + debug_assert!(exponent <= ALP_MAX_EXPONENT_F32); + debug_assert!(factor <= exponent); + ( + ALP_POW10_F32[exponent as usize], + ALP_NEG_POW10_F32[factor as usize], + ) + } + + fn apply_scale(self, scale: Self::Scale) -> Self { + (self * scale.0) * scale.1 + } + + fn is_impossible_to_encode(self) -> bool { + // The infinities need no separate test: they fall outside the limits. + self.is_nan() + || !(Self::ENCODING_LOWER_LIMIT..=Self::ENCODING_UPPER_LIMIT).contains(&self) + || (self == 0.0 && self.is_sign_negative()) + } + + fn fast_round(self) -> i32 { + ((self + Self::MAGIC_NUMBER) - Self::MAGIC_NUMBER) as i32 + } +} + +impl AlpFloat for f64 { + type Exact = u64; + type Scale = (f64, f64); + + const MAX_EXPONENT: u8 = ALP_MAX_EXPONENT_F64; + const MAGIC_NUMBER: Self = 6755399441055744.0; // 2^51 + 2^52 + const ENCODING_UPPER_LIMIT: Self = 9223372036854774784.0; + const ENCODING_LOWER_LIMIT: Self = -9223372036854774784.0; + const ENCODING_SENTINEL: i64 = 9223372036854774784; + + fn decode_scale(exponent: u8, factor: u8) -> Self::Scale { + debug_assert!(exponent <= ALP_MAX_EXPONENT_F64); + debug_assert!(factor <= exponent); + ( + ALP_POW10_F64[factor as usize], + ALP_NEG_POW10_F64[exponent as usize], + ) + } + + fn decode_value(signed_encoded: i64, scale: Self::Scale) -> Self { + ((signed_encoded as f64) * scale.0) * scale.1 + } + + fn from_exact_bits(bits: Self::Exact) -> Self { + f64::from_bits(bits) + } + + fn to_exact_bits(self) -> Self::Exact { + self.to_bits() + } + + fn encode_scale(exponent: u8, factor: u8) -> Self::Scale { + debug_assert!(exponent <= ALP_MAX_EXPONENT_F64); + debug_assert!(factor <= exponent); + ( + ALP_POW10_F64[exponent as usize], + ALP_NEG_POW10_F64[factor as usize], + ) + } + + fn apply_scale(self, scale: Self::Scale) -> Self { + (self * scale.0) * scale.1 + } + + fn is_impossible_to_encode(self) -> bool { + // The infinities need no separate test: they fall outside the limits. + self.is_nan() + || !(Self::ENCODING_LOWER_LIMIT..=Self::ENCODING_UPPER_LIMIT).contains(&self) + || (self == 0.0 && self.is_sign_negative()) + } + + fn fast_round(self) -> i64 { + ((self + Self::MAGIC_NUMBER) - Self::MAGIC_NUMBER) as i64 + } +} diff --git a/parquet/src/encodings/decoding.rs b/parquet/src/encodings/decoding.rs index 0fb960412295..d56c75bed478 100644 --- a/parquet/src/encodings/decoding.rs +++ b/parquet/src/encodings/decoding.rs @@ -26,6 +26,7 @@ use super::rle::RleDecoder; use crate::basic::*; use crate::data_type::private::ParquetValueType; use crate::data_type::*; +use crate::encodings::decoding::alp_decoder::AlpDecoder; use crate::encodings::decoding::byte_stream_split_decoder::{ ByteStreamSplitDecoder, VariableWidthByteStreamSplitDecoder, }; @@ -33,6 +34,7 @@ use crate::errors::{ParquetError, Result}; use crate::schema::types::ColumnDescPtr; use crate::util::bit_util::{self, BitReader, FromBitpacked}; +pub(crate) mod alp_decoder; mod byte_stream_split_decoder; pub(crate) mod private { @@ -63,7 +65,8 @@ pub(crate) mod private { Encoding::RLE | Encoding::DELTA_BINARY_PACKED | Encoding::DELTA_BYTE_ARRAY - | Encoding::DELTA_LENGTH_BYTE_ARRAY => Err(general_err!( + | Encoding::DELTA_LENGTH_BYTE_ARRAY + | Encoding::ALP => Err(general_err!( "Encoding {} is not supported for type", encoding )), @@ -116,6 +119,7 @@ pub(crate) mod private { ) -> Result>> { match encoding { Encoding::BYTE_STREAM_SPLIT => Ok(Box::new(ByteStreamSplitDecoder::new())), + Encoding::ALP => Ok(Box::new(AlpDecoder::new())), _ => get_decoder_default(descr, encoding), } } @@ -127,6 +131,7 @@ pub(crate) mod private { ) -> Result>> { match encoding { Encoding::BYTE_STREAM_SPLIT => Ok(Box::new(ByteStreamSplitDecoder::new())), + Encoding::ALP => Ok(Box::new(AlpDecoder::new())), _ => get_decoder_default(descr, encoding), } } @@ -1299,6 +1304,8 @@ mod tests { create_and_check_decoder::(Encoding::DELTA_LENGTH_BYTE_ARRAY, None); create_and_check_decoder::(Encoding::DELTA_BYTE_ARRAY, None); create_and_check_decoder::(Encoding::RLE, None); + create_and_check_decoder::(Encoding::ALP, None); + create_and_check_decoder::(Encoding::ALP, None); // error when initializing create_and_check_decoder::( diff --git a/parquet/src/encodings/decoding/alp_decoder.rs b/parquet/src/encodings/decoding/alp_decoder.rs new file mode 100644 index 000000000000..2955adaa7340 --- /dev/null +++ b/parquet/src/encodings/decoding/alp_decoder.rs @@ -0,0 +1,1408 @@ +// 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::ops::Range; + +use bytes::Bytes; + +use crate::basic::Encoding; +use crate::data_type::DataType; +use crate::encodings::alp::{ + ALP_COMPRESSION_MODE, ALP_DEFAULT_LOG_VECTOR_SIZE, ALP_HEADER_SIZE, + ALP_INTEGER_ENCODING_FOR_BIT_PACK, ALP_MAX_EXPONENT_F32, ALP_MAX_EXPONENT_F64, + ALP_MAX_LOG_VECTOR_SIZE, ALP_MIN_LOG_VECTOR_SIZE, AlpExact, AlpFloat, AlpHeader, AlpInfo, + ForInfo, +}; +use crate::encodings::decoding::Decoder; +use crate::errors::{ParquetError, Result}; +use crate::util::bit_util::BitReader; + +/// Parsed view of one vector's metadata and data sections. +/// +/// Each data section is described by its start offset into the page body; the +/// section bytes themselves stay in the body and are decoded lazily when the +/// vector is decoded. Section lengths are fully determined by the fixed-size +/// metadata at the front of the vector (`bit_width` for `packed_values`, +/// `num_exceptions` for both exception sections), so only the start offset is +/// stored. +#[derive(Debug, Clone, Copy)] +struct AlpEncodedVectorView { + num_elements: u16, + alp_info: AlpInfo, + for_info: ForInfo, + packed_values: usize, + exception_positions: usize, + exception_values: usize, +} + +impl AlpEncodedVectorView { + fn expected_stored_size(&self) -> usize { + AlpInfo::STORED_SIZE + + ForInfo::::stored_size() + + self + .for_info + .get_data_stored_size(self.num_elements, self.alp_info.num_exceptions) + } + + /// Byte range of the bit-packed values section in the page body. + fn packed_values_range(&self) -> Range { + let len = self.for_info.get_bit_packed_size(self.num_elements); + self.packed_values..self.packed_values + len + } + + /// Byte range of the exception positions section (`u16` each) in the page body. + fn exception_positions_range(&self) -> Range { + let len = self.alp_info.num_exceptions as usize * std::mem::size_of::(); + self.exception_positions..self.exception_positions + len + } + + /// Byte range of the exception values section (`Exact::WIDTH` each) in the page body. + fn exception_values_range(&self) -> Range { + let len = self.alp_info.num_exceptions as usize * Exact::WIDTH; + self.exception_values..self.exception_values + len + } +} + +/// Parse and validate the 7-byte ALP page header: compression mode, integer +/// encoding, and vector-size range. +fn parse_alp_page_header(data: &[u8]) -> Result { + let header = AlpHeader::deserialize(data)?; + + if header.compression_mode != ALP_COMPRESSION_MODE { + return Err(general_err!( + "Invalid ALP page: unsupported compression mode {}", + header.compression_mode + )); + } + if header.integer_encoding != ALP_INTEGER_ENCODING_FOR_BIT_PACK { + return Err(general_err!( + "Invalid ALP page: unsupported integer encoding {}", + header.integer_encoding + )); + } + if header.vector_size < (1usize << ALP_MIN_LOG_VECTOR_SIZE) { + return Err(general_err!( + "Invalid ALP page: log_vector_size {} below min {}", + header.vector_size.trailing_zeros(), + ALP_MIN_LOG_VECTOR_SIZE + )); + } + if header.vector_size > (1usize << ALP_MAX_LOG_VECTOR_SIZE) { + return Err(general_err!( + "Invalid ALP page: log_vector_size {} exceeds max {}", + header.vector_size.trailing_zeros(), + ALP_MAX_LOG_VECTOR_SIZE + )); + } + + Ok(header) +} + +/// Read the little-endian `u32` vector offset at index `idx` from the offsets +/// section at the start of the page body. +fn read_offset(body: &[u8], idx: usize) -> Result { + let start = idx * std::mem::size_of::(); + let bytes = body + .get(start..start + std::mem::size_of::()) + .ok_or_else(|| general_err!("Invalid ALP page: offset index {} out of bounds", idx))?; + Ok(u32::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]) as usize) +} + +/// Parse a single vector section: +/// `[AlpInfo][ForInfo][PackedValues][ExceptionPositions][ExceptionValues]`. +fn parse_vector_view( + body: &[u8], + vector_start: usize, + vector_end: usize, + num_elements: u16, +) -> Result> { + let vector_bytes = &body[vector_start..vector_end]; + + let metadata_size = AlpInfo::STORED_SIZE + ForInfo::::stored_size(); + if vector_bytes.len() < metadata_size { + return Err(general_err!( + "Invalid ALP page: vector metadata too short, expected at least {} bytes, got {}", + metadata_size, + vector_bytes.len() + )); + } + + let alp_info = AlpInfo { + exponent: vector_bytes[0], + factor: vector_bytes[1], + num_exceptions: u16::from_le_bytes([vector_bytes[2], vector_bytes[3]]), + }; + + let max_exponent = if Exact::WIDTH == 4 { + ALP_MAX_EXPONENT_F32 + } else { + ALP_MAX_EXPONENT_F64 + }; + + if alp_info.exponent > max_exponent { + return Err(general_err!( + "Invalid ALP page: exponent {} exceeds max {}", + alp_info.exponent, + max_exponent + )); + } + + if alp_info.factor > alp_info.exponent { + return Err(general_err!( + "Invalid ALP page: factor {} exceeds exponent {}", + alp_info.factor, + alp_info.exponent + )); + } + + if alp_info.num_exceptions > num_elements { + return Err(general_err!( + "Invalid ALP page: num_exceptions {} exceeds vector num_elements {}", + alp_info.num_exceptions, + num_elements + )); + } + + let for_start = AlpInfo::STORED_SIZE; + let for_end = for_start + Exact::WIDTH; + let frame_of_reference = Exact::from_le_slice(&vector_bytes[for_start..for_end]); + let bit_width = vector_bytes[for_end]; + + if bit_width as usize > Exact::WIDTH * 8 { + return Err(general_err!( + "Invalid ALP page: bit width {} exceeds {}", + bit_width, + Exact::WIDTH * 8 + )); + } + + let for_info = ForInfo:: { + frame_of_reference, + bit_width, + }; + + let data_size = for_info.get_data_stored_size(num_elements, alp_info.num_exceptions); + let expected_size = metadata_size + data_size; + if vector_bytes.len() < expected_size { + return Err(general_err!( + "Invalid ALP page: vector data too short, expected at least {} bytes, got {}", + expected_size, + vector_bytes.len() + )); + } + if vector_bytes.len() > expected_size { + return Err(general_err!( + "Invalid ALP page: vector data too long, expected {} bytes, got {}", + expected_size, + vector_bytes.len() + )); + } + + let data = &vector_bytes[metadata_size..expected_size]; + let packed_size = for_info.get_bit_packed_size(num_elements); + let positions_size = alp_info.num_exceptions as usize * std::mem::size_of::(); + + // Section offsets relative to the start of the data section: packed values + // first, then exception positions, then exception values. + let positions_start = packed_size; + let values_start = positions_start + positions_size; + + // Validate exception positions without materializing them. They are decoded + // straight from the body when the vector is decoded; here we only enforce + // that every position is in range so the whole page is validated up front. + for chunk in data[positions_start..values_start].chunks_exact(2) { + let position = u16::from_le_bytes([chunk[0], chunk[1]]); + if position >= num_elements { + return Err(general_err!( + "Invalid ALP page: exception position {} out of bounds for vector length {}", + position, + num_elements + )); + } + } + + // Store each section's start offset into the page body. Lengths are derived + // from the vector metadata at decode time, so no end offset is stored. + let data_start = vector_start + metadata_size; + let packed_values = data_start; + let exception_positions = data_start + positions_start; + let exception_values = data_start + values_start; + + Ok(AlpEncodedVectorView { + num_elements, + alp_info, + for_info, + packed_values, + exception_positions, + exception_values, + }) +} + +/// Live decode state for the one vector currently being consumed. +/// +/// Holds the bit position inside that vector's packed values plus the +/// vector-level constants needed to turn each packed integer back into a float. +/// `delivered` is the vector-local index of the next element to produce, so +/// exception patches (which use vector-local positions) land in the right place +/// even when a vector is split across several `get`/`skip` calls. +struct CurrentVector { + reader: BitReader, + bit_width: u8, + frame_of_reference: Value::Exact, + scale: Value::Scale, + /// Number of this vector's elements not yet delivered or skipped. + remaining: usize, + /// Vector-local index of the next element to produce. + delivered: usize, + exception_positions: Bytes, + exception_values: Bytes, +} + +/// Largest slice decoded in one unpack-then-decode pass: the canonical ALP +/// vector size - 1024. +/// +/// The unpack scratch is sized to `min(vector_size, this)`, so vectors at the +/// default size or smaller are decoded whole, while larger (non-default) vectors +/// are decoded in canonical-vector-sized tiles. +/// +/// Bounding the tile to one canonical vector keeps the scratch L1-resident, which +/// is what makes the staged unpack-then-decode beat an in-place decode. +const DECODE_TILE_CAP: usize = 1 << ALP_DEFAULT_LOG_VECTOR_SIZE; + +/// Decode the next `out.len()` elements of the current vector into `out`, +/// patching any exceptions whose vector-local position falls in the +/// just-produced sub-range. +/// +/// Deltas are bulk-unpacked a tile at a time into the caller-provided `scratch` +/// via `get_batch` (which dispatches to the SIMD-friendly fixed-width `unpack` +/// kernels), then the inverse FOR and decimal decode run as one branchless, +/// state-free loop over that contiguous tile so the compiler can autovectorize +/// it. +fn decode_range( + cur: &mut CurrentVector, + scratch: &mut [Value::Exact], + out: &mut [Value], +) -> Result<()> { + let frame_of_reference = cur.frame_of_reference; + if cur.bit_width == 0 { + // Every packed delta is zero, so all values share `frame_of_reference`. + let signed = frame_of_reference.reinterpret_as_signed(); + out.fill(Value::decode_value(signed, cur.scale)); + } else { + let bit_width = cur.bit_width as usize; + let scale = cur.scale; + for chunk in out.chunks_mut(scratch.len()) { + let deltas = &mut scratch[..chunk.len()]; + let unpacked = cur.reader.get_batch::(deltas, bit_width); + if unpacked != chunk.len() { + return Err(general_err!( + "Invalid ALP page: not enough packed bits to decode vector" + )); + } + for (slot, &delta) in chunk.iter_mut().zip(deltas.iter()) { + let signed = delta + .wrapping_add(frame_of_reference) + .reinterpret_as_signed(); + *slot = Value::decode_value(signed, scale); + } + } + } + + // Patch exceptions landing in `[delivered, delivered + out.len())`. Positions + // were validated in bounds when the vector was parsed, and patching is a + // positional overwrite, so it is independent of exception ordering. + let lo = cur.delivered; + let hi = cur.delivered + out.len(); + for (pos_chunk, value_chunk) in cur + .exception_positions + .chunks_exact(std::mem::size_of::()) + .zip(cur.exception_values.chunks_exact(Value::Exact::WIDTH)) + { + let pos = u16::from_le_bytes([pos_chunk[0], pos_chunk[1]]) as usize; + if (lo..hi).contains(&pos) { + out[pos - lo] = Value::from_exact_bits(Value::Exact::from_le_slice(value_chunk)); + } + } + + cur.delivered = hi; + cur.remaining -= out.len(); + Ok(()) +} + +/// Decoder for ALP-encoded floating-point pages (`f32`/`f64`). +/// +/// Values are decoded directly into the caller's output buffer, one vector at a +/// time: `set_data` validates the page header, and each vector is parsed and +/// decoded on demand as the read cursor reaches it. +pub(crate) struct AlpDecoder +where + T::T: AlpFloat, + ::Exact: Send, +{ + /// Validated page header; drives vector sizing and count. + header: AlpHeader, + /// Page body following the 7-byte header: `[offsets][vector data...]`. + body: Bytes, + /// Index of the next vector to parse once `current` is exhausted. + next_vector_idx: usize, + /// Body offset where the next vector must begin; offsets are validated to be + /// contiguous as vectors are parsed. + expected_next_offset: usize, + /// Live decode state for the vector currently being consumed. + current: Option>, + /// Reused unpack buffer, sized once per page to `min(vector_size, cap)` and + /// shared across all vectors; holds bulk-unpacked deltas for one tile. + scratch: Vec<::Exact>, + /// Number of values still to be read from the page. + num_values: usize, +} + +impl AlpDecoder +where + T::T: AlpFloat, + ::Exact: Send, +{ + pub(crate) fn new() -> Self { + Self { + // Benign placeholder header; replaced on the first `set_data`. The + // cursor never consults it while `num_values == 0`. + header: AlpHeader { + compression_mode: 0, + integer_encoding: 0, + vector_size: 1, + num_elements: 0, + }, + body: Bytes::new(), + next_vector_idx: 0, + expected_next_offset: 0, + current: None, + scratch: Vec::new(), + num_values: 0, + } + } + + /// Parse the next vector on demand and install it as `current`: validate its + /// offset is contiguous with the previous vector, parse its metadata, build a + /// live bit reader over its packed values, and slice out its exception + /// sections. + fn load_current_vector(&mut self) -> Result<()> { + let idx = self.next_vector_idx; + let num_vectors = self.header.num_vectors(); + debug_assert!(idx < num_vectors, "load_current_vector with no vector left"); + + let body_ref = self.body.as_ref(); + let start = read_offset(body_ref, idx)?; + if start != self.expected_next_offset { + return Err(general_err!( + "Invalid ALP page: vector offset {} at index {} does not match expected {}", + start, + idx, + self.expected_next_offset + )); + } + let end = if idx + 1 < num_vectors { + read_offset(body_ref, idx + 1)? + } else { + body_ref.len() + }; + if end < start || end > body_ref.len() { + return Err(general_err!( + "Invalid ALP page: vector offset {} out of bounds at index {}", + end, + idx + )); + } + + let count = self.header.vector_num_elements(idx); + let view = parse_vector_view::<::Exact>(body_ref, start, end, count)?; + + // The next vector must start exactly where this one ends. + self.expected_next_offset = start + view.expected_stored_size(); + + let packed = self.body.slice(view.packed_values_range()); + let exception_positions = self.body.slice(view.exception_positions_range()); + let exception_values = self.body.slice(view.exception_values_range()); + self.current = Some(CurrentVector { + reader: BitReader::new(packed), + bit_width: view.for_info.bit_width, + frame_of_reference: view.for_info.frame_of_reference, + scale: ::decode_scale(view.alp_info.exponent, view.alp_info.factor), + remaining: count as usize, + delivered: 0, + exception_positions, + exception_values, + }); + self.next_vector_idx += 1; + Ok(()) + } +} + +impl Decoder for AlpDecoder +where + T::T: AlpFloat, + ::Exact: Send, +{ + /// Validate the page header and reset the read cursor. Vectors are parsed + /// and decoded lazily, on the first `get`/`skip`. + fn set_data(&mut self, data: Bytes, num_values: usize) -> Result<()> { + let header = parse_alp_page_header(data.as_ref())?; + + // `num_values` is an upper bound, not an exact count: only non-null values + // are encoded, and the caller passes the level count when the value count + // is not known up front. The header carries the authoritative count. + if header.num_elements > num_values { + return Err(general_err!( + "Invalid ALP page: header num_elements {} exceeds page num_values {}", + header.num_elements, + num_values + )); + } + let num_values = header.num_elements; + + let offsets_section_size = header + .num_vectors() + .checked_mul(std::mem::size_of::()) + .ok_or_else(|| general_err!("Invalid ALP page: offsets length overflow"))?; + + // `parse_alp_page_header` guarantees `data.len() >= ALP_HEADER_SIZE`. + let body = data.slice(ALP_HEADER_SIZE..); + if body.len() < offsets_section_size { + return Err(general_err!( + "Invalid ALP page: expected at least {} bytes for {} offsets, got {}", + offsets_section_size, + header.num_vectors(), + body.len() + )); + } + + // Size the reused unpack buffer to one vector, capped to stay L1-resident + // and to the page's value count so tiny pages don't over-allocate. The + // `.max(1)` keeps a non-empty tile so the `chunks_mut` decode never hits a + // zero stride (an empty page returns before any decode anyway). + let tile = header + .vector_size + .min(DECODE_TILE_CAP) + .min(num_values) + .max(1); + self.scratch.clear(); + self.scratch.resize(tile, Default::default()); + + self.header = header; + self.body = body; + self.next_vector_idx = 0; + self.expected_next_offset = offsets_section_size; + self.current = None; + self.num_values = num_values; + Ok(()) + } + + /// Read up to `buffer.len()` values, decoding each vector on demand directly + /// into `buffer`. + fn get(&mut self, buffer: &mut [T::T]) -> Result { + let target = buffer.len().min(self.num_values); + if target == 0 { + return Ok(0); + } + + let mut written = 0; + while written < target { + if self.current.as_ref().is_none_or(|c| c.remaining == 0) { + self.load_current_vector()?; + } + let cur = self.current.as_mut().unwrap(); + let n = cur.remaining.min(target - written); + decode_range::(cur, &mut self.scratch, &mut buffer[written..written + n])?; + written += n; + self.num_values -= n; + } + Ok(target) + } + + fn values_left(&self) -> usize { + self.num_values + } + + fn encoding(&self) -> Encoding { + Encoding::ALP + } + + /// Skip up to `num_values` values. + /// + /// Skipped-over vectors are not parsed, so a multi-vector skip is O(1) in + /// vectors skipped - and, unlike a sequential read, they are not validated. + fn skip(&mut self, num_values: usize) -> Result { + let to_skip = num_values.min(self.num_values); + if to_skip == 0 { + return Ok(0); + } + + let mut left = to_skip; + + // The current vector carries live bit-reader/cursor state, so a skip + // within it advances it in place rather than jumping. + if let Some(cur) = self.current.as_mut() { + let within = left.min(cur.remaining); + if cur.bit_width != 0 { + cur.reader.skip(within, cur.bit_width as usize); + } + cur.delivered += within; + cur.remaining -= within; + self.num_values -= within; + left -= within; + } + if left == 0 { + return Ok(to_skip); + } + + let vector_size = self.header.vector_size; + let num_elements = self.header.num_elements; + let target = (num_elements - self.num_values) + left; + + self.current = None; + self.num_values = num_elements - target; + + if target == num_elements { + self.next_vector_idx = self.header.num_vectors(); + return Ok(to_skip); + } + + // vector_size is a power of two: shift and mask instead of div/rem. + let landing = target >> vector_size.trailing_zeros(); + let within = target & (vector_size - 1); + + // Seed expected_next_offset so the jump satisfies load_current_vector's + // contiguity check; validation resumes from the landing vector. + self.next_vector_idx = landing; + self.expected_next_offset = read_offset(self.body.as_ref(), landing)?; + + // within == 0 lands on a boundary and is loaded lazily by the next + // get/skip; a mid-vector landing must be parsed and advanced now. + if within > 0 { + self.load_current_vector()?; + let cur = self.current.as_mut().unwrap(); + if cur.bit_width != 0 { + cur.reader.skip(within, cur.bit_width as usize); + } + cur.delivered = within; + cur.remaining -= within; + } + + Ok(to_skip) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::data_type::{DoubleType, FloatType}; + use crate::encodings::alp::{ + ALP_NEG_POW10_F32, ALP_NEG_POW10_F64, ALP_POW10_F32, ALP_POW10_F64, + }; + + fn make_alp_page_bytes( + compression_mode: u8, + integer_encoding: u8, + log_vector_size: u8, + num_elements: i32, + offsets: &[u32], + body_tail_len: usize, + ) -> Vec { + let mut out = Vec::with_capacity(ALP_HEADER_SIZE + offsets.len() * 4 + body_tail_len); + out.push(compression_mode); + out.push(integer_encoding); + out.push(log_vector_size); + out.extend_from_slice(&num_elements.to_le_bytes()); + for offset in offsets { + out.extend_from_slice(&offset.to_le_bytes()); + } + out.extend(std::iter::repeat_n(0u8, body_tail_len)); + out + } + + trait AppendLeBytes { + fn append_le_bytes(self, out: &mut Vec); + } + + impl AppendLeBytes for u32 { + fn append_le_bytes(self, out: &mut Vec) { + out.extend_from_slice(&self.to_le_bytes()); + } + } + + impl AppendLeBytes for u64 { + fn append_le_bytes(self, out: &mut Vec) { + out.extend_from_slice(&self.to_le_bytes()); + } + } + + struct VectorSpec<'a, Exact> { + exponent: u8, + factor: u8, + frame_of_reference: Exact, + bit_width: u8, + packed_values: &'a [u8], + exception_positions: &'a [u16], + exception_values: &'a [Exact], + } + + fn make_vector(spec: VectorSpec<'_, Exact>) -> Vec { + let num_exceptions = spec.exception_positions.len(); + assert_eq!(num_exceptions, spec.exception_values.len()); + assert!(u16::try_from(num_exceptions).is_ok()); + + let mut out = Vec::new(); + out.push(spec.exponent); + out.push(spec.factor); + out.extend_from_slice(&(num_exceptions as u16).to_le_bytes()); + spec.frame_of_reference.append_le_bytes(&mut out); + out.push(spec.bit_width); + out.extend_from_slice(spec.packed_values); + for position in spec.exception_positions { + out.extend_from_slice(&position.to_le_bytes()); + } + for value in spec.exception_values { + value.append_le_bytes(&mut out); + } + out + } + + /// Decode a full page through the streaming decoder. + fn decode_page(page: Vec, num_values: usize) -> Vec + where + T::T: AlpFloat, + ::Exact: Send, + { + let mut decoder = AlpDecoder::::new(); + decoder.set_data(Bytes::from(page), num_values).unwrap(); + let mut out = vec![T::T::default(); num_values]; + assert_eq!(decoder.get(&mut out).unwrap(), num_values); + out + } + + /// Drive the streaming decoder to its first error. Header errors surface + /// from `set_data`; per-vector and offset errors surface from `get`. + fn decode_err(page: Vec, num_values: usize) -> ParquetError + where + T::T: AlpFloat, + ::Exact: Send, + { + let mut decoder = AlpDecoder::::new(); + match decoder.set_data(Bytes::from(page), num_values) { + Err(e) => e, + Ok(()) => { + let mut out = vec![T::T::default(); num_values]; + decoder.get(&mut out).unwrap_err() + } + } + } + + fn make_page_from_vectors( + log_vector_size: u8, + num_elements: i32, + vectors: &[Vec], + ) -> Vec { + let vector_size = 1usize << log_vector_size; + let expected_num_vectors = if num_elements <= 0 { + 0 + } else { + (num_elements as usize).div_ceil(vector_size) + }; + assert_eq!(vectors.len(), expected_num_vectors); + + let offsets_section_size = vectors.len() * std::mem::size_of::(); + let mut offsets = Vec::with_capacity(vectors.len()); + let mut running_offset = offsets_section_size as u32; + for vector in vectors { + offsets.push(running_offset); + running_offset += vector.len() as u32; + } + + let mut page = make_alp_page_bytes(0, 0, log_vector_size, num_elements, &offsets, 0); + for vector in vectors { + page.extend_from_slice(vector); + } + page + } + + #[test] + fn test_alp_header_serialize_deserialize_round_trip() { + let header = AlpHeader { + compression_mode: ALP_COMPRESSION_MODE, + integer_encoding: ALP_INTEGER_ENCODING_FOR_BIT_PACK, + vector_size: 8, + num_elements: 1234, + }; + let bytes = header.serialize().unwrap(); + let parsed = AlpHeader::deserialize(&bytes).unwrap(); + assert_eq!(parsed.compression_mode, header.compression_mode); + assert_eq!(parsed.integer_encoding, header.integer_encoding); + assert_eq!(parsed.vector_size, header.vector_size); + assert_eq!(parsed.num_elements, header.num_elements); + } + + #[test] + fn test_alp_header_serialize_rejects_overflow() { + let header = AlpHeader { + compression_mode: 0, + integer_encoding: 0, + vector_size: 8, + num_elements: i32::MAX as usize + 1, + }; + let err = header.serialize().unwrap_err(); + assert!(err.to_string().contains("exceeds i32::MAX")); + } + + #[test] + fn test_alp_header_serialize_rejects_non_power_of_two_vector_size() { + let header = AlpHeader { + compression_mode: 0, + integer_encoding: 0, + vector_size: 100, + num_elements: 4, + }; + let err = header.serialize().unwrap_err(); + assert!(err.to_string().contains("is not a power of two")); + } + + #[test] + fn test_alp_header_deserialize_short_header() { + let err = AlpHeader::deserialize(&[0, 1, 2]).unwrap_err(); + assert!( + err.to_string() + .contains("Invalid ALP page: expected at least 7 bytes for header") + ); + } + + #[test] + fn test_alp_header_deserialize_negative_num_elements() { + let mut bytes = [0u8; ALP_HEADER_SIZE]; + bytes[2] = ALP_MIN_LOG_VECTOR_SIZE; + bytes[3..7].copy_from_slice(&(-1i32).to_le_bytes()); + let err = AlpHeader::deserialize(&bytes).unwrap_err(); + assert!( + err.to_string() + .contains("Invalid ALP page: num_elements -1 must be >= 0") + ); + } + + #[test] + fn test_alp_header_deserialize_log_vector_size_overflow() { + let mut bytes = [0u8; ALP_HEADER_SIZE]; + bytes[2] = 64; // 1 << 64 overflows usize + let err = AlpHeader::deserialize(&bytes).unwrap_err(); + assert!( + err.to_string() + .contains("too large to represent a vector size") + ); + } + + #[test] + fn test_decode_valid_page() { + let data = make_alp_page_bytes(0, 0, 3, 4, &[4], 13); + let decoded = decode_page::(data, 4); + assert_eq!(decoded, vec![0.0_f64; 4]); + } + + #[test] + fn test_set_data_rejects_small_vector_size() { + let data = make_alp_page_bytes(0, 0, 2, 1, &[4], 8); + let err = decode_err::(data, 1); + assert!( + err.to_string() + .contains("Invalid ALP page: log_vector_size 2 below min 3") + ); + } + + #[test] + fn test_set_data_rejects_big_vector_size() { + let data = make_alp_page_bytes(0, 0, 16, 1, &[4], 8); + let err = decode_err::(data, 1); + assert!( + err.to_string() + .contains("Invalid ALP page: log_vector_size 16 exceeds max 15") + ); + } + + #[test] + fn test_set_data_rejects_integer_encoding() { + let data = make_alp_page_bytes(0, 1, 3, 1, &[4], 8); + let err = decode_err::(data, 1); + assert!( + err.to_string() + .contains("Invalid ALP page: unsupported integer encoding 1") + ); + } + + #[test] + fn test_decode_rejects_invalid_exponent_f32() { + let vector = make_vector(VectorSpec { + exponent: 11, + factor: 0, + frame_of_reference: 0u32, + bit_width: 0, + packed_values: &[], + exception_positions: &[], + exception_values: &[], + }); + let page = make_page_from_vectors(3, 1, &[vector]); + let err = decode_err::(page, 1); + assert!( + err.to_string() + .contains("Invalid ALP page: exponent 11 exceeds max 10") + ); + } + + #[test] + fn test_decode_rejects_invalid_factor_f32() { + let vector = make_vector(VectorSpec { + exponent: 0, + factor: 11, + frame_of_reference: 0u32, + bit_width: 0, + packed_values: &[], + exception_positions: &[], + exception_values: &[], + }); + let page = make_page_from_vectors(3, 1, &[vector]); + let err = decode_err::(page, 1); + assert!( + err.to_string() + .contains("Invalid ALP page: factor 11 exceeds exponent 0") + ); + } + + #[test] + fn test_decode_rejects_factor_exceeds_exponent() { + let vector = make_vector(VectorSpec { + exponent: 2, + factor: 3, + frame_of_reference: 0u32, + bit_width: 0, + packed_values: &[], + exception_positions: &[], + exception_values: &[], + }); + let page = make_page_from_vectors(3, 1, &[vector]); + let err = decode_err::(page, 1); + assert!( + err.to_string() + .contains("Invalid ALP page: factor 3 exceeds exponent 2") + ); + } + + #[test] + fn test_decode_rejects_invalid_num_exceptions() { + let vector = make_vector(VectorSpec { + exponent: 0, + factor: 0, + frame_of_reference: 0u32, + bit_width: 0, + packed_values: &[], + exception_positions: &[0, 0], + exception_values: &[0, 0], + }); + let page = make_page_from_vectors(3, 1, &[vector]); + let err = decode_err::(page, 1); + assert!( + err.to_string() + .contains("Invalid ALP page: num_exceptions 2 exceeds vector num_elements 1") + ); + } + + #[test] + fn test_decode_rejects_invalid_exception_position() { + let vector = make_vector(VectorSpec { + exponent: 0, + factor: 0, + frame_of_reference: 0u32, + bit_width: 0, + packed_values: &[], + exception_positions: &[1], + exception_values: &[123], + }); + let page = make_page_from_vectors(3, 1, &[vector]); + let err = decode_err::(page, 1); + assert!( + err.to_string().contains( + "Invalid ALP page: exception position 1 out of bounds for vector length 1" + ) + ); + } + + #[test] + fn test_decode_rejects_non_contiguous_offset() { + let data = make_alp_page_bytes(0, 0, 3, 9, &[12, 8], 12); + let err = decode_err::(data, 9); + assert!( + err.to_string().contains( + "Invalid ALP page: vector offset 12 at index 0 does not match expected 8" + ) + ); + } + + #[test] + fn test_decode_rejects_truncated_vector() { + let mut vector = Vec::new(); + vector.push(0); + vector.push(0); + vector.extend_from_slice(&0u16.to_le_bytes()); + vector.extend_from_slice(&10u32.to_le_bytes()); + vector.push(1); + let page = make_page_from_vectors(3, 2, &[vector]); + let err = decode_err::(page, 2); + assert!( + err.to_string() + .contains("Invalid ALP page: vector data too short") + ); + } + + #[test] + fn test_decode_rejects_vector_too_long() { + let mut vector = make_vector(VectorSpec { + exponent: 0, + factor: 0, + frame_of_reference: 0u32, + bit_width: 0, + packed_values: &[], + exception_positions: &[], + exception_values: &[], + }); + vector.push(0xAB); + + let page = make_page_from_vectors(3, 1, &[vector]); + let err = decode_err::(page, 1); + assert!( + err.to_string() + .contains("Invalid ALP page: vector data too long, expected 9 bytes, got 10") + ); + } + + #[test] + fn test_decode_rejects_unclaimed_bytes() { + let vector = make_vector(VectorSpec { + exponent: 0, + factor: 0, + frame_of_reference: 0u32, + bit_width: 0, + packed_values: &[], + exception_positions: &[], + exception_values: &[], + }); + + // offsets section is 4 bytes for one vector, so offset=5 leaves one + // unclaimed byte between offsets and vector data. + let mut page = make_alp_page_bytes(0, 0, 3, 1, &[5], 0); + page.push(0); + page.extend_from_slice(&vector); + + let err = decode_err::(page, 1); + assert!( + err.to_string() + .contains("Invalid ALP page: vector offset 5 at index 0 does not match expected 4") + ); + } + + #[test] + fn test_parse_vector_view_exception_sections() { + let mut vector = Vec::new(); + + vector.push(2); + vector.push(0); + vector.extend_from_slice(&1u16.to_le_bytes()); + + vector.extend_from_slice(&10u64.to_le_bytes()); + vector.push(0); + + vector.extend_from_slice(&0u16.to_le_bytes()); + vector.extend_from_slice(&42.5_f64.to_le_bytes()); + + let body = Bytes::from(vector); + let view = parse_vector_view::(body.as_ref(), 0, body.len(), 1).unwrap(); + assert_eq!(view.num_elements, 1); + assert_eq!(view.alp_info.num_exceptions, 1); + assert_eq!(view.for_info.bit_width, 0); + + let positions: Vec = body[view.exception_positions_range()] + .chunks_exact(2) + .map(|c| u16::from_le_bytes([c[0], c[1]])) + .collect(); + let values: Vec = body[view.exception_values_range()] + .chunks_exact(8) + .map(|c| u64::from_le_bytes(c.try_into().unwrap())) + .collect(); + assert_eq!(positions, vec![0]); + assert_eq!(values, vec![42.5_f64.to_bits()]); + } + + #[test] + fn test_decode_page_values_f32_no_exceptions() { + let vector = make_vector(VectorSpec { + exponent: 0, + factor: 0, + frame_of_reference: 10u32, + bit_width: 2, + packed_values: &[0b1110_0100], + exception_positions: &[], + exception_values: &[], + }); + let page = make_page_from_vectors(3, 4, &[vector]); + let decoded = decode_page::(page, 4); + assert_eq!(decoded, vec![10.0, 11.0, 12.0, 13.0]); + } + + #[test] + fn test_decode_page_values_f64_multi_vector_with_exceptions() { + let vector0 = make_vector(VectorSpec { + exponent: 0, + factor: 0, + frame_of_reference: 10u64, + bit_width: 1, + packed_values: &[0b0000_0010], + exception_positions: &[1], + exception_values: &[42.5f64.to_bits()], + }); + let vector1 = make_vector(VectorSpec { + exponent: 0, + factor: 0, + frame_of_reference: 7u64, + bit_width: 0, + packed_values: &[], + exception_positions: &[], + exception_values: &[], + }); + let page = make_page_from_vectors(3, 9, &[vector0, vector1]); + let decoded = decode_page::(page, 9); + assert_eq!( + decoded, + vec![10.0, 42.5, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 7.0] + ); + } + + #[test] + fn test_decode_page_values_f32_edge_values_via_exceptions() { + let edge_values = [ + f32::NAN.to_bits(), + (-0.0f32).to_bits(), + f32::INFINITY.to_bits(), + ]; + let vector = make_vector(VectorSpec { + exponent: 0, + factor: 0, + frame_of_reference: 0u32, + bit_width: 0, + packed_values: &[], + exception_positions: &[0, 1, 2], + exception_values: &edge_values, + }); + let page = make_page_from_vectors(3, 3, &[vector]); + let decoded = decode_page::(page, 3); + + assert!(decoded[0].is_nan()); + assert_eq!(decoded[1], -0.0); + assert!(decoded[1].is_sign_negative()); + assert_eq!(decoded[2], f32::INFINITY); + } + + #[test] + fn test_decode_page_values_f32_two_step_decimal_multiply() { + let encoded = 1_970_570_984_i32; + let vector = make_vector(VectorSpec { + exponent: 1, + factor: 1, + frame_of_reference: encoded as u32, + bit_width: 0, + packed_values: &[], + exception_positions: &[], + exception_values: &[], + }); + let page = make_page_from_vectors(3, 1, &[vector]); + let decoded = decode_page::(page, 1); + + let expected_two_step = ((encoded as f32) * ALP_POW10_F32[1]) * ALP_NEG_POW10_F32[1]; + let one_step_scale = ALP_POW10_F32[1] * ALP_NEG_POW10_F32[1]; + let expected_one_step = (encoded as f32) * one_step_scale; + + assert_eq!(decoded[0].to_bits(), expected_two_step.to_bits()); + assert_ne!(decoded[0].to_bits(), expected_one_step.to_bits()); + } + + #[test] + fn test_decode_page_values_f64_two_step_decimal_multiply() { + let encoded = -3_900_047_474_048_127_703_i64; + let vector = make_vector(VectorSpec { + exponent: 1, + factor: 1, + frame_of_reference: encoded as u64, + bit_width: 0, + packed_values: &[], + exception_positions: &[], + exception_values: &[], + }); + let page = make_page_from_vectors(3, 1, &[vector]); + let decoded = decode_page::(page, 1); + + let expected_two_step = ((encoded as f64) * ALP_POW10_F64[1]) * ALP_NEG_POW10_F64[1]; + let one_step_scale = ALP_POW10_F64[1] * ALP_NEG_POW10_F64[1]; + let expected_one_step = (encoded as f64) * one_step_scale; + + assert_eq!(decoded[0].to_bits(), expected_two_step.to_bits()); + assert_ne!(decoded[0].to_bits(), expected_one_step.to_bits()); + } + + #[test] + fn test_alp_decoder_get_across_vectors() { + let vector0 = make_vector(VectorSpec { + exponent: 0, + factor: 0, + frame_of_reference: 10u32, + bit_width: 1, + packed_values: &[0b0000_0010], + exception_positions: &[], + exception_values: &[], + }); + let vector1 = make_vector(VectorSpec { + exponent: 0, + factor: 0, + frame_of_reference: 20u32, + bit_width: 1, + packed_values: &[0b0000_0010], + exception_positions: &[], + exception_values: &[], + }); + let page = make_page_from_vectors(3, 12, &[vector0, vector1]); + + let mut decoder = AlpDecoder::::new(); + decoder.set_data(Bytes::from(page), 12).unwrap(); + + let mut first = [0.0f32; 9]; + let read = decoder.get(&mut first).unwrap(); + assert_eq!(read, 9); + assert_eq!( + first, + [10.0, 11.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 20.0] + ); + assert_eq!(decoder.values_left(), 3); + + let mut second = [0.0f32; 3]; + let read = decoder.get(&mut second).unwrap(); + assert_eq!(read, 3); + assert_eq!(second, [21.0, 20.0, 20.0]); + assert_eq!(decoder.values_left(), 0); + } + + #[test] + fn test_alp_decoder_skip_across_vectors() { + let vector0 = make_vector(VectorSpec { + exponent: 0, + factor: 0, + frame_of_reference: 10u32, + bit_width: 1, + packed_values: &[0b0000_0010], + exception_positions: &[], + exception_values: &[], + }); + let vector1 = make_vector(VectorSpec { + exponent: 0, + factor: 0, + frame_of_reference: 20u32, + bit_width: 1, + packed_values: &[0b0000_0010], + exception_positions: &[], + exception_values: &[], + }); + let page = make_page_from_vectors(3, 12, &[vector0, vector1]); + + let mut decoder = AlpDecoder::::new(); + decoder.set_data(Bytes::from(page), 12).unwrap(); + + let skipped = decoder.skip(9).unwrap(); + assert_eq!(skipped, 9); + assert_eq!(decoder.values_left(), 3); + + let mut out = [0.0f32; 1]; + let read = decoder.get(&mut out).unwrap(); + assert_eq!(read, 1); + assert_eq!(out[0], 21.0); + assert_eq!(decoder.values_left(), 2); + } + + #[test] + fn test_alp_decoder_get_full_read() { + let vector0 = make_vector(VectorSpec { + exponent: 0, + factor: 0, + frame_of_reference: 10u32, + bit_width: 1, + packed_values: &[0b0000_0010], + exception_positions: &[], + exception_values: &[], + }); + let vector1 = make_vector(VectorSpec { + exponent: 0, + factor: 0, + frame_of_reference: 20u32, + bit_width: 1, + packed_values: &[0b0000_0010], + exception_positions: &[], + exception_values: &[], + }); + let page = make_page_from_vectors(3, 12, &[vector0, vector1]); + + let mut decoder = AlpDecoder::::new(); + decoder.set_data(Bytes::from(page), 12).unwrap(); + + let mut out = [0.0f32; 12]; + let read = decoder.get(&mut out).unwrap(); + assert_eq!(read, 12); + assert_eq!( + out, + [ + 10.0, 11.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 20.0, 21.0, 20.0, 20.0 + ] + ); + assert_eq!(decoder.values_left(), 0); + + // The exhausted decoder yields nothing more. + let mut extra = [0.0f32; 1]; + assert_eq!(decoder.get(&mut extra).unwrap(), 0); + } + + /// A skip must land where a sequential decode would, across boundaries and + /// over vectors carrying exceptions. `exponent = factor = 0` makes the + /// decode the identity `frame_of_reference + delta`, so expected values fall + /// out of the packed deltas. + #[test] + fn test_alp_decoder_skip_matches_sequential_decode() { + use crate::util::bit_util::BitWriter; + + fn pack(deltas: &[u64], bit_width: usize) -> Vec { + let mut w = BitWriter::new(deltas.len() * 2 + 16); + for &d in deltas { + w.put_value(d, bit_width); + } + w.flush(); + w.consume() + } + + struct Spec { + for_ref: u64, + bit_width: usize, + len: usize, + exc_pos: u16, + exc_val: f64, + } + let specs = [ + Spec { + for_ref: 1000, + bit_width: 4, + len: 1024, + exc_pos: 100, + exc_val: f64::NAN, + }, + Spec { + for_ref: 5000, + bit_width: 8, + len: 1024, + exc_pos: 300, + exc_val: f64::INFINITY, + }, + Spec { + for_ref: 100, + bit_width: 2, + len: 552, + exc_pos: 200, + exc_val: -0.0, + }, + ]; + + let mut vectors = Vec::new(); + let mut expected: Vec = Vec::new(); + for s in &specs { + let modulo = 1u64 << s.bit_width; + let deltas: Vec = (0..s.len).map(|j| (j as u64) % modulo).collect(); + let packed = pack(&deltas, s.bit_width); + vectors.push(make_vector(VectorSpec { + exponent: 0, + factor: 0, + frame_of_reference: s.for_ref, + bit_width: s.bit_width as u8, + packed_values: &packed, + exception_positions: &[s.exc_pos], + exception_values: &[s.exc_val.to_bits()], + })); + for (j, &d) in deltas.iter().enumerate() { + expected.push(if j as u16 == s.exc_pos { + s.exc_val + } else { + (s.for_ref + d) as f64 + }); + } + } + let n = expected.len(); + let page = make_page_from_vectors(10, n as i32, &vectors); + + let assert_bits = |actual: &[f64], expected: &[f64], ctx: &str| { + assert_eq!(actual.len(), expected.len(), "{ctx}: length"); + for (i, (a, e)) in actual.iter().zip(expected.iter()).enumerate() { + assert_eq!(a.to_bits(), e.to_bits(), "{ctx}: mismatch at {i}"); + } + }; + + // Sanity check: a straight sequential decode reproduces `expected`. + assert_bits( + &decode_page::(page.clone(), n), + &expected, + "sequential", + ); + + // Skip from the start: boundary-aligned (1024, 2048), mid-vector (1, 1025, + // 2222), just before the end (2599), and exactly to the end (2600). + for &k in &[0usize, 1, 1023, 1024, 1025, 2048, 2222, 2599, 2600] { + let mut d = AlpDecoder::::new(); + d.set_data(Bytes::from(page.clone()), n).unwrap(); + assert_eq!(d.skip(k).unwrap(), k); + assert_eq!(d.values_left(), n - k); + + let mut tail = vec![0.0f64; n - k]; + assert_eq!(d.get(&mut tail).unwrap(), n - k); + assert_bits(&tail, &expected[k..], &format!("skip {k}")); + } + + // Skip that starts mid-vector: consume part of vector 0, then skip across + // the vector-0/1 boundary before reading the rest. + let mut d = AlpDecoder::::new(); + d.set_data(Bytes::from(page.clone()), n).unwrap(); + let mut head = vec![0.0f64; 500]; + assert_eq!(d.get(&mut head).unwrap(), 500); + assert_bits(&head, &expected[..500], "head"); + assert_eq!(d.skip(700).unwrap(), 700); // 500..1200, crossing into vector 1 + assert_eq!(d.values_left(), n - 1200); + let mut tail = vec![0.0f64; n - 1200]; + assert_eq!(d.get(&mut tail).unwrap(), n - 1200); + assert_bits(&tail, &expected[1200..], "tail"); + + // Two consecutive skips, the first landing boundary-aligned (lazy, no + // parse), the second jumping again from that lazy position. + let mut d = AlpDecoder::::new(); + d.set_data(Bytes::from(page.clone()), n).unwrap(); + assert_eq!(d.skip(1024).unwrap(), 1024); // boundary -> lazy landing + assert_eq!(d.skip(600).unwrap(), 600); // jump again from the boundary + let mut tail = vec![0.0f64; n - 1624]; + assert_eq!(d.get(&mut tail).unwrap(), n - 1624); + assert_bits(&tail, &expected[1624..], "double skip"); + } +} diff --git a/parquet/src/encodings/encoding/alp_encoder.rs b/parquet/src/encodings/encoding/alp_encoder.rs new file mode 100644 index 000000000000..7e12a5883bf7 --- /dev/null +++ b/parquet/src/encodings/encoding/alp_encoder.rs @@ -0,0 +1,1030 @@ +// 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. + +//! ALP (Adaptive Lossless floating-Point) encoder. +//! +//! Spec: +//! +//! Values are buffered until the page is flushed, then encoded a vector at a +//! time into the page layout that [`AlpDecoder`] reads back: +//! +//! ```text +//! [AlpHeader][offsets][vector 0][vector 1]...[vector N-1] +//! ``` +//! +//! [`AlpDecoder`]: crate::encodings::decoding::alp_decoder::AlpDecoder + +use std::cmp::Reverse; + +use bytes::Bytes; + +use crate::basic::Encoding; +use crate::data_type::DataType; +use crate::encodings::alp::{ + ALP_COMPRESSION_MODE, ALP_DEFAULT_LOG_VECTOR_SIZE, ALP_HEADER_SIZE, + ALP_INTEGER_ENCODING_FOR_BIT_PACK, AlpExact, AlpFloat, AlpHeader, AlpInfo, ForInfo, +}; +use crate::encodings::encoding::Encoder; +use crate::errors::{ParquetError, Result}; +use crate::util::bit_util::{BitWriter, num_required_bits}; + +/// Vectors are written at the spec's default size, the ALP paper's 1024. +const VECTOR_SIZE: usize = 1 << ALP_DEFAULT_LOG_VECTOR_SIZE; + +/// Values sampled from a vector when estimating a candidate's encoded size. +const SAMPLES_PER_VECTOR: usize = 256; + +/// Vectors sampled from the first page to build the column chunk's candidate set. +const SAMPLE_VECTORS: usize = 8; + +/// Candidate `(exponent, factor)` pairs carried forward from the sampling pass. +const MAX_COMBINATIONS: usize = 5; + +/// Consecutive non-improving candidates after which per-vector selection stops. +const SAMPLING_EARLY_EXIT_THRESHOLD: usize = 4; + +/// Bits of overhead an exception costs: the value itself plus its `u16` position. +fn exception_bits() -> u64 { + (F::Exact::WIDTH as u64 * 8) + 16 +} + +/// One ALP decimal-encoding candidate: `encoded = round(value * 10^e * 10^-f)`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct ExponentAndFactor { + exponent: u8, + factor: u8, +} + +/// A candidate together with how often it won across the sampled vectors. +#[derive(Debug, Clone, Copy)] +struct Combination { + params: ExponentAndFactor, + num_appearances: u64, + estimated_size_bits: u64, +} + +/// Sort key ranking candidates, larger is better: more appearances as a vector's +/// best candidate, then smaller estimated size, then larger exponent, then +/// larger factor. +/// +/// The last two are tie-breaks taken from the ALP paper (Afroozeh et al., SIGMOD +/// 2023, section 3.1.2), which prefers higher exponents and factors without +/// justifying it further. +fn rank(c: &Combination) -> (u64, Reverse, u8, u8) { + ( + c.num_appearances, + Reverse(c.estimated_size_bits), + c.params.exponent, + c.params.factor, + ) +} + +fn is_better(c1: &Combination, c2: &Combination) -> bool { + rank(c1) > rank(c2) +} + +/// Estimate, in bits, what `params` would cost on `sample`. +/// +/// Encodes each value, checks whether it round-trips, and prices the result as +/// `num_values * bit_width + num_exceptions * exception_bits`, where `bit_width` +/// covers the FOR range of the values that did round-trip. +/// +/// Returns `None` when `penalize_exceptions` is set and fewer than two values +/// round-trip: such a candidate encodes almost everything as an exception, and +/// its FOR range is meaningless. +fn estimate_size_bits( + sample: &[F], + params: ExponentAndFactor, + penalize_exceptions: bool, +) -> Option { + let encode_scale = F::encode_scale(params.exponent, params.factor); + let decode_scale = F::decode_scale(params.exponent, params.factor); + + let mut num_exceptions = 0u64; + let mut min = None; + let mut max = None; + + for &value in sample { + let encoded = value.encode_value(encode_scale); + if F::decode_value(encoded, decode_scale) == value { + min = Some(min.map_or(encoded, |m: ::Signed| m.min(encoded))); + max = Some(max.map_or(encoded, |m: ::Signed| m.max(encoded))); + } else { + num_exceptions += 1; + } + } + + let num_values = sample.len() as u64; + let num_non_exceptions = num_values - num_exceptions; + if penalize_exceptions && num_non_exceptions < 2 { + return None; + } + + // With nothing to frame, there is no packed section at all: every value is + // stored as an exception. + let (Some(min), Some(max)) = (min, max) else { + return Some(num_values * exception_bits::()); + }; + + let range = + F::Exact::reinterpret_from_signed(max).wrapping_sub(F::Exact::reinterpret_from_signed(min)); + let bit_width = u64::from(num_required_bits(range.to_u64())); + + Some(num_values * bit_width + num_exceptions * exception_bits::()) +} + +/// Sample every `n`th value so the whole span is represented, capped at +/// [`SAMPLES_PER_VECTOR`] values. +fn sample_values(values: &[F], out: &mut Vec) { + out.clear(); + let stride = values.len().div_ceil(SAMPLES_PER_VECTOR).max(1); + out.extend(values.iter().step_by(stride).copied()); +} + +/// Build the column chunk's candidate set: the top [`MAX_COMBINATIONS`] pairs by +/// how often they win across sampled vectors. +/// +/// This is the first level of the ALP paper's two-level sampling. Each sampled +/// vector is searched exhaustively (66 pairs for `f32`, 190 for `f64`); the +/// winners are tallied, and the most frequent are carried forward so that each +/// vector only has to choose among a handful of candidates. +fn build_preset(values: &[F]) -> Vec { + let num_vectors = values.len().div_ceil(VECTOR_SIZE); + let vector_stride = num_vectors.div_ceil(SAMPLE_VECTORS).max(1); + + let mut sample = Vec::with_capacity(SAMPLES_PER_VECTOR); + let mut tally: Vec = Vec::new(); + + for vector in values.chunks(VECTOR_SIZE).step_by(vector_stride) { + sample_values(vector, &mut sample); + + // Start from the worst case - every value an exception, unpacked - so + // that any candidate that manages to encode anything beats it. + let mut best = Combination { + params: ExponentAndFactor { + exponent: F::MAX_EXPONENT, + factor: F::MAX_EXPONENT, + }, + num_appearances: 0, + estimated_size_bits: sample.len() as u64 + * (exception_bits::() + F::Exact::WIDTH as u64 * 8), + }; + + for exponent in 0..=F::MAX_EXPONENT { + for factor in 0..=exponent { + let params = ExponentAndFactor { exponent, factor }; + let Some(estimated_size_bits) = estimate_size_bits(&sample, params, true) else { + continue; + }; + let candidate = Combination { + params, + num_appearances: 0, + estimated_size_bits, + }; + if is_better(&candidate, &best) { + best = candidate; + } + } + } + + match tally.iter_mut().find(|c| c.params == best.params) { + Some(existing) => existing.num_appearances += 1, + None => tally.push(Combination { + num_appearances: 1, + ..best + }), + } + } + + // Size estimates from different vectors are not comparable with each other, + // so only the win counts rank the candidate set. + for combination in tally.iter_mut() { + combination.estimated_size_bits = 0; + } + tally.sort_by_key(|c| Reverse(rank(c))); + tally.truncate(MAX_COMBINATIONS); + + if tally.is_empty() { + // An empty page has nothing to sample; any valid pair will do. + return vec![ExponentAndFactor { + exponent: 0, + factor: 0, + }]; + } + tally.into_iter().map(|c| c.params).collect() +} + +/// Pick the candidate that encodes `vector` smallest. +/// +/// This is the second level of the two-level sampling: only the chunk's +/// candidates are tried, against a sample of the vector rather than all of it. +fn select_params( + vector: &[F], + preset: &[ExponentAndFactor], + sample: &mut Vec, +) -> ExponentAndFactor { + if preset.len() == 1 { + return preset[0]; + } + + sample_values(vector, sample); + + let mut best = preset[0]; + let mut best_size_bits = u64::MAX; + let mut worse_in_a_row = 0; + + for ¶ms in preset { + let Some(size_bits) = estimate_size_bits(sample, params, false) else { + continue; + }; + if size_bits >= best_size_bits { + worse_in_a_row += 1; + if worse_in_a_row == SAMPLING_EARLY_EXIT_THRESHOLD { + break; + } + continue; + } + best = params; + best_size_bits = size_bits; + worse_in_a_row = 0; + } + best +} + +/// Reusable per-vector buffers, kept across vectors and pages so encoding a page +/// does not allocate per vector. +struct Scratch { + /// Decimal-encoded integers, overwritten in place with their FOR deltas. + encoded: Vec<::Signed>, + /// Per-value exception flag (`1` = round-trip failed). A byte, not a bitset, + /// so the map loop's per-lane store stays vectorizable. + exc_mask: Vec, + exception_positions: Vec, + exception_values: Vec, + sample: Vec, +} + +impl Scratch { + fn new() -> Self { + Self { + encoded: Vec::new(), + exc_mask: Vec::new(), + exception_positions: Vec::new(), + exception_values: Vec::new(), + sample: Vec::new(), + } + } + + fn estimated_memory_size(&self) -> usize { + self.encoded.capacity() * std::mem::size_of::<::Signed>() + + self.exc_mask.capacity() + + self.exception_positions.capacity() * std::mem::size_of::() + + self.exception_values.capacity() * std::mem::size_of::() + + self.sample.capacity() * std::mem::size_of::() + } +} + +/// Encode one vector and append it to `out`: +/// `[AlpInfo][ForInfo][PackedValues][ExceptionPositions][ExceptionValues]`. +fn encode_vector( + values: &[F], + params: ExponentAndFactor, + scratch: &mut Scratch, + out: &mut Vec, +) -> Result<()> { + let encode_scale = F::encode_scale(params.exponent, params.factor); + let decode_scale = F::decode_scale(params.exponent, params.factor); + + let Scratch { + encoded, + exc_mask, + exception_positions, + exception_values, + .. + } = scratch; + + let zero = F::Exact::default().reinterpret_as_signed(); + let n_values = values.len(); + encoded.resize(n_values, zero); + exc_mask.resize(n_values, 0); + exception_values.clear(); + + // Comparing the *decoded* value to the input (not the input to itself) flags + // -0.0, which encodes to +0.0. + for ((&value, enc_slot), mask_slot) in values + .iter() + .zip(encoded.iter_mut()) + .zip(exc_mask.iter_mut()) + { + let encoded_value = value.encode_value(encode_scale); + *enc_slot = encoded_value; + *mask_slot = u8::from(F::decode_value(encoded_value, decode_scale) != value); + } + + // Branchless compaction: always write the index, advance only when flagged. + exception_positions.resize(n_values, 0); + let mut num_exceptions_usize = 0usize; + for (idx, &is_exception) in exc_mask.iter().enumerate() { + exception_positions[num_exceptions_usize] = idx as u16; + num_exceptions_usize += usize::from(is_exception); + } + exception_positions.truncate(num_exceptions_usize); + + let num_exceptions = u16::try_from(exception_positions.len()).map_err(|_| { + general_err!( + "Invalid ALP vector: {} exceptions exceeds u16::MAX", + exception_positions.len() + ) + })?; + + // Fill each exception's packed slot with a real value (not the sentinel) so + // the FOR range - and thus the bit width - stays tight; the true value goes + // in the exception section. + let placeholder = first_non_exception_value::(encoded, exception_positions); + for &position in exception_positions.iter() { + exception_values.push(values[position as usize]); + encoded[position as usize] = placeholder; + } + + let min = encoded.iter().copied().min().unwrap_or(zero); + let max = encoded.iter().copied().max().unwrap_or(zero); + + let frame_of_reference = F::Exact::reinterpret_from_signed(min); + let range = F::Exact::reinterpret_from_signed(max).wrapping_sub(frame_of_reference); + let bit_width = num_required_bits(range.to_u64()); + + let alp_info = AlpInfo { + exponent: params.exponent, + factor: params.factor, + num_exceptions, + }; + let for_info = ForInfo:: { + frame_of_reference, + bit_width, + }; + alp_info.extend_serialized(out); + for_info.extend_serialized(out); + + // PackedValues: FOR deltas, LSB-first, as the spec requires. A zero bit + // width means every value equals the frame of reference, so nothing is + // stored. + if bit_width > 0 { + let mut writer = BitWriter::new_from_buf(std::mem::take(out)); + for &encoded_value in encoded.iter() { + let delta = + F::Exact::reinterpret_from_signed(encoded_value).wrapping_sub(frame_of_reference); + writer.put_value(delta.to_u64(), bit_width as usize); + } + // Pads to a byte boundary, giving exactly the ceil(n * bit_width / 8) + // bytes the decoder derives from the metadata. + *out = writer.consume(); + } + + for &position in exception_positions.iter() { + out.extend_from_slice(&position.to_le_bytes()); + } + for &value in exception_values.iter() { + value.to_exact_bits().extend_le_bytes(out); + } + + Ok(()) +} + +/// The encoded integer of the first value that is not an exception, or zero if +/// every value is one. `exception_positions` is ascending, so the first index it +/// skips over is the first non-exception. +fn first_non_exception_value( + encoded: &[::Signed], + exception_positions: &[u16], +) -> ::Signed { + let mut candidate = 0usize; + for &position in exception_positions { + if position as usize != candidate { + break; + } + candidate += 1; + } + encoded + .get(candidate) + .copied() + .unwrap_or_else(|| F::Exact::default().reinterpret_as_signed()) +} + +/// Encode `values` as one ALP page. +fn encode_page( + values: &[F], + preset: &[ExponentAndFactor], + scratch: &mut Scratch, +) -> Result> { + let header = AlpHeader { + compression_mode: ALP_COMPRESSION_MODE, + integer_encoding: ALP_INTEGER_ENCODING_FOR_BIT_PACK, + vector_size: VECTOR_SIZE, + num_elements: values.len(), + }; + let num_vectors = header.num_vectors(); + + let mut page = Vec::with_capacity(ALP_HEADER_SIZE + num_vectors * 4 + values.len() * 4); + page.extend_from_slice(&header.serialize()?); + + // Offsets are only known once each vector is encoded, so leave room and + // backfill. + let offsets_start = page.len(); + page.resize(offsets_start + num_vectors * std::mem::size_of::(), 0); + + for (idx, vector) in values.chunks(VECTOR_SIZE).enumerate() { + // Offsets are relative to the start of the page body, which follows the + // fixed-size header. + let offset = u32::try_from(page.len() - ALP_HEADER_SIZE) + .map_err(|_| general_err!("Invalid ALP page: body exceeds u32 offset range"))?; + let offset_at = offsets_start + idx * std::mem::size_of::(); + page[offset_at..offset_at + 4].copy_from_slice(&offset.to_le_bytes()); + + let params = select_params(vector, preset, &mut scratch.sample); + encode_vector(vector, params, scratch, &mut page)?; + } + + Ok(page) +} + +/// A page encoded incrementally, one vector at a time. +/// +/// Once the column chunk's preset is known (after the first page), later pages do +/// not need all their values at once: each vector is encoded and appended to +/// `body` as soon as its [`VECTOR_SIZE`] values arrive, and the raw floats are +/// dropped. Only a sub-vector remainder is held in `carry` between `put`s. The +/// header and offset array - which depend on the final vector count - are +/// prepended in [`StreamingPage::finish`]. +/// +/// The bytes produced are identical to encoding the same values in one pass with +/// [`encode_page`]: same vectors, same per-vector parameters, same contiguous +/// offsets. What changes is peak memory - the whole page of raw floats is never +/// resident - and that the encode work is spread across `put`s. +struct StreamingPage { + /// Encoded vectors, concatenated; becomes the page body after the offsets. + body: Vec, + /// Body-relative start offset of each completed vector. + vector_offsets: Vec, + /// Values not yet forming a complete vector, carried across `put`s. + carry: Vec, + /// Total values appended to this page so far. + count: usize, +} + +impl StreamingPage { + fn new() -> Self { + Self { + body: Vec::new(), + vector_offsets: Vec::new(), + carry: Vec::new(), + count: 0, + } + } + + fn estimated_memory_size(&self) -> usize { + self.body.capacity() + + self.vector_offsets.capacity() * std::mem::size_of::() + + self.carry.capacity() * std::mem::size_of::() + } + + /// Encode one complete vector and append it to `body`. + fn push_vector( + &mut self, + vector: &[F], + preset: &[ExponentAndFactor], + scratch: &mut Scratch, + ) -> Result<()> { + let body_start = u32::try_from(self.body.len()) + .map_err(|_| general_err!("Invalid ALP page: body exceeds u32 offset range"))?; + self.vector_offsets.push(body_start); + let params = select_params(vector, preset, &mut scratch.sample); + encode_vector(vector, params, scratch, &mut self.body) + } + + /// Buffer `values`, encoding and dropping every complete vector they form. + fn put( + &mut self, + mut values: &[F], + preset: &[ExponentAndFactor], + scratch: &mut Scratch, + ) -> Result<()> { + self.count += values.len(); + + // Finish a vector left partially filled by a previous `put`. + if !self.carry.is_empty() { + let need = VECTOR_SIZE - self.carry.len(); + if values.len() < need { + self.carry.extend_from_slice(values); + return Ok(()); + } + let (head, tail) = values.split_at(need); + self.carry.extend_from_slice(head); + // Move the carry out so the vector slice does not alias `self`; the + // allocation is handed back, cleared, for the next partial to reuse. + let mut vector = std::mem::take(&mut self.carry); + self.push_vector(&vector, preset, scratch)?; + vector.clear(); + self.carry = vector; + values = tail; + } + + // Encode whole vectors straight from the input, no copy. + let mut chunks = values.chunks_exact(VECTOR_SIZE); + for vector in chunks.by_ref() { + self.push_vector(vector, preset, scratch)?; + } + self.carry.extend_from_slice(chunks.remainder()); + Ok(()) + } + + /// Encode the trailing partial vector, then assemble `[header][offsets][body]` + /// and reset for the next page. + fn finish( + &mut self, + preset: &[ExponentAndFactor], + scratch: &mut Scratch, + ) -> Result> { + if !self.carry.is_empty() { + let mut vector = std::mem::take(&mut self.carry); + self.push_vector(&vector, preset, scratch)?; + vector.clear(); + self.carry = vector; + } + + let num_vectors = self.vector_offsets.len(); + let offsets_section = num_vectors * std::mem::size_of::(); + let header = AlpHeader { + compression_mode: ALP_COMPRESSION_MODE, + integer_encoding: ALP_INTEGER_ENCODING_FOR_BIT_PACK, + vector_size: VECTOR_SIZE, + num_elements: self.count, + }; + + let mut page = Vec::with_capacity(ALP_HEADER_SIZE + offsets_section + self.body.len()); + page.extend_from_slice(&header.serialize()?); + for &body_start in &self.vector_offsets { + // Offsets are measured from the start of the offset array: the array's + // own size plus the vector's position within the body. + let page_offset = u32::try_from(offsets_section + body_start as usize) + .map_err(|_| general_err!("Invalid ALP page: body exceeds u32 offset range"))?; + page.extend_from_slice(&page_offset.to_le_bytes()); + } + page.extend_from_slice(&self.body); + + self.body.clear(); + self.vector_offsets.clear(); + self.count = 0; + Ok(page) + } +} + +/// Encoder for ALP-encoded floating-point pages (`f32`/`f64`). +/// +/// The first page is buffered whole: ALP samples it to choose the column chunk's +/// candidate `(exponent, factor)` set, which needs all of the page's data. Once +/// that preset is fixed, later pages are encoded incrementally, a vector at a +/// time, without ever holding the whole page of raw floats (see [`StreamingPage`]). +pub struct AlpEncoder +where + T::T: AlpFloat, +{ + /// Values buffered for the first page, until the preset is built. Empty once + /// streaming begins. + values: Vec, + /// Candidate `(exponent, factor)` pairs for this column chunk, sampled once + /// from the first page and reused for the rest of the chunk. `None` until the + /// first page is flushed; its presence is what switches `put` to streaming. + preset: Option>, + scratch: Scratch, + /// The page built incrementally, used for every page after the first. + streaming: StreamingPage, +} + +impl AlpEncoder +where + T::T: AlpFloat, +{ + pub(crate) fn new() -> Self { + Self { + values: Vec::new(), + preset: None, + scratch: Scratch::new(), + streaming: StreamingPage::new(), + } + } + + /// Values buffered for the current page: the first page lives in `values`, + /// later pages in the streaming buffer. Used only for size estimates. + fn current_page_len(&self) -> usize { + if self.preset.is_none() { + self.values.len() + } else { + self.streaming.count + } + } +} + +impl Encoder for AlpEncoder +where + T::T: AlpFloat, +{ + fn put(&mut self, values: &[T::T]) -> Result<()> { + let Self { + values: buffer, + preset, + scratch, + streaming, + } = self; + match preset.as_deref() { + // First page: buffer until the preset can be built on flush. + None => buffer.extend_from_slice(values), + // Later pages: encode incrementally against the fixed preset. + Some(preset) => streaming.put(values, preset, scratch)?, + } + Ok(()) + } + + fn encoding(&self) -> Encoding { + Encoding::ALP + } + + fn estimated_data_encoded_size(&self) -> usize { + // Encoded size is not known until the parameters are chosen, so bound it + // by the unencoded size: a vector never encodes larger than storing every + // value as an exception. Bounding by value count (not the running encoded + // size) keeps the writer's page-boundary decisions identical whether the + // page is buffered or streamed. + let len = self.current_page_len(); + let num_vectors = len.div_ceil(VECTOR_SIZE); + ALP_HEADER_SIZE + + num_vectors + * (std::mem::size_of::() + + AlpInfo::STORED_SIZE + + ForInfo::<::Exact>::stored_size()) + + len * (::Exact::WIDTH + std::mem::size_of::()) + } + + fn estimated_memory_size(&self) -> usize { + self.values.capacity() * std::mem::size_of::() + + self.preset.as_ref().map_or(0, |p| { + p.capacity() * std::mem::size_of::() + }) + + self.scratch.estimated_memory_size() + + self.streaming.estimated_memory_size() + } + + fn flush_buffer(&mut self) -> Result { + let Self { + values, + preset, + scratch, + streaming, + } = self; + + // The first flush builds the preset from the whole buffered page and + // encodes it in one pass; that also arms streaming for later pages. + let page = match preset { + None => { + let built = build_preset(values); + let page = encode_page(values, &built, scratch)?; + values.clear(); + *preset = Some(built); + page + } + Some(preset) => streaming.finish(preset.as_slice(), scratch)?, + }; + Ok(page.into()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::data_type::{DoubleType, FloatType}; + use crate::encodings::decoding::Decoder; + use crate::encodings::decoding::alp_decoder::AlpDecoder; + + /// Encode `values` and read them back through the decoder. + fn roundtrip(values: &[T::T]) -> Vec + where + T::T: AlpFloat, + ::Exact: Send, + { + let mut encoder = AlpEncoder::::new(); + encoder.put(values).unwrap(); + let page = encoder.flush_buffer().unwrap(); + + let mut decoder = AlpDecoder::::new(); + decoder.set_data(page, values.len()).unwrap(); + let mut out = vec![T::T::default(); values.len()]; + assert_eq!(decoder.get(&mut out).unwrap(), values.len()); + out + } + + /// Assert bit-for-bit equality, so that NaN and -0.0 are held to the same + /// standard as every other value. + fn assert_bits_eq(actual: &[F], expected: &[F]) { + assert_eq!(actual.len(), expected.len(), "length mismatch"); + for (idx, (a, e)) in actual.iter().zip(expected.iter()).enumerate() { + assert_eq!( + a.to_exact_bits(), + e.to_exact_bits(), + "value mismatch at {idx}: expected {e:?}, got {a:?}" + ); + } + } + + #[test] + fn test_roundtrip_f64_decimals() { + let values: Vec = (0..500).map(|i| (i as f64) * 0.01 + 1.23).collect(); + assert_bits_eq(&roundtrip::(&values), &values); + } + + #[test] + fn test_roundtrip_f32_decimals() { + let values: Vec = (0..500).map(|i| (i as f32) * 0.5 + 1.25).collect(); + assert_bits_eq(&roundtrip::(&values), &values); + } + + /// Spans several vectors, including a short trailing one. + #[test] + fn test_roundtrip_multiple_vectors() { + let values: Vec = (0..2600).map(|i| (i as f64) * 0.001).collect(); + assert_bits_eq(&roundtrip::(&values), &values); + } + + /// The values ALP cannot represent must survive verbatim through the + /// exception path - including -0.0, which compares equal to +0.0 and would + /// be silently lost by a naive round-trip check. + #[test] + fn test_roundtrip_exceptions() { + let values = vec![ + 1.5f64, + f64::NAN, + 2.5, + f64::INFINITY, + -0.0, + f64::NEG_INFINITY, + 3.5, + 0.0, + f64::MAX, + f64::MIN, + ]; + let decoded = roundtrip::(&values); + assert_bits_eq(&decoded, &values); + assert!(decoded[1].is_nan()); + assert!(decoded[4].is_sign_negative()); + } + + /// Every value identical means a zero FOR range, so `bit_width` is 0 and no + /// packed section is written at all. + #[test] + fn test_roundtrip_all_identical() { + let values = vec![42.42f64; 3000]; + assert_bits_eq(&roundtrip::(&values), &values); + } + + #[test] + fn test_roundtrip_all_exceptions() { + let values = vec![f64::NAN; 100]; + let decoded = roundtrip::(&values); + assert!(decoded.iter().all(|v| v.is_nan())); + } + + #[test] + fn test_roundtrip_single_value() { + assert_bits_eq(&roundtrip::(&[3.25]), &[3.25]); + } + + #[test] + fn test_roundtrip_empty() { + let mut encoder = AlpEncoder::::new(); + let page = encoder.flush_buffer().unwrap(); + assert_eq!(page.len(), ALP_HEADER_SIZE); + + let mut decoder = AlpDecoder::::new(); + decoder.set_data(page, 0).unwrap(); + assert_eq!(decoder.values_left(), 0); + } + + /// The encoder must be reusable across pages, and the candidate set chosen on + /// the first page must still produce valid pages for later ones. + #[test] + fn test_roundtrip_multiple_pages() { + let mut encoder = AlpEncoder::::new(); + + for page_idx in 0..3 { + let values: Vec = (0..1500) + .map(|i| (i as f64) * 0.01 + (page_idx as f64)) + .collect(); + encoder.put(&values).unwrap(); + let page = encoder.flush_buffer().unwrap(); + + let mut decoder = AlpDecoder::::new(); + decoder.set_data(page, values.len()).unwrap(); + let mut out = vec![0.0f64; values.len()]; + assert_eq!(decoder.get(&mut out).unwrap(), values.len()); + assert_bits_eq(&out, &values); + } + } + + /// Two-decimal data must encode with no exceptions and pack far below the 64 + /// bits an unencoded double takes. + /// + /// The parameter to check is `exponent - factor`, not `exponent`: the encoded + /// integers depend only on the effective scale `10^(exponent - factor)`, so + /// (3, 1) and (2, 0) encode identically and tie on estimated size. The ALP + /// paper's tie-break then prefers the larger exponent and factor, which is + /// why the pair chosen here is not simply (2, 0). + #[test] + fn test_selects_decimal_parameters() { + let values: Vec = (0..1024).map(|i| (i as f64) * 0.01).collect(); + + let mut encoder = AlpEncoder::::new(); + encoder.put(&values).unwrap(); + let page = encoder.flush_buffer().unwrap(); + + // [header][one u32 offset][exponent][factor][num_exceptions]... + let exponent = page[ALP_HEADER_SIZE + 4]; + let factor = page[ALP_HEADER_SIZE + 5]; + let num_exceptions = + u16::from_le_bytes([page[ALP_HEADER_SIZE + 6], page[ALP_HEADER_SIZE + 7]]); + + assert_eq!( + exponent - factor, + 2, + "two-decimal data should encode at an effective scale of 10^2, \ + got exponent {exponent} factor {factor}" + ); + assert_eq!(num_exceptions, 0, "two-decimal data should not except"); + + let plain_size = values.len() * std::mem::size_of::(); + assert!( + page.len() * 4 < plain_size, + "expected at least 4x compression, got {} bytes vs {plain_size} plain", + page.len() + ); + } + + /// A single exception must not blow up the frame of reference: its slot is + /// filled with a real encoded value, so the bit width stays tight. + #[test] + fn test_exception_placeholder_keeps_bit_width_tight() { + let mut values: Vec = (0..1024).map(|i| (i as f64) * 0.01).collect(); + values[500] = f64::NAN; + + let mut encoder = AlpEncoder::::new(); + encoder.put(&values).unwrap(); + let page = encoder.flush_buffer().unwrap(); + + // [header][offset][AlpInfo(4)][frame_of_reference(8)][bit_width(1)] + let bit_width = page[ALP_HEADER_SIZE + 4 + AlpInfo::STORED_SIZE + 8]; + assert!( + bit_width <= 17, + "one exception should not widen the frame; got bit_width {bit_width}" + ); + + assert_bits_eq(&roundtrip::(&values), &values); + } + + /// Streaming pages (every page after the first) must be byte-for-byte + /// identical to encoding the same values in one pass with the same preset. + #[test] + fn test_streaming_matches_buffered() { + let page1: Vec = (0..3000).map(|i| (i as f64) * 0.01 + 1.23).collect(); + let page2: Vec = (0..3000).map(|i| (i as f64) * 0.03 - 7.0).collect(); + + let mut encoder = AlpEncoder::::new(); + encoder.put(&page1).unwrap(); + let _ = encoder.flush_buffer().unwrap(); // builds and caches the preset + + // Feed page 2 in irregular chunks that cross vector boundaries. + for chunk in page2.chunks(997) { + encoder.put(chunk).unwrap(); + } + let streamed = encoder.flush_buffer().unwrap(); + + // The encoder built its preset from page 1; reproduce it to encode page 2 + // in a single pass as the reference. + let preset = build_preset(&page1); + let mut scratch = Scratch::::new(); + let reference = encode_page(&page2, &preset, &mut scratch).unwrap(); + + assert_eq!( + streamed.as_ref(), + reference.as_slice(), + "streamed page differs from single-pass encoding" + ); + } + + /// The carry logic must reassemble vectors correctly no matter how `put` calls + /// land relative to vector boundaries: sizes below, at, and above a vector. + #[test] + fn test_streaming_irregular_puts() { + let page1: Vec = (0..2048).map(|i| (i as f64) * 0.01).collect(); + let page2: Vec = (0..5000).map(|i| (i as f64) * 0.01 + 100.0).collect(); + + let mut encoder = AlpEncoder::::new(); + encoder.put(&page1).unwrap(); + let _ = encoder.flush_buffer().unwrap(); + + let sizes = [1usize, 1023, 2, 1024, 1025, 7, 900, 118]; + let (mut offset, mut i) = (0usize, 0usize); + while offset < page2.len() { + let n = sizes[i % sizes.len()].min(page2.len() - offset); + encoder.put(&page2[offset..offset + n]).unwrap(); + offset += n; + i += 1; + } + let page = encoder.flush_buffer().unwrap(); + + let mut decoder = AlpDecoder::::new(); + decoder.set_data(page, page2.len()).unwrap(); + let mut out = vec![0.0f64; page2.len()]; + assert_eq!(decoder.get(&mut out).unwrap(), page2.len()); + assert_bits_eq(&out, &page2); + } +} + +#[cfg(test)] +mod conformance { + use super::*; + use crate::column::page::Page; + use crate::data_type::FloatType; + use crate::encodings::decoding::Decoder; + use crate::encodings::decoding::alp_decoder::AlpDecoder; + use crate::file::reader::{FileReader, SerializedFileReader}; + + /// Re-encode the values from a page written by the arrow-cpp ALP + /// implementation and require our bytes to match theirs exactly. + /// + /// The parameter search is explicitly non-normative - the spec says any + /// valid (exponent, factor) pair decodes correctly - so this is not a + /// conformance requirement in itself. It is a much sharper test than one: + /// byte identity pins the decimal parameters, the exception set, the + /// placeholder substitution, the frame of reference, the bit width and the + /// packing order all at once, against an independent implementation. + #[test] + fn test_matches_cpp_reference_page() { + let path = format!( + "{}/alp_float_arade.parquet", + arrow::util::test_util::parquet_test_data() + ); + let reader = SerializedFileReader::new(std::fs::File::open(&path).unwrap()).unwrap(); + let mut page_reader = reader + .get_row_group(0) + .unwrap() + .get_column_page_reader(0) + .unwrap(); + + let mut pages_checked = 0; + while let Some(page) = page_reader.get_next_page().unwrap() { + let Page::DataPage { + buf, + num_values, + encoding: Encoding::ALP, + .. + } = &page + else { + continue; + }; + let num_values = *num_values as usize; + + // A v1 data page stores its definition levels ahead of the encoded + // values: a 4-byte length, then that many bytes. + let def_len = u32::from_le_bytes(buf.as_ref()[0..4].try_into().unwrap()) as usize; + let cpp_page = buf.slice(4 + def_len..); + + let mut decoder = AlpDecoder::::new(); + decoder.set_data(cpp_page.clone(), num_values).unwrap(); + let mut values = vec![0.0f32; num_values]; + assert_eq!(decoder.get(&mut values).unwrap(), num_values); + + let mut encoder = AlpEncoder::::new(); + encoder.put(&values).unwrap(); + let our_page = encoder.flush_buffer().unwrap(); + + assert_eq!( + our_page.as_ref(), + cpp_page.as_ref(), + "re-encoded page differs from the arrow-cpp reference page" + ); + pages_checked += 1; + } + assert!(pages_checked > 0, "no ALP data page found in the test file"); + } +} diff --git a/parquet/src/encodings/encoding/mod.rs b/parquet/src/encodings/encoding/mod.rs index eeabcf4ba5ce..cece022fb1c4 100644 --- a/parquet/src/encodings/encoding/mod.rs +++ b/parquet/src/encodings/encoding/mod.rs @@ -27,10 +27,12 @@ use crate::errors::{ParquetError, Result}; use crate::schema::types::ColumnDescPtr; use crate::util::bit_util::{BitWriter, num_required_bits}; +use alp_encoder::AlpEncoder; use byte_stream_split_encoder::{ByteStreamSplitEncoder, VariableWidthByteStreamSplitEncoder}; use bytes::Bytes; pub use dict_encoder::DictEncoder; +mod alp_encoder; mod byte_stream_split_encoder; mod dict_encoder; @@ -84,26 +86,90 @@ pub fn get_encoder( encoding: Encoding, descr: &ColumnDescPtr, ) -> Result>> { - let encoder: Box> = match encoding { - Encoding::PLAIN => Box::new(PlainEncoder::new()), - Encoding::RLE_DICTIONARY | Encoding::PLAIN_DICTIONARY => { - return Err(general_err!( - "Cannot initialize this encoding through this function" - )); + ::get_encoder(descr, encoding) +} + +pub(crate) mod private { + use super::*; + + /// A trait that allows getting an [`Encoder`] implementation for a [`DataType`] + /// with the corresponding [`ParquetValueType`]. This is necessary to support + /// [`Encoder`] implementations that may not be applicable for all [`DataType`] + /// and by extension all [`ParquetValueType`], such as ALP, which encodes only + /// floating-point columns. + /// + /// [`ParquetValueType`]: crate::data_type::private::ParquetValueType + pub trait GetEncoder { + fn get_encoder>( + descr: &ColumnDescPtr, + encoding: Encoding, + ) -> Result>> { + get_encoder_default(descr, encoding) } - Encoding::RLE => Box::new(RleValueEncoder::new()), - Encoding::DELTA_BINARY_PACKED => Box::new(DeltaBitPackEncoder::new()), - Encoding::DELTA_LENGTH_BYTE_ARRAY => Box::new(DeltaLengthByteArrayEncoder::new()), - Encoding::DELTA_BYTE_ARRAY => Box::new(DeltaByteArrayEncoder::new()), - Encoding::BYTE_STREAM_SPLIT => match T::get_physical_type() { - Type::FIXED_LEN_BYTE_ARRAY => Box::new(VariableWidthByteStreamSplitEncoder::new( - descr.type_length(), - )), - _ => Box::new(ByteStreamSplitEncoder::new()), - }, - e => return Err(nyi_err!("Encoding {} is not supported", e)), - }; - Ok(encoder) + } + + fn get_encoder_default( + descr: &ColumnDescPtr, + encoding: Encoding, + ) -> Result>> { + let encoder: Box> = match encoding { + Encoding::PLAIN => Box::new(PlainEncoder::new()), + Encoding::RLE_DICTIONARY | Encoding::PLAIN_DICTIONARY => { + return Err(general_err!( + "Cannot initialize this encoding through this function" + )); + } + Encoding::RLE => Box::new(RleValueEncoder::new()), + Encoding::DELTA_BINARY_PACKED => Box::new(DeltaBitPackEncoder::new()), + Encoding::DELTA_LENGTH_BYTE_ARRAY => Box::new(DeltaLengthByteArrayEncoder::new()), + Encoding::DELTA_BYTE_ARRAY => Box::new(DeltaByteArrayEncoder::new()), + Encoding::BYTE_STREAM_SPLIT => match T::get_physical_type() { + Type::FIXED_LEN_BYTE_ARRAY => Box::new(VariableWidthByteStreamSplitEncoder::new( + descr.type_length(), + )), + _ => Box::new(ByteStreamSplitEncoder::new()), + }, + Encoding::ALP => { + return Err(general_err!( + "Encoding {} is not supported for type", + encoding + )); + } + e => return Err(nyi_err!("Encoding {} is not supported", e)), + }; + Ok(encoder) + } + + impl GetEncoder for bool {} + impl GetEncoder for i32 {} + impl GetEncoder for i64 {} + impl GetEncoder for Int96 {} + impl GetEncoder for ByteArray {} + impl GetEncoder for FixedLenByteArray {} + + impl GetEncoder for f32 { + fn get_encoder>( + descr: &ColumnDescPtr, + encoding: Encoding, + ) -> Result>> { + match encoding { + Encoding::ALP => Ok(Box::new(AlpEncoder::new())), + _ => get_encoder_default(descr, encoding), + } + } + } + + impl GetEncoder for f64 { + fn get_encoder>( + descr: &ColumnDescPtr, + encoding: Encoding, + ) -> Result>> { + match encoding { + Encoding::ALP => Ok(Box::new(AlpEncoder::new())), + _ => get_encoder_default(descr, encoding), + } + } + } } // ---------------------------------------------------------------------- diff --git a/parquet/src/encodings/mod.rs b/parquet/src/encodings/mod.rs index 894c4fb961ee..cb23625a8251 100644 --- a/parquet/src/encodings/mod.rs +++ b/parquet/src/encodings/mod.rs @@ -15,7 +15,9 @@ // specific language governing permissions and limitations // under the License. +mod alp; pub mod decoding; pub mod encoding; pub mod levels; + experimental!(pub(crate) mod rle); diff --git a/parquet/tests/arrow_reader/alp.rs b/parquet/tests/arrow_reader/alp.rs new file mode 100644 index 000000000000..734aae6bc7b9 --- /dev/null +++ b/parquet/tests/arrow_reader/alp.rs @@ -0,0 +1,258 @@ +// 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 arrow::compute::concat_batches; +use arrow::util::test_util::parquet_test_data; +use arrow_array::cast::as_primitive_array; +use arrow_array::types::{Float32Type, Float64Type}; +use arrow_array::{Array, Float32Array, Float64Array, RecordBatch}; +use arrow_csv::ReaderBuilder as CsvReaderBuilder; +use arrow_schema::{DataType, Field, Schema}; +use bytes::Bytes; +use parquet::arrow::ArrowWriter; +use parquet::arrow::arrow_reader::ArrowReaderBuilder; +use parquet::basic::Encoding; +use parquet::file::properties::{WriterProperties, WriterVersion}; +use std::fs::File; +use std::path::PathBuf; +use std::sync::Arc; + +#[test] +fn test_read_f32_alp() { + let data_dir = PathBuf::from(parquet_test_data()); + let parquet_path = data_dir.join("alp_float_arade.parquet"); + let expected_csv_path = data_dir.join("alp_arade_expect.csv"); + + let expected = read_expected_csv_batch(&expected_csv_path); + let actual = read_parquet_batch(&parquet_path); + + assert_eq!(actual.schema(), expected.schema(), "schema mismatch"); + assert_eq!( + actual.num_columns(), + expected.num_columns(), + "column mismatch" + ); + assert_eq!(actual.num_rows(), expected.num_rows(), "row count mismatch"); + + for col_idx in 0..actual.num_columns() { + let col_name = actual.schema().field(col_idx).name().clone(); + let actual_col = as_primitive_array::(actual.column(col_idx).as_ref()); + let expected_col = as_primitive_array::(expected.column(col_idx).as_ref()); + + for row_idx in 0..actual.num_rows() { + assert_eq!( + actual_col.is_valid(row_idx), + expected_col.is_valid(row_idx), + "null mismatch at column {col_name} row {row_idx}" + ); + if actual_col.is_valid(row_idx) { + let actual_value = actual_col.value(row_idx); + let expected_value = expected_col.value(row_idx); + assert!( + actual_value.to_bits() == expected_value.to_bits(), + "bit mismatch at column {col_name} row {row_idx}: expected={expected_value} actual={actual_value}" + ); + } + } + } +} + +/// Write the arade values with the ALP encoder and read them back, over real +/// float data rather than synthetic decimals. +/// +/// This checks losslessness only, not compression: these are `f32` values whose +/// encoded integers need 31 bits and which except 6.4% of the time, so ALP is +/// larger than PLAIN here, a property of the data rather than the encoder. +/// ALP's win on arade in the paper is on the `f64` version of the dataset. +#[test] +fn test_write_f32_alp_roundtrip() { + let data_dir = PathBuf::from(parquet_test_data()); + let expected = read_expected_csv_batch(&data_dir.join("alp_arade_expect.csv")); + + let alp_bytes = write_batch(&expected, Encoding::ALP); + let actual = read_parquet_bytes(alp_bytes); + assert_eq!(actual.num_rows(), expected.num_rows(), "row count mismatch"); + + for col_idx in 0..expected.num_columns() { + let col_name = expected.schema().field(col_idx).name().clone(); + let actual_col = as_primitive_array::(actual.column(col_idx).as_ref()); + let expected_col = as_primitive_array::(expected.column(col_idx).as_ref()); + + for row_idx in 0..expected.num_rows() { + assert_eq!( + actual_col.is_valid(row_idx), + expected_col.is_valid(row_idx), + "null mismatch at column {col_name} row {row_idx}" + ); + if expected_col.is_valid(row_idx) { + // Bitwise, so that NaN and -0.0 are held to the same standard. + assert_eq!( + actual_col.value(row_idx).to_bits(), + expected_col.value(row_idx).to_bits(), + "bit mismatch at column {col_name} row {row_idx}" + ); + } + } + } +} + +/// Round-trip a nullable column under both data page versions. +/// +/// The versions differ in what the reader can tell the value decoder. A v2 page +/// header carries `num_nulls`, so the decoder is handed the exact count of +/// encoded values; a v1 header carries only `num_values`, which counts nulls, so +/// the decoder receives the level count instead. Only non-null values are +/// encoded either way, which is why the ALP header's element count - not the +/// count the reader passes in - is what governs the decode. +/// +/// The exception values are here deliberately: they make the encoded and +/// unencoded value counts differ from the level count in two different ways at +/// once. +#[test] +fn test_alp_roundtrip_page_versions_with_nulls() { + let f32_values: Vec> = (0..3000) + .map(|i| match i % 100 { + 0 => None, + 7 => Some(f32::NAN), + 13 => Some(-0.0), + _ => Some(i as f32 * 0.01), + }) + .collect(); + let f64_values: Vec> = (0..3000) + .map(|i| match i % 100 { + 5 => None, + 11 => Some(f64::INFINITY), + _ => Some(i as f64 * 0.001), + }) + .collect(); + + let schema = Arc::new(Schema::new(vec![ + Field::new("f32", DataType::Float32, true), + Field::new("f64", DataType::Float64, true), + ])); + let batch = RecordBatch::try_new( + schema, + vec![ + Arc::new(Float32Array::from(f32_values.clone())), + Arc::new(Float64Array::from(f64_values.clone())), + ], + ) + .unwrap(); + + for version in [WriterVersion::PARQUET_1_0, WriterVersion::PARQUET_2_0] { + let props = WriterProperties::builder() + .set_dictionary_enabled(false) + .set_encoding(Encoding::ALP) + .set_writer_version(version) + .build(); + + let mut buf = Vec::new(); + let mut writer = ArrowWriter::try_new(&mut buf, batch.schema(), Some(props)).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + + let actual = read_parquet_bytes(buf.into()); + assert_eq!( + actual.num_rows(), + batch.num_rows(), + "{version:?}: row count" + ); + + let actual_f32 = as_primitive_array::(actual.column(0).as_ref()); + for (row, expected) in f32_values.iter().enumerate() { + match expected { + None => assert!( + actual_f32.is_null(row), + "{version:?}: f32 row {row} not null" + ), + // Bitwise, so NaN and -0.0 are held to the same standard. + Some(value) => assert_eq!( + actual_f32.value(row).to_bits(), + value.to_bits(), + "{version:?}: f32 row {row}" + ), + } + } + + let actual_f64 = as_primitive_array::(actual.column(1).as_ref()); + for (row, expected) in f64_values.iter().enumerate() { + match expected { + None => assert!( + actual_f64.is_null(row), + "{version:?}: f64 row {row} not null" + ), + Some(value) => assert_eq!( + actual_f64.value(row).to_bits(), + value.to_bits(), + "{version:?}: f64 row {row}" + ), + } + } + } +} + +fn write_batch(batch: &RecordBatch, encoding: Encoding) -> Bytes { + let props = WriterProperties::builder() + .set_dictionary_enabled(false) + .set_encoding(encoding) + .build(); + + let mut buf = Vec::new(); + let mut writer = ArrowWriter::try_new(&mut buf, batch.schema(), Some(props)).unwrap(); + writer.write(batch).unwrap(); + writer.close().unwrap(); + buf.into() +} + +fn read_parquet_bytes(bytes: Bytes) -> RecordBatch { + let reader = ArrowReaderBuilder::try_new(bytes).unwrap().build().unwrap(); + let batches: Vec = reader.map(|b| b.unwrap()).collect(); + assert!(!batches.is_empty(), "expected non-empty parquet batch set"); + concat_batches(batches[0].schema_ref(), &batches).unwrap() +} + +fn alp_schema() -> Arc { + Arc::new(Schema::new(vec![ + Field::new("value1", DataType::Float32, true), + Field::new("value2", DataType::Float32, true), + Field::new("value3", DataType::Float32, true), + Field::new("value4", DataType::Float32, true), + ])) +} + +fn read_parquet_batch(path: &PathBuf) -> RecordBatch { + let file = File::open(path).unwrap(); + let reader = ArrowReaderBuilder::try_new(file).unwrap().build().unwrap(); + let mut batches = Vec::new(); + for batch in reader { + batches.push(batch.unwrap()); + } + assert!(!batches.is_empty(), "expected non-empty parquet batch set"); + concat_batches(batches[0].schema_ref(), &batches).unwrap() +} + +fn read_expected_csv_batch(path: &PathBuf) -> RecordBatch { + let file = File::open(path).unwrap(); + let schema = alp_schema(); + let reader = CsvReaderBuilder::new(schema.clone()) + .with_header(true) + .build(file) + .unwrap(); + let batches: Vec = reader.map(|b| b.unwrap()).collect(); + assert!(!batches.is_empty(), "expected non-empty csv batch set"); + concat_batches(&schema, &batches).unwrap() +} diff --git a/parquet/tests/arrow_reader/mod.rs b/parquet/tests/arrow_reader/mod.rs index 404bebc05d0b..dc8cb614888d 100644 --- a/parquet/tests/arrow_reader/mod.rs +++ b/parquet/tests/arrow_reader/mod.rs @@ -38,6 +38,7 @@ use parquet::file::properties::{ use std::sync::Arc; use tempfile::NamedTempFile; +mod alp; mod bad_data; mod bloom_filter; #[cfg(feature = "crc")]