Skip to content

GH-55448: [C++][Parquet] Optimize ALP and make vector sizes configurable #55449

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wgtmac wants to merge 2 commits into
base: main
Choose a base branch
from
+446 −247
Open

GH-55448: [C++][Parquet] Optimize ALP and make vector sizes configurable #55449

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ffe72de
GH-55448: [C++][Parquet] Optimize ALP and make vector sizes configurable
wgtmac Nov 9, 2026
19bda38
add EncodingOptions abstraction
wgtmac Nov 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
71 changes: 29 additions & 42 deletions cpp/src/arrow/util/alp/alp_codec.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -153,61 +153,45 @@ Result<int64_t> EncodePage(std::span<const T> inputs, const AlpEncodingPreset& p
std::span<uint8_t> body = output.subspan(AlpHeader::kSize);
const int64_t element_count = static_cast<int64_t>(inputs.size());

// 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;
for (int64_t remaining_elements = element_count; remaining_elements > 0;) {
const int64_t elements_to_encode =
std::min(static_cast<int64_t>(vector_size), remaining_elements);
encoded_vectors.push_back(AlpCompression<T>::Compress(
inputs.subspan(static_cast<size_t>(input_offset),
static_cast<size_t>(elements_to_encode)),
preset));
input_offset += elements_to_encode;
remaining_elements -= elements_to_encode;
}

// Phase 2: calculate sizes and offsets.
const int64_t per_vector_metadata_size =
AlpInfo::kStoredSize + GetIntegerEncodingMetadataSize<T>(preset.integer_encoding);
const int64_t offsets_section_size =
num_vectors * static_cast<int64_t>(sizeof(AlpFormatConstants::OffsetType));

std::vector<AlpFormatConstants::OffsetType> vector_offsets;
vector_offsets.reserve(num_vectors);
if (offsets_section_size > static_cast<int64_t>(body.size())) {
return Status::Invalid("ALP output buffer too small for offsets: ", body.size(),
" < ", offsets_section_size);
}

int64_t current_offset = offsets_section_size;
for (const auto& vec : encoded_vectors) {
int64_t input_offset = 0;
for (int64_t vector_index = 0; vector_index < num_vectors; ++vector_index) {
const int64_t remaining_elements = element_count - input_offset;
const int64_t elements_to_encode =
std::min(static_cast<int64_t>(vector_size), remaining_elements);
const auto vec = AlpCompression<T>::Compress(
inputs.subspan(static_cast<size_t>(input_offset),
static_cast<size_t>(elements_to_encode)),
preset);
const int64_t stored_size = per_vector_metadata_size + vec.GetDataStoredSize();
if (current_offset >
static_cast<int64_t>(
std::numeric_limits<AlpFormatConstants::OffsetType>::max())) {
return Status::Invalid("ALP encoded data exceeds the uint32 offset range");
}
vector_offsets.push_back(static_cast<AlpFormatConstants::OffsetType>(current_offset));
current_offset += per_vector_metadata_size + vec.GetDataStoredSize();
if (stored_size > static_cast<int64_t>(body.size()) - current_offset) {
return Status::Invalid("ALP output buffer too small: ", body.size(), " < ",
current_offset + stored_size);
}
const auto offset = static_cast<AlpFormatConstants::OffsetType>(current_offset);
util::SafeStore(body.data() + vector_index * sizeof(AlpFormatConstants::OffsetType),
bit_util::ToLittleEndian(offset));
vec.Store(body.subspan(static_cast<size_t>(current_offset),
static_cast<size_t>(stored_size)));
current_offset += stored_size;
input_offset += elements_to_encode;
}
const int64_t body_size = current_offset;
if (body_size > static_cast<int64_t>(body.size())) {
return Status::Invalid("ALP output buffer too small: ", body.size(), " < ",
body_size);
}

// Phase 3: write the offset table.
uint8_t* offset_ptr = body.data();
for (const auto& offset : vector_offsets) {
util::SafeStore(offset_ptr, bit_util::ToLittleEndian(offset));
offset_ptr += sizeof(AlpFormatConstants::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];
const int64_t stored_size = per_vector_metadata_size + vec.GetDataStoredSize();
vec.Store(body.subspan(vector_offsets[i], static_cast<size_t>(stored_size)));
}
AlpHeader header{};
header.compression_mode = static_cast<uint8_t>(AlpMode::kAlp);
header.integer_encoding = static_cast<uint8_t>(AlpIntegerEncoding::kForBitPack);
Expand All @@ -228,8 +212,11 @@ Result<int64_t> EncodePage(std::span<const T> inputs, const AlpEncodingPreset& p
template <AlpFloatingType T>
Result<AlpEncodingPreset> AlpCodec<T>::MakePreset(std::span<const T> inputs) {
RETURN_NOT_OK(ValidateElementCount(inputs.size()));
if (inputs.empty()) {
return AlpEncodingPreset::MakeDefault();
}

AlpSampler<T> sampler;
AlpSampler<T> sampler(static_cast<int64_t>(inputs.size()));
sampler.AddSample(inputs);
return sampler.MakePreset();
}
Expand Down
109 changes: 60 additions & 49 deletions cpp/src/arrow/util/alp/alp_compression.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
#include "arrow/util/alp/alp_compression_internal.h"

#include <algorithm>
#include <array>
#include <cmath>
#include <cstring>
#include <limits>
Expand Down Expand Up @@ -58,6 +59,20 @@ class AlpInlines {
using EncodedUnsigned = typename Constants::EncodedUnsigned;
using EncodedSigned = typename Constants::EncodedSigned;

struct Multipliers {
T encode_exponent;
T encode_factor;
T decode_factor;
T decode_exponent;
};

static Multipliers GetMultipliers(const AlpExponentAndFactor exponent_and_factor) {
return {Constants::GetExponent(exponent_and_factor.exponent),
Constants::GetFactor(exponent_and_factor.factor),
Constants::GetExponent(exponent_and_factor.factor),
Constants::GetFactor(exponent_and_factor.exponent)};
Comment on lines +70 to +73

@pitrou pitrou Nov 10, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Unrelated to this PR, but this reads weird until one realizes the definitions of GetExponent and GetFactor (misnamed functions IMHO).

}

// Returns true when the scaled value cannot be encoded.
static bool IsImpossibleToEncode(const T n) {
// The two limit comparisons already cover both infinities: the encoding
Expand Down Expand Up @@ -89,22 +104,15 @@ class AlpInlines {
return FastRound(n);
}

// Encodes one value using the exponent/factor pair.
static EncodedSigned 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);
static EncodedSigned EncodeValue(const T value, const Multipliers& multipliers) {
return NumberToInt(value * multipliers.encode_exponent * multipliers.encode_factor);
}

// Decodes one encoded integer using the exponent/factor pair.
static T DecodeValue(const EncodedSigned encoded_value,
const AlpExponentAndFactor exponent_and_factor) {
const Multipliers& multipliers) {
// The cast to T is needed to prevent a signed integer overflow.
return static_cast<T>(encoded_value) *
Constants::GetExponent(exponent_and_factor.factor) *
Constants::GetFactor(exponent_and_factor.exponent);
return static_cast<T>(encoded_value) * multipliers.decode_factor *
multipliers.decode_exponent;
}
};

Expand Down Expand Up @@ -149,6 +157,7 @@ std::optional<int64_t> EstimateCompressedSizeBits(
static constexpr int kBatchSize = 8;
const size_t n = inputs.size();
int64_t num_exceptions = 0;
const auto multipliers = AlpInlines<T>::GetMultipliers(exponent_and_factor);

size_t i = 0;
for (; i + kBatchSize <= n; i += kBatchSize) {
Expand All @@ -157,12 +166,11 @@ std::optional<int64_t> EstimateCompressedSizeBits(

// Pass 1: Encode (vectorizable: multiply + magic-number round)
for (int j = 0; j < kBatchSize; j++) {
encoded_values[j] = AlpInlines<T>::EncodeValue(inputs[i + j], exponent_and_factor);
encoded_values[j] = AlpInlines<T>::EncodeValue(inputs[i + j], multipliers);
}
// Pass 2: Decode (vectorizable: int->float cast + multiply)
for (int j = 0; j < kBatchSize; j++) {
decoded_values[j] =
AlpInlines<T>::DecodeValue(encoded_values[j], exponent_and_factor);
decoded_values[j] = AlpInlines<T>::DecodeValue(encoded_values[j], multipliers);
}
// Pass 3: Compare and accumulate min/max/exceptions
for (int j = 0; j < kBatchSize; j++) {
Expand All @@ -177,9 +185,8 @@ std::optional<int64_t> EstimateCompressedSizeBits(
// Scalar tail for remaining elements
for (; i < n; i++) {
const EncodedSigned encoded_value =
AlpInlines<T>::EncodeValue(inputs[i], exponent_and_factor);
const T decoded_value =
AlpInlines<T>::DecodeValue(encoded_value, exponent_and_factor);
AlpInlines<T>::EncodeValue(inputs[i], multipliers);
const T decoded_value = AlpInlines<T>::DecodeValue(encoded_value, multipliers);
if (decoded_value == inputs[i]) {
max_encoded_value = std::max(encoded_value, max_encoded_value);
min_encoded_value = std::min(encoded_value, min_encoded_value);
Expand Down Expand Up @@ -304,21 +311,6 @@ AlpEncodingPreset AlpCompression<T>::MakePreset(
return AlpEncodingPreset{std::move(combinations), estimated_compressed_size_bytes};
}

template <AlpFloatingType T>
std::vector<T> AlpCompression<T>::CreateSample(std::span<const T> inputs) {
// Pick an equidistant stride so the sample count is capped at kSamplesPerChunk.
const int32_t idx_increments = std::max<int32_t>(
1, static_cast<int32_t>(std::ceil(static_cast<double>(inputs.size()) /
AlpSamplingOptions::kSamplesPerChunk)));
const size_t increment = static_cast<size_t>(idx_increments);
std::vector<T> sample;
sample.reserve((inputs.size() + increment - 1) / increment);
for (size_t i = 0; i < inputs.size(); i += idx_increments) {
sample.push_back(inputs[i]);
}
return sample;
}

template <AlpFloatingType T>
AlpExponentAndFactor AlpCompression<T>::FindBestExponentAndFactor(
std::span<const T> inputs, const std::vector<AlpExponentAndFactor>& combinations) {
Expand All @@ -331,7 +323,17 @@ AlpExponentAndFactor AlpCompression<T>::FindBestExponentAndFactor(
return combinations.front();
}

const std::vector<T> sample = CreateSample(inputs);
// The sample is capped at 256 values. Keep it on the stack to avoid a heap
// allocation for every vector parameter search.
std::array<T, AlpSamplingOptions::kSamplesPerChunk> sample_storage;

@pitrou pitrou Nov 10, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Unrelated, but it's weird that it's called AlpSamplingOptions while it contains compile-time constants.

const int32_t idx_increments = std::max<int32_t>(
1, static_cast<int32_t>(std::ceil(static_cast<double>(inputs.size()) /
AlpSamplingOptions::kSamplesPerChunk)));
size_t sample_count = 0;
for (size_t i = 0; i < inputs.size(); i += idx_increments) {
sample_storage[sample_count++] = inputs[i];
}
const std::span<const T> sample(sample_storage.data(), sample_count);

AlpExponentAndFactor best_exponent_and_factor;
int64_t best_total_bits = std::numeric_limits<int64_t>::max();
Expand Down Expand Up @@ -375,6 +377,7 @@ typename AlpCompression<T>::EncodingResult AlpCompression<T>::EncodeVector(
for_deltas.reserve(inputs.size());
std::vector<T> exceptions;
std::vector<AlpFormatConstants::PositionType> exception_positions;
const auto multipliers = AlpInlines<T>::GetMultipliers(exponent_and_factor);

EncodedSigned min_encoded_value = std::numeric_limits<EncodedSigned>::max();
EncodedSigned max_encoded_value = std::numeric_limits<EncodedSigned>::min();
Expand All @@ -383,10 +386,8 @@ typename AlpCompression<T>::EncodingResult AlpCompression<T>::EncodeVector(
// while the remaining values define the FOR range.
int64_t input_offset = 0;
for (const T input : inputs) {
const EncodedSigned encoded_value =
AlpInlines<T>::EncodeValue(input, exponent_and_factor);
const T decoded_value =
AlpInlines<T>::DecodeValue(encoded_value, exponent_and_factor);
const EncodedSigned encoded_value = AlpInlines<T>::EncodeValue(input, multipliers);
const T decoded_value = AlpInlines<T>::DecodeValue(encoded_value, multipliers);
for_deltas.push_back(static_cast<EncodedUnsigned>(encoded_value));

if (decoded_value != input) {
Expand Down Expand Up @@ -524,16 +525,25 @@ template <typename TargetType>
void AlpCompression<T>::PatchExceptions(
std::span<const T> exceptions,
std::span<const AlpFormatConstants::PositionType> exception_positions,
std::span<TargetType> outputs) {
const bool positions_strictly_increasing, std::span<TargetType> outputs) {
ARROW_DCHECK_EQ(exceptions.size(), exception_positions.size());
// `ivdep` assumes the positions are distinct. The encoder writes them in
// order, and loading a page rejects positions that do not increase.
int64_t exception_idx = 0;
ARROW_ALP_UNROLL_AND_ASSUME_INDEPENDENT
for (const AlpFormatConstants::PositionType exception_position : exception_positions) {
ARROW_DCHECK_LT(exception_position, outputs.size());
outputs[exception_position] = static_cast<T>(exceptions[exception_idx]);
exception_idx++;
const size_t num_exceptions = exception_positions.size();
const auto patch_exception = [&](size_t i) {
const auto position = exception_positions[i];
ARROW_DCHECK_LT(position, outputs.size());
outputs[position] = static_cast<T>(exceptions[i]);
};
if (positions_strictly_increasing) {
// The encoder emits strictly increasing positions. Use the vectorized
// fast path for that common case.
Comment on lines +537 to +538

@pitrou pitrou Nov 10, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Was the "vectorized fast path" measured to bring substantial benefits? After all, there should not be many exceptions...

@wgtmac wgtmac Nov 10, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

#48345 (comment) mentioned testing unroll factors of 2, 4 and 8, but provided no measurements specific to exception patching.

When I made this change, I have verified that the code path has been vectorized by gcc on godbolt. But I didn't verify its performance gain because I used Clang locally which failed to do this :/

ARROW_ALP_UNROLL_AND_ASSUME_INDEPENDENT
for (size_t i = 0; i < num_exceptions; ++i) {
patch_exception(i);
}
} else {
for (size_t i = 0; i < num_exceptions; ++i) {
patch_exception(i);
}
}
}

Expand Down Expand Up @@ -563,17 +573,18 @@ void AlpCompression<T>::Decompress(const AlpEncodedVectorView<T>& encoded_view,
// Read the exponent/factor once. AlpInfo stores them as uint8_t, which may
// alias the output stores, so reading them inside the loop blocks vectorization.
const AlpExponentAndFactor exponent_and_factor = alp_info.GetExponentAndFactor();
const auto multipliers = AlpInlines<T>::GetMultipliers(exponent_and_factor);
for (size_t i = 0; i < output.size(); ++i) {
const EncodedUnsigned unfored_value = scratch[i] + frame_of_ref;
const EncodedSigned signed_value = util::SafeCopy<EncodedSigned>(unfored_value);
output[i] = AlpInlines<T>::DecodeValue(signed_value, exponent_and_factor);
output[i] = AlpInlines<T>::DecodeValue(signed_value, multipliers);
}
Comment on lines 577 to 581

@pitrou pitrou Nov 10, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is the code path where I would expect unrolling and/or vectorization to be beneficial, but benchmarks don't seem conclusive:

BM_AlpDecodingFloat/1024          815 ns          815 ns       345130 bytes_per_second=4.68104Gi/s items_per_second=1.25656G/s
BM_AlpDecodingFloat/4096         2681 ns         2681 ns       102889 bytes_per_second=5.69219Gi/s items_per_second=1.52798G/s
BM_AlpDecodingFloat/32768       20041 ns        20034 ns        13779 bytes_per_second=6.09326Gi/s items_per_second=1.63565G/s
BM_AlpDecodingFloat/65536       40186 ns        40183 ns         6992 bytes_per_second=6.07578Gi/s items_per_second=1.63096G/s

BM_AlpDecodingDouble/1024         878 ns          878 ns       318022 bytes_per_second=8.68788Gi/s items_per_second=1.16607G/s
BM_AlpDecodingDouble/4096        2870 ns         2869 ns        96807 bytes_per_second=10.6353Gi/s items_per_second=1.42745G/s
BM_AlpDecodingDouble/32768      21227 ns        21216 ns        13090 bytes_per_second=11.5073Gi/s items_per_second=1.54448G/s
BM_AlpDecodingDouble/65536      42987 ns        42934 ns         6532 bytes_per_second=11.3728Gi/s items_per_second=1.52643G/s


PatchExceptions<TargetType>(
{encoded_view.exceptions().data(), encoded_view.exceptions().size()},
{encoded_view.exception_positions().data(),
encoded_view.exception_positions().size()},
output);
encoded_view.exception_positions_strictly_increasing(), output);
}

// Template instantiations
Expand Down
4 changes: 1 addition & 3 deletions cpp/src/arrow/util/alp/alp_compression_internal.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,8 +79,6 @@ class ARROW_EXPORT AlpCompression {
std::span<EncodedUnsigned> integer_scratch);

private:
static std::vector<T> CreateSample(std::span<const T> inputs);

static AlpExponentAndFactor FindBestExponentAndFactor(
std::span<const T> inputs, const std::vector<AlpExponentAndFactor>& combinations);

Expand Down Expand Up @@ -113,7 +111,7 @@ class ARROW_EXPORT AlpCompression {
static void PatchExceptions(
std::span<const T> exceptions,
std::span<const AlpFormatConstants::PositionType> exception_positions,
std::span<TargetType> outputs);
bool positions_strictly_increasing, std::span<TargetType> outputs);
};

// The marker on a class template does not reach its member templates: each is
Expand Down
11 changes: 4 additions & 7 deletions cpp/src/arrow/util/alp/alp_constants_internal.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,9 +38,6 @@ enum class AlpMode : uint8_t { kAlp = 0 };
enum class AlpIntegerEncoding : uint8_t { kForBitPack = 0 };

struct AlpFormatConstants {
/// Recommended default vector size.
static constexpr int32_t kDefaultVectorSize = 1024;

/// Minimum log2(vector size), corresponding to 8 elements.
static constexpr uint8_t kMinLogVectorSize = 3;

Expand All @@ -63,14 +60,14 @@ struct AlpSamplingOptions {
/// Input elements per sampling chunk.
static constexpr int64_t kChunkSize = 4096;

/// Elements per sampling interval.
static constexpr int64_t kIntervalSize = 122880;
/// Maximum prefix length examined in each sampled chunk.
static constexpr int64_t kSampleWindowSize = 1024;

/// Samples collected per chunk.
static constexpr int64_t kSamplesPerChunk = 256;

/// Chunks sampled per interval.
static constexpr int64_t kSampleChunksPerInterval = 8;
/// Target number of sampled chunks per window.
static constexpr int64_t kSampleChunksPerWindow = 8;

/// Consecutive non-improving combinations before an early exit.
static constexpr uint8_t kEarlyExitThreshold = 4;
Expand Down
Loading
Loading