prtkgaur commented on code in PR #48345: URL: https://github.com/apache/arrow/pull/48345#discussion_r3953307161
########## 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); Review Comment: This one no longer applies: the header has no version field, and the function it was attached to is gone. Version negotiation is handled at the encoding level by the preview flag instead. ########## cpp/src/arrow/util/alp/alp.cc: ########## @@ -0,0 +1,1039 @@ +// 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.h" + +#include <cmath> +#include <cstring> +#include <functional> +#include <iostream> +#include <map> + +#include "arrow/util/alp/alp_constants.h" +#include "arrow/util/bit_stream_utils_internal.h" +#include "arrow/util/bit_util.h" +#include "arrow/util/endian.h" +#include "arrow/util/bpacking_internal.h" +#include "arrow/util/logging.h" +#include "arrow/util/small_vector.h" +#include "arrow/util/span.h" +#include "arrow/util/ubsan.h" + +namespace arrow { +namespace util { +namespace alp { + +// ALP serialization uses memcpy for multi-byte integers (frame_of_reference, +// num_exceptions, offsets) and assumes little-endian byte order on disk. +static_assert(ARROW_LITTLE_ENDIAN, + "ALP serialization assumes little-endian byte order"); + +// ---------------------------------------------------------------------- +// AlpEncodedVectorInfo implementation (non-templated, 4 bytes) + +void AlpEncodedVectorInfo::Store(arrow::util::span<char> output_buffer) const { + ARROW_CHECK(output_buffer.size() >= GetStoredSize()) + << "alp_vector_info_output_too_small: " << output_buffer.size() << " vs " + << GetStoredSize(); + + char* ptr = output_buffer.data(); + + // exponent, factor: 1 byte each + *ptr++ = static_cast<char>(exponent); + *ptr++ = static_cast<char>(factor); + + // num_exceptions: 2 bytes + std::memcpy(ptr, &num_exceptions, sizeof(num_exceptions)); +} + +AlpEncodedVectorInfo AlpEncodedVectorInfo::Load( + arrow::util::span<const char> input_buffer) { + ARROW_CHECK(input_buffer.size() >= GetStoredSize()) + << "alp_vector_info_input_too_small: " << input_buffer.size() << " vs " + << GetStoredSize(); + + AlpEncodedVectorInfo result{}; + const char* ptr = input_buffer.data(); + + // exponent, factor: 1 byte each + result.exponent = static_cast<uint8_t>(*ptr++); + result.factor = static_cast<uint8_t>(*ptr++); + + // num_exceptions: 2 bytes + std::memcpy(&result.num_exceptions, ptr, sizeof(result.num_exceptions)); Review Comment: Same change as the load thread — the stores use `SafeStore`. ########## cpp/src/arrow/util/alp/alp.cc: ########## @@ -0,0 +1,1039 @@ +// 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.h" + +#include <cmath> +#include <cstring> +#include <functional> +#include <iostream> +#include <map> + +#include "arrow/util/alp/alp_constants.h" +#include "arrow/util/bit_stream_utils_internal.h" +#include "arrow/util/bit_util.h" +#include "arrow/util/endian.h" +#include "arrow/util/bpacking_internal.h" +#include "arrow/util/logging.h" +#include "arrow/util/small_vector.h" +#include "arrow/util/span.h" +#include "arrow/util/ubsan.h" + +namespace arrow { +namespace util { +namespace alp { + +// ALP serialization uses memcpy for multi-byte integers (frame_of_reference, +// num_exceptions, offsets) and assumes little-endian byte order on disk. +static_assert(ARROW_LITTLE_ENDIAN, + "ALP serialization assumes little-endian byte order"); + +// ---------------------------------------------------------------------- +// AlpEncodedVectorInfo implementation (non-templated, 4 bytes) + +void AlpEncodedVectorInfo::Store(arrow::util::span<char> output_buffer) const { + ARROW_CHECK(output_buffer.size() >= GetStoredSize()) + << "alp_vector_info_output_too_small: " << output_buffer.size() << " vs " + << GetStoredSize(); + + char* ptr = output_buffer.data(); + + // exponent, factor: 1 byte each + *ptr++ = static_cast<char>(exponent); + *ptr++ = static_cast<char>(factor); + + // num_exceptions: 2 bytes + std::memcpy(ptr, &num_exceptions, sizeof(num_exceptions)); +} + +AlpEncodedVectorInfo AlpEncodedVectorInfo::Load( + arrow::util::span<const char> input_buffer) { + ARROW_CHECK(input_buffer.size() >= GetStoredSize()) + << "alp_vector_info_input_too_small: " << input_buffer.size() << " vs " + << GetStoredSize(); + + AlpEncodedVectorInfo result{}; + const char* ptr = input_buffer.data(); + + // exponent, factor: 1 byte each + result.exponent = static_cast<uint8_t>(*ptr++); + result.factor = static_cast<uint8_t>(*ptr++); + + // num_exceptions: 2 bytes + std::memcpy(&result.num_exceptions, ptr, sizeof(result.num_exceptions)); + + return result; +} + +// ---------------------------------------------------------------------- +// AlpEncodedForVectorInfo implementation (templated, 5/9 bytes) + +template <typename T> +void AlpEncodedForVectorInfo<T>::Store(arrow::util::span<char> output_buffer) const { + ARROW_CHECK(output_buffer.size() >= GetStoredSize()) + << "alp_for_vector_info_output_too_small: " << output_buffer.size() << " vs " + << GetStoredSize(); + + char* ptr = output_buffer.data(); + + // frame_of_reference: 4 bytes for float, 8 bytes for double + std::memcpy(ptr, &frame_of_reference, sizeof(frame_of_reference)); Review Comment: Both of these go through the helpers now — `SafeLoadAs` for the scalar reads and `SafeStore` for the writes, in every serialization path. `Load` also returns a `Result`, so the size check and the read sit together rather than the caller having to remember one of them. The `memcpy` calls still in the file are bulk copies of the bit-packed payload and the exception arrays, which the single-object helpers don't cover. ########## cpp/src/parquet/encoding_alp_benchmark.cc: ########## Review Comment: Done — the ALP benchmark file is out of this branch and will come back as its own PR. ########## 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; + } + + if (output_offset + this_vector_elements > decomp_element_count) { + return Status::Invalid("ALP decode output buffer too small: offset=", + output_offset, " + elements=", this_vector_elements, + " > capacity=", decomp_element_count); + } + + // Validate offset is within bounds and enough buffer remains for metadata + const uint64_t vector_offset = vector_offsets[vector_index]; + constexpr uint64_t kMinVectorMetadataSize = + AlpEncodedVectorInfo::kStoredSize + AlpEncodedForVectorInfo<T>::kStoredSize; + if (vector_offset + kMinVectorMetadataSize > comp_size) { + return Status::Invalid("ALP vector offset out of bounds or insufficient buffer " + "for metadata: offset=", vector_offset, + ", metadata_size=", kMinVectorMetadataSize, + ", buffer_size=", comp_size); + } + + // Jump directly to this vector using its offset + const char* vector_start = comp + vector_offset; + const uint64_t remaining = comp_size - vector_offset; + + // Read AlpInfo (interleaved) + const AlpEncodedVectorInfo alp_info = + AlpEncodedVectorInfo::Load({vector_start, remaining}); + const char* ptr = vector_start + AlpEncodedVectorInfo::kStoredSize; + + // Decode based on integer encoding type + switch (integer_encoding) { + case AlpIntegerEncoding::kForBitPack: { + // Read ForInfo (interleaved) + const AlpEncodedForVectorInfo<T> for_info = + AlpEncodedForVectorInfo<T>::Load( + {ptr, remaining - AlpEncodedVectorInfo::kStoredSize}); + ptr += AlpEncodedForVectorInfo<T>::kStoredSize; + + // Validate enough buffer remains for the data section + const uint64_t data_remaining = comp_size - static_cast<uint64_t>(ptr - comp); + const uint64_t data_size = + for_info.GetDataStoredSize(this_vector_elements, alp_info.num_exceptions); + if (data_size > data_remaining) { Review Comment: Agreed, and they have moved. Each metadata `Load` validates the buffer it was handed and returns a `Result`, so the bounds check for a field lives next to the field rather than in the caller, and the offset chain is validated once when the reader is opened. The wrapper layer these guards were duplicated in is gone. ########## cpp/src/arrow/util/alp/alp.h: ########## @@ -0,0 +1,843 @@ +// 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 <vector> + +#include "arrow/util/alp/alp_constants.h" +#include "arrow/util/small_vector.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_wrapper.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 }; + +// ---------------------------------------------------------------------- +// 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 | +/// +------------------------------------------+ +struct AlpEncodedVectorInfo { + /// Exponent used for decimal encoding (multiply by 10^exponent) + uint8_t exponent = 0; + /// Factor used for decimal encoding (divide by 10^factor) + uint8_t factor = 0; + /// Number of exceptions stored in this vector + uint16_t num_exceptions = 0; + + /// Size of the serialized portion (4 bytes, fixed) + static constexpr uint64_t kStoredSize = 4; + + /// \brief Store the ALP metadata into an output buffer + void Store(arrow::util::span<char> output_buffer) const; + + /// \brief Load ALP metadata from an input buffer + static AlpEncodedVectorInfo Load(arrow::util::span<const char> input_buffer); + + /// \brief Get serialized size of the ALP metadata + static uint64_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); } +}; + +// ---------------------------------------------------------------------- +// 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> +struct AlpEncodedForVectorInfo { + static_assert(std::is_same_v<T, float> || std::is_same_v<T, double>, + "AlpEncodedForVectorInfo only supports float and double"); + + /// Use uint32_t for float, uint64_t for double (matches encoded integer size) + using ExactType = typename AlpTypedConstants<T>::FloatingToExact; + + /// Delta used for frame of reference encoding (4 bytes for float, 8 for double) + ExactType frame_of_reference = 0; + /// Bitwidth used for bitpacking + uint8_t bit_width = 0; + + /// Size of the serialized portion (5 bytes for float, 9 for double) + static constexpr uint64_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] bit_width bits per element + /// \return the size in bytes of the bitpacked data + static uint64_t GetBitPackedSize(uint16_t num_elements, uint8_t bit_width) { + return (static_cast<uint64_t>(num_elements) * bit_width + 7) / 8; + } + + /// \brief Store the FOR metadata into an output buffer + void Store(arrow::util::span<char> output_buffer) const; + + /// \brief Load FOR metadata from an input buffer + static AlpEncodedForVectorInfo Load(arrow::util::span<const char> input_buffer); + + /// \brief Get serialized size of the FOR metadata + static uint64_t GetStoredSize() { return kStoredSize; } + + /// \brief Get the size of the data section (packed values + exceptions) + /// + /// \param[in] num_elements number of elements in this vector + /// \param[in] num_exceptions number of exceptions (from AlpEncodedVectorInfo) + /// \return the size in bytes of packed values + exception positions + exceptions + uint64_t GetDataStoredSize(uint16_t num_elements, uint16_t num_exceptions) const { + const uint64_t bit_packed_size = GetBitPackedSize(num_elements, bit_width); + return bit_packed_size + + num_exceptions * (sizeof(AlpConstants::PositionType) + sizeof(T)); + } + + bool operator==(const AlpEncodedForVectorInfo& other) const { + return frame_of_reference == other.frame_of_reference && + bit_width == other.bit_width; + } + + bool operator!=(const AlpEncodedForVectorInfo& other) const { return !(*this == other); } +}; + +// ---------------------------------------------------------------------- +// AlpEncodedVector + +/// \class AlpEncodedVector +/// \brief A compressed ALP vector with metadata +/// +/// Per-vector data layout: +/// +/// +------------------------------------------------------------+ +/// | AlpEncodedVector<T> Data Layout | +/// +------------------------------------------------------------+ +/// | Section | Size (bytes) | Description | +/// +-----------------------+----------------------+-------------+ +/// | 1. AlpInfo | 4B (fixed) | ALP meta | +/// +-----------------------+----------------------+-------------+ +/// | 2. ForInfo | 6B (float) or | FOR meta | +/// | | 10B (double) | | +/// +-----------------------+----------------------+-------------+ +/// | 3. Packed Values | bit_packed_size | Bitpacked | +/// | (compressed data) | (computed) | integers | +/// +-----------------------+----------------------+-------------+ +/// | 4. Exception Pos | num_exceptions * 2 | uint16_t[] | +/// | (indices) | (variable) | positions | +/// +-----------------------+----------------------+-------------+ +/// | 5. Exception Values | num_exceptions * | T[] (float/| +/// | (original floats) | sizeof(T) | double) | +/// +------------------------------------------------------------+ +/// +/// Page-level layout (offset-based interleaved for O(1) random access): +/// +/// +------------------------------------------------------------+ +/// | Page Layout | +/// +------------------------------------------------------------+ +/// | [Header (8B)] | +/// | [Offset₀ | Offset₁ | ... | Offsetₙ₋₁] ← Vector offsets | +/// | [Vector₀][Vector₁]...[Vectorₙ₋₁] ← Interleaved | +/// +------------------------------------------------------------+ +/// where each Vector = [AlpInfo | ForInfo | Data] +/// +/// The offset-based layout enables: +/// - O(1) random access to any vector via offset lookup +/// - Better locality for single-vector decompression +/// - Parallel decompression without coordination +/// +/// Example for 1024 floats with 5 exceptions and bit_width=8: +/// - AlpInfo: 4 bytes (fixed) +/// - ForInfo: 6 bytes (float) +/// - Packed Values: 1024 bytes (1024 * 8 bits / 8) +/// - Exception Pos: 10 bytes (5 * 2) +/// - Exception Values: 20 bytes (5 * 4) +/// Total: 1064 bytes +template <typename T> +class AlpEncodedVector { + public: + /// ALP-specific metadata (exponent, factor, num_exceptions) + AlpEncodedVectorInfo alp_info; + /// FOR-specific metadata (frame_of_reference, bit_width) + AlpEncodedForVectorInfo<T> for_info; + /// Number of elements in this vector (not serialized; from page header) + uint16_t num_elements = 0; + /// Successfully encoded and bitpacked data + arrow::internal::StaticVector<uint8_t, AlpConstants::kAlpVectorSize * sizeof(T)> + packed_values; + /// Float values that could not be converted successfully + arrow::internal::StaticVector<T, AlpConstants::kAlpVectorSize> exceptions; + /// Positions of the exceptions in the decompressed vector + arrow::internal::StaticVector<uint16_t, AlpConstants::kAlpVectorSize> exception_positions; + + /// Total metadata size (AlpInfo + ForInfo) + static constexpr uint64_t kMetadataStoredSize = + AlpEncodedVectorInfo::kStoredSize + AlpEncodedForVectorInfo<T>::kStoredSize; + + /// \brief Get the size of the vector if stored into a sequential memory block + /// + /// \return the stored size in bytes + uint64_t GetStoredSize() const; + + /// \brief Get the stored size for given metadata and element count + /// + /// \param[in] alp_info the ALP metadata + /// \param[in] for_info the FOR metadata + /// \param[in] num_elements the number of elements in this vector + /// \return the stored size in bytes + static uint64_t GetStoredSize(const AlpEncodedVectorInfo& alp_info, + const AlpEncodedForVectorInfo<T>& for_info, + uint16_t num_elements); + + /// \brief Get the number of elements in this vector + /// + /// \return number of elements + uint64_t GetNumElements() const { return num_elements; } + + /// \brief Store the compressed vector in a compact format into an output buffer + /// + /// Stores [AlpInfo][ForInfo][PackedValues][ExceptionPositions][ExceptionValues] + /// + /// \param[out] output_buffer the buffer to store the compressed data into + void Store(arrow::util::span<char> output_buffer) const; + + /// \brief Store only the data section (without metadata) into an output buffer + /// + /// Stores [PackedValues][ExceptionPositions][ExceptionValues] + /// Used when metadata (AlpInfo, ForInfo) is written separately. + /// + /// \param[out] output_buffer the buffer to store the data section into + void StoreDataOnly(arrow::util::span<char> output_buffer) const; + + /// \brief Get the size of the data section only (without metadata) + /// + /// \return the size in bytes of packed values + exception positions + exceptions + uint64_t GetDataStoredSize() const { + return for_info.GetDataStoredSize(num_elements, alp_info.num_exceptions); + } + + /// \brief Load a compressed vector from a compact format from an input buffer + /// + /// \param[in] input_buffer the buffer to load from + /// \param[in] num_elements the number of elements (from page header) + /// \return the loaded AlpEncodedVector + static AlpEncodedVector Load(arrow::util::span<const char> input_buffer, + uint16_t num_elements); + + bool operator==(const AlpEncodedVector<T>& other) const; +}; + +// ---------------------------------------------------------------------- +// AlpEncodedVectorView + +/// \class AlpEncodedVectorView +/// \brief A view into compressed ALP data optimized for decompression +/// +/// Unlike AlpEncodedVector which copies all data into internal buffers, +/// AlpEncodedVectorView uses zero-copy for the large packed values array +/// while copying the small exception arrays into aligned storage. +/// +/// The packed values are accessed via a span (zero-copy) since they are +/// byte arrays with no alignment requirements. Exception positions and +/// values are copied into aligned StaticVectors because: +/// 1. The serialized data may not be properly aligned for uint16_t/T access +/// 2. Exceptions are rare (typically < 5%), so copying is negligible +/// 3. This avoids undefined behavior from misaligned memory access +/// +/// Use LoadView() to create a view, then pass to DecompressVectorView(). +/// The underlying buffer must remain valid for the lifetime of the view +/// (for packed_values access). +template <typename T> +struct AlpEncodedVectorView { + /// ALP-specific metadata (exponent, factor, num_exceptions) + AlpEncodedVectorInfo alp_info; + /// FOR-specific metadata (frame_of_reference, bit_width) + AlpEncodedForVectorInfo<T> for_info; + /// Number of elements in this vector (not serialized; from page header) + uint16_t num_elements = 0; Review Comment: The spec carries it now: `log_vector_size` is normatively restricted to the inclusive range [3, 15] in the ALP encoding specification in parquet-format, so a reader is required to reject anything outside that range rather than meeting it as an overflow. The reason is that per-vector element counts are held in a `uint16`, which 2^16 does not fit. ########## cpp/src/arrow/util/alp/alp_codec.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) { Review Comment: I'd rather keep it. The merged spec reserves the byte for future variants such as ALP-RD and requires a reader to reject any value it does not recognize, so dropping it here would put the implementation out of step with the format. It also buys a clear failure today: a page written by a later variant is refused by name instead of being parsed as ALP and yielding wrong values, which is covered by a test. -- 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]
