This is an automated email from the ASF dual-hosted git repository.
alamb pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-rs.git
The following commit(s) were added to refs/heads/main by this push:
new cb682a2f35 [Parquet] ALP encoder/decoder support (#9372)
cb682a2f35 is described below
commit cb682a2f35e5bf1c610c8a762975ba933d4701f1
Author: Kosta Tarasov <[email protected]>
AuthorDate: Fri Sep 4 08:08:10 2026 -0400
[Parquet] ALP encoder/decoder support (#9372)
# Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax.
-->
- Part of https://github.com/apache/arrow-rs/issues/10541
- Closes #8748 .
- Related spec PR: https://github.com/apache/parquet-format/pull/557
# Rationale for this change
check issue
<!--
Why are you proposing this change? If this is already explained clearly
in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand
your changes and offer better suggestions for fixes.
-->
# What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here but it
is sometimes worth providing a summary of the individual changes in this
PR.
-->
# Are these changes tested?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example,
are they covered by existing tests)?
-->
# Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be
updated before approving the PR.
If there are any breaking changes to public APIs, please call them out.
-->
---------
Co-authored-by: Andrew Lamb <[email protected]>
Co-authored-by: Claude Opus 4.8 (1M context) <[email protected]>
---
parquet/src/arrow/arrow_writer/mod.rs | 2 +-
parquet/src/basic.rs | 10 +
parquet/src/bin/parquet-rewrite.rs | 4 +
parquet/src/data_type.rs | 1 +
parquet/src/encodings/alp.rs | 600 ++++++++++
parquet/src/encodings/decoding.rs | 9 +-
parquet/src/encodings/decoding/alp_decoder.rs | 1549 +++++++++++++++++++++++++
parquet/src/encodings/encoding/alp_encoder.rs | 1081 +++++++++++++++++
parquet/src/encodings/encoding/mod.rs | 114 +-
parquet/src/encodings/mod.rs | 2 +
parquet/tests/arrow_reader/parquet_testing.rs | 59 +-
11 files changed, 3407 insertions(+), 24 deletions(-)
diff --git a/parquet/src/arrow/arrow_writer/mod.rs
b/parquet/src/arrow/arrow_writer/mod.rs
index 267a52d4c9..0bb19ffa11 100644
--- a/parquet/src/arrow/arrow_writer/mod.rs
+++ b/parquet/src/arrow/arrow_writer/mod.rs
@@ -3406,7 +3406,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 4e55525c99..95babaf018 100644
--- a/parquet/src/basic.rs
+++ b/parquet/src/basic.rs
@@ -451,6 +451,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;
}
);
@@ -471,6 +475,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)),
}
}
@@ -610,6 +615,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}"),
}
}
@@ -1970,6 +1976,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]
@@ -2315,6 +2322,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();
@@ -2452,6 +2461,7 @@ mod tests {
Encoding::PLAIN_DICTIONARY,
Encoding::RLE_DICTIONARY,
Encoding::BYTE_STREAM_SPLIT,
+ Encoding::ALP,
];
encodings_roundtrip(encodings.into());
}
diff --git a/parquet/src/bin/parquet-rewrite.rs
b/parquet/src/bin/parquet-rewrite.rs
index c10eb26deb..0369d7553a 100644
--- a/parquet/src/bin/parquet-rewrite.rs
+++ b/parquet/src/bin/parquet-rewrite.rs
@@ -130,6 +130,9 @@ enum EncodingArgs {
/// Encoding for fixed-width data.
ByteStreamSplit,
+
+ /// Adaptive Lossless floating-Point encoding for FLOAT and DOUBLE.
+ Alp,
}
#[expect(deprecated)]
@@ -145,6 +148,7 @@ impl From<EncodingArgs> for Encoding {
EncodingArgs::DeltaByteArray => Self::DELTA_BYTE_ARRAY,
EncodingArgs::RleDictionary => Self::RLE_DICTIONARY,
EncodingArgs::ByteStreamSplit => Self::BYTE_STREAM_SPLIT,
+ EncodingArgs::Alp => Self::ALP,
}
}
}
diff --git a/parquet/src/data_type.rs b/parquet/src/data_type.rs
index ef8c52f41a..1f53505d52 100644
--- a/parquet/src/data_type.rs
+++ b/parquet/src/data_type.rs
@@ -718,6 +718,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 0000000000..ba7d56f31f
--- /dev/null
+++ b/parquet/src/encodings/alp.rs
@@ -0,0 +1,600 @@
+// 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:
<https://github.com/apache/parquet-format/blob/master/Encodings.md#adaptive-lossless-floating-point-alp--10>
+//!
+//! # 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<Self> {
+ 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<u8>) {
+ 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<Exact: AlpExact> {
+ pub(crate) frame_of_reference: Exact,
+ pub(crate) bit_width: u8,
+}
+
+impl<Exact: AlpExact> ForInfo<Exact> {
+ 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<u8>) {
+ 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::<u16>()
+ + num_exceptions as usize * Exact::WIDTH
+ }
+}
+
+/// Exact integer type used by FOR reconstruction: `u32` for `f32`, `u64` for
+/// `f64`. Integer-side counterpart of [`AlpFloat`], which selects its partner
+/// via [`AlpFloat::Exact`].
+///
+/// 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<u8>);
+}
+
+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<u8>) {
+ 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<u8>) {
+ 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,
+];
+
+/// Floating-point type being ALP encoded or decoded: `f32` or `f64`.
+///
+/// Each implementation pairs with an [`AlpExact`] integer of the same width
+/// ([`AlpFloat::Exact`]: `u32` for `f32`, `u64` for `f64`). `AlpFloat` owns
the
+/// float side of the codec, the decimal scaling and rounding that turn floats
+/// into exact integers and back, while [`AlpExact`] owns the integer side, the
+/// FOR and bit-packing arithmetic on those encoded values.
+pub(crate) trait AlpFloat:
+ Copy + Default + PartialEq + std::ops::Mul<Output = Self>
+{
+ 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`),
+ /// i.e. `1.5 * 2^mantissa_bits`. See [`AlpFloat::fast_round`].
+ 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: <Self::Exact as AlpExact>::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: <Self::Exact as AlpExact>::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: an add and a subtract
+ /// in plain floating-point math, cheaper than a `round()` call and free to
+ /// autovectorize. How the scaled value is rounded decides whether decoding
+ /// reproduces the input, so rounding to nearest maximizes the values that
+ /// pass the caller's round-trip check instead of becoming exceptions. This
+ /// single-form variant is the one the ALP reference implementation uses.
+ ///
+ /// Mechanics: 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. The 1.5 coefficient in
+ /// `magic = 1.5 * 2^mantissa_bits` is what keeps the sum in that binade
for
+ /// negative `x` too, without a signed fix-up. Values large enough to be
+ /// mis-rounded just fail the round-trip check and become exceptions. The
+ /// add/sub must not be simplified away: it *is* the rounding.
+ fn fast_round(self) -> <Self::Exact as AlpExact>::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) -> <Self::Exact as
AlpExact>::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 = 2147483500.0;
+ const ENCODING_LOWER_LIMIT: Self = -2147483500.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 = 9223372036854775000.0;
+ const ENCODING_LOWER_LIMIT: Self = -9223372036854775000.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 ad88ae3415..c884f3c2e0 100644
--- a/parquet/src/encodings/decoding.rs
+++ b/parquet/src/encodings/decoding.rs
@@ -26,6 +26,7 @@ use super::rle::{MAX_RLE_DICTIONARY_BIT_WIDTH, 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<Box<dyn Decoder<T>>> {
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<Box<dyn Decoder<T>>> {
match encoding {
Encoding::BYTE_STREAM_SPLIT =>
Ok(Box::new(ByteStreamSplitDecoder::new())),
+ Encoding::ALP => Ok(Box::new(AlpDecoder::new())),
_ => get_decoder_default(descr, encoding),
}
}
@@ -1298,6 +1303,8 @@ mod tests {
create_and_check_decoder::<ByteArrayType>(Encoding::DELTA_LENGTH_BYTE_ARRAY,
None);
create_and_check_decoder::<ByteArrayType>(Encoding::DELTA_BYTE_ARRAY,
None);
create_and_check_decoder::<BoolType>(Encoding::RLE, None);
+ create_and_check_decoder::<FloatType>(Encoding::ALP, None);
+ create_and_check_decoder::<DoubleType>(Encoding::ALP, None);
// error when initializing
create_and_check_decoder::<Int32Type>(
diff --git a/parquet/src/encodings/decoding/alp_decoder.rs
b/parquet/src/encodings/decoding/alp_decoder.rs
new file mode 100644
index 0000000000..0fb44ffc34
--- /dev/null
+++ b/parquet/src/encodings/decoding/alp_decoder.rs
@@ -0,0 +1,1549 @@
+// 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<Exact: AlpExact> {
+ num_elements: u16,
+ alp_info: AlpInfo,
+ for_info: ForInfo<Exact>,
+ packed_values: usize,
+ exception_positions: usize,
+ exception_values: usize,
+}
+
+impl<Exact: AlpExact> AlpEncodedVectorView<Exact> {
+ fn expected_stored_size(&self) -> usize {
+ AlpInfo::STORED_SIZE
+ + ForInfo::<Exact>::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<usize> {
+ 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<usize> {
+ let len = self.alp_info.num_exceptions as usize *
std::mem::size_of::<u16>();
+ 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<usize> {
+ 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<AlpHeader> {
+ 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<usize> {
+ let start = idx * std::mem::size_of::<u32>();
+ let bytes = body
+ .get(start..start + std::mem::size_of::<u32>())
+ .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<Exact: AlpExact>(
+ body: &[u8],
+ vector_start: usize,
+ vector_end: usize,
+ num_elements: u16,
+) -> Result<AlpEncodedVectorView<Exact>> {
+ let vector_bytes = &body[vector_start..vector_end];
+
+ let metadata_size = AlpInfo::STORED_SIZE + ForInfo::<Exact>::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::<Exact> {
+ 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::<u16>();
+
+ // 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<Value: AlpFloat> {
+ 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<Value: AlpFloat>(
+ cur: &mut CurrentVector<Value>,
+ 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::<Value::Exact>(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();
+ // Both buffers were sliced to exactly `num_exceptions` whole entries at
+ // parse time. A mismatch here would make `zip` silently drop exceptions.
+ debug_assert_eq!(
+ cur.exception_positions.len() % std::mem::size_of::<u16>(),
+ 0
+ );
+ debug_assert_eq!(cur.exception_values.len() % Value::Exact::WIDTH, 0);
+ debug_assert_eq!(
+ cur.exception_positions.len() / std::mem::size_of::<u16>(),
+ cur.exception_values.len() / Value::Exact::WIDTH,
+ );
+ for (pos_chunk, value_chunk) in cur
+ .exception_positions
+ .chunks_exact(std::mem::size_of::<u16>())
+ .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<T: DataType>
+where
+ T::T: AlpFloat,
+ <T::T as 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<CurrentVector<T::T>>,
+ /// 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<<T::T as AlpFloat>::Exact>,
+ /// Number of values still to be read from the page.
+ num_values: usize,
+}
+
+impl<T: DataType> AlpDecoder<T>
+where
+ T::T: AlpFloat,
+ <T::T as 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::<<T::T as AlpFloat>::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: <T::T as AlpFloat>::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<T: DataType> Decoder<T> for AlpDecoder<T>
+where
+ T::T: AlpFloat,
+ <T::T as 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::<u32>())
+ .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<usize> {
+ 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::<T::T>(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<usize> {
+ 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<u8> {
+ 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<u8>);
+ }
+
+ impl AppendLeBytes for u32 {
+ fn append_le_bytes(self, out: &mut Vec<u8>) {
+ out.extend_from_slice(&self.to_le_bytes());
+ }
+ }
+
+ impl AppendLeBytes for u64 {
+ fn append_le_bytes(self, out: &mut Vec<u8>) {
+ 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<Exact: AppendLeBytes + Copy>(spec: VectorSpec<'_, Exact>)
-> Vec<u8> {
+ 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<T: DataType>(page: Vec<u8>, num_values: usize) -> Vec<T::T>
+ where
+ T::T: AlpFloat,
+ <T::T as AlpFloat>::Exact: Send,
+ {
+ let mut decoder = AlpDecoder::<T>::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<T: DataType>(page: Vec<u8>, num_values: usize) ->
ParquetError
+ where
+ T::T: AlpFloat,
+ <T::T as AlpFloat>::Exact: Send,
+ {
+ let mut decoder = AlpDecoder::<T>::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<u8>],
+ ) -> Vec<u8> {
+ 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::<u32>();
+ 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::<DoubleType>(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::<DoubleType>(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::<DoubleType>(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::<DoubleType>(data, 1);
+ assert!(
+ err.to_string()
+ .contains("Invalid ALP page: unsupported integer encoding 1")
+ );
+ }
+
+ #[test]
+ fn test_set_data_rejects_compression_mode() {
+ let data = make_alp_page_bytes(1, 0, 3, 1, &[4], 8);
+ let err = decode_err::<DoubleType>(data, 1);
+ assert!(
+ err.to_string()
+ .contains("Invalid ALP page: unsupported compression mode 1")
+ );
+ }
+
+ #[test]
+ fn test_set_data_rejects_num_elements_exceeding_num_values() {
+ let data = make_alp_page_bytes(0, 0, 3, 5, &[4], 8);
+ let err = decode_err::<FloatType>(data, 2);
+ assert!(
+ err.to_string()
+ .contains("Invalid ALP page: header num_elements 5 exceeds
page num_values 2")
+ );
+ }
+
+ #[test]
+ fn test_set_data_rejects_short_offsets_section() {
+ // 9 elements at vector size 8 means two vectors, so 8 bytes of
offsets.
+ let data = make_alp_page_bytes(0, 0, 3, 9, &[4], 0);
+ let err = decode_err::<FloatType>(data, 9);
+ assert!(
+ err.to_string()
+ .contains("Invalid ALP page: expected at least 8 bytes for 2
offsets, got 4")
+ );
+ }
+
+ #[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::<FloatType>(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::<FloatType>(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::<FloatType>(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::<FloatType>(page, 1);
+ assert!(
+ err.to_string()
+ .contains("Invalid ALP page: num_exceptions 2 exceeds vector
num_elements 1")
+ );
+ }
+
+ #[test]
+ fn test_decode_rejects_short_vector_metadata() {
+ // Three bytes cannot hold the 9-byte f32 vector header.
+ let page = make_page_from_vectors(3, 1, &[vec![0u8; 3]]);
+ let err = decode_err::<FloatType>(page, 1);
+ assert!(err.to_string().contains(
+ "Invalid ALP page: vector metadata too short, expected at least 9
bytes, got 3"
+ ));
+ }
+
+ #[test]
+ fn test_decode_rejects_bit_width_overflow() {
+ let vector = make_vector(VectorSpec {
+ exponent: 0,
+ factor: 0,
+ frame_of_reference: 0u32,
+ bit_width: 33,
+ packed_values: &[],
+ exception_positions: &[],
+ exception_values: &[],
+ });
+ let page = make_page_from_vectors(3, 1, &[vector]);
+ let err = decode_err::<FloatType>(page, 1);
+ assert!(
+ err.to_string()
+ .contains("Invalid ALP page: bit width 33 exceeds 32")
+ );
+ }
+
+ #[test]
+ fn test_decode_rejects_out_of_bounds_vector_offset() {
+ let vector0 = make_vector(VectorSpec {
+ exponent: 0,
+ factor: 0,
+ frame_of_reference: 0u32,
+ bit_width: 1,
+ packed_values: &[0],
+ exception_positions: &[],
+ exception_values: &[],
+ });
+ // The second vector's offset points far past the end of the body.
+ let mut page = make_alp_page_bytes(0, 0, 3, 12, &[8, 200], 0);
+ page.extend_from_slice(&vector0);
+ let err = decode_err::<FloatType>(page, 12);
+ assert!(
+ err.to_string()
+ .contains("Invalid ALP page: vector offset 200 out of bounds
at index 0")
+ );
+ }
+
+ #[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::<FloatType>(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::<DoubleType>(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::<FloatType>(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::<FloatType>(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::<FloatType>(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::<u64>(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<u16> = body[view.exception_positions_range()]
+ .chunks_exact(2)
+ .map(|c| u16::from_le_bytes([c[0], c[1]]))
+ .collect();
+ let values: Vec<u64> = 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::<FloatType>(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::<DoubleType>(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::<FloatType>(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::<FloatType>(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::<DoubleType>(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::<FloatType>::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::<FloatType>::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);
+ }
+
+ /// A skip absorbed entirely by the vector already being decoded. `skip`
has
+ /// two halves: an offset-array jump when the request leaves the open
vector
+ /// (covered by `test_alp_decoder_skip_across_vectors`) and an in-place
+ /// advance of the open vector's live bit-reader state. This exercises the
+ /// in-place half alone.
+ #[test]
+ fn test_alp_decoder_skip_within_current_vector() {
+ 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::<FloatType>::new();
+ decoder.set_data(Bytes::from(page), 12).unwrap();
+
+ // Reading two values opens vector 0 and leaves it mid-flight with six
+ // values remaining, so the skip below must advance it in place.
+ let mut head = [0.0f32; 2];
+ assert_eq!(decoder.get(&mut head).unwrap(), 2);
+ assert_eq!(head, [10.0, 11.0]);
+
+ // Positions 2..5 fit inside those six, so the skip returns early
+ // without consulting the offset array.
+ assert_eq!(decoder.skip(3).unwrap(), 3);
+ assert_eq!(decoder.values_left(), 7);
+
+ // Resuming mid-vector and reading across the boundary proves the bit
+ // reader advanced exactly three values: any other count would shift
+ // the delta-1 values (11.0, 21.0) into the wrong slots.
+ let mut tail = [0.0f32; 4];
+ assert_eq!(decoder.get(&mut tail).unwrap(), 4);
+ assert_eq!(tail, [10.0, 10.0, 10.0, 20.0]);
+ }
+
+ #[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::<FloatType>::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<u8> {
+ 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<f64> = Vec::new();
+ for s in &specs {
+ let modulo = 1u64 << s.bit_width;
+ let deltas: Vec<u64> = (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::<DoubleType>(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::<DoubleType>::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::<DoubleType>::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::<DoubleType>::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 0000000000..aa8682d981
--- /dev/null
+++ b/parquet/src/encodings/encoding/alp_encoder.rs
@@ -0,0 +1,1081 @@
+// 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:
<https://github.com/apache/parquet-format/blob/master/Encodings.md#adaptive-lossless-floating-point-alp--10>
+//!
+//! 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<F: AlpFloat>() -> 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<u64>, 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<F: AlpFloat>(
+ sample: &[F],
+ params: ExponentAndFactor,
+ penalize_exceptions: bool,
+) -> Option<u64> {
+ 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: <F::Exact as AlpExact>::Signed|
m.min(encoded)));
+ max = Some(max.map_or(encoded, |m: <F::Exact as AlpExact>::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::<F>());
+ };
+
+ 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::<F>())
+}
+
+/// Sample every `n`th value so the whole span is represented, capped at
+/// [`SAMPLES_PER_VECTOR`] values.
+fn sample_values<F: AlpFloat>(values: &[F], out: &mut Vec<F>) {
+ 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<F: AlpFloat>(values: &[F]) -> Vec<ExponentAndFactor> {
+ 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<Combination> = 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>() + 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 &mut tally {
+ 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<F: AlpFloat>(
+ vector: &[F],
+ preset: &[ExponentAndFactor],
+ sample: &mut Vec<F>,
+) -> 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<F: AlpFloat> {
+ /// Decimal-encoded integers, overwritten in place with their FOR deltas.
+ encoded: Vec<<F::Exact as AlpExact>::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<u8>,
+ exception_positions: Vec<u16>,
+ exception_values: Vec<F>,
+ sample: Vec<F>,
+}
+
+impl<F: AlpFloat> Scratch<F> {
+ 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::<<F::Exact as
AlpExact>::Signed>()
+ + self.exc_mask.capacity()
+ + self.exception_positions.capacity() * std::mem::size_of::<u16>()
+ + self.exception_values.capacity() * std::mem::size_of::<F>()
+ + self.sample.capacity() * std::mem::size_of::<F>()
+ }
+}
+
+/// Encode one vector and append it to `out`:
+/// `[AlpInfo][ForInfo][PackedValues][ExceptionPositions][ExceptionValues]`.
+fn encode_vector<F: AlpFloat>(
+ values: &[F],
+ params: ExponentAndFactor,
+ scratch: &mut Scratch<F>,
+ out: &mut Vec<u8>,
+) -> 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::<F>(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::<F::Exact> {
+ 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<F: AlpFloat>(
+ encoded: &[<F::Exact as AlpExact>::Signed],
+ exception_positions: &[u16],
+) -> <F::Exact as AlpExact>::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<F: AlpFloat>(
+ values: &[F],
+ preset: &[ExponentAndFactor],
+ scratch: &mut Scratch<F>,
+) -> Result<Vec<u8>> {
+ 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::<u32>(), 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::<u32>();
+ 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<F: AlpFloat> {
+ /// Encoded vectors, concatenated; becomes the page body after the offsets.
+ body: Vec<u8>,
+ /// Body-relative start offset of each completed vector.
+ vector_offsets: Vec<u32>,
+ /// Values not yet forming a complete vector, carried across `put`s.
+ carry: Vec<F>,
+ /// Total values appended to this page so far.
+ count: usize,
+}
+
+impl<F: AlpFloat> StreamingPage<F> {
+ 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::<u32>()
+ + self.carry.capacity() * std::mem::size_of::<F>()
+ }
+
+ /// Encode one complete vector and append it to `body`.
+ fn push_vector(
+ &mut self,
+ vector: &[F],
+ preset: &[ExponentAndFactor],
+ scratch: &mut Scratch<F>,
+ ) -> 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<F>,
+ ) -> 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<F>,
+ ) -> Result<Vec<u8>> {
+ 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::<u32>();
+ 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<T: DataType>
+where
+ T::T: AlpFloat,
+{
+ /// Values buffered for the first page, until the preset is built. Empty
once
+ /// streaming begins.
+ values: Vec<T::T>,
+ /// 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<Vec<ExponentAndFactor>>,
+ scratch: Scratch<T::T>,
+ /// The page built incrementally, used for every page after the first.
+ streaming: StreamingPage<T::T>,
+}
+
+impl<T: DataType> AlpEncoder<T>
+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<T: DataType> Encoder<T> for AlpEncoder<T>
+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. In the
+ // worst case, almost every value is an exception while two encodable
+ // values force a full-width packed FOR range. Bound both the packed
+ // placeholder/value and the raw exception value by the exact type's
+ // width, plus the exception position. 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::<u32>()
+ + AlpInfo::STORED_SIZE
+ + ForInfo::<<T::T as AlpFloat>::Exact>::stored_size())
+ + len * (2 * <T::T as AlpFloat>::Exact::WIDTH +
std::mem::size_of::<u16>())
+ }
+
+ fn estimated_memory_size(&self) -> usize {
+ self.values.capacity() * std::mem::size_of::<T::T>()
+ + self.preset.as_ref().map_or(0, |p| {
+ p.capacity() * std::mem::size_of::<ExponentAndFactor>()
+ })
+ + self.scratch.estimated_memory_size()
+ + self.streaming.estimated_memory_size()
+ }
+
+ fn flush_buffer(&mut self) -> Result<Bytes> {
+ 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<T: DataType>(values: &[T::T]) -> Vec<T::T>
+ where
+ T::T: AlpFloat,
+ <T::T as AlpFloat>::Exact: Send,
+ {
+ let mut encoder = AlpEncoder::<T>::new();
+ encoder.put(values).unwrap();
+ let page = encoder.flush_buffer().unwrap();
+
+ let mut decoder = AlpDecoder::<T>::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<F: AlpFloat + std::fmt::Debug>(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<f64> = (0..500).map(|i| (i as f64) * 0.01 +
1.23).collect();
+ assert_bits_eq(&roundtrip::<DoubleType>(&values), &values);
+ }
+
+ #[test]
+ fn test_roundtrip_f32_decimals() {
+ let values: Vec<f32> = (0..500).map(|i| (i as f32) * 0.5 +
1.25).collect();
+ assert_bits_eq(&roundtrip::<FloatType>(&values), &values);
+ }
+
+ /// Spans several vectors, including a short trailing one.
+ #[test]
+ fn test_roundtrip_multiple_vectors() {
+ let values: Vec<f64> = (0..2600).map(|i| (i as f64) * 0.001).collect();
+ assert_bits_eq(&roundtrip::<DoubleType>(&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::<DoubleType>(&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::<DoubleType>(&values), &values);
+ }
+
+ #[test]
+ fn test_roundtrip_all_exceptions() {
+ let values = vec![f64::NAN; 100];
+ let decoded = roundtrip::<DoubleType>(&values);
+ assert!(decoded.iter().all(|v| v.is_nan()));
+ }
+
+ #[test]
+ fn test_roundtrip_single_value() {
+ assert_bits_eq(&roundtrip::<DoubleType>(&[3.25]), &[3.25]);
+ }
+
+ #[test]
+ fn test_roundtrip_empty() {
+ let mut encoder = AlpEncoder::<DoubleType>::new();
+ let page = encoder.flush_buffer().unwrap();
+ assert_eq!(page.len(), ALP_HEADER_SIZE);
+
+ let mut decoder = AlpDecoder::<DoubleType>::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::<DoubleType>::new();
+
+ for page_idx in 0..3 {
+ let values: Vec<f64> = (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::<DoubleType>::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<f64> = (0..1024).map(|i| (i as f64) * 0.01).collect();
+
+ let mut encoder = AlpEncoder::<DoubleType>::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::<f64>();
+ 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<f64> = (0..1024).map(|i| (i as f64) *
0.01).collect();
+ values[500] = f64::NAN;
+
+ let mut encoder = AlpEncoder::<DoubleType>::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::<DoubleType>(&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<f64> = (0..3000).map(|i| (i as f64) * 0.01 +
1.23).collect();
+ let page2: Vec<f64> = (0..3000).map(|i| (i as f64) * 0.03 -
7.0).collect();
+
+ let mut encoder = AlpEncoder::<DoubleType>::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::<f64>::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<f64> = (0..2048).map(|i| (i as f64) * 0.01).collect();
+ let page2: Vec<f64> = (0..5000).map(|i| (i as f64) * 0.01 +
100.0).collect();
+
+ let mut encoder = AlpEncoder::<DoubleType>::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::<DoubleType>::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);
+ }
+
+ /// A streaming page ending exactly on a vector boundary must finish with
an
+ /// empty carry buffer.
+ #[test]
+ fn test_streaming_page_ends_on_vector_boundary() {
+ // Flushing a throwaway first page builds the preset and switches the
+ // encoder to streaming. Only the second page is under test.
+ let preset_page: Vec<f64> = (0..100).map(|i| (i as f64) *
0.01).collect();
+ let values: Vec<f64> = (0..2 * VECTOR_SIZE).map(|i| (i as f64) *
0.01).collect();
+
+ let mut encoder = AlpEncoder::<DoubleType>::new();
+ encoder.put(&preset_page).unwrap();
+ let _ = encoder.flush_buffer().unwrap();
+
+ let empty_page_estimate = encoder.estimated_data_encoded_size();
+ encoder.put(&values).unwrap();
+ // The size estimate for a streamed page comes from the running count.
+ assert!(encoder.estimated_data_encoded_size() > empty_page_estimate);
+ // The 2048 values streamed through as two complete vectors, so this
+ // flush reaches `StreamingPage::finish` with an empty carry: the page
+ // must assemble correctly without the trailing-partial-vector encode
+ // that every other streaming test ends with.
+ let page = encoder.flush_buffer().unwrap();
+
+ let mut decoder = AlpDecoder::<DoubleType>::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);
+ }
+
+ /// The memory estimate must track the buffers in both encoder states: the
+ /// buffered first page and the streaming state after it.
+ #[test]
+ fn test_estimated_memory_size_tracks_buffers() {
+ let mut encoder = AlpEncoder::<DoubleType>::new();
+ // A fresh encoder has allocated nothing.
+ assert_eq!(encoder.estimated_memory_size(), 0);
+
+ let values: Vec<f64> = (0..1500).map(|i| (i as f64) * 0.01).collect();
+ // First-page state: only the raw-value buffer is live (scratch and
+ // streaming are untouched), so the estimate must cover its bytes.
+ encoder.put(&values).unwrap();
+ assert!(encoder.estimated_memory_size() >= values.len() *
std::mem::size_of::<f64>());
+
+ // The flushed first-page buffer and encoding scratch space keep their
+ // capacities, and the preset is retained. Capture that full baseline.
+ let _ = encoder.flush_buffer().unwrap();
+ let after_flush = encoder.estimated_memory_size();
+
+ // Streaming state: the streaming-page buffers now hold encoded
vectors,
+ // which must be counted on top of that baseline.
+ encoder.put(&values).unwrap();
+ assert!(encoder.estimated_memory_size() > after_flush);
+ }
+
+ /// The size estimate must include packed placeholders as well as raw
+ /// exception values when both sections coexist at their widest.
+ #[test]
+ fn test_estimated_data_size_covers_wide_packed_values_with_exceptions() {
+ let mut values = vec![f64::NAN; VECTOR_SIZE];
+ // Both positions are sampled when building the preset. At exponent and
+ // factor zero their integer range is exactly 2^63, forcing a 64-bit
+ // packed section; every other value remains an exception.
+ values[0] = f64::ENCODING_LOWER_LIMIT;
+ values[4] = 1024.0;
+
+ let mut encoder = AlpEncoder::<DoubleType>::new();
+ encoder.put(&values).unwrap();
+ let estimated_size = encoder.estimated_data_encoded_size();
+ let page = encoder.flush_buffer().unwrap();
+
+ let vector_start = ALP_HEADER_SIZE + std::mem::size_of::<u32>();
+ let num_exceptions = u16::from_le_bytes([page[vector_start + 2],
page[vector_start + 3]]);
+ let bit_width = page[vector_start + AlpInfo::STORED_SIZE + <f64 as
AlpFloat>::Exact::WIDTH];
+ assert_eq!(num_exceptions as usize, VECTOR_SIZE - 2);
+ assert_eq!(bit_width, 64);
+ assert!(
+ estimated_size >= page.len(),
+ "estimated {estimated_size} bytes, encoded {} bytes",
+ page.len()
+ );
+ }
+
+ /// Fill the preset to `MAX_COMBINATIONS` and include a vector no candidate
+ /// can encode, exercising the all-exception size estimate and the early
+ /// exit of the per-vector candidate search.
+ #[test]
+ fn test_full_preset_with_all_exception_vector() {
+ // Nine vectors are sampled at stride two, so the even-indexed ones
+ // decide the preset: four distinct decimal scales, plus the all-NaN
+ // ninth vector, which no candidate can encode and which therefore
+ // contributes the worst-case fallback pair as the fifth candidate.
+ let scales = [0.01, 0.01, 0.001, 0.01, 0.0001, 0.01, 0.00001, 0.01];
+ let mut values = Vec::with_capacity(9 * VECTOR_SIZE);
+ for scale in scales {
+ values.extend((0..VECTOR_SIZE).map(|i| (i as f64) * scale));
+ }
+ values.extend(std::iter::repeat_n(f64::NAN, VECTOR_SIZE));
+
+ let preset = build_preset(&values);
+ assert_eq!(preset.len(), MAX_COMBINATIONS);
+
+ let nan_vector = &values[8 * VECTOR_SIZE..];
+ let mut sample = Vec::new();
+ sample_values(nan_vector, &mut sample);
+ let candidate_costs: Vec<_> = preset
+ .iter()
+ .map(|¶ms| estimate_size_bits(&sample, params, false))
+ .collect();
+ assert!(candidate_costs.windows(2).all(|costs| costs[0] == costs[1]));
+
+ // Encoding the NaN vector prices all five candidates at the same
+ // all-exception size, so the per-vector search stops early after
+ // `SAMPLING_EARLY_EXIT_THRESHOLD` non-improving candidates. The
+ // round-trip proves the page survives both paths losslessly.
+ assert_bits_eq(&roundtrip::<DoubleType>(&values), &values);
+ }
+}
diff --git a/parquet/src/encodings/encoding/mod.rs
b/parquet/src/encodings/encoding/mod.rs
index f6235696dc..09ad15fd01 100644
--- a/parquet/src/encodings/encoding/mod.rs
+++ b/parquet/src/encodings/encoding/mod.rs
@@ -28,10 +28,12 @@ use crate::schema::types::ColumnDescPtr;
use crate::util::bit_util::{BitWriter, num_required_bits};
use crate::util::prefix::common_prefix_length;
+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;
@@ -85,27 +87,92 @@ pub fn get_encoder<T: DataType>(
encoding: Encoding,
descr: &ColumnDescPtr,
) -> Result<Box<dyn Encoder<T>>> {
- let encoder: Box<dyn Encoder<T>> = 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"
- ));
+ <T::T as private::GetEncoder>::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<T: DataType<T = Self>>(
+ descr: &ColumnDescPtr,
+ encoding: Encoding,
+ ) -> Result<Box<dyn Encoder<T>>> {
+ 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()),
- },
- #[expect(deprecated, reason = "BIT_PACKED is the encoding we reject
here")]
- e @ Encoding::BIT_PACKED => return Err(nyi_err!("Encoding {} is not
supported", e)),
- };
- Ok(encoder)
+ }
+
+ fn get_encoder_default<T: DataType>(
+ descr: &ColumnDescPtr,
+ encoding: Encoding,
+ ) -> Result<Box<dyn Encoder<T>>> {
+ let encoder: Box<dyn Encoder<T>> = 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 {} only supports FLOAT and DOUBLE, got {}",
+ encoding,
+ T::get_physical_type()
+ ));
+ }
+ #[expect(deprecated, reason = "BIT_PACKED is the encoding we
reject here")]
+ e @ Encoding::BIT_PACKED => 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<T: DataType<T = Self>>(
+ descr: &ColumnDescPtr,
+ encoding: Encoding,
+ ) -> Result<Box<dyn Encoder<T>>> {
+ match encoding {
+ Encoding::ALP => Ok(Box::new(AlpEncoder::new())),
+ _ => get_encoder_default(descr, encoding),
+ }
+ }
+ }
+
+ impl GetEncoder for f64 {
+ fn get_encoder<T: DataType<T = Self>>(
+ descr: &ColumnDescPtr,
+ encoding: Encoding,
+ ) -> Result<Box<dyn Encoder<T>>> {
+ match encoding {
+ Encoding::ALP => Ok(Box::new(AlpEncoder::new())),
+ _ => get_encoder_default(descr, encoding),
+ }
+ }
+ }
}
// ----------------------------------------------------------------------
@@ -794,6 +861,13 @@ mod tests {
"Cannot initialize this encoding through this function"
)),
);
+ create_and_check_encoder::<Int32Type>(
+ 0,
+ Encoding::ALP,
+ Some(general_err!(
+ "Encoding ALP only supports FLOAT and DOUBLE, got INT32"
+ )),
+ );
// unsupported
#[expect(deprecated)]
diff --git a/parquet/src/encodings/mod.rs b/parquet/src/encodings/mod.rs
index 894c4fb961..cb23625a82 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/parquet_testing.rs
b/parquet/tests/arrow_reader/parquet_testing.rs
index cbc7aa3eae..ef92e6b70f 100644
--- a/parquet/tests/arrow_reader/parquet_testing.rs
+++ b/parquet/tests/arrow_reader/parquet_testing.rs
@@ -19,14 +19,69 @@
//!
//! [parquet-testing]: https://github.com/apache/parquet-testing
+use arrow::util::test_util::parquet_test_data;
use arrow_array::cast::AsArray;
-use arrow_array::{Array, BinaryArray, Int64Array, StringArray, types};
-use arrow_schema::{Field, Schema, TimeUnit};
+use arrow_array::{Array, ArrayRef, BinaryArray, Int64Array, RecordBatch,
StringArray, types};
+use arrow_schema::{ArrowError, Field, Schema, TimeUnit};
use parquet::arrow::arrow_reader::{ArrowReaderOptions,
ParquetRecordBatchReaderBuilder};
use parquet::basic::{LogicalType, Type as PhysicalType};
use std::fs::File;
+use std::path::PathBuf;
use std::sync::Arc;
+/// The ALP test data file has 8 columns, each containing the same 9032 values.
+///
+/// The float_plain and double_plain columns are encoded with `PLAIN` + zstd
+/// compression, and serve as a reference for the other columns which are
+/// encoded with `ALP` encoding.
+///
+/// Ensure the values in the other column come back the same as the reference
columns
+///
+/// | Column | Encoding |
Rationale / coverage
|
+///
|-------------------------------------|---------------------------|-----------------------------------------------------------------------------------|
+/// | `float_plain`, `double_plain` | `PLAIN` + zstd |
In-file reference: readers can bit-compare the ALP columns against these
|
+/// | `float_alp_1024`, `double_alp_1024` | `ALP`, 1024-value vectors | The
default vector size of 1024 values |
+/// | `float_alp_4096`, `double_alp_4096` | `ALP`, 4096-value vectors |
Readers must honor `log_vector_size` from the page header rather than assume
1024 |
+/// | `float_alp_32`, `double_alp_32` | `ALP`, 32-value vectors | Many
vectors per page, stresses the per-vector metadata loop |
+#[test]
+#[cfg_attr(miri, ignore)] // Zstd calls native C functions unsupported by Miri
+fn test_alp_extended() {
+ let alp_extended =
PathBuf::from(parquet_test_data()).join("alp_extended.zstd.parquet");
+ let file = File::open(alp_extended).unwrap();
+ let reader = ParquetRecordBatchReaderBuilder::try_new(file)
+ .unwrap()
+ .build()
+ .unwrap();
+
+ let batches: Vec<_> = reader.into_iter().collect::<Result<Vec<_>,
_>>().unwrap();
+ let total_rows = batches.iter().map(|batch|
batch.num_rows()).sum::<usize>();
+ assert_eq!(total_rows, 9032);
+ batches
+ .iter()
+ .for_each(|batch| assert_eq!(batch.num_columns(), 8));
+
+ // compare float values to the reference
+ let float_plain = column(&batches, "float_plain").unwrap();
+ assert_eq!(float_plain, column(&batches, "float_alp_1024").unwrap());
+ assert_eq!(float_plain, column(&batches, "float_alp_4096").unwrap());
+ assert_eq!(float_plain, column(&batches, "float_alp_32").unwrap());
+
+ // compare double values to the reference
+ let double_plain = column(&batches, "double_plain").unwrap();
+ assert_eq!(double_plain, column(&batches, "double_alp_1024").unwrap());
+ assert_eq!(double_plain, column(&batches, "double_alp_4096").unwrap());
+ assert_eq!(double_plain, column(&batches, "double_alp_32").unwrap());
+}
+
+fn column(batches: &[RecordBatch], column_name: &str) -> Result<Vec<ArrayRef>,
ArrowError> {
+ let mut columns = Vec::new();
+ for batch in batches {
+ let array = batch.column(batch.schema().index_of(column_name)?);
+ columns.push(array.clone());
+ }
+ Ok(columns)
+}
+
#[test]
fn test_int96_from_spark_file_with_provided_schema() {
// int96_from_spark.parquet was written based on Spark's microsecond
timestamps which trade
