prtkgaur commented on code in PR #48345: URL: https://github.com/apache/arrow/pull/48345#discussion_r3920940990
########## cpp/src/arrow/util/alp/alp.h: ########## @@ -0,0 +1,795 @@ +// 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. + +// Adaptive Lossless floating-Point (ALP) compression implementation + +#pragma once + +#include <optional> +#include <vector> + +#include "arrow/result.h" +#include "arrow/status.h" +#include "arrow/util/alp/alp_constants.h" +#include "arrow/util/span.h" + +namespace arrow { +namespace util { +namespace alp { + +// ---------------------------------------------------------------------- +// ALP Overview +// +// IMPORTANT: For abstract interfaces or examples how to use ALP, consult +// alp_codec.h. +// This is our implementation of the adaptive lossless floating-point +// compression for decimals (ALP) (https://dl.acm.org/doi/10.1145/3626717). +// It works by converting a float into a decimal (if possible). The exponent +// and factor are chosen per vector. Each float is converted using +// c(f) = int64(f * 10^exponent * 10^-factor). The converted floats are then +// encoded via a delta frame of reference and bitpacked. Every exception, +// where the conversion/reconversion changes the value of the float, is stored +// separately and has to be patched into the decompressed vector afterwards. +// +// ========================================================================== +// ALP COMPRESSION/DECOMPRESSION PIPELINE +// ========================================================================== +// +// COMPRESSION FLOW: +// ----------------- +// +// Input: float/double array +// | +// v +// +------------------------------------------------------------------+ +// | 1. SAMPLING & PRESET GENERATION | +// | * Sample vectors from dataset | +// | * Try all exponent/factor combinations (e, f) | +// | * Select best k combinations for preset | +// +------------------------------------+-----------------------------+ +// | preset.combinations +// v +// +------------------------------------------------------------------+ +// | 2. PER-VECTOR COMPRESSION | +// | a) Find best (e,f) from preset for this vector | +// | b) Encode: encoded[i] = int64(value[i] * 10^e * 10^-f) | +// | c) Verify: if decode(encoded[i]) != value[i] -> exception | +// | d) Replace exceptions with placeholder value | +// +------------------------------------+-----------------------------+ +// | encoded integers + exceptions +// v +// +------------------------------------------------------------------+ +// | 3. FRAME OF REFERENCE (FOR) | +// | * Find min value in encoded integers | +// | * Subtract min from all values: delta[i] = encoded[i] - min | +// +------------------------------------+-----------------------------+ +// | delta values (smaller range) +// v +// +------------------------------------------------------------------+ +// | 4. BIT PACKING | +// | * Calculate bit_width = log2(max_delta) | +// | * Pack each value into bit_width bits | +// | * Result: tightly packed binary data | +// +------------------------------------+-----------------------------+ +// | packed bytes +// v +// +------------------------------------------------------------------+ +// | 5. SERIALIZATION (offset-based interleaved layout) | +// | [Header][Offsets...][Vector₀][Vector₁]... | +// | where each Vector = [AlpInfo|ForInfo|Data] | +// +------------------------------------------------------------------+ +// +// +// DECOMPRESSION FLOW: +// ------------------- +// +// Serialized bytes -> AlpEncodedVector::Load() +// | +// v +// +------------------------------------------------------------------+ +// | 1. BIT UNPACKING | +// | * Extract bit_width from metadata | +// | * Unpack each value from bit_width bits -> delta values | +// +------------------------------------+-----------------------------+ +// | delta values +// v +// +------------------------------------------------------------------+ +// | 2. REVERSE FRAME OF REFERENCE (unFOR) | +// | * Add back min: encoded[i] = delta[i] + frame_of_reference | +// +------------------------------------+-----------------------------+ +// | encoded integers +// v +// +------------------------------------------------------------------+ +// | 3. DECODE | +// | * Apply inverse formula: value[i] = encoded[i] * 10^-e * 10^f | +// +------------------------------------+-----------------------------+ +// | decoded floats (with placeholders) +// v +// +------------------------------------------------------------------+ +// | 4. PATCH EXCEPTIONS | +// | * Replace values at exception_positions[] with exceptions[] | +// +------------------------------------+-----------------------------+ +// | +// v +// Output: Original float/double array (lossless!) +// +// ========================================================================== + +// ---------------------------------------------------------------------- +// AlpMode + +/// \brief ALP compression mode +/// +/// Currently only ALP (decimal compression) is implemented. +enum class AlpMode { kAlp = 0 }; + +// ---------------------------------------------------------------------- +// AlpExponentAndFactor + +/// \brief Helper struct to encapsulate the exponent and factor +struct AlpExponentAndFactor { + uint8_t exponent{0}; + uint8_t factor{0}; + + bool operator==(const AlpExponentAndFactor& other) const { + return exponent == other.exponent && factor == other.factor; + } + + /// \brief Comparison operator for deterministic std::map ordering + bool operator<(const AlpExponentAndFactor& other) const { + if (exponent != other.exponent) return exponent < other.exponent; + return factor < other.factor; + } +}; + +// ---------------------------------------------------------------------- +// AlpEncodedVectorInfo (non-templated, ALP core metadata) + +/// \brief ALP-specific metadata for an encoded vector (non-templated) +/// +/// Contains the metadata specific to ALP's float-to-integer conversion: +/// - exponent/factor: parameters for decimal encoding +/// - num_exceptions: count of values that couldn't be losslessly encoded +/// +/// This struct is the same size regardless of the floating-point type (float/double). +/// It is separate from the integer encoding metadata (e.g., FOR) to allow +/// different integer encodings to be used in the future. +/// +/// Serialization format (4 bytes): +/// +/// +------------------------------------------+ +/// | AlpEncodedVectorInfo (4 bytes) | +/// +------------------------------------------+ +/// | Offset | Field | Size | +/// +---------+---------------------+----------+ +/// | 0 | exponent (uint8_t) | 1 byte | +/// | 1 | factor (uint8_t) | 1 byte | +/// | 2 | num_exceptions | 2 bytes | +/// +------------------------------------------+ +class AlpEncodedVectorInfo { + public: + AlpEncodedVectorInfo() = default; + AlpEncodedVectorInfo(uint8_t exponent, uint8_t factor, int16_t num_exceptions) + : exponent_(exponent), factor_(factor), num_exceptions_(num_exceptions) {} + + uint8_t exponent() const { return exponent_; } + uint8_t factor() const { return factor_; } + int16_t num_exceptions() const { return num_exceptions_; } + + void set_exponent(uint8_t exponent) { exponent_ = exponent; } + void set_factor(uint8_t factor) { factor_ = factor; } + void set_num_exceptions(int16_t num_exceptions) { num_exceptions_ = num_exceptions; } + + /// Size of the serialized portion (4 bytes, fixed) + static constexpr int64_t kStoredSize = + sizeof(uint8_t) + sizeof(uint8_t) + sizeof(int16_t); + static_assert(kStoredSize == 4, "AlpEncodedVectorInfo stored size must be 4 bytes"); + + /// \brief Store the ALP metadata into an output buffer + /// + /// \pre output_buffer.size() >= kStoredSize + void Store(arrow::util::span<uint8_t> output_buffer) const; + + /// \brief Load ALP metadata from an input buffer + /// + /// \return the loaded metadata, or Status::Invalid if the buffer is too small + static Result<AlpEncodedVectorInfo> Load(arrow::util::span<const uint8_t> input_buffer); + + /// \brief Get serialized size of the ALP metadata + static int64_t GetStoredSize() { return kStoredSize; } + + /// \brief Get exponent and factor as a combined struct + AlpExponentAndFactor GetExponentAndFactor() const { + return AlpExponentAndFactor{exponent_, factor_}; + } + + bool operator==(const AlpEncodedVectorInfo& other) const { + return exponent_ == other.exponent_ && factor_ == other.factor_ && + num_exceptions_ == other.num_exceptions_; + } + + bool operator!=(const AlpEncodedVectorInfo& other) const { return !(*this == other); } + + private: + uint8_t exponent_ = 0; + uint8_t factor_ = 0; + int16_t num_exceptions_ = 0; +}; + +// ---------------------------------------------------------------------- +// AlpEncodedForVectorInfo (templated, FOR integer encoding metadata) + +/// \brief FOR (Frame of Reference) encoding metadata for an encoded vector +/// +/// Contains the metadata specific to FOR bit-packing integer encoding: +/// - frame_of_reference: minimum value subtracted from all encoded integers +/// - bit_width: number of bits used to pack each delta value +/// +/// This struct is templated because frame_of_reference size depends on T: +/// - float: uint32_t frame_of_reference (4 bytes) +/// - double: uint64_t frame_of_reference (8 bytes) +/// +/// Serialization format for float (5 bytes): +/// +/// +------------------------------------------+ +/// | AlpEncodedForVectorInfo<float> (5B) | +/// +------------------------------------------+ +/// | Offset | Field | Size | +/// +---------+---------------------+----------+ +/// | 0 | frame_of_reference | 4 bytes | +/// | 4 | bit_width (uint8_t)| 1 byte | +/// +------------------------------------------+ +/// +/// Serialization format for double (9 bytes): +/// +/// +------------------------------------------+ +/// | AlpEncodedForVectorInfo<double> (9B) | +/// +------------------------------------------+ +/// | Offset | Field | Size | +/// +---------+---------------------+----------+ +/// | 0 | frame_of_reference | 8 bytes | +/// | 8 | bit_width (uint8_t)| 1 byte | +/// +------------------------------------------+ +/// +/// \tparam T the floating point type (float or double) +template <typename T> +class AlpEncodedForVectorInfo { + static_assert(std::is_same_v<T, float> || std::is_same_v<T, double>, + "AlpEncodedForVectorInfo only supports float and double"); + + public: + /// Use uint32_t for float, uint64_t for double (matches encoded integer size) + using ExactType = typename AlpTypedConstants<T>::FloatingToExact; + + AlpEncodedForVectorInfo() = default; + AlpEncodedForVectorInfo(ExactType frame_of_reference, uint8_t bit_width) + : frame_of_reference_(frame_of_reference), bit_width_(bit_width) {} + + ExactType frame_of_reference() const { return frame_of_reference_; } + uint8_t bit_width() const { return bit_width_; } + + void set_frame_of_reference(ExactType frame_of_reference) { + frame_of_reference_ = frame_of_reference; + } + void set_bit_width(uint8_t bit_width) { bit_width_ = bit_width; } + + /// Size of the serialized portion (5 bytes for float, 9 for double) + static constexpr int64_t kStoredSize = sizeof(ExactType) + 1; + + /// \brief Compute the bitpacked size in bytes from num_elements and bit_width + /// + /// \param[in] num_elements number of elements in this vector + /// \param[in] bw bits per element + /// \return the size in bytes of the bitpacked data + static int64_t GetBitPackedSize(int32_t num_elements, uint8_t bw) { Review Comment: The hand-rolled helper is gone. Its callers compute the packed size with `bit_util::BytesForBits(num_elements * bit_width)` directly, so it's in `bit_util` after all rather than relocated. ########## cpp/src/parquet/CMakeLists.txt: ########## @@ -442,6 +442,11 @@ add_parquet_benchmark(bloom_filter_benchmark SOURCES bloom_filter_benchmark.cc add_parquet_benchmark(column_reader_benchmark) add_parquet_benchmark(column_io_benchmark) add_parquet_benchmark(encoding_benchmark) +add_parquet_benchmark(encoding_alp_benchmark) + +add_executable(generate-alp-parquet + ${PROJECT_SOURCE_DIR}/src/arrow/util/alp/generate_alp_parquet.cc) Review Comment: Removed rather than moved — the generator is gone and the `add_executable` with it, so there's no cross-directory reference left. The ALP benchmarks went into the existing `encoding_benchmark.cc`. ########## cpp/src/arrow/util/alp/alp_wrapper.cc: ########## @@ -0,0 +1,435 @@ +// 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. + +#include "arrow/util/alp/alp_wrapper.h" + +#include <cmath> +#include <optional> + +#include "arrow/util/alp/alp.h" +#include "arrow/util/alp/alp_constants.h" +#include "arrow/util/alp/alp_sampler.h" +#include "arrow/util/logging.h" +#include "arrow/util/ubsan.h" + +namespace arrow { +namespace util { +namespace alp { + +namespace { + +// ---------------------------------------------------------------------- +// AlpHeader + +/// \brief Header structure for ALP compression blocks +/// +/// Contains page-level metadata for ALP compression. The num_elements field +/// stores the total element count for the page, allowing per-vector element +/// counts to be inferred (all vectors except the last have vector_size elements). +/// +/// Note: num_elements is uint32_t because Parquet page headers use i32 for num_values. +/// See: https://github.com/apache/parquet-format/blob/master/src/main/thrift/parquet.thrift +/// +/// Note: log_vector_size stores the base-2 logarithm of the vector size. +/// The actual vector size is computed as: 1u << log_vector_size (i.e., 2^log_vector_size). +/// For example, log_vector_size=10 means vector_size=1024. +/// This allows representing any power-of-2 vector size up to 2^255 in a single byte. +/// +/// Header format (version 1): +/// +/// +---------------------------------------------------+ +/// | AlpHeader (8 bytes) | +/// +---------------------------------------------------+ +/// | Offset | Field | Size | +/// +---------+---------------------+-------------------+ +/// | 0 | version | 1 byte (uint8) | +/// | 1 | compression_mode | 1 byte (uint8) | +/// | 2 | integer_encoding | 1 byte (uint8) | +/// | 3 | log_vector_size | 1 byte (uint8) | +/// | 4 | num_elements | 4 bytes (uint32) | +/// +---------------------------------------------------+ +/// +/// Page-level layout (metadata-at-start for efficient random access): +/// +/// +-------------------------------------------------------------------+ +/// | [AlpHeader (8B)] | +/// | [VectorInfo₀ | VectorInfo₁ | ... | VectorInfoₙ] ← Metadata | +/// | [Data₀ | Data₁ | ... | Dataₙ] ← Data sections | +/// +-------------------------------------------------------------------+ +/// +/// This layout enables O(1) random access to any vector by: +/// 1. Reading all VectorInfo first (contiguous, cache-friendly) +/// 2. Computing data offsets from VectorInfo +/// 3. Seeking directly to the target vector's data +/// +/// \note version must remain the first field to allow reading the rest +/// of the header based on version number. +struct AlpHeader { + /// Version number. Must remain the first field for version-based parsing. + uint8_t version = 0; + /// Compression mode (currently only kAlp is supported). + uint8_t compression_mode = static_cast<uint8_t>(AlpMode::kAlp); + /// Integer encoding method used (currently only kForBitPack is supported). + uint8_t integer_encoding = static_cast<uint8_t>(AlpIntegerEncoding::kForBitPack); + /// Log base 2 of vector size. Actual vector size = 1u << log_vector_size. + /// For example: 10 means 2^10 = 1024 elements per vector. + uint8_t log_vector_size = 0; + /// Total number of elements in the page (uint32_t since Parquet uses i32). + /// Per-vector element count is inferred: vector_size for all but the last vector. + uint32_t num_elements = 0; + + /// \brief Get the size in bytes of the AlpHeader for a version + /// + /// \param[in] v the version number + /// \return the size in bytes + static constexpr size_t GetSizeForVersion(uint8_t v) { + // Version 1 header is 8 bytes + return (v == 1) ? 8 : 0; + } + + /// \brief Check whether the given version is valid + /// + /// \param[in] v the version to check + /// \return the version if valid, otherwise asserts + static uint8_t IsValidVersion(uint8_t v) { + ARROW_CHECK(v == 1) << "invalid_version: " << static_cast<int>(v); + return v; + } + + /// \brief Compute the actual vector size from log_vector_size + /// + /// \return the vector size (2^log_vector_size) + uint32_t GetVectorSize() const { return 1u << log_vector_size; } + + /// \brief Compute log base 2 of a power-of-2 value + /// + /// \param[in] value a power-of-2 value + /// \return the log base 2 of value + static uint8_t Log2(uint32_t value) { + ARROW_CHECK(value > 0 && (value & (value - 1)) == 0) + << "value_must_be_power_of_2: " << value; + uint8_t log = 0; + while ((1u << log) < value) { + ++log; + } + return log; + } + + /// \brief Calculate the number of elements for a given vector index + /// + /// \param[in] vector_index the 0-based index of the vector + /// \return the number of elements in this vector + uint16_t GetVectorNumElements(uint64_t vector_index) const { + const uint32_t vector_size = GetVectorSize(); + const uint64_t num_full_vectors = num_elements / vector_size; + const uint64_t remainder = num_elements % vector_size; + if (vector_index < num_full_vectors) { + return static_cast<uint16_t>(vector_size); // Full vector + } else if (vector_index == num_full_vectors && remainder > 0) { + return static_cast<uint16_t>(remainder); // Last partial vector + } + return 0; // Invalid index + } + + /// \brief Get the AlpMode enum from the stored uint8_t + AlpMode GetCompressionMode() const { + return static_cast<AlpMode>(compression_mode); + } + + /// \brief Get the AlpIntegerEncoding enum from the stored uint8_t + AlpIntegerEncoding GetIntegerEncoding() const { + return static_cast<AlpIntegerEncoding>(integer_encoding); + } +}; + +} // namespace + +// ---------------------------------------------------------------------- +// AlpWrapper::AlpHeader definition + +template <typename T> +struct AlpWrapper<T>::AlpHeader : public ::arrow::util::alp::AlpHeader { +}; + +// ---------------------------------------------------------------------- +// AlpWrapper implementation + +template <typename T> +typename AlpWrapper<T>::AlpHeader AlpWrapper<T>::LoadHeader( + const char* comp, size_t comp_size) { + ARROW_CHECK(comp_size >= 1) << "alp_loadHeader_compSize_too_small_for_version"; + uint8_t version; + std::memcpy(&version, comp, sizeof(version)); + AlpHeader::IsValidVersion(version); + const size_t header_size = AlpHeader::GetSizeForVersion(version); + ARROW_CHECK(comp_size >= header_size) << "alp_loadHeader_compSize_too_small"; + AlpHeader header{}; + std::memcpy(&header, comp, header_size); + return header; +} + +template <typename T> +void AlpWrapper<T>::Encode(const T* decomp, size_t decomp_size, char* comp, + size_t* comp_size, std::optional<AlpMode> enforce_mode) { + ARROW_CHECK(decomp_size % sizeof(T) == 0) << "alp_encode_input_must_be_multiple_of_T"; + const uint64_t element_count = decomp_size / sizeof(T); + const uint8_t version = + AlpHeader::IsValidVersion(AlpConstants::kAlpVersion); + + AlpSampler<T> sampler; + sampler.AddSample({decomp, element_count}); + auto sampling_result = sampler.Finalize(); + + // Make room to store header afterwards. + char* encoded_header = comp; + const size_t header_size = AlpHeader::GetSizeForVersion(version); + comp += header_size; + const uint64_t remaining_compressed_size = *comp_size - header_size; + + const CompressionProgress compression_progress = + EncodeAlp(decomp, element_count, comp, remaining_compressed_size, + sampling_result.alp_preset); + + AlpHeader header{}; + header.version = version; + header.compression_mode = static_cast<uint8_t>(AlpMode::kAlp); + header.integer_encoding = static_cast<uint8_t>(AlpIntegerEncoding::kForBitPack); + header.log_vector_size = AlpHeader::Log2(AlpConstants::kAlpVectorSize); + header.num_elements = static_cast<uint32_t>(element_count); + + std::memcpy(encoded_header, &header, header_size); + *comp_size = header_size + compression_progress.num_compressed_bytes_produced; +} + +template <typename T> +template <typename TargetType> +void AlpWrapper<T>::Decode(TargetType* decomp, uint32_t num_elements, const char* comp, + size_t comp_size) { + const AlpHeader header = LoadHeader(comp, comp_size); + const uint32_t vector_size = header.GetVectorSize(); + ARROW_CHECK(vector_size == AlpConstants::kAlpVectorSize) Review Comment: Audited. The one you flagged is gone: the vector size comes off the header as `log_vector_size` now, and `LoadHeader` range-checks it and returns `Status::Invalid`. Ten remain, and none is reachable from untrusted input: every wire field goes through `Load`, which returns `Result`. Four are buffer-size preconditions in the `Store` paths, on a buffer this code sized one call earlier, and two more there are consistency invariants — the exception count against the exception vector, and the final offset against the section size. One is `Log2`'s power-of-two precondition, reachable only after `ValidateVectorSize` has rejected bad input; one is a sample-count invariant in `GetAlpSamplingParameters`. The two in the decode path are the ones worth naming: `Decode` asserts the per-vector lengths summed back to the element count — the header-vs-argument disagreement that could break it is already rejected with `Status::Invalid` just above — and `VectorLength` asserts its index is in range. Say so and I'll convert those two anyway. ########## cpp/src/arrow/util/alp/alp.h: ########## @@ -0,0 +1,849 @@ +// 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. + +// Adaptive Lossless floating-Point (ALP) compression implementation + +#pragma once + +#include <optional> +#include <vector> + +#include "arrow/result.h" +#include "arrow/status.h" +#include "arrow/util/alp/alp_constants.h" +#include "arrow/util/span.h" + +namespace arrow { +namespace util { +namespace alp { + +// ---------------------------------------------------------------------- +// ALP Overview +// +// IMPORTANT: For abstract interfaces or examples how to use ALP, consult +// alp_codec.h. +// This is our implementation of the adaptive lossless floating-point +// compression for decimals (ALP) (https://dl.acm.org/doi/10.1145/3626717). +// It works by converting a float into a decimal (if possible). The exponent +// and factor are chosen per vector. Each float is converted using +// c(f) = int64(f * 10^exponent * 10^-factor). The converted floats are then +// encoded via a delta frame of reference and bitpacked. Every exception, +// where the conversion/reconversion changes the value of the float, is stored +// separately and has to be patched into the decompressed vector afterwards. +// +// ========================================================================== +// ALP COMPRESSION/DECOMPRESSION PIPELINE +// ========================================================================== +// +// COMPRESSION FLOW: +// ----------------- +// +// Input: float/double array +// | +// v +// +------------------------------------------------------------------+ +// | 1. SAMPLING & PRESET GENERATION | +// | * Sample vectors from dataset | +// | * Try all exponent/factor combinations (e, f) | +// | * Select best k combinations for preset | +// +------------------------------------+-----------------------------+ +// | preset.combinations +// v +// +------------------------------------------------------------------+ +// | 2. PER-VECTOR COMPRESSION | +// | a) Find best (e,f) from preset for this vector | +// | b) Encode: encoded[i] = int64(value[i] * 10^e * 10^-f) | +// | c) Verify: if decode(encoded[i]) != value[i] -> exception | +// | d) Replace exceptions with placeholder value | +// +------------------------------------+-----------------------------+ +// | encoded integers + exceptions +// v +// +------------------------------------------------------------------+ +// | 3. FRAME OF REFERENCE (FOR) | +// | * Find min value in encoded integers | +// | * Subtract min from all values: delta[i] = encoded[i] - min | +// +------------------------------------+-----------------------------+ +// | delta values (smaller range) +// v +// +------------------------------------------------------------------+ +// | 4. BIT PACKING | +// | * Calculate bit_width = log2(max_delta) | +// | * Pack each value into bit_width bits | +// | * Result: tightly packed binary data | +// +------------------------------------+-----------------------------+ +// | packed bytes +// v +// +------------------------------------------------------------------+ +// | 5. SERIALIZATION (offset-based interleaved layout) | +// | [Header][Offsets...][Vector₀][Vector₁]... | +// | where each Vector = [AlpInfo|ForInfo|Data] | +// +------------------------------------------------------------------+ +// +// +// DECOMPRESSION FLOW: +// ------------------- +// +// Serialized bytes -> AlpEncodedVector::Load() +// | +// v +// +------------------------------------------------------------------+ +// | 1. BIT UNPACKING | +// | * Extract bit_width from metadata | +// | * Unpack each value from bit_width bits -> delta values | +// +------------------------------------+-----------------------------+ +// | delta values +// v +// +------------------------------------------------------------------+ +// | 2. REVERSE FRAME OF REFERENCE (unFOR) | +// | * Add back min: encoded[i] = delta[i] + frame_of_reference | +// +------------------------------------+-----------------------------+ +// | encoded integers +// v +// +------------------------------------------------------------------+ +// | 3. DECODE | +// | * Apply inverse formula: value[i] = encoded[i] * 10^-e * 10^f | +// +------------------------------------+-----------------------------+ +// | decoded floats (with placeholders) +// v +// +------------------------------------------------------------------+ +// | 4. PATCH EXCEPTIONS | +// | * Replace values at exception_positions[] with exceptions[] | +// +------------------------------------+-----------------------------+ +// | +// v +// Output: Original float/double array (lossless!) +// +// ========================================================================== + +// ---------------------------------------------------------------------- +// AlpMode + +/// \brief ALP compression mode +/// +/// Currently only ALP (decimal compression) is implemented. +enum class AlpMode { kAlp }; Review Comment: Done — it's `enum class AlpMode : uint8_t { kAlp = 0 };` now. ########## cpp/src/arrow/util/alp/alp_wrapper.cc: ########## @@ -0,0 +1,532 @@ +// 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. + +#include "arrow/util/alp/alp_wrapper.h" + +#include <cmath> +#include <optional> + +#include "arrow/result.h" +#include "arrow/status.h" +#include "arrow/util/alp/alp.h" +#include "arrow/util/alp/alp_constants.h" +#include "arrow/util/alp/alp_sampler.h" +#include "arrow/util/endian.h" +#include "arrow/util/logging.h" +#include "arrow/util/ubsan.h" + +namespace arrow { +namespace util { +namespace alp { + +// ALP serialization uses memcpy for multi-byte integers (header fields, +// offsets, frame_of_reference) and assumes little-endian byte order on disk. +static_assert(ARROW_LITTLE_ENDIAN, + "ALP serialization assumes little-endian byte order"); + +namespace { + +// ---------------------------------------------------------------------- +// AlpHeader + +/// \brief Header structure for ALP compression blocks +/// +/// Contains page-level metadata for ALP compression. The num_elements field +/// stores the total element count for the page, allowing per-vector element +/// counts to be inferred (all vectors except the last have vector_size elements). +/// +/// Note: num_elements is uint32_t because Parquet page headers use i32 for num_values. +/// See: https://github.com/apache/parquet-format/blob/master/src/main/thrift/parquet.thrift +/// +/// Note: log_vector_size stores the base-2 logarithm of the vector size. +/// The actual vector size is computed as: 1u << log_vector_size (i.e., 2^log_vector_size). +/// For example, log_vector_size=10 means vector_size=1024. +/// This allows representing any power-of-2 vector size up to 2^255 in a single byte. +/// +/// Header format (7 bytes): +/// +/// +---------------------------------------------------+ +/// | AlpHeader (7 bytes) | +/// +---------------------------------------------------+ +/// | Offset | Field | Size | +/// +---------+---------------------+-------------------+ +/// | 0 | compression_mode | 1 byte (uint8) | +/// | 1 | integer_encoding | 1 byte (uint8) | +/// | 2 | log_vector_size | 1 byte (uint8) | +/// | 3 | num_elements | 4 bytes (uint32) | +/// +---------------------------------------------------+ +/// +/// Page-level layout (offset-based interleaved for O(1) random access): +/// +/// +-------------------------------------------------------------------+ +/// | [AlpHeader (7B)] | +/// | [Offset₀ | Offset₁ | ... | Offsetₙ₋₁] ← Vector offsets | +/// | [Vector₀][Vector₁]...[Vectorₙ₋₁] ← Interleaved data | +/// +-------------------------------------------------------------------+ +/// where each Vector = [AlpInfo | ForInfo | Data] +/// +/// This layout enables O(1) random access to any vector by: +/// 1. Reading the offset for target vector (direct lookup) +/// 2. Jumping to that offset to read metadata + data together +struct AlpHeader { + /// Compression mode (currently only kAlp is supported). + uint8_t compression_mode = static_cast<uint8_t>(AlpMode::kAlp); + /// Integer encoding method used (currently only kForBitPack is supported). + uint8_t integer_encoding = static_cast<uint8_t>(AlpIntegerEncoding::kForBitPack); + /// Log base 2 of vector size. Actual vector size = 1u << log_vector_size. + /// For example: 10 means 2^10 = 1024 elements per vector. + uint8_t log_vector_size = 0; + /// Total number of elements in the page (uint32_t since Parquet uses i32). + /// Per-vector element count is inferred: vector_size for all but the last vector. + uint32_t num_elements = 0; + + /// Size of the serialized header in bytes. + static constexpr size_t kSize = 7; + + /// \brief Calculate the number of vectors from total elements and vector size + /// + /// \return number of vectors (full + partial if any) + uint32_t GetNumVectors() const { + const uint32_t vector_size = GetVectorSize(); + return (num_elements + vector_size - 1) / vector_size; + } + + /// \brief Get the size of the offsets section + /// + /// \return size in bytes of the offsets array (num_vectors * sizeof(OffsetType)) + uint64_t GetOffsetsSectionSize() const { + return static_cast<uint64_t>(GetNumVectors()) * sizeof(AlpConstants::OffsetType); + } + + /// \brief Compute the actual vector size from log_vector_size + /// + /// \return the vector size (2^log_vector_size) + uint32_t GetVectorSize() const { return 1u << log_vector_size; } + + /// \brief Compute log base 2 of a power-of-2 value + /// + /// \param[in] value a power-of-2 value + /// \return the log base 2 of value + static uint8_t Log2(uint32_t value) { + ARROW_CHECK(value > 0 && (value & (value - 1)) == 0) + << "value_must_be_power_of_2: " << value; + return static_cast<uint8_t>(__builtin_ctz(value)); + } + + /// \brief Calculate the number of elements for a given vector index + /// + /// \param[in] vector_index the 0-based index of the vector + /// \return the number of elements in this vector + uint16_t GetVectorNumElements(uint64_t vector_index) const { + const uint32_t vector_size = GetVectorSize(); + const uint64_t num_full_vectors = num_elements / vector_size; + const uint64_t remainder = num_elements % vector_size; + if (vector_index < num_full_vectors) { + return static_cast<uint16_t>(vector_size); // Full vector + } else if (vector_index == num_full_vectors && remainder > 0) { + return static_cast<uint16_t>(remainder); // Last partial vector + } + ARROW_CHECK(false) << "alp_invalid_vector_index: " << vector_index + << " (num_vectors=" << GetNumVectors() << ")"; + return 0; // Unreachable, but silences compiler warning + } + + /// \brief Get the AlpMode enum from the stored uint8_t + AlpMode GetCompressionMode() const { + return static_cast<AlpMode>(compression_mode); + } + + /// \brief Get the AlpIntegerEncoding enum from the stored uint8_t + AlpIntegerEncoding GetIntegerEncoding() const { + return static_cast<AlpIntegerEncoding>(integer_encoding); + } +}; + +} // namespace + +// ---------------------------------------------------------------------- +// AlpCodec::AlpHeader definition + +template <typename T> +struct AlpCodec<T>::AlpHeader : public ::arrow::util::alp::AlpHeader { +}; + +// ---------------------------------------------------------------------- +// AlpCodec implementation + +template <typename T> +auto AlpCodec<T>::LoadHeader(const char* comp, size_t comp_size) + -> Result<AlpHeader> { + if (comp_size < AlpHeader::kSize) { + return Status::Invalid("ALP compressed buffer too small for header: ", comp_size, + " < ", AlpHeader::kSize); + } + AlpHeader header{}; + std::memcpy(&header.compression_mode, comp, 3); + std::memcpy(&header.num_elements, comp + 3, sizeof(header.num_elements)); + return header; +} + +template <typename T> +auto AlpCodec<T>::CreateSamplingPreset(const T* decomp, size_t decomp_size) + -> AlpSamplerResult { + ARROW_CHECK(decomp_size % sizeof(T) == 0) << "alp_encode_input_must_be_multiple_of_T"; + const uint64_t element_count = decomp_size / sizeof(T); + + AlpSampler<T> sampler; + sampler.AddSample({decomp, element_count}); + return sampler.Finalize(); +} + +template <typename T> +void AlpCodec<T>::EncodeWithPreset(const T* decomp, size_t decomp_size, char* comp, + size_t* comp_size, const AlpSamplerResult& preset) { + ARROW_CHECK(decomp_size % sizeof(T) == 0) << "alp_encode_input_must_be_multiple_of_T"; + const uint64_t element_count = decomp_size / sizeof(T); + + // Make room to store header afterwards. + char* encoded_header = comp; + comp += AlpHeader::kSize; + const uint64_t remaining_compressed_size = *comp_size - AlpHeader::kSize; + + const CompressionProgress compression_progress = + EncodeAlp(decomp, element_count, comp, remaining_compressed_size, + preset.alp_parameters); + + AlpHeader header{}; + header.compression_mode = static_cast<uint8_t>(AlpMode::kAlp); + header.integer_encoding = static_cast<uint8_t>(AlpIntegerEncoding::kForBitPack); + header.log_vector_size = AlpHeader::Log2(AlpConstants::kAlpVectorSize); + header.num_elements = static_cast<uint32_t>(element_count); + + std::memcpy(encoded_header, &header.compression_mode, 3); + std::memcpy(encoded_header + 3, &header.num_elements, sizeof(header.num_elements)); + *comp_size = AlpHeader::kSize + compression_progress.num_compressed_bytes_produced; +} + +template <typename T> +void AlpCodec<T>::Encode(const T* decomp, size_t decomp_size, char* comp, + size_t* comp_size, std::optional<AlpMode> enforce_mode) { + // Sample the data and encode with the preset + auto sampling_result = CreateSamplingPreset(decomp, decomp_size); + EncodeWithPreset(decomp, decomp_size, comp, comp_size, sampling_result); +} + +template <typename T> +template <typename TargetType> +Status AlpCodec<T>::Decode(int32_t num_elements, const char* comp, size_t comp_size, + TargetType* decomp) { + ARROW_ASSIGN_OR_RAISE(const AlpHeader header, LoadHeader(comp, comp_size)); + if (header.log_vector_size > AlpConstants::kMaxLogVectorSize) { + return Status::Invalid("ALP log_vector_size too large: ", + static_cast<int>(header.log_vector_size), + " > ", static_cast<int>(AlpConstants::kMaxLogVectorSize), + " (would overflow uint16_t element count)"); + } + const uint32_t vector_size = header.GetVectorSize(); + if (vector_size != AlpConstants::kAlpVectorSize) { + return Status::Invalid("Unsupported ALP vector_size: ", vector_size, + " (only ", AlpConstants::kAlpVectorSize, " is supported)"); + } + + const char* compression_body = comp + AlpHeader::kSize; + const uint64_t compression_body_size = comp_size - AlpHeader::kSize; + + if (header.GetCompressionMode() != AlpMode::kAlp) { + return Status::Invalid("Unsupported ALP compression mode: ", + static_cast<int>(header.compression_mode)); + } + + ARROW_RETURN_NOT_OK( + DecodeAlp<TargetType>(num_elements, compression_body, compression_body_size, + header.GetIntegerEncoding(), vector_size, + header.num_elements, decomp) + .status()); + return Status::OK(); +} + +template Status AlpCodec<float>::Decode(int32_t num_elements, const char* comp, + size_t comp_size, float* decomp); +template Status AlpCodec<float>::Decode(int32_t num_elements, const char* comp, + size_t comp_size, double* decomp); +template Status AlpCodec<double>::Decode(int32_t num_elements, const char* comp, + size_t comp_size, double* decomp); + +template <typename T> +int64_t AlpCodec<T>::GetMaxCompressedSize(int64_t uncompressed_size) { + ARROW_CHECK(uncompressed_size >= 0 && uncompressed_size % sizeof(T) == 0) + << "alp_decompressed_size_not_multiple_of_T"; + const uint64_t element_count = static_cast<uint64_t>(uncompressed_size) / sizeof(T); + uint64_t max_alp_size = AlpHeader::kSize; + + const uint64_t vectors_count = + static_cast<uint64_t>(std::ceil(static_cast<double>(element_count) / AlpConstants::kAlpVectorSize)); + + // Add offsets section (4 bytes per vector) + max_alp_size += vectors_count * sizeof(AlpConstants::OffsetType); + + // Add per-vector metadata sizes: AlpInfo (4 bytes) + ForInfo (5/9 bytes) + max_alp_size += + (AlpEncodedVectorInfo::kStoredSize + AlpEncodedForVectorInfo<T>::kStoredSize) * vectors_count; + + // Worst case: everything is an exception, except two values that are chosen + // with large difference to make FOR encoding for placeholders impossible. + // Values/placeholders. + max_alp_size += element_count * sizeof(T); + // Exceptions. + max_alp_size += element_count * sizeof(T); + // Exception positions. + max_alp_size += element_count * sizeof(AlpConstants::PositionType); + + return static_cast<int64_t>(max_alp_size); +} + +template <typename T> +auto AlpCodec<T>::EncodeAlp(const T* decomp, uint64_t element_count, char* comp, + size_t comp_size, const AlpEncodingParameters& combinations) + -> CompressionProgress { + // OFFSET-BASED LAYOUT + // [Offset₀ | Offset₁ | ... | Offsetₙ₋₁] ← Byte offsets to each vector (4B each) + // [AlpInfo₀ | ForInfo₀ | Data₀] ← Vector 0 (interleaved) + // [AlpInfo₁ | ForInfo₁ | Data₁] ← Vector 1 + // ... + // [AlpInfoₙ₋₁ | ForInfoₙ₋₁ | Dataₙ₋₁] ← Vector n-1 + // + // Benefits: + // - O(1) random access to any vector (no cumulative offset computation) + // - Better locality for single-vector access (metadata + data together) + // - Enables parallel decompression without coordination + + // Phase 1: Compress all vectors and collect them + std::vector<AlpEncodedVector<T>> encoded_vectors; + const uint64_t num_vectors = + (element_count + AlpConstants::kAlpVectorSize - 1) / AlpConstants::kAlpVectorSize; + encoded_vectors.reserve(num_vectors); + + uint64_t input_offset = 0; + for (uint64_t remaining_elements = element_count; remaining_elements > 0; + remaining_elements -= std::min(AlpConstants::kAlpVectorSize, remaining_elements)) { + const uint64_t elements_to_encode = + std::min(AlpConstants::kAlpVectorSize, remaining_elements); + encoded_vectors.push_back(AlpCompression<T>::CompressVector( + decomp + input_offset, static_cast<uint16_t>(elements_to_encode), combinations)); + input_offset += elements_to_encode; + } + + // Phase 2: Calculate sizes and offsets + const AlpIntegerEncoding integer_encoding = combinations.integer_encoding; + const uint64_t per_vector_metadata_size = + AlpEncodedVectorInfo::kStoredSize + GetIntegerEncodingMetadataSize<T>(integer_encoding); + + // Offsets section comes first (after header, which is written by Encode()) + const uint64_t offsets_section_size = + num_vectors * sizeof(AlpConstants::OffsetType); + + // Calculate total size and per-vector offsets + std::vector<AlpConstants::OffsetType> vector_offsets; + vector_offsets.reserve(num_vectors); + + // First vector starts right after the offsets section + uint64_t current_offset = offsets_section_size; + for (const auto& vec : encoded_vectors) { + // Store offset to this vector (relative to start of body, after header) + vector_offsets.push_back(static_cast<AlpConstants::OffsetType>(current_offset)); + // Advance by metadata + data size + current_offset += per_vector_metadata_size + vec.GetDataStoredSize(); + } + const uint64_t total_size = current_offset; + + if (total_size > comp_size) { + return CompressionProgress{0, 0}; + } + + // Phase 3: Write offsets section + char* offset_ptr = comp; + for (const auto& offset : vector_offsets) { + std::memcpy(offset_ptr, &offset, sizeof(AlpConstants::OffsetType)); + offset_ptr += sizeof(AlpConstants::OffsetType); + } + + // Phase 4: Write interleaved vectors [AlpInfo | ForInfo | Data] + for (size_t i = 0; i < encoded_vectors.size(); i++) { + const auto& vec = encoded_vectors[i]; + char* vector_start = comp + vector_offsets[i]; + + // Write AlpInfo + vec.alp_info.Store({vector_start, AlpEncodedVectorInfo::kStoredSize}); + char* ptr = vector_start + AlpEncodedVectorInfo::kStoredSize; + + // Write ForInfo (or other integer encoding metadata) + switch (integer_encoding) { + case AlpIntegerEncoding::kForBitPack: { + vec.for_info.Store({ptr, AlpEncodedForVectorInfo<T>::kStoredSize}); + ptr += AlpEncodedForVectorInfo<T>::kStoredSize; + } break; + + default: + ARROW_CHECK(false) << "unsupported_integer_encoding: " + << static_cast<int>(integer_encoding); + break; + } + + // Write data (packed values + exception positions + exception values) + const uint64_t data_size = vec.GetDataStoredSize(); + vec.StoreDataOnly({ptr, data_size}); + } + + return CompressionProgress{static_cast<int64_t>(total_size), + static_cast<int64_t>(element_count)}; +} + +template <typename T> +template <typename TargetType> +auto AlpCodec<T>::DecodeAlp(size_t decomp_element_count, + const char* comp, size_t comp_size, + AlpIntegerEncoding integer_encoding, + uint32_t vector_size, uint32_t total_elements, + TargetType* decomp) + -> Result<DecompressionProgress> { + // OFFSET-BASED LAYOUT: + // [Offset₀ | Offset₁ | ... | Offsetₙ₋₁] ← Byte offsets to each vector (4B each) + // [AlpInfo₀ | ForInfo₀ | Data₀] ← Vector 0 (interleaved) + // [AlpInfo₁ | ForInfo₁ | Data₁] ← Vector 1 + // ... + // + // Benefits: + // - O(1) random access to any vector (no cumulative offset computation) + // - Better locality for single-vector access (metadata + data together) + // - Enables parallel decompression without coordination + + // Calculate number of vectors + const uint32_t num_vectors = + (total_elements + vector_size - 1) / vector_size; + + if (num_vectors == 0) { + return DecompressionProgress{0, 0}; + } + + const uint64_t offsets_section_size = + static_cast<uint64_t>(num_vectors) * sizeof(AlpConstants::OffsetType); + if (comp_size < offsets_section_size) { + return Status::Invalid("ALP compressed buffer too small for offsets section: ", + comp_size, " < ", offsets_section_size); + } + + // Read all offsets + std::vector<AlpConstants::OffsetType> vector_offsets(num_vectors); + std::memcpy(vector_offsets.data(), comp, + num_vectors * sizeof(AlpConstants::OffsetType)); + + // Decode each vector using its offset for O(1) random access + uint64_t output_offset = 0; + uint64_t bytes_consumed = offsets_section_size; + + for (uint32_t vector_index = 0; vector_index < num_vectors; vector_index++) { + // Calculate number of elements in this vector + const uint64_t num_full_vectors = total_elements / vector_size; + const uint64_t remainder = total_elements % vector_size; + uint16_t this_vector_elements; + if (vector_index < num_full_vectors) { + this_vector_elements = static_cast<uint16_t>(vector_size); + } else if (vector_index == num_full_vectors && remainder > 0) { + this_vector_elements = static_cast<uint16_t>(remainder); + } else { + this_vector_elements = 0; Review Comment: The `0` is gone rather than marked. `GetVectorNumElements` returns `Result<int32_t>`, so an out-of-range index comes back as `Status::Invalid` instead of a zero-length vector. ########## cpp/src/arrow/util/alp/alp_codec.h: ########## @@ -0,0 +1,185 @@ +// 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. + +// High-level wrapper interface for ALP compression + +#pragma once + +#include <cstddef> +#include <optional> + +#include "arrow/result.h" +#include "arrow/status.h" +#include "arrow/util/alp/alp.h" +#include "arrow/util/alp/alp_sampler.h" + +namespace arrow { +namespace util { +namespace alp { + +// ---------------------------------------------------------------------- +// AlpCodec + +/// \class AlpCodec +/// \brief High-level interface for ALP compression +/// +/// AlpCodec is an interface for Adaptive Lossless floating-Point Compression +/// (ALP) (https://dl.acm.org/doi/10.1145/3626717). For encoding, it samples +/// the data and applies decimal compression (Alp) to floating point values. +/// This class acts as a wrapper around the vector-based interfaces of +/// AlpSampler and Alp. +/// +/// \tparam T the floating point type (float or double) +template <typename T> +class AlpCodec { + public: + /// Type alias for the sampler result containing encoding presets + using AlpSamplerResult = typename AlpSampler<T>::AlpSamplerResult; + + /// \brief Create a sampling preset from input data + /// + /// This samples the input data and generates an encoding preset that can be + /// reused for encoding. This is useful when you want to pre-compute the preset + /// outside of the benchmark loop or encode multiple batches with the same preset. + /// + /// \param[in] decomp pointer to the input data to sample + /// \param[in] decomp_size size of decomp in bytes. + /// This needs to be a multiple of sizeof(T). + /// \return the sampling result containing the encoding preset + static AlpSamplerResult CreateSamplingPreset(const T* decomp, size_t decomp_size); + + /// \brief Encode floating point values using a pre-computed preset + /// + /// This encodes the data using a preset that was previously computed via + /// CreateSamplingPreset(). This avoids the sampling overhead during encoding. + /// + /// \param[in] decomp pointer to the input that is to be encoded + /// \param[in] decomp_size size of decomp in bytes. + /// This needs to be a multiple of sizeof(T). + /// \param[out] comp pointer to the memory region we will encode into. + /// Must be at least GetMaxCompressedSize(decomp_size) bytes. + /// \param[in,out] comp_size the actual size of the encoded data in bytes, + /// expects the size of comp as input. If this is too small, + /// this is set to 0 and we bail out. + /// \param[in] preset the pre-computed sampling result from CreateSamplingPreset() + static void EncodeWithPreset(const T* decomp, size_t decomp_size, char* comp, + size_t* comp_size, const AlpSamplerResult& preset); + + /// \brief Encode floating point values using ALP decimal compression + /// + /// \param[in] decomp pointer to the input that is to be encoded + /// \param[in] decomp_size size of decomp in bytes. + /// This needs to be a multiple of sizeof(T). + /// \param[out] comp pointer to the memory region we will encode into. + /// Must be at least GetMaxCompressedSize(decomp_size) bytes. + /// \param[in,out] comp_size the actual size of the encoded data in bytes, + /// expects the size of comp as input. If this is too small, + /// this is set to 0 and we bail out. + /// \param[in] enforce_mode reserved for future use. + /// Currently only AlpMode::kAlp is supported. + static void Encode(const T* decomp, size_t decomp_size, char* comp, + size_t* comp_size, + std::optional<AlpMode> enforce_mode = std::nullopt); + + /// \brief Decode floating point values + /// + /// \param[in] num_elements number of elements to decode (from page header) + /// \param[in] comp pointer to the input that is to be decoded + /// \param[in] comp_size size of the input in bytes (from page header) + /// \param[out] decomp pointer to the memory region we will decode into. + /// The caller is responsible for ensuring this is big enough + /// to hold num_elements values. + /// \return Status::OK on success, or an error if the compressed data is malformed + /// \tparam TargetType the type that is used to store the output. + /// May not be a narrowing conversion from T. + template <typename TargetType> + static Status Decode(int32_t num_elements, const char* comp, size_t comp_size, + TargetType* decomp); + + /// \brief Get the maximum compressed size of an uncompressed buffer + /// + /// \param[in] uncompressed_size the size of the uncompressed buffer in bytes + /// \return the maximum size of the compressed buffer + static int64_t GetMaxCompressedSize(int64_t uncompressed_size); + + private: + struct AlpHeader; + + /// \brief Tracks the progress of a compression operation + /// + /// Used to report how much data was consumed and produced during encoding. + struct CompressionProgress { + /// Number of compressed bytes written to output + int64_t num_compressed_bytes_produced = 0; + /// Number of input elements consumed + int64_t num_uncompressed_elements_taken = 0; + }; + + /// \brief Tracks the progress of a decompression operation + /// + /// Used to report how much data was consumed and produced during decoding. + struct DecompressionProgress { + /// Number of decompressed elements written + int64_t num_decompressed_elements_produced = 0; + /// Number of compressed bytes consumed + int64_t num_compressed_bytes_taken = 0; + }; + + /// \brief Compress a buffer using the ALP variant + /// + /// \param[in] decomp array of floating point numbers to compress + /// \param[in] element_count the number of floating point numbers + /// \param[out] comp the buffer to be compressed into + /// \param[in] comp_size the size of the compression buffer + /// \param[in] combinations the encoding preset to use + /// \return the compression progress + static CompressionProgress EncodeAlp(const T* decomp, uint64_t element_count, Review Comment: Done — `EncodeAlp` takes `const T* input` and writes to `output`, and no abbreviated `comp`/`decomp` names are left anywhere in the codec. ########## cpp/src/arrow/util/alp/alp_codec.h: ########## @@ -0,0 +1,185 @@ +// 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. + +// High-level wrapper interface for ALP compression + +#pragma once + +#include <cstddef> +#include <optional> + +#include "arrow/result.h" +#include "arrow/status.h" +#include "arrow/util/alp/alp.h" +#include "arrow/util/alp/alp_sampler.h" + +namespace arrow { +namespace util { +namespace alp { + +// ---------------------------------------------------------------------- +// AlpCodec + +/// \class AlpCodec +/// \brief High-level interface for ALP compression +/// +/// AlpCodec is an interface for Adaptive Lossless floating-Point Compression +/// (ALP) (https://dl.acm.org/doi/10.1145/3626717). For encoding, it samples +/// the data and applies decimal compression (Alp) to floating point values. +/// This class acts as a wrapper around the vector-based interfaces of +/// AlpSampler and Alp. +/// +/// \tparam T the floating point type (float or double) +template <typename T> +class AlpCodec { + public: + /// Type alias for the sampler result containing encoding presets + using AlpSamplerResult = typename AlpSampler<T>::AlpSamplerResult; + + /// \brief Create a sampling preset from input data + /// + /// This samples the input data and generates an encoding preset that can be + /// reused for encoding. This is useful when you want to pre-compute the preset + /// outside of the benchmark loop or encode multiple batches with the same preset. + /// + /// \param[in] decomp pointer to the input data to sample + /// \param[in] decomp_size size of decomp in bytes. + /// This needs to be a multiple of sizeof(T). + /// \return the sampling result containing the encoding preset + static AlpSamplerResult CreateSamplingPreset(const T* decomp, size_t decomp_size); + + /// \brief Encode floating point values using a pre-computed preset + /// + /// This encodes the data using a preset that was previously computed via + /// CreateSamplingPreset(). This avoids the sampling overhead during encoding. + /// + /// \param[in] decomp pointer to the input that is to be encoded + /// \param[in] decomp_size size of decomp in bytes. + /// This needs to be a multiple of sizeof(T). + /// \param[out] comp pointer to the memory region we will encode into. + /// Must be at least GetMaxCompressedSize(decomp_size) bytes. + /// \param[in,out] comp_size the actual size of the encoded data in bytes, + /// expects the size of comp as input. If this is too small, + /// this is set to 0 and we bail out. + /// \param[in] preset the pre-computed sampling result from CreateSamplingPreset() + static void EncodeWithPreset(const T* decomp, size_t decomp_size, char* comp, + size_t* comp_size, const AlpSamplerResult& preset); + + /// \brief Encode floating point values using ALP decimal compression + /// + /// \param[in] decomp pointer to the input that is to be encoded + /// \param[in] decomp_size size of decomp in bytes. + /// This needs to be a multiple of sizeof(T). + /// \param[out] comp pointer to the memory region we will encode into. + /// Must be at least GetMaxCompressedSize(decomp_size) bytes. + /// \param[in,out] comp_size the actual size of the encoded data in bytes, + /// expects the size of comp as input. If this is too small, + /// this is set to 0 and we bail out. + /// \param[in] enforce_mode reserved for future use. + /// Currently only AlpMode::kAlp is supported. + static void Encode(const T* decomp, size_t decomp_size, char* comp, + size_t* comp_size, + std::optional<AlpMode> enforce_mode = std::nullopt); + + /// \brief Decode floating point values + /// + /// \param[in] num_elements number of elements to decode (from page header) + /// \param[in] comp pointer to the input that is to be decoded + /// \param[in] comp_size size of the input in bytes (from page header) + /// \param[out] decomp pointer to the memory region we will decode into. + /// The caller is responsible for ensuring this is big enough + /// to hold num_elements values. + /// \return Status::OK on success, or an error if the compressed data is malformed + /// \tparam TargetType the type that is used to store the output. + /// May not be a narrowing conversion from T. + template <typename TargetType> + static Status Decode(int32_t num_elements, const char* comp, size_t comp_size, + TargetType* decomp); + + /// \brief Get the maximum compressed size of an uncompressed buffer + /// + /// \param[in] uncompressed_size the size of the uncompressed buffer in bytes + /// \return the maximum size of the compressed buffer + static int64_t GetMaxCompressedSize(int64_t uncompressed_size); + + private: + struct AlpHeader; + + /// \brief Tracks the progress of a compression operation + /// + /// Used to report how much data was consumed and produced during encoding. + struct CompressionProgress { + /// Number of compressed bytes written to output + int64_t num_compressed_bytes_produced = 0; + /// Number of input elements consumed + int64_t num_uncompressed_elements_taken = 0; + }; + + /// \brief Tracks the progress of a decompression operation + /// + /// Used to report how much data was consumed and produced during decoding. + struct DecompressionProgress { + /// Number of decompressed elements written + int64_t num_decompressed_elements_produced = 0; + /// Number of compressed bytes consumed + int64_t num_compressed_bytes_taken = 0; + }; + + /// \brief Compress a buffer using the ALP variant + /// + /// \param[in] decomp array of floating point numbers to compress + /// \param[in] element_count the number of floating point numbers + /// \param[out] comp the buffer to be compressed into + /// \param[in] comp_size the size of the compression buffer + /// \param[in] combinations the encoding preset to use + /// \return the compression progress + static CompressionProgress EncodeAlp(const T* decomp, uint64_t element_count, + char* comp, size_t comp_size, + const AlpEncodingParameters& combinations); + + /// \brief Decompress a buffer using the ALP variant + /// + /// \param[in] decomp_element_count the number of floats to decompress + /// \param[in] comp the compressed buffer to be decompressed + /// \param[in] comp_size the size of the compressed data + /// \param[in] integer_encoding the bit packing layout used + /// \param[in] vector_size the number of elements per vector (from header) + /// \param[in] total_elements the total number of elements in the page (from header). + /// Uses uint32_t since Parquet page headers use i32 for num_values. + /// \param[out] decomp the buffer to be decompressed into + /// \return the decompression progress, or an error if the compressed data is malformed + /// \tparam TargetType the type that is used to store the output. + /// May not be a narrowing conversion from T. + template <typename TargetType> + static Result<DecompressionProgress> DecodeAlp(size_t decomp_element_count, + const char* comp, size_t comp_size, + AlpIntegerEncoding integer_encoding, + uint32_t vector_size, Review Comment: Signed now — the public entry point is `Decode(int32_t num_elements, const uint8_t* input, int64_t input_size, TargetType* output)`, and the vector size comes off the header as `int32_t`. ########## cpp/src/arrow/util/alp/alp_wrapper.h: ########## @@ -0,0 +1,148 @@ +// 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. + +// High-level wrapper interface for ALP compression + +#pragma once + +#include <cstddef> +#include <optional> + +#include "arrow/util/alp/alp.h" + +namespace arrow { +namespace util { +namespace alp { + +// ---------------------------------------------------------------------- +// AlpWrapper + +/// \class AlpWrapper +/// \brief High-level interface for ALP compression +/// +/// AlpWrapper is an interface for Adaptive Lossless floating-Point Compression +/// (ALP) (https://dl.acm.org/doi/10.1145/3626717). For encoding, it samples +/// the data and applies decimal compression (Alp) to floating point values. +/// This class acts as a wrapper around the vector-based interfaces of +/// AlpSampler and Alp. +/// +/// \tparam T the floating point type (float or double) +template <typename T> +class AlpWrapper { + public: + /// \brief Encode floating point values using ALP decimal compression + /// + /// \param[in] decomp pointer to the input that is to be encoded + /// \param[in] decomp_size size of decomp in bytes. + /// This needs to be a multiple of sizeof(T). + /// \param[out] comp pointer to the memory region we will encode into. + /// The caller is responsible for ensuring this is big enough. + /// \param[in,out] comp_size the actual size of the encoded data in bytes, + /// expects the size of comp as input. If this is too small, + /// this is set to 0 and we bail out. + /// \param[in] enforce_mode reserved for future use. + /// Currently only AlpMode::kAlp is supported. + static void Encode(const T* decomp, size_t decomp_size, char* comp, + size_t* comp_size, + std::optional<AlpMode> enforce_mode = std::nullopt); + + /// \brief Decode floating point values + /// + /// \param[out] decomp pointer to the memory region we will decode into. + /// The caller is responsible for ensuring this is big enough + /// to hold num_elements values. + /// \param[in] num_elements number of elements to decode (from page header). + /// Uses uint32_t since Parquet page headers use i32 for num_values. + /// \param[in] comp pointer to the input that is to be decoded + /// \param[in] comp_size size of the input in bytes (from page header) + /// \tparam TargetType the type that is used to store the output. + /// May not be a narrowing conversion from T. + template <typename TargetType> + static void Decode(TargetType* decomp, uint32_t num_elements, const char* comp, + size_t comp_size); + + /// \brief Get the maximum compressed size of an uncompressed buffer + /// + /// \param[in] decomp_size the size of the uncompressed buffer in bytes + /// \return the maximum size of the compressed buffer + static uint64_t GetMaxCompressedSize(uint64_t decomp_size); + + private: + struct AlpHeader; + + /// \brief Tracks the progress of a compression operation + /// + /// Used to report how much data was consumed and produced during encoding. + struct CompressionProgress { + /// Number of compressed bytes written to output + uint64_t num_compressed_bytes_produced = 0; + /// Number of input elements consumed + uint64_t num_uncompressed_elements_taken = 0; + }; + + /// \brief Tracks the progress of a decompression operation + /// + /// Used to report how much data was consumed and produced during decoding. + struct DecompressionProgress { + /// Number of decompressed elements written + uint64_t num_decompressed_elements_produced = 0; + /// Number of compressed bytes consumed + uint64_t num_compressed_bytes_taken = 0; + }; + + /// \brief Compress a buffer using the ALP variant + /// + /// \param[in] decomp array of floating point numbers to compress + /// \param[in] element_count the number of floating point numbers + /// \param[out] comp the buffer to be compressed into + /// \param[in] comp_size the size of the compression buffer + /// \param[in] combinations the encoding preset to use + /// \return the compression progress + static CompressionProgress EncodeAlp(const T* decomp, uint64_t element_count, + char* comp, size_t comp_size, + const AlpEncodingPreset& combinations); + + /// \brief Decompress a buffer using the ALP variant + /// + /// \param[out] decomp the buffer to be decompressed into Review Comment: Done on the public API. On the private `EncodeAlp` helper the output pointer is followed by its capacity, `uint8_t* output, int64_t output_size`, since the two travel together. ########## cpp/src/arrow/util/alp/alp_codec.h: ########## @@ -0,0 +1,185 @@ +// 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. + +// High-level wrapper interface for ALP compression + +#pragma once + +#include <cstddef> +#include <optional> + +#include "arrow/result.h" +#include "arrow/status.h" +#include "arrow/util/alp/alp.h" +#include "arrow/util/alp/alp_sampler.h" + +namespace arrow { +namespace util { +namespace alp { + +// ---------------------------------------------------------------------- +// AlpCodec + +/// \class AlpCodec +/// \brief High-level interface for ALP compression +/// +/// AlpCodec is an interface for Adaptive Lossless floating-Point Compression +/// (ALP) (https://dl.acm.org/doi/10.1145/3626717). For encoding, it samples +/// the data and applies decimal compression (Alp) to floating point values. +/// This class acts as a wrapper around the vector-based interfaces of +/// AlpSampler and Alp. +/// +/// \tparam T the floating point type (float or double) +template <typename T> +class AlpCodec { + public: + /// Type alias for the sampler result containing encoding presets + using AlpSamplerResult = typename AlpSampler<T>::AlpSamplerResult; + + /// \brief Create a sampling preset from input data + /// + /// This samples the input data and generates an encoding preset that can be + /// reused for encoding. This is useful when you want to pre-compute the preset + /// outside of the benchmark loop or encode multiple batches with the same preset. + /// + /// \param[in] decomp pointer to the input data to sample + /// \param[in] decomp_size size of decomp in bytes. + /// This needs to be a multiple of sizeof(T). + /// \return the sampling result containing the encoding preset + static AlpSamplerResult CreateSamplingPreset(const T* decomp, size_t decomp_size); + + /// \brief Encode floating point values using a pre-computed preset + /// + /// This encodes the data using a preset that was previously computed via + /// CreateSamplingPreset(). This avoids the sampling overhead during encoding. + /// + /// \param[in] decomp pointer to the input that is to be encoded + /// \param[in] decomp_size size of decomp in bytes. + /// This needs to be a multiple of sizeof(T). + /// \param[out] comp pointer to the memory region we will encode into. + /// Must be at least GetMaxCompressedSize(decomp_size) bytes. + /// \param[in,out] comp_size the actual size of the encoded data in bytes, + /// expects the size of comp as input. If this is too small, + /// this is set to 0 and we bail out. + /// \param[in] preset the pre-computed sampling result from CreateSamplingPreset() + static void EncodeWithPreset(const T* decomp, size_t decomp_size, char* comp, + size_t* comp_size, const AlpSamplerResult& preset); + + /// \brief Encode floating point values using ALP decimal compression + /// + /// \param[in] decomp pointer to the input that is to be encoded + /// \param[in] decomp_size size of decomp in bytes. + /// This needs to be a multiple of sizeof(T). + /// \param[out] comp pointer to the memory region we will encode into. + /// Must be at least GetMaxCompressedSize(decomp_size) bytes. + /// \param[in,out] comp_size the actual size of the encoded data in bytes, + /// expects the size of comp as input. If this is too small, + /// this is set to 0 and we bail out. + /// \param[in] enforce_mode reserved for future use. + /// Currently only AlpMode::kAlp is supported. + static void Encode(const T* decomp, size_t decomp_size, char* comp, + size_t* comp_size, + std::optional<AlpMode> enforce_mode = std::nullopt); + + /// \brief Decode floating point values + /// + /// \param[in] num_elements number of elements to decode (from page header) + /// \param[in] comp pointer to the input that is to be decoded + /// \param[in] comp_size size of the input in bytes (from page header) + /// \param[out] decomp pointer to the memory region we will decode into. + /// The caller is responsible for ensuring this is big enough + /// to hold num_elements values. + /// \return Status::OK on success, or an error if the compressed data is malformed + /// \tparam TargetType the type that is used to store the output. + /// May not be a narrowing conversion from T. + template <typename TargetType> + static Status Decode(int32_t num_elements, const char* comp, size_t comp_size, + TargetType* decomp); + + /// \brief Get the maximum compressed size of an uncompressed buffer + /// + /// \param[in] uncompressed_size the size of the uncompressed buffer in bytes + /// \return the maximum size of the compressed buffer + static int64_t GetMaxCompressedSize(int64_t uncompressed_size); + + private: + struct AlpHeader; + + /// \brief Tracks the progress of a compression operation + /// + /// Used to report how much data was consumed and produced during encoding. + struct CompressionProgress { + /// Number of compressed bytes written to output + int64_t num_compressed_bytes_produced = 0; + /// Number of input elements consumed + int64_t num_uncompressed_elements_taken = 0; + }; + + /// \brief Tracks the progress of a decompression operation + /// + /// Used to report how much data was consumed and produced during decoding. + struct DecompressionProgress { + /// Number of decompressed elements written + int64_t num_decompressed_elements_produced = 0; + /// Number of compressed bytes consumed + int64_t num_compressed_bytes_taken = 0; + }; + + /// \brief Compress a buffer using the ALP variant + /// + /// \param[in] decomp array of floating point numbers to compress + /// \param[in] element_count the number of floating point numbers + /// \param[out] comp the buffer to be compressed into + /// \param[in] comp_size the size of the compression buffer + /// \param[in] combinations the encoding preset to use + /// \return the compression progress + static CompressionProgress EncodeAlp(const T* decomp, uint64_t element_count, + char* comp, size_t comp_size, Review Comment: Done — `Encode` and `Decode` both take the output last. `EncodeAlp` ends `uint8_t* output, int64_t output_size`, keeping the buffer with its capacity. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
