prtkgaur commented on code in PR #48345: URL: https://github.com/apache/arrow/pull/48345#discussion_r3919374305
########## cpp/src/arrow/util/alp/alp_wrapper.h: ########## @@ -0,0 +1,148 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// High-level wrapper interface for ALP compression + +#pragma once + +#include <cstddef> +#include <optional> + +#include "arrow/util/alp/alp.h" + +namespace arrow { +namespace util { +namespace alp { + +// ---------------------------------------------------------------------- +// AlpWrapper + +/// \class AlpWrapper +/// \brief High-level interface for ALP compression +/// +/// AlpWrapper is an interface for Adaptive Lossless floating-Point Compression +/// (ALP) (https://dl.acm.org/doi/10.1145/3626717). For encoding, it samples +/// the data and applies decimal compression (Alp) to floating point values. +/// This class acts as a wrapper around the vector-based interfaces of +/// AlpSampler and Alp. +/// +/// \tparam T the floating point type (float or double) +template <typename T> +class AlpWrapper { + public: + /// \brief Encode floating point values using ALP decimal compression + /// + /// \param[in] decomp pointer to the input that is to be encoded + /// \param[in] decomp_size size of decomp in bytes. + /// This needs to be a multiple of sizeof(T). + /// \param[out] comp pointer to the memory region we will encode into. + /// The caller is responsible for ensuring this is big enough. Review Comment: Done — `alp_codec.h:95-112` spells out the buffer contract: `output` must be at least `GetMaxCompressedSize(num_elements, vector_size)` bytes, `*output_size` must satisfy the same bound on input and carries the encoded size back out, and it says what undersizing costs you (a partial write plus an out-of-bounds header write). ########## cpp/src/arrow/util/alp/alp_wrapper.h: ########## @@ -0,0 +1,148 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// High-level wrapper interface for ALP compression + +#pragma once + +#include <cstddef> +#include <optional> + +#include "arrow/util/alp/alp.h" + +namespace arrow { +namespace util { +namespace alp { + +// ---------------------------------------------------------------------- +// AlpWrapper + +/// \class AlpWrapper +/// \brief High-level interface for ALP compression +/// +/// AlpWrapper is an interface for Adaptive Lossless floating-Point Compression +/// (ALP) (https://dl.acm.org/doi/10.1145/3626717). For encoding, it samples +/// the data and applies decimal compression (Alp) to floating point values. +/// This class acts as a wrapper around the vector-based interfaces of +/// AlpSampler and Alp. +/// +/// \tparam T the floating point type (float or double) +template <typename T> +class AlpWrapper { + public: + /// \brief Encode floating point values using ALP decimal compression + /// + /// \param[in] decomp pointer to the input that is to be encoded + /// \param[in] decomp_size size of decomp in bytes. + /// This needs to be a multiple of sizeof(T). + /// \param[out] comp pointer to the memory region we will encode into. + /// The caller is responsible for ensuring this is big enough. + /// \param[in,out] comp_size the actual size of the encoded data in bytes, + /// expects the size of comp as input. If this is too small, + /// this is set to 0 and we bail out. + /// \param[in] enforce_mode reserved for future use. + /// Currently only AlpMode::kAlp is supported. + static void Encode(const T* decomp, size_t decomp_size, char* comp, + size_t* comp_size, + std::optional<AlpMode> enforce_mode = std::nullopt); + + /// \brief Decode floating point values + /// + /// \param[out] decomp pointer to the memory region we will decode into. + /// The caller is responsible for ensuring this is big enough + /// to hold num_elements values. + /// \param[in] num_elements number of elements to decode (from page header). + /// Uses uint32_t since Parquet page headers use i32 for num_values. + /// \param[in] comp pointer to the input that is to be decoded + /// \param[in] comp_size size of the input in bytes (from page header) + /// \tparam TargetType the type that is used to store the output. + /// May not be a narrowing conversion from T. + template <typename TargetType> + static void Decode(TargetType* decomp, uint32_t num_elements, const char* comp, + size_t comp_size); + + /// \brief Get the maximum compressed size of an uncompressed buffer + /// + /// \param[in] decomp_size the size of the uncompressed buffer in bytes + /// \return the maximum size of the compressed buffer + static uint64_t GetMaxCompressedSize(uint64_t decomp_size); + + private: + struct AlpHeader; + + /// \brief Tracks the progress of a compression operation + /// + /// Used to report how much data was consumed and produced during encoding. + struct CompressionProgress { + /// Number of compressed bytes written to output + uint64_t num_compressed_bytes_produced = 0; + /// Number of input elements consumed + uint64_t num_uncompressed_elements_taken = 0; + }; + + /// \brief Tracks the progress of a decompression operation + /// + /// Used to report how much data was consumed and produced during decoding. + struct DecompressionProgress { + /// Number of decompressed elements written + uint64_t num_decompressed_elements_produced = 0; + /// Number of compressed bytes consumed + uint64_t num_compressed_bytes_taken = 0; + }; + + /// \brief Compress a buffer using the ALP variant + /// + /// \param[in] decomp array of floating point numbers to compress + /// \param[in] element_count the number of floating point numbers + /// \param[out] comp the buffer to be compressed into + /// \param[in] comp_size the size of the compression buffer + /// \param[in] combinations the encoding preset to use + /// \return the compression progress + static CompressionProgress EncodeAlp(const T* decomp, uint64_t element_count, + char* comp, size_t comp_size, + const AlpEncodingPreset& combinations); + + /// \brief Decompress a buffer using the ALP variant + /// + /// \param[out] decomp the buffer to be decompressed into Review Comment: Done on the public API and on the private `EncodeAlp` helper, which had the same asymmetry. ########## cpp/src/arrow/util/alp/alp_wrapper.h: ########## @@ -0,0 +1,148 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// High-level wrapper interface for ALP compression + +#pragma once + +#include <cstddef> +#include <optional> + +#include "arrow/util/alp/alp.h" + +namespace arrow { +namespace util { +namespace alp { + +// ---------------------------------------------------------------------- +// AlpWrapper + +/// \class AlpWrapper +/// \brief High-level interface for ALP compression +/// +/// AlpWrapper is an interface for Adaptive Lossless floating-Point Compression +/// (ALP) (https://dl.acm.org/doi/10.1145/3626717). For encoding, it samples +/// the data and applies decimal compression (Alp) to floating point values. +/// This class acts as a wrapper around the vector-based interfaces of +/// AlpSampler and Alp. +/// +/// \tparam T the floating point type (float or double) +template <typename T> +class AlpWrapper { + public: + /// \brief Encode floating point values using ALP decimal compression + /// + /// \param[in] decomp pointer to the input that is to be encoded + /// \param[in] decomp_size size of decomp in bytes. + /// This needs to be a multiple of sizeof(T). + /// \param[out] comp pointer to the memory region we will encode into. + /// The caller is responsible for ensuring this is big enough. + /// \param[in,out] comp_size the actual size of the encoded data in bytes, + /// expects the size of comp as input. If this is too small, + /// this is set to 0 and we bail out. + /// \param[in] enforce_mode reserved for future use. + /// Currently only AlpMode::kAlp is supported. + static void Encode(const T* decomp, size_t decomp_size, char* comp, + size_t* comp_size, + std::optional<AlpMode> enforce_mode = std::nullopt); + + /// \brief Decode floating point values + /// + /// \param[out] decomp pointer to the memory region we will decode into. + /// The caller is responsible for ensuring this is big enough + /// to hold num_elements values. + /// \param[in] num_elements number of elements to decode (from page header). + /// Uses uint32_t since Parquet page headers use i32 for num_values. + /// \param[in] comp pointer to the input that is to be decoded + /// \param[in] comp_size size of the input in bytes (from page header) + /// \tparam TargetType the type that is used to store the output. + /// May not be a narrowing conversion from T. + template <typename TargetType> + static void Decode(TargetType* decomp, uint32_t num_elements, const char* comp, + size_t comp_size); + + /// \brief Get the maximum compressed size of an uncompressed buffer + /// + /// \param[in] decomp_size the size of the uncompressed buffer in bytes + /// \return the maximum size of the compressed buffer + static uint64_t GetMaxCompressedSize(uint64_t decomp_size); + + private: + struct AlpHeader; + + /// \brief Tracks the progress of a compression operation + /// + /// Used to report how much data was consumed and produced during encoding. + struct CompressionProgress { + /// Number of compressed bytes written to output + uint64_t num_compressed_bytes_produced = 0; Review Comment: Done. ########## cpp/src/parquet/decoder.cc: ########## @@ -2323,6 +2327,121 @@ class ByteStreamSplitDecoder<FLBAType> : public ByteStreamSplitDecoderBase<FLBAT } }; +// ---------------------------------------------------------------------- +// ALP decoder (Adaptive Lossless floating-Point) + +template <typename DType> +class AlpDecoder : public TypedDecoderImpl<DType> { + public: + using Base = TypedDecoderImpl<DType>; + using T = typename DType::c_type; + + explicit AlpDecoder(const ColumnDescriptor* descr) + : Base(descr, Encoding::ALP), current_offset_{0}, needs_decode_{false} { + static_assert(std::is_same<T, float>::value || std::is_same<T, double>::value, + "ALP only supports float and double types"); + } + + void SetData(int num_values, const uint8_t* data, int len) final { + Base::SetData(num_values, data, len); + current_offset_ = 0; + needs_decode_ = (len > 0 && num_values > 0); + decoded_buffer_.clear(); + } + + int Decode(T* buffer, int max_values) override { + // Fast path: decode directly into output buffer if requesting all values + if (needs_decode_ && max_values >= this->num_values_) { + ::arrow::util::alp::AlpWrapper<T>::Decode( Review Comment: Agreed in principle, deferring to a follow-up. `AlpDecoder` sits under `TypedDecoder`, whose `SetData`/`Decode` are `void`/`int` across every existing encoding — changing ALP alone would make it the odd one out, and changing the interface is a wider Parquet-C++ change than this PR should carry. It throws `ParquetException` today, which is what the other decoders do, so the error does reach the caller. ########## cpp/src/arrow/util/alp/alp_wrapper.h: ########## @@ -0,0 +1,148 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// High-level wrapper interface for ALP compression + +#pragma once + +#include <cstddef> +#include <optional> + +#include "arrow/util/alp/alp.h" + +namespace arrow { +namespace util { +namespace alp { + +// ---------------------------------------------------------------------- +// AlpWrapper + +/// \class AlpWrapper +/// \brief High-level interface for ALP compression +/// +/// AlpWrapper is an interface for Adaptive Lossless floating-Point Compression +/// (ALP) (https://dl.acm.org/doi/10.1145/3626717). For encoding, it samples +/// the data and applies decimal compression (Alp) to floating point values. +/// This class acts as a wrapper around the vector-based interfaces of +/// AlpSampler and Alp. +/// +/// \tparam T the floating point type (float or double) +template <typename T> +class AlpWrapper { Review Comment: Taken — `AlpWrapper` is `AlpCodec`, and the files are `alp_codec.h` / `alp_codec.cc`. That rename is why a lot of the threads on this PR now show as outdated. ########## cpp/src/arrow/util/alp/alp.cc: ########## @@ -0,0 +1,1035 @@ +// 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/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 { + +// ---------------------------------------------------------------------- +// 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)); + ptr += sizeof(frame_of_reference); + + // bit_width: 1 byte + *ptr = static_cast<char>(bit_width); +} + +template <typename T> +AlpEncodedForVectorInfo<T> AlpEncodedForVectorInfo<T>::Load( + arrow::util::span<const char> input_buffer) { + ARROW_CHECK(input_buffer.size() >= GetStoredSize()) + << "alp_for_vector_info_input_too_small: " << input_buffer.size() << " vs " + << GetStoredSize(); + + AlpEncodedForVectorInfo<T> result{}; + const char* ptr = input_buffer.data(); + + // frame_of_reference: 4 bytes for float, 8 bytes for double + std::memcpy(&result.frame_of_reference, ptr, sizeof(result.frame_of_reference)); + ptr += sizeof(result.frame_of_reference); + + // bit_width: 1 byte + result.bit_width = static_cast<uint8_t>(*ptr); + + return result; +} + +// Explicit template instantiations for AlpEncodedForVectorInfo +template struct AlpEncodedForVectorInfo<float>; +template struct AlpEncodedForVectorInfo<double>; + +// ---------------------------------------------------------------------- +// AlpEncodedVector implementation + +template <typename T> +void AlpEncodedVector<T>::Store(arrow::util::span<char> output_buffer) const { + const uint64_t overall_size = GetStoredSize(); + ARROW_CHECK(output_buffer.size() >= overall_size) + << "alp_bit_packed_vector_store_output_too_small: " << output_buffer.size() + << " vs " << overall_size; + + uint64_t offset = 0; + + // Store AlpInfo (4 bytes) + alp_info.Store({output_buffer.data() + offset, AlpEncodedVectorInfo::kStoredSize}); + offset += AlpEncodedVectorInfo::kStoredSize; + + // Store ForInfo (6/10 bytes) + for_info.Store({output_buffer.data() + offset, AlpEncodedForVectorInfo<T>::kStoredSize}); + offset += AlpEncodedForVectorInfo<T>::kStoredSize; + + // Store data section + StoreDataOnly({output_buffer.data() + offset, output_buffer.size() - offset}); +} + +template <typename T> +void AlpEncodedVector<T>::StoreDataOnly(arrow::util::span<char> output_buffer) const { + const uint64_t data_size = GetDataStoredSize(); + ARROW_CHECK(output_buffer.size() >= data_size) + << "alp_bit_packed_vector_store_data_output_too_small: " << output_buffer.size() + << " vs " << data_size; + + ARROW_CHECK(alp_info.num_exceptions == exceptions.size() && + alp_info.num_exceptions == exception_positions.size()) + << "alp_bit_packed_vector_store_num_exceptions_mismatch: " + << alp_info.num_exceptions << " vs " << exceptions.size() << " vs " + << exception_positions.size(); + + uint64_t offset = 0; + + // Compute bit_packed_size from num_elements and bit_width + const uint64_t bit_packed_size = + AlpEncodedForVectorInfo<T>::GetBitPackedSize(num_elements, for_info.bit_width); + + // Store all successfully compressed values first. + std::memcpy(output_buffer.data() + offset, packed_values.data(), bit_packed_size); + offset += bit_packed_size; + + // Store exception positions. + const uint64_t exception_position_size = + alp_info.num_exceptions * sizeof(AlpConstants::PositionType); + std::memcpy(output_buffer.data() + offset, exception_positions.data(), + exception_position_size); + offset += exception_position_size; + + // Store exception values. + const uint64_t exception_size = alp_info.num_exceptions * sizeof(T); + std::memcpy(output_buffer.data() + offset, exceptions.data(), exception_size); + offset += exception_size; + + ARROW_CHECK(offset == data_size) + << "alp_bit_packed_vector_data_size_mismatch: " << offset << " vs " << data_size; +} + +template <typename T> +AlpEncodedVector<T> AlpEncodedVector<T>::Load( + arrow::util::span<const char> input_buffer, uint16_t num_elements) { + ARROW_CHECK(num_elements <= AlpConstants::kAlpVectorSize) + << "alp_compression_state_element_count_too_large: " << num_elements << " vs " + << AlpConstants::kAlpVectorSize; + + AlpEncodedVector<T> result; + uint64_t input_offset = 0; + + // Load AlpInfo (4 bytes) + result.alp_info = AlpEncodedVectorInfo::Load( + {input_buffer.data() + input_offset, AlpEncodedVectorInfo::kStoredSize}); + input_offset += AlpEncodedVectorInfo::kStoredSize; + + // Load ForInfo (6/10 bytes) + result.for_info = AlpEncodedForVectorInfo<T>::Load( + {input_buffer.data() + input_offset, AlpEncodedForVectorInfo<T>::kStoredSize}); + input_offset += AlpEncodedForVectorInfo<T>::kStoredSize; + + result.num_elements = num_elements; + + const uint64_t overall_size = + GetStoredSize(result.alp_info, result.for_info, num_elements); + + ARROW_CHECK(input_buffer.size() >= overall_size) + << "alp_compression_state_input_too_small: " << input_buffer.size() << " vs " + << overall_size; + + // Compute bit_packed_size from num_elements and bit_width + const uint64_t bit_packed_size = + AlpEncodedForVectorInfo<T>::GetBitPackedSize(num_elements, result.for_info.bit_width); + + // Optimization: Use UnsafeResize to avoid zero-initialization before memcpy. + // This is safe for POD types since we immediately overwrite with memcpy. + result.packed_values.UnsafeResize(bit_packed_size); + std::memcpy(result.packed_values.data(), input_buffer.data() + input_offset, + bit_packed_size); + input_offset += bit_packed_size; + + result.exception_positions.UnsafeResize(result.alp_info.num_exceptions); + const uint64_t exception_position_size = + result.alp_info.num_exceptions * sizeof(AlpConstants::PositionType); + std::memcpy(result.exception_positions.data(), input_buffer.data() + input_offset, + exception_position_size); + input_offset += exception_position_size; + + result.exceptions.UnsafeResize(result.alp_info.num_exceptions); + const uint64_t exception_size = result.alp_info.num_exceptions * sizeof(T); + std::memcpy(result.exceptions.data(), input_buffer.data() + input_offset, + exception_size); + return result; +} + +template <typename T> +uint64_t AlpEncodedVector<T>::GetStoredSize() const { + return GetStoredSize(alp_info, for_info, num_elements); +} + +template <typename T> +uint64_t AlpEncodedVector<T>::GetStoredSize(const AlpEncodedVectorInfo& alp_info, + const AlpEncodedForVectorInfo<T>& for_info, + uint16_t num_elements) { + const uint64_t bit_packed_size = + AlpEncodedForVectorInfo<T>::GetBitPackedSize(num_elements, for_info.bit_width); + return AlpEncodedVectorInfo::kStoredSize + AlpEncodedForVectorInfo<T>::kStoredSize + + bit_packed_size + + alp_info.num_exceptions * (sizeof(AlpConstants::PositionType) + sizeof(T)); +} + +template <typename T> +bool AlpEncodedVector<T>::operator==(const AlpEncodedVector<T>& other) const { + if (alp_info != other.alp_info || for_info != other.for_info || + num_elements != other.num_elements) { + return false; + } + if (packed_values.size() != other.packed_values.size() || + !std::equal(packed_values.begin(), packed_values.end(), other.packed_values.begin())) { + return false; + } + if (exceptions.size() != other.exceptions.size() || + !std::equal(exceptions.begin(), exceptions.end(), other.exceptions.begin())) { + return false; + } + if (exception_positions.size() != other.exception_positions.size() || + !std::equal(exception_positions.begin(), exception_positions.end(), + other.exception_positions.begin())) { + return false; + } + return true; +} + +// ---------------------------------------------------------------------- +// AlpEncodedVectorView implementation + +template <typename T> +AlpEncodedVectorView<T> AlpEncodedVectorView<T>::LoadView( + arrow::util::span<const char> input_buffer, uint16_t num_elements) { + ARROW_CHECK(num_elements <= AlpConstants::kAlpVectorSize) + << "alp_view_element_count_too_large: " << num_elements << " vs " + << AlpConstants::kAlpVectorSize; + + AlpEncodedVectorView<T> result; + uint64_t input_offset = 0; + + // Load AlpInfo (4 bytes) + result.alp_info = AlpEncodedVectorInfo::Load( + {input_buffer.data() + input_offset, AlpEncodedVectorInfo::kStoredSize}); + input_offset += AlpEncodedVectorInfo::kStoredSize; + + // Load ForInfo (6/10 bytes) + result.for_info = AlpEncodedForVectorInfo<T>::Load( + {input_buffer.data() + input_offset, AlpEncodedForVectorInfo<T>::kStoredSize}); + input_offset += AlpEncodedForVectorInfo<T>::kStoredSize; + + result.num_elements = num_elements; + + const uint64_t overall_size = + AlpEncodedVector<T>::GetStoredSize(result.alp_info, result.for_info, num_elements); + + ARROW_CHECK(input_buffer.size() >= overall_size) + << "alp_view_input_too_small: " << input_buffer.size() << " vs " << overall_size; + + // Load data section (after metadata) + AlpEncodedVectorView<T> data_view = LoadViewDataOnly( + {input_buffer.data() + input_offset, input_buffer.size() - input_offset}, + result.alp_info, result.for_info, num_elements); + + // Copy the loaded data into result + result.packed_values = data_view.packed_values; + result.exception_positions = std::move(data_view.exception_positions); + result.exceptions = std::move(data_view.exceptions); + + return result; +} + +template <typename T> +AlpEncodedVectorView<T> AlpEncodedVectorView<T>::LoadViewDataOnly( + arrow::util::span<const char> input_buffer, const AlpEncodedVectorInfo& alp_info, + const AlpEncodedForVectorInfo<T>& for_info, uint16_t num_elements) { + ARROW_CHECK(num_elements <= AlpConstants::kAlpVectorSize) + << "alp_view_data_only_element_count_too_large: " << num_elements << " vs " + << AlpConstants::kAlpVectorSize; + + AlpEncodedVectorView<T> result; + result.alp_info = alp_info; + result.for_info = for_info; + result.num_elements = num_elements; + + const uint64_t data_size = for_info.GetDataStoredSize(num_elements, alp_info.num_exceptions); + ARROW_CHECK(input_buffer.size() >= data_size) + << "alp_view_data_only_input_too_small: " << input_buffer.size() << " vs " + << data_size; + + uint64_t input_offset = 0; + + // Compute bit_packed_size from num_elements and bit_width + const uint64_t bit_packed_size = + AlpEncodedForVectorInfo<T>::GetBitPackedSize(num_elements, for_info.bit_width); + + // Zero-copy for packed values (bytes have no alignment requirements) + result.packed_values = { + reinterpret_cast<const uint8_t*>(input_buffer.data() + input_offset), + bit_packed_size}; + input_offset += bit_packed_size; + + // Copy exception positions into aligned storage to avoid UB from misaligned access. + // Exceptions are rare (typically < 5%), so the copy overhead is negligible. + const uint64_t exception_position_size = + alp_info.num_exceptions * sizeof(AlpConstants::PositionType); + result.exception_positions.UnsafeResize(alp_info.num_exceptions); + std::memcpy(result.exception_positions.data(), input_buffer.data() + input_offset, + exception_position_size); + input_offset += exception_position_size; + + // Copy exception values into aligned storage to avoid UB from misaligned access. + const uint64_t exception_size = alp_info.num_exceptions * sizeof(T); + result.exceptions.UnsafeResize(alp_info.num_exceptions); + std::memcpy(result.exceptions.data(), input_buffer.data() + input_offset, + exception_size); + + return result; +} + +template <typename T> +uint64_t AlpEncodedVectorView<T>::GetStoredSize() const { + return AlpEncodedVector<T>::GetStoredSize(alp_info, for_info, num_elements); +} + +template struct AlpEncodedVectorView<float>; +template struct AlpEncodedVectorView<double>; + +// ---------------------------------------------------------------------- +// AlpMetadataCache implementation + +template <typename T> +AlpMetadataCache<T> AlpMetadataCache<T>::Load( + uint32_t num_vectors, uint32_t vector_size, uint32_t total_elements, + AlpIntegerEncoding integer_encoding, + arrow::util::span<const char> alp_metadata_buffer, + arrow::util::span<const char> int_encoding_metadata_buffer) { + AlpMetadataCache<T> cache; + + if (num_vectors == 0) { + return cache; + } + + const uint64_t alp_info_size = AlpEncodedVectorInfo::kStoredSize; + const uint64_t expected_alp_size = num_vectors * alp_info_size; + + ARROW_CHECK(alp_metadata_buffer.size() >= expected_alp_size) + << "alp_metadata_cache_alp_buffer_too_small: " << alp_metadata_buffer.size() + << " vs " << expected_alp_size; + + cache.alp_infos_.reserve(num_vectors); + cache.cumulative_data_offsets_.reserve(num_vectors); + cache.vector_num_elements_.reserve(num_vectors); + + // Calculate number of full vectors and remainder + const uint32_t num_full_vectors = total_elements / vector_size; + const uint32_t remainder = total_elements % vector_size; + + // Load integer encoding metadata based on encoding type + switch (integer_encoding) { + case AlpIntegerEncoding::kForBitPack: { + const uint64_t for_info_size = AlpEncodedForVectorInfo<T>::kStoredSize; + const uint64_t expected_for_size = num_vectors * for_info_size; + ARROW_CHECK(int_encoding_metadata_buffer.size() >= expected_for_size) + << "alp_metadata_cache_for_buffer_too_small: " + << int_encoding_metadata_buffer.size() << " vs " << expected_for_size; + cache.for_infos_.reserve(num_vectors); + + uint64_t cumulative_offset = 0; + for (uint32_t i = 0; i < num_vectors; i++) { + // Load AlpInfo + const AlpEncodedVectorInfo alp_info = AlpEncodedVectorInfo::Load( + {alp_metadata_buffer.data() + i * alp_info_size, alp_info_size}); + cache.alp_infos_.push_back(alp_info); + + // Load ForInfo for kForBitPack encoding + const AlpEncodedForVectorInfo<T> for_info = AlpEncodedForVectorInfo<T>::Load( + {int_encoding_metadata_buffer.data() + i * for_info_size, for_info_size}); + cache.for_infos_.push_back(for_info); + + // Calculate number of elements for this vector + const uint16_t this_vector_elements = + (i < num_full_vectors) ? static_cast<uint16_t>(vector_size) + : static_cast<uint16_t>(remainder); + cache.vector_num_elements_.push_back(this_vector_elements); + + // Store cumulative offset (offset to start of this vector's data) + cache.cumulative_data_offsets_.push_back(cumulative_offset); + + // Advance offset by this vector's data size + cumulative_offset += + for_info.GetDataStoredSize(this_vector_elements, alp_info.num_exceptions); + } + cache.total_data_size_ = cumulative_offset; + } break; + + default: + ARROW_CHECK(false) << "unsupported_integer_encoding: " + << static_cast<int>(integer_encoding); + break; + } + + return cache; +} + +template class AlpMetadataCache<float>; +template class AlpMetadataCache<double>; + +template class AlpEncodedVector<float>; +template class AlpEncodedVector<double>; + +// ---------------------------------------------------------------------- +// Internal helper classes + +namespace { + +/// \brief Helper class for encoding/decoding individual values +template <typename T> +class AlpInlines : private AlpConstants { + public: + using Constants = AlpTypedConstants<T>; + using ExactType = typename Constants::FloatingToExact; + using SignedExactType = typename Constants::FloatingToSignedExact; + + /// \brief Check if float is a special value that cannot be converted + static inline bool IsImpossibleToEncode(const T n) { + // We do not have to check for positive or negative infinity, since + // std::numeric_limits<T>::infinity() > std::numeric_limits<T>::max() + // and vice versa for negative infinity. + return std::isnan(n) || n > Constants::kEncodingUpperLimit || + n < Constants::kEncodingLowerLimit || + (n == 0.0 && std::signbit(n)); // Verification for -0.0 + } + + /// \brief Convert a float to an int without rounding + static inline auto FastRound(T n) -> SignedExactType { + n = n + Constants::kMagicNumber - Constants::kMagicNumber; + return static_cast<SignedExactType>(n); + } + + /// \brief Fast way to round float to nearest integer + static inline auto NumberToInt(T n) -> SignedExactType { + if (IsImpossibleToEncode(n)) { + return static_cast<SignedExactType>(Constants::kEncodingUpperLimit); + } + return FastRound(n); + } + + /// \brief Convert a float into an int using encoding options + static inline SignedExactType EncodeValue( + const T value, const AlpExponentAndFactor exponent_and_factor) { + const T tmp_encoded_value = value * + Constants::GetExponent(exponent_and_factor.exponent) * + Constants::GetFactor(exponent_and_factor.factor); + return NumberToInt(tmp_encoded_value); + } + + /// \brief Reconvert an int to a float using encoding options + static inline T DecodeValue(const SignedExactType encoded_value, + const AlpExponentAndFactor exponent_and_factor) { + // The cast to T is needed to prevent a signed integer overflow. + return static_cast<T>(encoded_value) * GetFactor(exponent_and_factor.factor) * + Constants::GetFactor(exponent_and_factor.exponent); + } +}; + +/// \brief Helper struct for tracking compression combinations +struct AlpCombination { + AlpExponentAndFactor exponent_and_factor; + uint64_t num_appearances{0}; + uint64_t estimated_compression_size{0}; +}; + +/// \brief Compare two ALP combinations to determine which is better +/// +/// Return true if c1 is a better combination than c2. +/// First criteria is number of times it appears as best combination. +/// Second criteria is the estimated compression size. +/// Third criteria is bigger exponent. +/// Fourth criteria is bigger factor. +bool CompareAlpCombinations(const AlpCombination& c1, const AlpCombination& c2) { + return (c1.num_appearances > c2.num_appearances) || + (c1.num_appearances == c2.num_appearances && + (c1.estimated_compression_size < c2.estimated_compression_size)) || + ((c1.num_appearances == c2.num_appearances && + c1.estimated_compression_size == c2.estimated_compression_size) && + (c2.exponent_and_factor.exponent < c1.exponent_and_factor.exponent)) || + ((c1.num_appearances == c2.num_appearances && + c1.estimated_compression_size == c2.estimated_compression_size && + c2.exponent_and_factor.exponent == c1.exponent_and_factor.exponent) && + (c2.exponent_and_factor.factor < c1.exponent_and_factor.factor)); +} + +} // namespace + +// ---------------------------------------------------------------------- +// AlpCompression implementation + +template <typename T> +std::optional<uint64_t> AlpCompression<T>::EstimateCompressedSize( + const std::vector<T>& input_vector, + const AlpExponentAndFactor exponent_and_factor, + const bool penalize_exceptions) { + // Dry compress a vector (ideally a sample) to estimate ALP compression size + // given an exponent and factor. + SignedExactType max_encoded_value = std::numeric_limits<SignedExactType>::min(); + SignedExactType min_encoded_value = std::numeric_limits<SignedExactType>::max(); + + uint64_t num_exceptions = 0; + uint64_t num_non_exceptions = 0; + for (const T& value : input_vector) { + const SignedExactType encoded_value = + AlpInlines<T>::EncodeValue(value, exponent_and_factor); + T decoded_value = AlpInlines<T>::DecodeValue(encoded_value, exponent_and_factor); + if (decoded_value == value) { + num_non_exceptions++; + max_encoded_value = std::max(encoded_value, max_encoded_value); + min_encoded_value = std::min(encoded_value, min_encoded_value); + continue; + } + num_exceptions++; + } + + // We penalize combinations which yield almost all exceptions. + if (penalize_exceptions && num_non_exceptions < 2) { + return std::nullopt; + } + + // Evaluate factor/exponent compression size (we optimize for FOR). + const ExactType delta = (static_cast<ExactType>(max_encoded_value) - + static_cast<ExactType>(min_encoded_value)); + + const uint32_t estimated_bits_per_value = + static_cast<uint32_t>(std::ceil(std::log2(delta + 1))); + uint64_t estimated_compression_size = input_vector.size() * estimated_bits_per_value; + estimated_compression_size += + num_exceptions * (kExactTypeBitSize + (sizeof(PositionType) * 8)); + return estimated_compression_size; +} + +template <typename T> +AlpEncodingPreset AlpCompression<T>::CreateEncodingPreset( + const std::vector<std::vector<T>>& vectors_sampled) { + // Find the best combinations of factor-exponent from each sampled vector. + // This function is called once per segment. + // This operates over ALP first level samples. + static constexpr uint64_t kMaxCombinationCount = + (Constants::kMaxExponent + 1) * (Constants::kMaxExponent + 2) / 2; + + std::map<AlpExponentAndFactor, uint64_t> best_k_combinations_hash; + + uint64_t best_compressed_size_bits = std::numeric_limits<uint64_t>::max(); + // For each vector sampled. + for (const std::vector<T>& sampled_vector : vectors_sampled) { + const uint64_t num_samples = sampled_vector.size(); + const AlpExponentAndFactor best_encoding_options{Constants::kMaxExponent, + Constants::kMaxExponent}; + + // Start optimization with worst possible total bits from compression. + const uint64_t best_total_bits = + (num_samples * (kExactTypeBitSize + sizeof(PositionType) * 8)) + + (num_samples * kExactTypeBitSize); + + // N of appearances is irrelevant at this phase; we search for best compression. + AlpCombination best_combination{best_encoding_options, 0, best_total_bits}; + // Try all combinations to find the one which minimizes compression size. + for (uint8_t exp_idx = 0; exp_idx <= Constants::kMaxExponent; exp_idx++) { + for (uint8_t factor_idx = 0; factor_idx <= exp_idx; factor_idx++) { + const AlpExponentAndFactor current_exponent_and_factor{exp_idx, factor_idx}; + std::optional<uint64_t> estimated_compression_size = EstimateCompressedSize( Review Comment: Agreed with your follow-up, and that's exactly the case — turning an extreme value into an exception can shrink the bit width for everything else, so a zero-exception combination isn't necessarily the cheapest. Leaving the full search in place. ########## cpp/src/arrow/util/type_fwd.h: ########## @@ -55,7 +55,8 @@ struct Compression { LZ4_FRAME, LZO, BZ2, - LZ4_HADOOP + LZ4_HADOOP, + ALP Review Comment: Agreed, and the file is no longer touched by this PR — ALP is registered as a Parquet encoding, not in the `Codec` enum. ########## cpp/src/arrow/util/alp/alp_wrapper.h: ########## @@ -0,0 +1,148 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// High-level wrapper interface for ALP compression + +#pragma once + +#include <cstddef> +#include <optional> + +#include "arrow/util/alp/alp.h" + +namespace arrow { +namespace util { +namespace alp { + +// ---------------------------------------------------------------------- +// AlpWrapper + +/// \class AlpWrapper +/// \brief High-level interface for ALP compression +/// +/// AlpWrapper is an interface for Adaptive Lossless floating-Point Compression +/// (ALP) (https://dl.acm.org/doi/10.1145/3626717). For encoding, it samples +/// the data and applies decimal compression (Alp) to floating point values. +/// This class acts as a wrapper around the vector-based interfaces of +/// AlpSampler and Alp. +/// +/// \tparam T the floating point type (float or double) +template <typename T> +class AlpWrapper { + public: + /// \brief Encode floating point values using ALP decimal compression + /// + /// \param[in] decomp pointer to the input that is to be encoded + /// \param[in] decomp_size size of decomp in bytes. + /// This needs to be a multiple of sizeof(T). + /// \param[out] comp pointer to the memory region we will encode into. + /// The caller is responsible for ensuring this is big enough. + /// \param[in,out] comp_size the actual size of the encoded data in bytes, + /// expects the size of comp as input. If this is too small, + /// this is set to 0 and we bail out. + /// \param[in] enforce_mode reserved for future use. + /// Currently only AlpMode::kAlp is supported. + static void Encode(const T* decomp, size_t decomp_size, char* comp, + size_t* comp_size, + std::optional<AlpMode> enforce_mode = std::nullopt); + + /// \brief Decode floating point values + /// + /// \param[out] decomp pointer to the memory region we will decode into. + /// The caller is responsible for ensuring this is big enough + /// to hold num_elements values. + /// \param[in] num_elements number of elements to decode (from page header). + /// Uses uint32_t since Parquet page headers use i32 for num_values. + /// \param[in] comp pointer to the input that is to be decoded + /// \param[in] comp_size size of the input in bytes (from page header) + /// \tparam TargetType the type that is used to store the output. + /// May not be a narrowing conversion from T. + template <typename TargetType> + static void Decode(TargetType* decomp, uint32_t num_elements, const char* comp, + size_t comp_size); + + /// \brief Get the maximum compressed size of an uncompressed buffer + /// + /// \param[in] decomp_size the size of the uncompressed buffer in bytes + /// \return the maximum size of the compressed buffer + static uint64_t GetMaxCompressedSize(uint64_t decomp_size); Review Comment: Done — spelled out, and `int64_t`. ########## cpp/src/parquet/decoder.cc: ########## @@ -2323,6 +2327,121 @@ class ByteStreamSplitDecoder<FLBAType> : public ByteStreamSplitDecoderBase<FLBAT } }; +// ---------------------------------------------------------------------- +// ALP decoder (Adaptive Lossless floating-Point) + +template <typename DType> +class AlpDecoder : public TypedDecoderImpl<DType> { + public: + using Base = TypedDecoderImpl<DType>; + using T = typename DType::c_type; + + explicit AlpDecoder(const ColumnDescriptor* descr) + : Base(descr, Encoding::ALP), current_offset_{0}, needs_decode_{false} { + static_assert(std::is_same<T, float>::value || std::is_same<T, double>::value, + "ALP only supports float and double types"); + } + + void SetData(int num_values, const uint8_t* data, int len) final { + Base::SetData(num_values, data, len); + current_offset_ = 0; + needs_decode_ = (len > 0 && num_values > 0); Review Comment: Added exactly that, in `AlpDecoder::SetData`: ```cpp if (num_values > 0 && len <= 0) { throw ParquetException("ALP SetData: num_values=" + std::to_string(num_values) + " but len=" + std::to_string(len)); } ``` ########## cpp/src/arrow/util/alp/alp.h: ########## @@ -0,0 +1,856 @@ +// 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 (metadata-at-start layout for random access) | +// | [Header][VectorInfo₀|VectorInfo₁|...][Data₀|Data₁|...] | +// | All VectorInfo first, then all data sections consecutively. | +// +------------------------------------------------------------------+ +// +// +// 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 (grouped metadata-at-start for efficient random access): +/// +/// +------------------------------------------------------------+ +/// | Page Layout | +/// +------------------------------------------------------------+ +/// | [Header (8B)] | +/// | [AlpInfo₀ | AlpInfo₁ | ... | AlpInfoₙ] ← ALP metadata | +/// | [ForInfo₀ | ForInfo₁ | ... | ForInfoₙ] ← FOR metadata | +/// | [Data₀ | Data₁ | ... | Dataₙ] ← Compressed | +/// +------------------------------------------------------------+ +/// +/// The grouped metadata layout enables O(1) random access and separates +/// ALP-specific metadata from integer encoding metadata (FOR). +/// +/// 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] + /// Use this for the grouped layout where metadata is stored 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; + /// View into bitpacked data (zero-copy, bytes have no alignment requirements) + arrow::util::span<const uint8_t> packed_values; + /// Exception positions (copied into aligned storage to avoid UB from misaligned access) + arrow::internal::StaticVector<uint16_t, AlpConstants::kAlpVectorSize> exception_positions; + /// Exception values (copied into aligned storage to avoid UB from misaligned access) + arrow::internal::StaticVector<T, AlpConstants::kAlpVectorSize> exceptions; + + /// \brief Create a zero-copy view from a compact format input buffer + /// + /// Expects format: [AlpInfo][ForInfo][PackedValues][ExceptionPositions][ExceptionValues] + /// + /// \param[in] input_buffer the buffer to create a view into + /// \param[in] num_elements the number of elements (from page header) + /// \return the view into the compressed data + static AlpEncodedVectorView LoadView(arrow::util::span<const char> input_buffer, + uint16_t num_elements); + + /// \brief Create a zero-copy view from data-only buffer (metadata provided separately) + /// + /// Use this for the grouped layout where AlpInfo and ForInfo are stored separately. + /// Expects format: [PackedValues][ExceptionPositions][ExceptionValues] (no metadata) + /// + /// \param[in] input_buffer the buffer containing only the data section + /// \param[in] alp_info the ALP metadata (loaded separately) + /// \param[in] for_info the FOR metadata (loaded separately) + /// \param[in] num_elements the number of elements (from page header) + /// \return the view into the compressed data + static AlpEncodedVectorView LoadViewDataOnly(arrow::util::span<const char> input_buffer, + const AlpEncodedVectorInfo& alp_info, + const AlpEncodedForVectorInfo<T>& for_info, + uint16_t num_elements); + + /// \brief Get the stored size of this vector in the buffer + /// + /// \return the stored size in bytes (includes AlpInfo + ForInfo + data) + uint64_t GetStoredSize() 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); + } +}; + +// ---------------------------------------------------------------------- +// AlpIntegerEncoding + +/// \brief Integer encoding method used after ALP decimal encoding +/// +/// Currently only FOR+BitPack is implemented. Future encodings can be added +/// by extending this enum and adding corresponding metadata structs. +enum class AlpIntegerEncoding : uint8_t { kForBitPack = 0 }; + +/// \brief Get the per-vector metadata size for a given integer encoding +/// +/// \tparam T the floating point type (float or double) +/// \param[in] encoding the integer encoding method +/// \return size in bytes of the per-vector metadata for this encoding +template <typename T> +inline uint64_t GetIntegerEncodingMetadataSize(AlpIntegerEncoding encoding) { + switch (encoding) { + case AlpIntegerEncoding::kForBitPack: + return AlpEncodedForVectorInfo<T>::kStoredSize; + default: + ARROW_CHECK(false) << "unknown_integer_encoding: " << static_cast<int>(encoding); + return 0; + } +} + +// ---------------------------------------------------------------------- +// AlpMetadataCache + +/// \class AlpMetadataCache +/// \brief Cache for vector metadata to enable O(1) random access to any vector +/// +/// With the grouped metadata layout, ALP metadata and FOR metadata are stored +/// in separate contiguous sections after the header. This class loads both +/// metadata types into memory and precomputes cumulative data offsets, +/// enabling O(1) access to any vector's data. +/// +/// Page layout: +/// [Header][AlpInfos...][ForInfos...][Data...] +/// +/// Usage: +/// \code +/// // Load metadata from compressed buffer +/// AlpMetadataCache<T> cache = AlpMetadataCache<T>::Load( +/// num_vectors, vector_size, total_elements, alp_metadata, for_metadata); +/// +/// // Access metadata for any vector in O(1) +/// const auto& alp_info = cache.GetAlpInfo(vector_idx); +/// const auto& for_info = cache.GetForInfo(vector_idx); +/// +/// // Get offset to any vector's data in O(1) +/// uint64_t data_offset = cache.GetVectorDataOffset(vector_idx); +/// +/// // Get number of elements in a specific vector +/// uint16_t num_elements = cache.GetVectorNumElements(vector_idx); +/// \endcode +/// +/// \tparam T the floating point type (float or double) +template <typename T> +class AlpMetadataCache { + public: + /// \brief Load all metadata from separate ALP and integer encoding metadata buffers + /// + /// \param[in] num_vectors number of vectors in the block + /// \param[in] vector_size size of each full vector (typically 1024) + /// \param[in] total_elements total number of elements across all vectors + /// \param[in] integer_encoding the integer encoding method used (determines metadata format) + /// \param[in] alp_metadata_buffer buffer containing all AlpEncodedVectorInfo contiguously + /// \param[in] int_encoding_metadata_buffer buffer containing integer encoding metadata + /// (AlpEncodedForVectorInfo for kForBitPack) + /// \return a metadata cache with all metadata and precomputed offsets + static AlpMetadataCache Load(uint32_t num_vectors, uint32_t vector_size, + uint32_t total_elements, + AlpIntegerEncoding integer_encoding, + arrow::util::span<const char> alp_metadata_buffer, + arrow::util::span<const char> int_encoding_metadata_buffer); + + /// \brief Get ALP metadata for vector at given index + /// + /// \param[in] vector_idx index of the vector (0 to num_vectors-1) + /// \return reference to the vector's ALP metadata + const AlpEncodedVectorInfo& GetAlpInfo(uint32_t vector_idx) const { + ARROW_CHECK(vector_idx < alp_infos_.size()) + << "vector_index_out_of_range: " << vector_idx; + return alp_infos_[vector_idx]; + } + + /// \brief Get FOR metadata for vector at given index + /// + /// \param[in] vector_idx index of the vector (0 to num_vectors-1) + /// \return reference to the vector's FOR metadata + const AlpEncodedForVectorInfo<T>& GetForInfo(uint32_t vector_idx) const { + ARROW_CHECK(vector_idx < for_infos_.size()) + << "vector_index_out_of_range: " << vector_idx; + return for_infos_[vector_idx]; + } + + /// \brief Get offset to vector's data from start of data section + /// + /// \param[in] vector_idx index of the vector (0 to num_vectors-1) + /// \return byte offset from start of data section to this vector's data + uint64_t GetVectorDataOffset(uint32_t vector_idx) const { + ARROW_CHECK(vector_idx < cumulative_data_offsets_.size()) + << "vector_index_out_of_range: " << vector_idx; + return cumulative_data_offsets_[vector_idx]; + } + + /// \brief Get number of elements in vector at given index + /// + /// \param[in] vector_idx index of the vector (0 to num_vectors-1) + /// \return number of elements in this vector + uint16_t GetVectorNumElements(uint32_t vector_idx) const { + ARROW_CHECK(vector_idx < vector_num_elements_.size()) + << "vector_index_out_of_range: " << vector_idx; + return vector_num_elements_[vector_idx]; + } + + /// \brief Get number of vectors in the cache + /// + /// \return number of vectors + uint32_t GetNumVectors() const { return static_cast<uint32_t>(alp_infos_.size()); } + + /// \brief Get total size of the data section in bytes + /// + /// \return total data size + uint64_t GetTotalDataSize() const { return total_data_size_; } + + /// \brief Get total size of the ALP metadata section in bytes + /// + /// \return total ALP metadata size (num_vectors * AlpEncodedVectorInfo::kStoredSize) + uint64_t GetAlpMetadataSectionSize() const { + return alp_infos_.size() * AlpEncodedVectorInfo::kStoredSize; + } + + /// \brief Get total size of the FOR metadata section in bytes + /// + /// \return total FOR metadata size (num_vectors * AlpEncodedForVectorInfo<T>::kStoredSize) + uint64_t GetForMetadataSectionSize() const { + return for_infos_.size() * AlpEncodedForVectorInfo<T>::kStoredSize; + } + + /// \brief Get total size of all metadata sections in bytes + /// + /// \return total metadata size (ALP + FOR) + uint64_t GetTotalMetadataSectionSize() const { + return GetAlpMetadataSectionSize() + GetForMetadataSectionSize(); + } + + private: + std::vector<AlpEncodedVectorInfo> alp_infos_; // ALP metadata per vector + std::vector<AlpEncodedForVectorInfo<T>> for_infos_; // FOR metadata per vector + std::vector<uint64_t> cumulative_data_offsets_; // Offset from data section start + std::vector<uint16_t> vector_num_elements_; // Number of elements in each vector + uint64_t total_data_size_ = 0; // Total size of data section +}; + +// ---------------------------------------------------------------------- +// AlpEncodingPreset + +/// \brief Preset for ALP compression +/// +/// Helper struct for compression. Before a larger amount of data is compressed, +/// a preset is generated, which contains multiple combinations of exponents and +/// factors. For each vector that is compressed, one of the combinations of this +/// preset is chosen dynamically. +struct AlpEncodingPreset { + /// Combinations of exponents and factors + std::vector<AlpExponentAndFactor> combinations; + /// Best compressed size for the preset + uint64_t best_compressed_size = 0; Review Comment: Done. ########## cpp/src/parquet/decoder.cc: ########## @@ -2323,6 +2327,121 @@ class ByteStreamSplitDecoder<FLBAType> : public ByteStreamSplitDecoderBase<FLBAT } }; +// ---------------------------------------------------------------------- +// ALP decoder (Adaptive Lossless floating-Point) + +template <typename DType> +class AlpDecoder : public TypedDecoderImpl<DType> { + public: + using Base = TypedDecoderImpl<DType>; + using T = typename DType::c_type; + + explicit AlpDecoder(const ColumnDescriptor* descr) + : Base(descr, Encoding::ALP), current_offset_{0}, needs_decode_{false} { + static_assert(std::is_same<T, float>::value || std::is_same<T, double>::value, + "ALP only supports float and double types"); + } + + void SetData(int num_values, const uint8_t* data, int len) final { + Base::SetData(num_values, data, len); + current_offset_ = 0; + needs_decode_ = (len > 0 && num_values > 0); + decoded_buffer_.clear(); + } + + int Decode(T* buffer, int max_values) override { + // Fast path: decode directly into output buffer if requesting all values + if (needs_decode_ && max_values >= this->num_values_) { + ::arrow::util::alp::AlpWrapper<T>::Decode( + buffer, static_cast<uint64_t>(this->num_values_), + reinterpret_cast<const char*>(this->data_), this->len_); + + const int decoded = this->num_values_; + this->num_values_ = 0; + needs_decode_ = false; + return decoded; + } + + // Slow path: partial read - decode to intermediate buffer + // ALP Bit unpacker needs batches of 64 + if (needs_decode_) { + decoded_buffer_.resize(this->num_values_); + ::arrow::util::alp::AlpWrapper<T>::Decode( + decoded_buffer_.data(), static_cast<uint64_t>(this->num_values_), + reinterpret_cast<const char*>(this->data_), this->len_); + needs_decode_ = false; + } + + // Copy from intermediate buffer + const int values_to_decode = std::min( + max_values, + static_cast<int>(decoded_buffer_.size() - current_offset_)); + + if (values_to_decode > 0) { + std::memcpy(buffer, decoded_buffer_.data() + current_offset_, + values_to_decode * sizeof(T)); + current_offset_ += values_to_decode; + this->num_values_ -= values_to_decode; + } + + return values_to_decode; + } + + int DecodeArrow(int num_values, int null_count, const uint8_t* valid_bits, + int64_t valid_bits_offset, + typename EncodingTraits<DType>::Accumulator* builder) override { + const int values_to_decode = num_values - null_count; + if (ARROW_PREDICT_FALSE(this->num_values_ < values_to_decode)) { + ParquetException::EofException("ALP DecodeArrow: Not enough values available. " + "Available: " + std::to_string(this->num_values_) + + ", Requested: " + std::to_string(values_to_decode)); + } + + // Decode if needed (DecodeArrow always needs intermediate buffer for nulls) + if (needs_decode_) { + decoded_buffer_.resize(this->num_values_); + ::arrow::util::alp::AlpWrapper<T>::Decode( + decoded_buffer_.data(), static_cast<uint64_t>(this->num_values_), Review Comment: Done, in the shape you described. `DecodeArrow` calls `builder->Reserve(num_values)`, decodes into `builder->GetMutableValue(builder->length() + null_count)` so the values land packed to the right in the builder's own storage, then either `UnsafeAdvance(num_values)` when there are no nulls or `SpacedExpandLeftward` followed by `UnsafeAdvance(num_values, valid_bits, valid_bits_offset)`. No intermediate buffer and no copy on that path. The peak memory came down separately, because decoding a page at a time was the other half of this. `AlpCodec<T>::VectorReader` validates the header and offset chain once in `Open` and then decodes any single vector on demand, so the decoder holds one vector — 4 or 8 KB — instead of a page, and that scratch is only touched when a batch starts or ends mid-vector. It's pool-backed now too. `TYPED_TEST(TestAlpEncoding, BatchedDecode)` covers six batch plans over a 2000-value page as the regression net. @wgtmac raised the same point independently on `decoder.cc:2387`; there's more detail in that thread. -- 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]
