prtkgaur commented on code in PR #48345: URL: https://github.com/apache/arrow/pull/48345#discussion_r3921143794
########## cpp/src/arrow/CMakeLists.txt: ########## @@ -654,6 +654,12 @@ if(ARROW_WITH_ZSTD) list(APPEND ARROW_UTIL_SRCS util/compression_zstd.cc) endif() +# ALP (for Parquet encoder/decoder) Review Comment: Added to `meson.build` too. The CMake side had a second problem worth mentioning: the ALP sources were in `ARROW_UTIL_SRCS` and also listed in the `alp-test` target, so the test compiled them a second time instead of linking the library's copies. They build once now, and the test is registered in both build systems. ########## cpp/src/arrow/util/alp/alp_codec.cc: ########## @@ -0,0 +1,583 @@ +// 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_codec.h" + +#include <bit> +#include <cmath> +#include <limits> + +#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/bit_util.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 int32_t to match Parquet page headers (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: 1 << log_vector_size (i.e., +/// 2^log_vector_size). For example, log_vector_size=10 means vector_size=1024. +/// LoadHeader rejects anything outside [kMinLogVectorSize, kMaxLogVectorSize], +/// so the representable vector sizes are 2^3 (8) through 2^15 (32768). +/// +/// 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 (int32) | +/// +---------------------------------------------------+ +/// +/// Page-level layout (offset-based interleaved for O(1) random access): +/// +/// +-------------------------------------------------------------------+ +/// | [AlpHeader (7B)] | +/// | [Offset₀ | Offset₁ | ... | Offsetₙ₋₁] ← Vector offsets | +/// | [Vector₀][Vector₁]...[Vectorₙ₋₁] ← Concatenated | +/// +-------------------------------------------------------------------+ +/// 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 = 1 << 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 (int32_t to match Parquet's i32 num_values). + /// Per-vector element count is inferred: vector_size for all but the last vector. + int32_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) + int32_t GetNumVectors() const { + const int32_t vector_size = GetVectorSize(); + return static_cast<int32_t>(::arrow::bit_util::CeilDiv(num_elements, vector_size)); + } + + /// \brief Get the size of the offsets section + /// + /// \return size in bytes of the offsets array (num_vectors * sizeof(OffsetType)) + int64_t GetOffsetsSectionSize() const { + return static_cast<int64_t>(GetNumVectors()) * sizeof(AlpConstants::OffsetType); + } + + /// \brief Compute the actual vector size from log_vector_size + /// + /// \return the vector size (2^log_vector_size) + int32_t GetVectorSize() const { return 1 << 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 + /// \pre value is positive and a power of two. Callers reach this only after + /// ValidateVectorSize has already rejected other inputs with + /// Status::Invalid, so a violation here is a programmer error rather + /// than malformed data; it is enforced with `ARROW_CHECK`, which aborts. + static uint8_t Log2(int32_t value) { + ARROW_CHECK(value > 0 && std::has_single_bit(static_cast<uint32_t>(value))) + << "value_must_be_power_of_2: " << value; + return static_cast<uint8_t>(std::countr_zero(static_cast<uint32_t>(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, or error if index is out of range + Result<int32_t> GetVectorNumElements(int32_t vector_index) const { + const int32_t vector_size = GetVectorSize(); + const int32_t num_full_vectors = num_elements / vector_size; + const int32_t remainder = num_elements % vector_size; + if (vector_index < num_full_vectors) { + return vector_size; // Full vector + } else if (vector_index == num_full_vectors && remainder > 0) { + return remainder; // Last partial vector + } + return Status::Invalid("ALP invalid vector index: ", vector_index, + " (num_vectors=", GetNumVectors(), ")"); + } + + /// \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> +Result<typename AlpCodec<T>::AlpHeader> AlpCodec<T>::LoadHeader(const uint8_t* input, + int64_t input_size) { + if (input_size < static_cast<int64_t>(AlpHeader::kSize)) { + return Status::Invalid("ALP compressed buffer too small for header: ", input_size, + " < ", AlpHeader::kSize); + } + AlpHeader header{}; + header.compression_mode = util::SafeLoadAs<uint8_t>(input); + header.integer_encoding = util::SafeLoadAs<uint8_t>(input + 1); + header.log_vector_size = util::SafeLoadAs<uint8_t>(input + 2); + header.num_elements = util::SafeLoadAs<int32_t>(input + 3); + + if (header.compression_mode != static_cast<uint8_t>(AlpMode::kAlp)) { + return Status::Invalid("ALP unsupported compression mode: ", + static_cast<int>(header.compression_mode)); + } + if (header.integer_encoding != static_cast<uint8_t>(AlpIntegerEncoding::kForBitPack)) { + return Status::Invalid("ALP unsupported integer encoding: ", + static_cast<int>(header.integer_encoding)); + } + if (header.log_vector_size < AlpConstants::kMinLogVectorSize || + header.log_vector_size > AlpConstants::kMaxLogVectorSize) { + return Status::Invalid( + "ALP invalid log_vector_size: ", static_cast<int>(header.log_vector_size), + " (must be in [", static_cast<int>(AlpConstants::kMinLogVectorSize), ", ", + static_cast<int>(AlpConstants::kMaxLogVectorSize), "])"); + } + if (header.num_elements < 0) { + return Status::Invalid("ALP invalid num_elements: ", header.num_elements); + } + return header; +} + +template <typename T> +Result<typename AlpCodec<T>::AlpSamplerResult> AlpCodec<T>::CreateSamplingPreset( + const T* input, int64_t num_elements) { + if (num_elements < 0) { + return Status::Invalid("ALP num_elements must be non-negative, got ", num_elements); + } + + AlpSampler<T> sampler; + sampler.AddSample({input, static_cast<size_t>(num_elements)}); + return sampler.Finalize(); +} + +namespace { + +/// \brief Validate a caller-supplied vector_size against the format spec +/// +/// The spec constrains `log_vector_size` to the inclusive range +/// [kMinLogVectorSize, kMaxLogVectorSize], so the vector size must be a power +/// of two within [2^3, 2^15]. +Status ValidateVectorSize(int32_t vector_size) { + constexpr int32_t kMin = 1 << AlpConstants::kMinLogVectorSize; + constexpr int32_t kMax = 1 << AlpConstants::kMaxLogVectorSize; + if (vector_size <= 0 || !std::has_single_bit(static_cast<uint32_t>(vector_size))) { + return Status::Invalid("ALP vector_size must be a positive power of 2, got ", + vector_size); + } + if (vector_size < kMin || vector_size > kMax) { + return Status::Invalid("ALP vector_size must be in [", kMin, ", ", kMax, "], got ", + vector_size); + } + return Status::OK(); +} + +} // namespace + +template <typename T> +Status AlpCodec<T>::EncodeWithPreset(const T* input, int64_t num_elements, + const AlpSamplerResult& preset, int32_t vector_size, + uint8_t* output, int64_t* output_size) { + if (num_elements < 0) { + return Status::Invalid("ALP num_elements must be non-negative, got ", num_elements); + } + if (num_elements > std::numeric_limits<int32_t>::max()) { + return Status::Invalid("ALP num_elements exceeds INT32_MAX, got ", num_elements); + } + RETURN_NOT_OK(ValidateVectorSize(vector_size)); + + // Make room to store header afterwards. + uint8_t* encoded_header = output; + uint8_t* body = output + AlpHeader::kSize; + const int64_t remaining_output_size = + *output_size - static_cast<int64_t>(AlpHeader::kSize); + + const CompressionProgress compression_progress = + EncodeAlp(input, num_elements, preset.alp_parameters, vector_size, body, + remaining_output_size); + + 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(vector_size); + header.num_elements = static_cast<int32_t>(num_elements); + + util::SafeStore(encoded_header + 0, header.compression_mode); + util::SafeStore(encoded_header + 1, header.integer_encoding); + util::SafeStore(encoded_header + 2, header.log_vector_size); + util::SafeStore(encoded_header + 3, header.num_elements); + *output_size = static_cast<int64_t>(AlpHeader::kSize) + + compression_progress.num_compressed_bytes_produced; + return Status::OK(); +} + +template <typename T> +Status AlpCodec<T>::Encode(const T* input, int64_t num_elements, int32_t vector_size, + uint8_t* output, int64_t* output_size) { + ARROW_ASSIGN_OR_RAISE(auto sampling_result, CreateSamplingPreset(input, num_elements)); + return EncodeWithPreset(input, num_elements, sampling_result, vector_size, output, + output_size); +} + +template <typename T> +Status AlpCodec<T>::Encode(const T* input, int64_t num_elements, uint8_t* output, + int64_t* output_size) { + return Encode(input, num_elements, AlpConstants::kAlpVectorSize, output, output_size); +} + +template <typename T> +template <typename TargetType> +Status AlpCodec<T>::Decode(int32_t num_elements, const uint8_t* input, int64_t input_size, + TargetType* output) { + ARROW_ASSIGN_OR_RAISE(const AlpHeader header, LoadHeader(input, input_size)); + const int32_t vector_size = header.GetVectorSize(); + + const uint8_t* body = input + AlpHeader::kSize; + const int64_t body_size = input_size - static_cast<int64_t>(AlpHeader::kSize); + + ARROW_RETURN_NOT_OK(DecodeAlp<TargetType>(num_elements, body, body_size, Review Comment: Required now — `Decode` fails when the two disagree in either direction, since a smaller header count leaves part of `output` unwritten and a larger one overruns it. `AlpRobustnessTest.HeaderElementCountMismatch` drives both sides, including the zero case you called out. The decoder passing the wrong count in the first place was the other half of this; that's your `SetData` comment below. ########## cpp/src/arrow/util/alp/alp_test.cc: ########## @@ -0,0 +1,1912 @@ +// 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 <cmath> +#include <cstdint> +#include <cstring> +#include <random> +#include <string> +#include <vector> + +#include <gtest/gtest.h> + +#include "arrow/testing/gtest_util.h" +#include "arrow/util/alp/alp.h" +#include "arrow/util/alp/alp_codec.h" +#include "arrow/util/alp/alp_constants.h" +#include "arrow/util/alp/alp_sampler.h" +#include "arrow/util/bit_stream_utils_internal.h" +#include "arrow/util/bit_util.h" +#include "arrow/util/bpacking_internal.h" + +namespace arrow { +namespace util { +namespace alp { + +// ============================================================================ +// Test helpers +// ============================================================================ + +// Compares two floating-point ranges by bit pattern, not by operator==. +// ALP is a lossless codec, so its tests must verify bit-exact recovery: +// `0.0 == -0.0` (different bits) and `NaN != NaN` (identical bits) make +// `EXPECT_THAT(out, ElementsAreArray(in))` the wrong check here. On +// mismatch the failure message names the index and prints both the +// value and the underlying hex bits. +template <typename T> +::testing::AssertionResult IsBitwiseEqual(const std::vector<T>& actual, + const std::vector<T>& expected) { + static_assert(std::is_floating_point<T>::value, + "IsBitwiseEqual is for float/double only"); + using Bits = typename std::conditional<sizeof(T) == 4, uint32_t, uint64_t>::type; + if (actual.size() != expected.size()) { + return ::testing::AssertionFailure() << "size mismatch: actual=" << actual.size() + << " expected=" << expected.size(); + } + for (size_t i = 0; i < actual.size(); ++i) { + Bits a_bits = 0, e_bits = 0; + std::memcpy(&a_bits, &actual[i], sizeof(T)); + std::memcpy(&e_bits, &expected[i], sizeof(T)); + if (a_bits != e_bits) { + return ::testing::AssertionFailure() + << "bit-mismatch at index " << i << ": actual=" << actual[i] << " (bits 0x" + << std::hex << a_bits << "), expected=" << std::dec << expected[i] + << " (bits 0x" << std::hex << e_bits << ")"; + } + } + return ::testing::AssertionSuccess(); +} + +// ============================================================================ +// ALP Constants Tests +// ============================================================================ + +TEST(AlpConstantsTest, SamplerConstants) { + EXPECT_GT(AlpConstants::kSamplerVectorSize, 0); + EXPECT_GT(AlpConstants::kSamplerRowgroupSize, 0); + EXPECT_GT(AlpConstants::kSamplerSamplesPerVector, 0); +} + +// ============================================================================ +// AlpIntegerEncoding Tests +// ============================================================================ + +TEST(AlpIntegerEncodingTest, GetIntegerEncodingMetadataSize) { + // Verify helper returns correct sizes for kForBitPack + EXPECT_EQ(GetIntegerEncodingMetadataSize<float>(AlpIntegerEncoding::kForBitPack), + AlpEncodedForVectorInfo<float>::kStoredSize); + EXPECT_EQ(GetIntegerEncodingMetadataSize<double>(AlpIntegerEncoding::kForBitPack), + AlpEncodedForVectorInfo<double>::kStoredSize); + + // Verify actual byte sizes (frame_of_reference + bit_width, no reserved) + EXPECT_EQ(GetIntegerEncodingMetadataSize<float>(AlpIntegerEncoding::kForBitPack), 5); + EXPECT_EQ(GetIntegerEncodingMetadataSize<double>(AlpIntegerEncoding::kForBitPack), 9); +} + +// ============================================================================ +// ALP Compression Tests +// ============================================================================ + +template <typename T> +class AlpCompressionTest : public ::testing::Test { + protected: + void TestCompressDecompress(const std::vector<T>& input) { + AlpCompression<T> compressor; + + // Compress + AlpEncodingParameters preset{}; // Default preset + auto encoded = compressor.CompressVector(input.data(), input.size(), preset); + + // Decompress + std::vector<T> output(input.size()); + compressor.DecompressVector(encoded, AlpIntegerEncoding::kForBitPack, output.data()); + + // Verify bit-exact recovery (important for -0.0, NaN; see IsBitwiseEqual). + ASSERT_EQ(output.size(), input.size()); + EXPECT_TRUE(IsBitwiseEqual(output, input)); + } +}; + +using CompressionTestTypes = ::testing::Types<float, double>; +TYPED_TEST_SUITE(AlpCompressionTest, CompressionTestTypes); + +TYPED_TEST(AlpCompressionTest, SimpleSequence) { + std::vector<TypeParam> input(64); + for (size_t i = 0; i < input.size(); ++i) { + input[i] = static_cast<TypeParam>(i + 1); + } + this->TestCompressDecompress(input); +} + +TYPED_TEST(AlpCompressionTest, DecimalValues) { + std::vector<TypeParam> input(64); + for (size_t i = 0; i < input.size(); ++i) { + input[i] = static_cast<TypeParam>(i) + static_cast<TypeParam>(0.5); + } + this->TestCompressDecompress(input); +} + +TYPED_TEST(AlpCompressionTest, SmallValues) { + std::vector<TypeParam> input(64); + for (size_t i = 0; i < input.size(); ++i) { + input[i] = static_cast<TypeParam>(0.001) * static_cast<TypeParam>(i + 1); + } + this->TestCompressDecompress(input); +} + +TYPED_TEST(AlpCompressionTest, MixedValues) { + // Exact binary fractions, so the values themselves are representable in both + // float and double and any loss would come from the codec, not the literals. + std::vector<TypeParam> input = { + static_cast<TypeParam>(100.5), static_cast<TypeParam>(200.25), + static_cast<TypeParam>(300.125), static_cast<TypeParam>(400.0625), + static_cast<TypeParam>(500.03125), static_cast<TypeParam>(600.015625), + static_cast<TypeParam>(700.0078125), static_cast<TypeParam>(800.00390625)}; + this->TestCompressDecompress(input); +} + +TYPED_TEST(AlpCompressionTest, RandomValues) { + std::mt19937 rng(42); + std::uniform_real_distribution<TypeParam> dist(static_cast<TypeParam>(0.0), + static_cast<TypeParam>(1000.0)); + + std::vector<TypeParam> input(64); + for (auto& v : input) { + v = dist(rng); + } + + this->TestCompressDecompress(input); +} + +TYPED_TEST(AlpCompressionTest, HighPrecision) { + // More decimal digits than float can hold, so for float these values arrive + // already rounded and mostly become exceptions; the codec must still return + // them bit for bit. + std::vector<TypeParam> input(64); + for (size_t i = 0; i < input.size(); ++i) { + input[i] = static_cast<TypeParam>(1.123456789) * static_cast<TypeParam>(i + 1); + } + this->TestCompressDecompress(input); +} + +TYPED_TEST(AlpCompressionTest, VerySmallValues) { + std::vector<TypeParam> input(64); + for (size_t i = 0; i < input.size(); ++i) { + input[i] = static_cast<TypeParam>(1e-10) * static_cast<TypeParam>(i + 1); + } + this->TestCompressDecompress(input); +} + +// ============================================================================ +// Integration Tests +// ============================================================================ + +// Cover both float and double so we don't drift apart between the two +// instantiations, and exercise a wide range plus a few extreme values so the +// test stresses the ALP-encodable / exception fallback boundary, not just the +// happy path. +template <typename T> +class AlpIntegrationTest : public ::testing::Test {}; + +using IntegrationTestTypes = ::testing::Types<float, double>; +TYPED_TEST_SUITE(AlpIntegrationTest, IntegrationTestTypes); + +TYPED_TEST(AlpIntegrationTest, RandomAndExtremes) { + std::mt19937 rng(12345); + + // Wide uniform spread: 10^-20 .. 10^20. ALP's encoder formula is + // `int64(v * 10^e * 10^-f)`, so values outside the range where that fits + // in int64 must fall through to the exception path. Lossless recovery + // here is exactly the contract being tested. + std::uniform_real_distribution<TypeParam> dist(static_cast<TypeParam>(-1e20), + static_cast<TypeParam>(1e20)); + + std::vector<TypeParam> input(1024); + for (auto& v : input) { + v = dist(rng); + } + + // Sprinkle in extreme/boundary values that the uniform RNG won't generate. + // These should round-trip via the exception path. + const std::array<TypeParam, 10> extremes = { + std::numeric_limits<TypeParam>::lowest(), + std::numeric_limits<TypeParam>::max(), + std::numeric_limits<TypeParam>::min(), // smallest normal + std::numeric_limits<TypeParam>::denorm_min(), // smallest subnormal + static_cast<TypeParam>(0.0), + static_cast<TypeParam>(-0.0), + static_cast<TypeParam>(1e-30), + static_cast<TypeParam>(-1e30), + static_cast<TypeParam>(1.234567890123456789), + static_cast<TypeParam>(0.1) + static_cast<TypeParam>(0.2), // classic float trap + }; + for (size_t i = 0; i < extremes.size(); ++i) { + input[i * 64] = extremes[i]; + } + + AlpCompression<TypeParam> compressor; + AlpEncodingParameters preset{}; + auto encoded = compressor.CompressVector(input.data(), input.size(), preset); + + std::vector<TypeParam> output(input.size()); + compressor.DecompressVector(encoded, AlpIntegerEncoding::kForBitPack, output.data()); + + // Bit-exact: lossless contract holds for everything, including the + // extremes that went through the exception path. + EXPECT_TRUE(IsBitwiseEqual(output, input)); +} + +// ============================================================================ +// AlpEncodedVectorInfo Serialization Tests +// ============================================================================ + +TEST(AlpEncodedVectorInfoTest, StoreLoadRoundTrip) { + // Test AlpEncodedVectorInfo (4 bytes) + AlpEncodedVectorInfo info{}; + info.set_exponent(5); + info.set_factor(3); + info.set_num_exceptions(10); + + std::vector<uint8_t> buffer(AlpEncodedVectorInfo::kStoredSize + 10); + info.Store({buffer.data(), buffer.size()}); + + ASSERT_OK_AND_ASSIGN(AlpEncodedVectorInfo loaded, + AlpEncodedVectorInfo::Load({buffer.data(), buffer.size()})); + EXPECT_EQ(info, loaded); + EXPECT_EQ(loaded.exponent(), 5); + EXPECT_EQ(loaded.factor(), 3); + EXPECT_EQ(loaded.num_exceptions(), 10); +} + +TEST(AlpEncodedForVectorInfoTest, StoreLoadRoundTripFloat) { + // Test AlpEncodedForVectorInfo<float> (6 bytes) + AlpEncodedForVectorInfo<float> info{}; + info.set_frame_of_reference(0x12345678U); + info.set_bit_width(12); + + std::vector<uint8_t> buffer(AlpEncodedForVectorInfo<float>::kStoredSize + 10); + info.Store({buffer.data(), buffer.size()}); + + ASSERT_OK_AND_ASSIGN( + AlpEncodedForVectorInfo<float> loaded, + AlpEncodedForVectorInfo<float>::Load({buffer.data(), buffer.size()})); + EXPECT_EQ(info, loaded); + EXPECT_EQ(loaded.frame_of_reference(), 0x12345678U); + EXPECT_EQ(loaded.bit_width(), 12); +} + +TEST(AlpEncodedForVectorInfoTest, StoreLoadRoundTripDouble) { + // Test AlpEncodedForVectorInfo<double> (10 bytes) + AlpEncodedForVectorInfo<double> info{}; + info.set_frame_of_reference(0x123456789ABCDEF0ULL); + info.set_bit_width(20); + + std::vector<uint8_t> buffer(AlpEncodedForVectorInfo<double>::kStoredSize + 10); + info.Store({buffer.data(), buffer.size()}); + + ASSERT_OK_AND_ASSIGN( + AlpEncodedForVectorInfo<double> loaded, + AlpEncodedForVectorInfo<double>::Load({buffer.data(), buffer.size()})); + EXPECT_EQ(info, loaded); + EXPECT_EQ(loaded.frame_of_reference(), 0x123456789ABCDEF0ULL); + EXPECT_EQ(loaded.bit_width(), 20); +} + +TEST(AlpEncodedVectorInfoTest, Size) { + // AlpEncodedVectorInfo is fixed at 4 bytes + EXPECT_EQ(AlpEncodedVectorInfo::kStoredSize, 4); + EXPECT_EQ(AlpEncodedVectorInfo::GetStoredSize(), 4); +} + +TEST(AlpEncodedForVectorInfoTest, Size) { + // AlpEncodedForVectorInfo: float=5 bytes, double=9 bytes + // (frame_of_reference is 4 bytes for float, 8 bytes for double, + 1 byte for bit_width) + EXPECT_EQ(AlpEncodedForVectorInfo<float>::kStoredSize, 5); + EXPECT_EQ(AlpEncodedForVectorInfo<float>::GetStoredSize(), 5); + EXPECT_EQ(AlpEncodedForVectorInfo<double>::kStoredSize, 9); + EXPECT_EQ(AlpEncodedForVectorInfo<double>::GetStoredSize(), 9); +} + +// ============================================================================ +// Edge Case Tests +// ============================================================================ + +template <typename T> +class AlpEdgeCaseTest : public ::testing::Test { + protected: + void TestCompressDecompress(const std::vector<T>& input) { + AlpCompression<T> compressor; + AlpEncodingParameters preset{}; + auto encoded = compressor.CompressVector(input.data(), input.size(), preset); + + std::vector<T> output(input.size()); + compressor.DecompressVector(encoded, AlpIntegerEncoding::kForBitPack, output.data()); + + ASSERT_EQ(output.size(), input.size()); + // Verify bit-exact recovery (important for -0.0, NaN; see IsBitwiseEqual). + EXPECT_TRUE(IsBitwiseEqual(output, input)); + } +}; + +using EdgeCaseTestTypes = ::testing::Types<float, double>; +TYPED_TEST_SUITE(AlpEdgeCaseTest, EdgeCaseTestTypes); + +TYPED_TEST(AlpEdgeCaseTest, SingleElement) { + std::vector<TypeParam> input = {static_cast<TypeParam>(42.5)}; + this->TestCompressDecompress(input); +} + +TYPED_TEST(AlpEdgeCaseTest, EmptyInput) { + // Test zero elements - empty vector + // The wrapper API requires decomp_size to be a multiple of sizeof(T), + // and 0 is a valid multiple. This tests the boundary condition. + std::vector<TypeParam> input; + + int64_t max_size = AlpCodec<TypeParam>::GetMaxCompressedSize(0).ValueOrDie(); + std::vector<uint8_t> buffer(max_size > 0 ? max_size : 8); // Ensure some buffer + int64_t comp_size = static_cast<int64_t>(buffer.size()); + + ASSERT_OK(AlpCodec<TypeParam>::Encode(input.data(), 0, buffer.data(), &comp_size)); + + // Decode zero elements + std::vector<TypeParam> output; + ASSERT_OK(AlpCodec<TypeParam>::template Decode<TypeParam>(0, buffer.data(), comp_size, + output.data())); + + // Both should be empty + EXPECT_EQ(input.size(), output.size()); + EXPECT_EQ(input.size(), 0); +} + +TYPED_TEST(AlpEdgeCaseTest, TwoElements) { + std::vector<TypeParam> input = {static_cast<TypeParam>(1.5), + static_cast<TypeParam>(2.5)}; + this->TestCompressDecompress(input); +} + +TYPED_TEST(AlpEdgeCaseTest, ExactVectorSize) { + // Test exactly kAlpVectorSize elements (1024) + std::vector<TypeParam> input(AlpConstants::kAlpVectorSize); + for (size_t i = 0; i < input.size(); ++i) { + input[i] = static_cast<TypeParam>(i) * static_cast<TypeParam>(0.1); + } + this->TestCompressDecompress(input); +} + +TYPED_TEST(AlpEdgeCaseTest, JustUnderVectorSize) { + // Test kAlpVectorSize - 1 elements (1023) + std::vector<TypeParam> input(AlpConstants::kAlpVectorSize - 1); + for (size_t i = 0; i < input.size(); ++i) { + input[i] = static_cast<TypeParam>(i) * static_cast<TypeParam>(0.1); + } + this->TestCompressDecompress(input); +} + +TYPED_TEST(AlpEdgeCaseTest, JustOverVectorSize) { + // Test kAlpVectorSize + 1 elements (1025) - requires multiple vectors + std::vector<TypeParam> input(AlpConstants::kAlpVectorSize + 1); + for (size_t i = 0; i < input.size(); ++i) { + input[i] = static_cast<TypeParam>(i) * static_cast<TypeParam>(0.1); + } + // For multi-vector, we need to process in chunks + AlpCompression<TypeParam> compressor; + AlpEncodingParameters preset{}; + + // Process first vector + auto encoded1 = + compressor.CompressVector(input.data(), AlpConstants::kAlpVectorSize, preset); + std::vector<TypeParam> output1(AlpConstants::kAlpVectorSize); + compressor.DecompressVector(encoded1, AlpIntegerEncoding::kForBitPack, output1.data()); + + // Process remaining element + auto encoded2 = + compressor.CompressVector(input.data() + AlpConstants::kAlpVectorSize, 1, preset); + std::vector<TypeParam> output2(1); + compressor.DecompressVector(encoded2, AlpIntegerEncoding::kForBitPack, output2.data()); + + // Verify (first vector covers input[0:kAlpVectorSize], second covers the + // trailing element). + EXPECT_TRUE(IsBitwiseEqual( + output1, std::vector<TypeParam>(input.begin(), + input.begin() + AlpConstants::kAlpVectorSize))); + EXPECT_TRUE(IsBitwiseEqual( + output2, std::vector<TypeParam>{input[AlpConstants::kAlpVectorSize]})); +} + +// ============================================================================ +// Special Values Tests +// ============================================================================ + +TYPED_TEST(AlpEdgeCaseTest, SpecialValues) { + // Test NaN, Inf, -Inf, -0.0 + std::vector<TypeParam> input = { + static_cast<TypeParam>(0.0), + static_cast<TypeParam>(-0.0), + std::numeric_limits<TypeParam>::infinity(), + -std::numeric_limits<TypeParam>::infinity(), + std::numeric_limits<TypeParam>::quiet_NaN(), + }; + this->TestCompressDecompress(input); +} + +TYPED_TEST(AlpEdgeCaseTest, NegativeZero) { + // -0.0 should be preserved bit-exactly + std::vector<TypeParam> input(100); + for (size_t i = 0; i < input.size(); ++i) { + input[i] = (i % 2 == 0) ? static_cast<TypeParam>(0.0) : static_cast<TypeParam>(-0.0); + } + this->TestCompressDecompress(input); +} + +TYPED_TEST(AlpEdgeCaseTest, AllNaN) { + // All NaN values - all become exceptions + std::vector<TypeParam> input(64); + for (auto& v : input) { + v = std::numeric_limits<TypeParam>::quiet_NaN(); + } + this->TestCompressDecompress(input); +} + +TYPED_TEST(AlpEdgeCaseTest, AllInfinity) { + // All infinity values + std::vector<TypeParam> input(64); + for (size_t i = 0; i < input.size(); ++i) { + input[i] = (i % 2 == 0) ? std::numeric_limits<TypeParam>::infinity() + : -std::numeric_limits<TypeParam>::infinity(); + } + this->TestCompressDecompress(input); +} + +// ============================================================================ +// Compression Characteristics Tests +// ============================================================================ + +TYPED_TEST(AlpEdgeCaseTest, ConstantValues) { + // All same values - should compress very well (bitWidth = 0) + std::vector<TypeParam> input(1024); + std::fill(input.begin(), input.end(), static_cast<TypeParam>(123.456)); + this->TestCompressDecompress(input); +} + +TYPED_TEST(AlpEdgeCaseTest, MixedCompressibleAndExceptions) { + // Mix of compressible decimals and exceptions + std::vector<TypeParam> input(1024); + for (size_t i = 0; i < input.size(); ++i) { + if (i % 10 == 0) { + input[i] = std::numeric_limits<TypeParam>::quiet_NaN(); + } else if (i % 20 == 5) { + input[i] = std::numeric_limits<TypeParam>::infinity(); + } else { + input[i] = static_cast<TypeParam>(i) * static_cast<TypeParam>(0.01); + } + } + this->TestCompressDecompress(input); +} + +// ============================================================================ +// Boundary Value Tests +// ============================================================================ + +TYPED_TEST(AlpEdgeCaseTest, MaxMinValues) { + std::vector<TypeParam> input = {std::numeric_limits<TypeParam>::max(), + std::numeric_limits<TypeParam>::min(), + std::numeric_limits<TypeParam>::lowest(), + std::numeric_limits<TypeParam>::denorm_min(), + std::numeric_limits<TypeParam>::epsilon(), + -std::numeric_limits<TypeParam>::max(), + -std::numeric_limits<TypeParam>::min(), + -std::numeric_limits<TypeParam>::denorm_min(), + -std::numeric_limits<TypeParam>::epsilon(), + static_cast<TypeParam>(0.0)}; + this->TestCompressDecompress(input); +} + +TYPED_TEST(AlpEdgeCaseTest, Subnormals) { + // Test subnormal (denormalized) floating point values + std::vector<TypeParam> input(100); + TypeParam subnormal = std::numeric_limits<TypeParam>::denorm_min(); + for (size_t i = 0; i < input.size(); ++i) { + input[i] = subnormal * static_cast<TypeParam>(i + 1); + } + this->TestCompressDecompress(input); +} + +TYPED_TEST(AlpEdgeCaseTest, LargeDecimals) { + // Test large decimal values that should still be compressible + std::vector<TypeParam> input(1024); + for (size_t i = 0; i < input.size(); ++i) { + input[i] = static_cast<TypeParam>(1000000.0) + + static_cast<TypeParam>(i) * static_cast<TypeParam>(0.01); + } + this->TestCompressDecompress(input); +} + +TYPED_TEST(AlpEdgeCaseTest, SmallDecimals) { + // Test very small decimal values + std::vector<TypeParam> input(1024); + for (size_t i = 0; i < input.size(); ++i) { + input[i] = static_cast<TypeParam>(0.000001) * static_cast<TypeParam>(i + 1); + } + this->TestCompressDecompress(input); +} + +TYPED_TEST(AlpEdgeCaseTest, NegativeValues) { + // Test negative values + std::vector<TypeParam> input(1024); + for (size_t i = 0; i < input.size(); ++i) { + input[i] = -static_cast<TypeParam>(i) * static_cast<TypeParam>(0.5); + } + this->TestCompressDecompress(input); +} + +TYPED_TEST(AlpEdgeCaseTest, AlternatingSignValues) { + // Test values alternating between positive and negative + std::vector<TypeParam> input(1024); + for (size_t i = 0; i < input.size(); ++i) { + TypeParam sign = + (i % 2 == 0) ? static_cast<TypeParam>(1.0) : static_cast<TypeParam>(-1.0); + input[i] = sign * static_cast<TypeParam>(i) * static_cast<TypeParam>(0.1); + } + this->TestCompressDecompress(input); +} + +// ============================================================================ +// AlpEncodedVector Store/Load Tests +// ============================================================================ + +template <typename T> +class AlpEncodedVectorTest : public ::testing::Test {}; + +TYPED_TEST_SUITE(AlpEncodedVectorTest, EdgeCaseTestTypes); + +TYPED_TEST(AlpEncodedVectorTest, StoreLoadRoundTrip) { + // Create a sample encoded vector + AlpCompression<TypeParam> compressor; + AlpEncodingParameters preset{}; + + std::vector<TypeParam> input(64); + for (size_t i = 0; i < input.size(); ++i) { + input[i] = static_cast<TypeParam>(i) * static_cast<TypeParam>(0.5); + } + + auto encoded = compressor.CompressVector(input.data(), input.size(), preset); + + // Store + std::vector<uint8_t> buffer(encoded.GetStoredSize()); + encoded.Store({buffer.data(), buffer.size()}); + + // Load (pass num_elements since it's not stored in the buffer) + ASSERT_OK_AND_ASSIGN(auto loaded, AlpEncodedVector<TypeParam>::Load( + {buffer.data(), buffer.size()}, + static_cast<uint16_t>(input.size()))); + + // Verify metadata + EXPECT_EQ(encoded.alp_info(), loaded.alp_info()); + EXPECT_EQ(encoded.for_info(), loaded.for_info()); + + // Decompress loaded and verify + std::vector<TypeParam> output(input.size()); + compressor.DecompressVector(loaded, AlpIntegerEncoding::kForBitPack, output.data()); + + EXPECT_TRUE(IsBitwiseEqual(output, input)); +} + +TYPED_TEST(AlpEncodedVectorTest, GetStoredSizeConsistency) { + AlpCompression<TypeParam> compressor; + AlpEncodingParameters preset{}; + + std::vector<TypeParam> input(128); + for (size_t i = 0; i < input.size(); ++i) { + input[i] = static_cast<TypeParam>(i) * static_cast<TypeParam>(0.25); + } + + auto encoded = compressor.CompressVector(input.data(), input.size(), preset); + + // Verify GetStoredSize matches actual storage + std::vector<uint8_t> buffer(encoded.GetStoredSize()); Review Comment: Right, it was circular. `GetStoredSizeConsistency` now pads the buffer with guard bytes past the reported size and asserts `Store` left them alone, then loads the vector back from exactly that many bytes, compares both info structs and checks the decompressed values bitwise — so a wrong offset or width fails it. The input has a NaN exception every fourth value, so exception positions and values both contribute and the size isn't trivially the same number. ########## cpp/src/arrow/util/alp/alp_codec.cc: ########## @@ -0,0 +1,583 @@ +// 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_codec.h" + +#include <bit> +#include <cmath> +#include <limits> + +#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/bit_util.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 int32_t to match Parquet page headers (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: 1 << log_vector_size (i.e., +/// 2^log_vector_size). For example, log_vector_size=10 means vector_size=1024. +/// LoadHeader rejects anything outside [kMinLogVectorSize, kMaxLogVectorSize], +/// so the representable vector sizes are 2^3 (8) through 2^15 (32768). +/// +/// 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 (int32) | +/// +---------------------------------------------------+ +/// +/// Page-level layout (offset-based interleaved for O(1) random access): +/// +/// +-------------------------------------------------------------------+ +/// | [AlpHeader (7B)] | +/// | [Offset₀ | Offset₁ | ... | Offsetₙ₋₁] ← Vector offsets | +/// | [Vector₀][Vector₁]...[Vectorₙ₋₁] ← Concatenated | +/// +-------------------------------------------------------------------+ +/// 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 = 1 << 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 (int32_t to match Parquet's i32 num_values). + /// Per-vector element count is inferred: vector_size for all but the last vector. + int32_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) + int32_t GetNumVectors() const { + const int32_t vector_size = GetVectorSize(); + return static_cast<int32_t>(::arrow::bit_util::CeilDiv(num_elements, vector_size)); + } + + /// \brief Get the size of the offsets section + /// + /// \return size in bytes of the offsets array (num_vectors * sizeof(OffsetType)) + int64_t GetOffsetsSectionSize() const { + return static_cast<int64_t>(GetNumVectors()) * sizeof(AlpConstants::OffsetType); + } + + /// \brief Compute the actual vector size from log_vector_size + /// + /// \return the vector size (2^log_vector_size) + int32_t GetVectorSize() const { return 1 << 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 + /// \pre value is positive and a power of two. Callers reach this only after + /// ValidateVectorSize has already rejected other inputs with + /// Status::Invalid, so a violation here is a programmer error rather + /// than malformed data; it is enforced with `ARROW_CHECK`, which aborts. + static uint8_t Log2(int32_t value) { + ARROW_CHECK(value > 0 && std::has_single_bit(static_cast<uint32_t>(value))) + << "value_must_be_power_of_2: " << value; + return static_cast<uint8_t>(std::countr_zero(static_cast<uint32_t>(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, or error if index is out of range + Result<int32_t> GetVectorNumElements(int32_t vector_index) const { + const int32_t vector_size = GetVectorSize(); + const int32_t num_full_vectors = num_elements / vector_size; + const int32_t remainder = num_elements % vector_size; + if (vector_index < num_full_vectors) { + return vector_size; // Full vector + } else if (vector_index == num_full_vectors && remainder > 0) { + return remainder; // Last partial vector + } + return Status::Invalid("ALP invalid vector index: ", vector_index, + " (num_vectors=", GetNumVectors(), ")"); + } + + /// \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> +Result<typename AlpCodec<T>::AlpHeader> AlpCodec<T>::LoadHeader(const uint8_t* input, + int64_t input_size) { + if (input_size < static_cast<int64_t>(AlpHeader::kSize)) { + return Status::Invalid("ALP compressed buffer too small for header: ", input_size, + " < ", AlpHeader::kSize); + } + AlpHeader header{}; + header.compression_mode = util::SafeLoadAs<uint8_t>(input); + header.integer_encoding = util::SafeLoadAs<uint8_t>(input + 1); + header.log_vector_size = util::SafeLoadAs<uint8_t>(input + 2); + header.num_elements = util::SafeLoadAs<int32_t>(input + 3); + + if (header.compression_mode != static_cast<uint8_t>(AlpMode::kAlp)) { + return Status::Invalid("ALP unsupported compression mode: ", + static_cast<int>(header.compression_mode)); + } + if (header.integer_encoding != static_cast<uint8_t>(AlpIntegerEncoding::kForBitPack)) { + return Status::Invalid("ALP unsupported integer encoding: ", + static_cast<int>(header.integer_encoding)); + } + if (header.log_vector_size < AlpConstants::kMinLogVectorSize || + header.log_vector_size > AlpConstants::kMaxLogVectorSize) { + return Status::Invalid( + "ALP invalid log_vector_size: ", static_cast<int>(header.log_vector_size), + " (must be in [", static_cast<int>(AlpConstants::kMinLogVectorSize), ", ", + static_cast<int>(AlpConstants::kMaxLogVectorSize), "])"); + } + if (header.num_elements < 0) { + return Status::Invalid("ALP invalid num_elements: ", header.num_elements); + } + return header; +} + +template <typename T> +Result<typename AlpCodec<T>::AlpSamplerResult> AlpCodec<T>::CreateSamplingPreset( + const T* input, int64_t num_elements) { + if (num_elements < 0) { + return Status::Invalid("ALP num_elements must be non-negative, got ", num_elements); + } + + AlpSampler<T> sampler; + sampler.AddSample({input, static_cast<size_t>(num_elements)}); + return sampler.Finalize(); +} + +namespace { + +/// \brief Validate a caller-supplied vector_size against the format spec +/// +/// The spec constrains `log_vector_size` to the inclusive range +/// [kMinLogVectorSize, kMaxLogVectorSize], so the vector size must be a power +/// of two within [2^3, 2^15]. +Status ValidateVectorSize(int32_t vector_size) { + constexpr int32_t kMin = 1 << AlpConstants::kMinLogVectorSize; + constexpr int32_t kMax = 1 << AlpConstants::kMaxLogVectorSize; + if (vector_size <= 0 || !std::has_single_bit(static_cast<uint32_t>(vector_size))) { + return Status::Invalid("ALP vector_size must be a positive power of 2, got ", + vector_size); + } + if (vector_size < kMin || vector_size > kMax) { + return Status::Invalid("ALP vector_size must be in [", kMin, ", ", kMax, "], got ", + vector_size); + } + return Status::OK(); +} + +} // namespace + +template <typename T> +Status AlpCodec<T>::EncodeWithPreset(const T* input, int64_t num_elements, + const AlpSamplerResult& preset, int32_t vector_size, + uint8_t* output, int64_t* output_size) { + if (num_elements < 0) { + return Status::Invalid("ALP num_elements must be non-negative, got ", num_elements); + } + if (num_elements > std::numeric_limits<int32_t>::max()) { + return Status::Invalid("ALP num_elements exceeds INT32_MAX, got ", num_elements); + } + RETURN_NOT_OK(ValidateVectorSize(vector_size)); + + // Make room to store header afterwards. + uint8_t* encoded_header = output; + uint8_t* body = output + AlpHeader::kSize; + const int64_t remaining_output_size = + *output_size - static_cast<int64_t>(AlpHeader::kSize); + + const CompressionProgress compression_progress = + EncodeAlp(input, num_elements, preset.alp_parameters, vector_size, body, + remaining_output_size); + + 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(vector_size); + header.num_elements = static_cast<int32_t>(num_elements); + + util::SafeStore(encoded_header + 0, header.compression_mode); + util::SafeStore(encoded_header + 1, header.integer_encoding); + util::SafeStore(encoded_header + 2, header.log_vector_size); + util::SafeStore(encoded_header + 3, header.num_elements); + *output_size = static_cast<int64_t>(AlpHeader::kSize) + + compression_progress.num_compressed_bytes_produced; + return Status::OK(); +} + +template <typename T> +Status AlpCodec<T>::Encode(const T* input, int64_t num_elements, int32_t vector_size, + uint8_t* output, int64_t* output_size) { + ARROW_ASSIGN_OR_RAISE(auto sampling_result, CreateSamplingPreset(input, num_elements)); + return EncodeWithPreset(input, num_elements, sampling_result, vector_size, output, + output_size); +} + +template <typename T> +Status AlpCodec<T>::Encode(const T* input, int64_t num_elements, uint8_t* output, + int64_t* output_size) { + return Encode(input, num_elements, AlpConstants::kAlpVectorSize, output, output_size); +} + +template <typename T> +template <typename TargetType> +Status AlpCodec<T>::Decode(int32_t num_elements, const uint8_t* input, int64_t input_size, + TargetType* output) { + ARROW_ASSIGN_OR_RAISE(const AlpHeader header, LoadHeader(input, input_size)); + const int32_t vector_size = header.GetVectorSize(); + + const uint8_t* body = input + AlpHeader::kSize; + const int64_t body_size = input_size - static_cast<int64_t>(AlpHeader::kSize); + + ARROW_RETURN_NOT_OK(DecodeAlp<TargetType>(num_elements, body, body_size, + header.GetIntegerEncoding(), vector_size, + header.num_elements, output) + .status()); + return Status::OK(); +} + +template Status AlpCodec<float>::Decode(int32_t num_elements, const uint8_t* input, + int64_t input_size, float* output); +template Status AlpCodec<float>::Decode(int32_t num_elements, const uint8_t* input, + int64_t input_size, double* output); +template Status AlpCodec<double>::Decode(int32_t num_elements, const uint8_t* input, + int64_t input_size, double* output); + +template <typename T> +Result<int64_t> AlpCodec<T>::GetMaxCompressedSize(int64_t num_elements, + int32_t vector_size) { + if (num_elements < 0) { + return Status::Invalid("ALP num_elements must be non-negative, got ", num_elements); + } + RETURN_NOT_OK(ValidateVectorSize(vector_size)); + int64_t max_alp_size = AlpHeader::kSize; + + const int64_t vectors_count = ::arrow::bit_util::CeilDiv(num_elements, vector_size); + + // 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 += num_elements * static_cast<int64_t>(sizeof(T)); + // Exceptions. + max_alp_size += num_elements * static_cast<int64_t>(sizeof(T)); + // Exception positions. + max_alp_size += num_elements * static_cast<int64_t>(sizeof(AlpConstants::PositionType)); + + return max_alp_size; +} + +template <typename T> +typename AlpCodec<T>::CompressionProgress AlpCodec<T>::EncodeAlp( + const T* input, int64_t element_count, const AlpEncodingParameters& preset, + int32_t vector_size, uint8_t* output, int64_t output_size) { + // 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 int64_t num_vectors = ::arrow::bit_util::CeilDiv(element_count, vector_size); + encoded_vectors.reserve(num_vectors); + + int64_t input_offset = 0; + const int64_t vs = vector_size; + for (int64_t remaining_elements = element_count; remaining_elements > 0; + remaining_elements -= std::min(vs, remaining_elements)) { + const int64_t elements_to_encode = std::min(vs, remaining_elements); + encoded_vectors.push_back(AlpCompression<T>::CompressVector( + input + input_offset, static_cast<uint16_t>(elements_to_encode), preset)); + input_offset += elements_to_encode; + } + + // Phase 2: Calculate sizes and offsets + const AlpIntegerEncoding integer_encoding = preset.integer_encoding; + const int64_t per_vector_metadata_size = + AlpEncodedVectorInfo::kStoredSize + + GetIntegerEncodingMetadataSize<T>(integer_encoding); + + // Offsets section comes first (after header, which is written by Encode()) + const int64_t offsets_section_size = + num_vectors * static_cast<int64_t>(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 + int64_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 int64_t total_size = current_offset; + + if (total_size > output_size) { + return CompressionProgress{0, 0}; + } + + // Phase 3: Write offsets section + uint8_t* offset_ptr = output; + for (const auto& offset : vector_offsets) { + util::SafeStore(offset_ptr, offset); + 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]; + uint8_t* vector_start = output + vector_offsets[i]; + + // Write AlpInfo + vec.alp_info().Store({vector_start, AlpEncodedVectorInfo::kStoredSize}); + uint8_t* ptr = vector_start + AlpEncodedVectorInfo::kStoredSize; + + // Write ForInfo — only kForBitPack is supported; validated at the API boundary + vec.for_info().Store({ptr, AlpEncodedForVectorInfo<T>::kStoredSize}); + ptr += AlpEncodedForVectorInfo<T>::kStoredSize; + + // Write data (packed values + exception positions + exception values) + const int64_t data_size = vec.GetDataStoredSize(); + vec.StoreDataOnly({ptr, static_cast<size_t>(data_size)}); + } + + return CompressionProgress{total_size, element_count}; +} + +template <typename T> +template <typename TargetType> +Result<typename AlpCodec<T>::DecompressionProgress> AlpCodec<T>::DecodeAlp( + int64_t num_elements, const uint8_t* input, int64_t input_size, + AlpIntegerEncoding integer_encoding, int32_t vector_size, int32_t total_elements, + TargetType* output) { + // 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 int32_t num_vectors = + static_cast<int32_t>(::arrow::bit_util::CeilDiv(total_elements, vector_size)); + + if (num_vectors == 0) { + return DecompressionProgress{0, 0}; + } + + const int64_t offsets_section_size = + static_cast<int64_t>(num_vectors) * sizeof(AlpConstants::OffsetType); + if (input_size < offsets_section_size) { + return Status::Invalid("ALP compressed buffer too small for offsets section: ", + input_size, " < ", offsets_section_size); + } + + // Sanity check: each vector must have at least its metadata. Reject obviously + // corrupted num_vectors before allocating (avoids OOM on malicious data). + constexpr int64_t kMinBytesPerVector = + AlpEncodedVectorInfo::kStoredSize + AlpEncodedForVectorInfo<T>::kStoredSize; + if (offsets_section_size + static_cast<int64_t>(num_vectors) * kMinBytesPerVector > + input_size) { + return Status::Invalid("ALP num_vectors inconsistent with buffer size: num_vectors=", + num_vectors, ", input_size=", input_size); + } + + // Read all offsets + std::vector<AlpConstants::OffsetType> vector_offsets(num_vectors); + std::memcpy(vector_offsets.data(), input, + num_vectors * sizeof(AlpConstants::OffsetType)); + + // Decode each vector using its offset for O(1) random access + int64_t output_offset = 0; + int64_t bytes_consumed = offsets_section_size; + + for (int32_t vector_index = 0; vector_index < num_vectors; vector_index++) { + // Calculate number of elements in this vector + const int32_t num_full_vectors = total_elements / vector_size; + const int32_t remainder = total_elements % vector_size; + int32_t this_vector_elements; + if (vector_index < num_full_vectors) { + this_vector_elements = vector_size; + } else if (vector_index == num_full_vectors && remainder > 0) { + this_vector_elements = remainder; + } else { + return Status::Invalid("ALP vector index out of range: ", vector_index, + " (total_elements=", total_elements, + ", vector_size=", vector_size, ")"); + } + + if (output_offset + this_vector_elements > num_elements) { + return Status::Invalid("ALP decode output buffer too small: offset=", output_offset, + " + elements=", this_vector_elements, + " > capacity=", num_elements); + } + + // Validate offset is within bounds and enough buffer remains for metadata Review Comment: The chain rule is checked now, exactly as you describe. `VectorReader::Open` checks the offsets in one pass against a running expected value, starting at `num_vectors * sizeof(OffsetType)` and adding each vector's size, and says which vector broke the chain. The end of the last vector is bound-checked against the buffer. `AlpRobustnessTest.CorruptedOffsetChain` covers duplicate, backward, gapped and skipped offsets, all chosen to stay inside the buffer so only the chain rule can reject them. ########## cpp/src/arrow/util/alp/alp_codec.h: ########## @@ -0,0 +1,222 @@ +// 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 codec interface for ALP compression + +#pragma once + +#include <cstddef> +#include <cstdint> +#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. Pre-computing the preset lets you keep sampling out + /// of a benchmark loop, or encode multiple batches with one preset. + /// + /// \param[in] input pointer to the input data to sample + /// \param[in] num_elements number of elements to sample + /// \return the sampling result containing the encoding preset, or + /// Status::Invalid if `num_elements < 0`. + static Result<AlpSamplerResult> CreateSamplingPreset(const T* input, + int64_t num_elements); + + /// \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] input pointer to the input to encode + /// \param[in] num_elements number of elements to encode + /// \param[in] preset the pre-computed sampling result from CreateSamplingPreset() + /// \param[in] vector_size number of elements per vector (must be a power of 2 + /// in the inclusive range [2^kMinLogVectorSize, 2^kMaxLogVectorSize], + /// i.e. 8 to 32768) + /// \param[out] output pointer to the memory region we will encode into. + /// Must be at least GetMaxCompressedSize(num_elements, vector_size) bytes. + /// Behavior is undefined if `output` is smaller. + /// \param[in,out] output_size on input, the size of `output` in bytes; on output, + /// the actual size of the encoded data. Must satisfy + /// `*output_size >= GetMaxCompressedSize(num_elements, vector_size)`; + /// undersizing leads to undefined behavior (a partial write to + /// `output` and an out-of-bounds write of the header). + /// \return Status::OK on success, or Status::Invalid if any precondition is + /// violated: `num_elements >= 0`, `num_elements <= INT32_MAX`, and + /// `vector_size` a power of two in + /// `[2^kMinLogVectorSize, 2^kMaxLogVectorSize]`. + static Status EncodeWithPreset(const T* input, int64_t num_elements, + const AlpSamplerResult& preset, int32_t vector_size, + uint8_t* output, int64_t* output_size); + + /// \brief Encode floating point values using ALP decimal compression + /// + /// \param[in] input pointer to the input to encode + /// \param[in] num_elements number of elements to encode + /// \param[in] vector_size number of elements per vector (must be a power of 2 + /// in the inclusive range [2^kMinLogVectorSize, 2^kMaxLogVectorSize], + /// i.e. 8 to 32768) + /// \param[out] output pointer to the memory region we will encode into. + /// Must be at least GetMaxCompressedSize(num_elements, vector_size) bytes. + /// Behavior is undefined if `output` is smaller. + /// \param[in,out] output_size on input, the size of `output` in bytes; on output, + /// the actual size of the encoded data. Must satisfy + /// `*output_size >= GetMaxCompressedSize(num_elements, vector_size)`; + /// undersizing leads to undefined behavior (a partial write to + /// `output` and an out-of-bounds write of the header). + /// \return Status::OK on success, or Status::Invalid if any precondition is + /// violated: `num_elements >= 0`, `num_elements <= INT32_MAX`, and + /// `vector_size` a power of two in + /// `[2^kMinLogVectorSize, 2^kMaxLogVectorSize]`. + static Status Encode(const T* input, int64_t num_elements, int32_t vector_size, + uint8_t* output, int64_t* output_size); + + /// \brief Convenience overload with default vector_size = kAlpVectorSize. + /// Same preconditions and error returns as the four-argument overload. + static Status Encode(const T* input, int64_t num_elements, uint8_t* output, + int64_t* output_size); + + /// \brief Decode floating point values + /// + /// \param[in] num_elements number of elements the caller expects, which is + /// also the capacity of `output` in elements. This comes from + /// Parquet's page header (`num_values`), not from the ALP header + /// embedded in `input`; if the ALP header's own element count + /// would overrun this capacity, decoding fails rather than + /// writing past `output`. + /// \param[in] input pointer to the compressed data + /// \param[in] input_size size of the compressed data in bytes + /// \param[out] output 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 uint8_t* input, int64_t input_size, + TargetType* output); + + /// \brief Get the maximum compressed size for a given number of elements + /// + /// \param[in] num_elements number of elements to compress + /// \param[in] vector_size number of elements per vector (must be a power of 2 + /// in the inclusive range [2^kMinLogVectorSize, 2^kMaxLogVectorSize], + /// i.e. 8 to 32768) + /// \return the maximum size of the compressed buffer in bytes, or + /// Status::Invalid if `num_elements < 0` or `vector_size` is not a + /// power of two in `[2^kMinLogVectorSize, 2^kMaxLogVectorSize]`. + static Result<int64_t> GetMaxCompressedSize( + int64_t num_elements, int32_t vector_size = AlpConstants::kAlpVectorSize); Review Comment: Removed. `GetMaxCompressedSize` takes both arguments explicitly and every call site names the vector size. `Encode` keeps a convenience form, but as a separate overload rather than a default argument, so the short version is a deliberate choice at the call site instead of an omission. ########## cpp/src/arrow/util/alp/alp.h: ########## @@ -0,0 +1,815 @@ +// 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 <span> +#include <vector> + +#include "arrow/result.h" +#include "arrow/status.h" +#include "arrow/util/alp/alp_constants.h" +#include "arrow/util/bit_util.h" + +namespace arrow { +namespace util { +namespace alp { + +// ---------------------------------------------------------------------- +// ALP Overview +// +// IMPORTANT: For abstract interfaces or examples how to use ALP, consult +// alp_codec.h. +// This file implements adaptive lossless floating-point compression for +// decimals (ALP) (https://dl.acm.org/doi/10.1145/3626717). ALP converts each +// float into a decimal where it can, using an exponent and factor chosen per +// vector: c(f) = int64(f * 10^exponent * 10^-factor). It then encodes the +// converted integers with a frame of reference and bit-packs them. +// A value the conversion cannot round-trip becomes an exception. ALP stores +// exceptions separately and patches them back into the vector after decoding. +// +// ========================================================================== +// 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 = std::bit_width(max_delta), i.e. the | +// | number of bits needed to hold 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. +/// +/// The underlying type is fixed at `uint8_t` because this enum is serialized +/// as a single byte in `AlpHeader::compression_mode`. +enum class AlpMode : uint8_t { kAlp = 0 }; Review Comment: Moved — `enum class AlpMode` is in the anonymous namespace of `alp_codec.cc` now, next to the header struct that serializes it, which is its only user. ########## cpp/src/arrow/util/alp/alp.cc: ########## @@ -0,0 +1,1026 @@ +// 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 <algorithm> +#include <bit> +#include <cmath> +#include <cstring> +#include <functional> +#include <map> +#include <span> + +#include "arrow/util/alp/alp_constants.h" +#include "arrow/util/bit_stream_utils_internal.h" +#include "arrow/util/bit_util.h" +#include "arrow/util/bpacking_internal.h" +#include "arrow/util/endian.h" +#include "arrow/util/logging.h" +#include "arrow/util/ubsan.h" + +namespace arrow { +namespace util { +namespace alp { + +// ALP serialization stores multi-byte integers (frame_of_reference, +// num_exceptions, offsets) via the util::SafeLoadAs/SafeStore helpers, which +// handle unaligned access, and uses memcpy for bulk array copies. Both assume +// little-endian byte order on disk. +static_assert(ARROW_LITTLE_ENDIAN, "ALP serialization assumes little-endian byte order"); Review Comment: You're right, and it was worse than strict — it refused to build at all on a big-endian host. Both `static_assert`s are gone; the store and load helpers `memcpy` under `if constexpr (ARROW_LITTLE_ENDIAN == 1)` and convert per element otherwise, so the little-endian path costs nothing. Only the metadata scalars and the exception and offset arrays needed it. The bit-packed payload is written by `bit_util::BitWriter` and read by `arrow::internal::unpack`, so it was already portable and is copied as is. ########## cpp/src/arrow/util/alp/alp_codec_internal.h: ########## @@ -0,0 +1,222 @@ +// 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 codec interface for ALP compression + +#pragma once + +#include <cstddef> +#include <cstdint> +#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 Review Comment: Renamed all four to `_internal.h`, following `rle_encoding_internal.h`. Agreed on the reasoning — performance work here is likely and none of it should be under a compatibility promise. On `ARROW_EXPORT`, I went to add it and found the ALP headers aren't installed by either build system, so there's no visibility problem to fix today, and the rename makes that permanent rather than accidental. Happy to add the annotations anyway if you'd like them as documentation of intent. On the directory, your read matches mine — the codec layer is general and only the page framing is Parquet's, so a split would make sense, as would moving the whole thing. I've left that for you to settle rather than change the same four files twice in one round; name the layout you want and it's a small change. ########## cpp/src/arrow/util/alp/alp_codec.cc: ########## @@ -0,0 +1,583 @@ +// 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_codec.h" + +#include <bit> +#include <cmath> +#include <limits> + +#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/bit_util.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 int32_t to match Parquet page headers (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: 1 << log_vector_size (i.e., +/// 2^log_vector_size). For example, log_vector_size=10 means vector_size=1024. +/// LoadHeader rejects anything outside [kMinLogVectorSize, kMaxLogVectorSize], +/// so the representable vector sizes are 2^3 (8) through 2^15 (32768). +/// +/// 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 (int32) | +/// +---------------------------------------------------+ +/// +/// Page-level layout (offset-based interleaved for O(1) random access): +/// +/// +-------------------------------------------------------------------+ +/// | [AlpHeader (7B)] | +/// | [Offset₀ | Offset₁ | ... | Offsetₙ₋₁] ← Vector offsets | +/// | [Vector₀][Vector₁]...[Vectorₙ₋₁] ← Concatenated | +/// +-------------------------------------------------------------------+ +/// 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 = 1 << 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 (int32_t to match Parquet's i32 num_values). + /// Per-vector element count is inferred: vector_size for all but the last vector. + int32_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) + int32_t GetNumVectors() const { + const int32_t vector_size = GetVectorSize(); + return static_cast<int32_t>(::arrow::bit_util::CeilDiv(num_elements, vector_size)); + } + + /// \brief Get the size of the offsets section + /// + /// \return size in bytes of the offsets array (num_vectors * sizeof(OffsetType)) + int64_t GetOffsetsSectionSize() const { + return static_cast<int64_t>(GetNumVectors()) * sizeof(AlpConstants::OffsetType); + } + + /// \brief Compute the actual vector size from log_vector_size + /// + /// \return the vector size (2^log_vector_size) + int32_t GetVectorSize() const { return 1 << 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 + /// \pre value is positive and a power of two. Callers reach this only after + /// ValidateVectorSize has already rejected other inputs with + /// Status::Invalid, so a violation here is a programmer error rather + /// than malformed data; it is enforced with `ARROW_CHECK`, which aborts. + static uint8_t Log2(int32_t value) { + ARROW_CHECK(value > 0 && std::has_single_bit(static_cast<uint32_t>(value))) + << "value_must_be_power_of_2: " << value; + return static_cast<uint8_t>(std::countr_zero(static_cast<uint32_t>(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, or error if index is out of range + Result<int32_t> GetVectorNumElements(int32_t vector_index) const { + const int32_t vector_size = GetVectorSize(); + const int32_t num_full_vectors = num_elements / vector_size; + const int32_t remainder = num_elements % vector_size; + if (vector_index < num_full_vectors) { + return vector_size; // Full vector + } else if (vector_index == num_full_vectors && remainder > 0) { + return remainder; // Last partial vector + } + return Status::Invalid("ALP invalid vector index: ", vector_index, + " (num_vectors=", GetNumVectors(), ")"); + } + + /// \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> +Result<typename AlpCodec<T>::AlpHeader> AlpCodec<T>::LoadHeader(const uint8_t* input, + int64_t input_size) { + if (input_size < static_cast<int64_t>(AlpHeader::kSize)) { + return Status::Invalid("ALP compressed buffer too small for header: ", input_size, + " < ", AlpHeader::kSize); + } + AlpHeader header{}; + header.compression_mode = util::SafeLoadAs<uint8_t>(input); + header.integer_encoding = util::SafeLoadAs<uint8_t>(input + 1); + header.log_vector_size = util::SafeLoadAs<uint8_t>(input + 2); + header.num_elements = util::SafeLoadAs<int32_t>(input + 3); + + if (header.compression_mode != static_cast<uint8_t>(AlpMode::kAlp)) { + return Status::Invalid("ALP unsupported compression mode: ", + static_cast<int>(header.compression_mode)); + } + if (header.integer_encoding != static_cast<uint8_t>(AlpIntegerEncoding::kForBitPack)) { + return Status::Invalid("ALP unsupported integer encoding: ", + static_cast<int>(header.integer_encoding)); + } + if (header.log_vector_size < AlpConstants::kMinLogVectorSize || + header.log_vector_size > AlpConstants::kMaxLogVectorSize) { + return Status::Invalid( + "ALP invalid log_vector_size: ", static_cast<int>(header.log_vector_size), + " (must be in [", static_cast<int>(AlpConstants::kMinLogVectorSize), ", ", + static_cast<int>(AlpConstants::kMaxLogVectorSize), "])"); + } + if (header.num_elements < 0) { + return Status::Invalid("ALP invalid num_elements: ", header.num_elements); + } + return header; +} + +template <typename T> +Result<typename AlpCodec<T>::AlpSamplerResult> AlpCodec<T>::CreateSamplingPreset( + const T* input, int64_t num_elements) { + if (num_elements < 0) { + return Status::Invalid("ALP num_elements must be non-negative, got ", num_elements); + } + + AlpSampler<T> sampler; + sampler.AddSample({input, static_cast<size_t>(num_elements)}); + return sampler.Finalize(); +} + +namespace { + +/// \brief Validate a caller-supplied vector_size against the format spec +/// +/// The spec constrains `log_vector_size` to the inclusive range +/// [kMinLogVectorSize, kMaxLogVectorSize], so the vector size must be a power +/// of two within [2^3, 2^15]. +Status ValidateVectorSize(int32_t vector_size) { + constexpr int32_t kMin = 1 << AlpConstants::kMinLogVectorSize; + constexpr int32_t kMax = 1 << AlpConstants::kMaxLogVectorSize; + if (vector_size <= 0 || !std::has_single_bit(static_cast<uint32_t>(vector_size))) { + return Status::Invalid("ALP vector_size must be a positive power of 2, got ", + vector_size); + } + if (vector_size < kMin || vector_size > kMax) { + return Status::Invalid("ALP vector_size must be in [", kMin, ", ", kMax, "], got ", + vector_size); + } + return Status::OK(); +} + +} // namespace + +template <typename T> +Status AlpCodec<T>::EncodeWithPreset(const T* input, int64_t num_elements, + const AlpSamplerResult& preset, int32_t vector_size, + uint8_t* output, int64_t* output_size) { + if (num_elements < 0) { + return Status::Invalid("ALP num_elements must be non-negative, got ", num_elements); + } + if (num_elements > std::numeric_limits<int32_t>::max()) { + return Status::Invalid("ALP num_elements exceeds INT32_MAX, got ", num_elements); + } + RETURN_NOT_OK(ValidateVectorSize(vector_size)); + + // Make room to store header afterwards. + uint8_t* encoded_header = output; + uint8_t* body = output + AlpHeader::kSize; + const int64_t remaining_output_size = + *output_size - static_cast<int64_t>(AlpHeader::kSize); + + const CompressionProgress compression_progress = + EncodeAlp(input, num_elements, preset.alp_parameters, vector_size, body, + remaining_output_size); + + 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(vector_size); + header.num_elements = static_cast<int32_t>(num_elements); + + util::SafeStore(encoded_header + 0, header.compression_mode); + util::SafeStore(encoded_header + 1, header.integer_encoding); + util::SafeStore(encoded_header + 2, header.log_vector_size); + util::SafeStore(encoded_header + 3, header.num_elements); + *output_size = static_cast<int64_t>(AlpHeader::kSize) + + compression_progress.num_compressed_bytes_produced; + return Status::OK(); +} + +template <typename T> +Status AlpCodec<T>::Encode(const T* input, int64_t num_elements, int32_t vector_size, + uint8_t* output, int64_t* output_size) { + ARROW_ASSIGN_OR_RAISE(auto sampling_result, CreateSamplingPreset(input, num_elements)); Review Comment: Fixed, and the check moved rather than being duplicated. `ValidateElementCount` is the first statement of `CreateSamplingPreset` now, which is where the span is formed. `EncodeWithPreset` validates too, so neither entry point depends on the other having done it. `AlpRobustnessTest.ElementCountAboveInt32Max` covers it. -- 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]
