pitrou commented on code in PR #55449:
URL: https://github.com/apache/arrow/pull/55449#discussion_r4522650106
##########
cpp/src/arrow/util/alp/alp_compression.cc:
##########
@@ -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;
Review Comment:
Unrelated, but it's weird that it's called `AlpSamplingOptions` while it
contains compile-time constants.
##########
cpp/src/arrow/util/alp/alp_compression.cc:
##########
@@ -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)};
Review Comment:
Unrelated to this PR, but this reads weird until one realizes the
definitions of `GetExponent` and `GetFactor` (misnamed functions IMHO).
##########
cpp/src/arrow/util/alp/alp_compression.cc:
##########
Review Comment:
Unrelated to this PR, but what's the point of the "early exit" thing?
Why would we not test all pre-selected combinations (especially as
`kMaxCombinations = 5`)?
##########
cpp/src/arrow/util/alp/alp_sampler_internal.h:
##########
@@ -46,9 +46,11 @@ class ARROW_EXPORT AlpSampler {
private:
/// Adds one sampling chunk. Only the first
- /// AlpFormatConstants::kDefaultVectorSize elements are examined.
+ /// AlpSamplingOptions::kSampleWindowSize elements are examined.
Review Comment:
Rather `sampling_window_size` elements, no?
##########
cpp/src/arrow/util/alp/alp_sampler.cc:
##########
Review Comment:
I'm curious, is `chunk_samples_` unbounded, or does it stop accumulating
after a certain number of samples? It's not clear by reading this code.
##########
cpp/src/arrow/util/alp/alp_test.cc:
##########
@@ -776,11 +774,34 @@ TYPED_TEST(AlpSamplerTest, PresetGenerationDecimalData) {
output, std::vector<TypeParam>(data.begin(), data.begin() +
kNumValues)));
}
+TYPED_TEST(AlpSamplerTest, SamplingWindowSize) {
+ constexpr int64_t kChunkSize = AlpSamplingOptions::kChunkSize;
+ const std::vector<TypeParam> nan_chunk(kChunkSize,
+
std::numeric_limits<TypeParam>::quiet_NaN());
+ const std::vector<TypeParam> decimal_chunk(kChunkSize,
static_cast<TypeParam>(1.25));
+ const auto decimal_preset =
AlpCompression<TypeParam>::MakePreset({decimal_chunk});
+
+ // The sampler targets eight chunks per window. An eight-chunk window samples
+ // both inputs, but only decimal_chunk can contribute usable parameters.
+ AlpSampler<TypeParam> every_chunk(/*sampling_window_size=*/8 * kChunkSize);
+ every_chunk.AddSample(nan_chunk);
+ every_chunk.AddSample(decimal_chunk);
+ EXPECT_EQ(decimal_preset.combinations,
every_chunk.MakePreset().combinations);
+
+ // A sixteen-chunk window samples every other chunk, skipping decimal_chunk.
+ // With only NaNs sampled, the preset falls back to identity.
+ AlpSampler<TypeParam> every_other_chunk(/*sampling_window_size=*/16 *
kChunkSize);
Review Comment:
Is this because `kSampleChunksPerWindow = 8`?
##########
cpp/src/arrow/util/alp/alp_compression.cc:
##########
@@ -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);
}
Review Comment:
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
```
##########
cpp/src/parquet/encoder.cc:
##########
@@ -1013,40 +1038,56 @@ class AlpEncoder : public EncoderImpl, virtual public
TypedEncoder<DType> {
public:
using T = typename DType::c_type;
using ArrowType = typename EncodingTraits<DType>::ArrowType;
+ using Codec = ::arrow::util::alp::AlpCodec<T>;
+ using EncodingPreset = ::arrow::util::alp::AlpEncodingPreset;
+ using Sampler = ::arrow::util::alp::AlpSampler<T>;
using TypedEncoder<DType>::Put;
static_assert(std::is_same_v<T, float> || std::is_same_v<T, double>,
"ALP only supports float and double types");
- explicit AlpEncoder(const ColumnDescriptor* descr,
- ::arrow::MemoryPool* pool =
::arrow::default_memory_pool())
- : EncoderImpl(descr, Encoding::ALP, pool), sink_{pool} {}
+ explicit AlpEncoder(const ColumnDescriptor* descr, ::arrow::MemoryPool* pool,
+ const AlpOptions& alp_options)
+ : EncoderImpl(descr, Encoding::ALP, pool),
+ sink_{pool},
+ alp_options_(alp_options),
+ sampler_(alp_options.sampling_window_size) {}
// TODO(GH-48701): use a ratio estimate instead of reporting the raw buffer
size
int64_t EstimatedDataEncodedSize() override { return sink_.length(); }
std::shared_ptr<Buffer> FlushValues() override {
- // TODO(GH-48701): allow configuring the vector size.
- constexpr auto kVectorSize =
- ::arrow::util::alp::AlpFormatConstants::kDefaultVectorSize;
-
- // An all-null optional page still has to produce a header-only page.
const int64_t num_elements = sink_.length() /
static_cast<int64_t>(sizeof(T));
+ const std::span<const T> input{reinterpret_cast<const T*>(sink_.data()),
+ static_cast<size_t>(num_elements)};
+
+ if (num_elements > 0 && !preset_.has_value()) {
+ sampler_.AddSample(input);
+ sampled_values_ += num_elements;
+ if (sampled_values_ >= alp_options_.sampling_window_size) {
+ preset_ = sampler_.MakePreset();
+ }
+ }
+
+ EncodingPreset page_preset;
+ if (!preset_.has_value()) {
+ // Use a page-specific sampler to avoid affecting the global preset.
Review Comment:
Why do we not use a page-specific sampler for all pages? Is it too expensive?
##########
cpp/src/parquet/encoder.cc:
##########
@@ -1062,13 +1103,23 @@ class AlpEncoder : public EncoderImpl, virtual public
TypedEncoder<DType> {
void PutSpaced(const T* src, int num_values, const uint8_t* valid_bits,
int64_t valid_bits_offset) override {
+ if (num_values == 0) {
+ return;
+ }
Review Comment:
Not sure this is useful, the same test is already in `Put`
##########
cpp/src/parquet/encoder.cc:
##########
@@ -1013,40 +1038,56 @@ class AlpEncoder : public EncoderImpl, virtual public
TypedEncoder<DType> {
public:
using T = typename DType::c_type;
using ArrowType = typename EncodingTraits<DType>::ArrowType;
+ using Codec = ::arrow::util::alp::AlpCodec<T>;
+ using EncodingPreset = ::arrow::util::alp::AlpEncodingPreset;
+ using Sampler = ::arrow::util::alp::AlpSampler<T>;
using TypedEncoder<DType>::Put;
static_assert(std::is_same_v<T, float> || std::is_same_v<T, double>,
"ALP only supports float and double types");
- explicit AlpEncoder(const ColumnDescriptor* descr,
- ::arrow::MemoryPool* pool =
::arrow::default_memory_pool())
- : EncoderImpl(descr, Encoding::ALP, pool), sink_{pool} {}
+ explicit AlpEncoder(const ColumnDescriptor* descr, ::arrow::MemoryPool* pool,
+ const AlpOptions& alp_options)
+ : EncoderImpl(descr, Encoding::ALP, pool),
+ sink_{pool},
+ alp_options_(alp_options),
+ sampler_(alp_options.sampling_window_size) {}
// TODO(GH-48701): use a ratio estimate instead of reporting the raw buffer
size
int64_t EstimatedDataEncodedSize() override { return sink_.length(); }
std::shared_ptr<Buffer> FlushValues() override {
- // TODO(GH-48701): allow configuring the vector size.
- constexpr auto kVectorSize =
- ::arrow::util::alp::AlpFormatConstants::kDefaultVectorSize;
-
- // An all-null optional page still has to produce a header-only page.
const int64_t num_elements = sink_.length() /
static_cast<int64_t>(sizeof(T));
+ const std::span<const T> input{reinterpret_cast<const T*>(sink_.data()),
+ static_cast<size_t>(num_elements)};
+
+ if (num_elements > 0 && !preset_.has_value()) {
+ sampler_.AddSample(input);
+ sampled_values_ += num_elements;
+ if (sampled_values_ >= alp_options_.sampling_window_size) {
+ preset_ = sampler_.MakePreset();
+ }
+ }
+
+ EncodingPreset page_preset;
+ if (!preset_.has_value()) {
+ // Use a page-specific sampler to avoid affecting the global preset.
+ Sampler page_sampler(alp_options_.sampling_window_size);
+ page_sampler.AddSample(input);
+ page_preset = page_sampler.MakePreset();
+ }
+ const auto& preset = preset_.has_value() ? *preset_ : page_preset;
+
PARQUET_ASSIGN_OR_THROW(
int64_t max_comp_size,
- ::arrow::util::alp::AlpCodec<T>::GetMaxCompressedSize(num_elements,
kVectorSize));
-
+ Codec::GetMaxCompressedSize(num_elements, alp_options_.vector_size));
PARQUET_ASSIGN_OR_THROW(
- std::shared_ptr<ResizableBuffer> compressed_buffer,
+ auto compressed_buffer,
::arrow::AllocateResizableBuffer(max_comp_size, this->memory_pool()));
-
- PARQUET_ASSIGN_OR_THROW(
- const int64_t compressed_size,
- ::arrow::util::alp::AlpCodec<T>::Encode(
- {reinterpret_cast<const T*>(sink_.data()),
static_cast<size_t>(num_elements)},
- kVectorSize,
- {compressed_buffer->mutable_data(),
static_cast<size_t>(max_comp_size)}));
-
+ PARQUET_ASSIGN_OR_THROW(const int64_t compressed_size,
+ Codec::Encode(input, preset,
alp_options_.vector_size,
+ {compressed_buffer->mutable_data(),
+
static_cast<size_t>(max_comp_size)}));
Review Comment:
Can write `compressed_buffer->mutable_span_as<uint8_t>()`
##########
cpp/src/arrow/util/alp/alp_compression.cc:
##########
@@ -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.
Review Comment:
Was the "vectorized fast path" measured to bring substantial benefits? After
all, there should not be many exceptions...
##########
cpp/src/parquet/encoder.cc:
##########
@@ -1062,13 +1103,23 @@ class AlpEncoder : public EncoderImpl, virtual public
TypedEncoder<DType> {
void PutSpaced(const T* src, int num_values, const uint8_t* valid_bits,
int64_t valid_bits_offset) override {
+ if (num_values == 0) {
+ return;
+ }
if (valid_bits != NULLPTR) {
- PARQUET_ASSIGN_OR_THROW(auto buffer, ::arrow::AllocateBuffer(num_values
* sizeof(T),
-
this->memory_pool()));
+ const int num_valid_values =
static_cast<int>(::arrow::internal::CountSetBits(
+ valid_bits, valid_bits_offset, static_cast<int64_t>(num_values)));
+ if (num_valid_values == 0) {
+ return;
+ }
+ PARQUET_ASSIGN_OR_THROW(
+ auto buffer,
+ ::arrow::AllocateBuffer(num_valid_values * sizeof(T),
this->memory_pool()));
T* data = buffer->template mutable_data_as<T>();
- const int num_valid_values = ::arrow::util::internal::SpacedCompress<T>(
+ const int compacted_values = ::arrow::util::internal::SpacedCompress<T>(
src, num_values, valid_bits, valid_bits_offset, data);
- Put(data, num_valid_values);
+ ARROW_DCHECK_EQ(compacted_values, num_valid_values);
+ Put(data, compacted_values);
Review Comment:
`Put` will merely copy `compacted_values` into `sink_`, so perhaps you can
call `SpacedCompress` directly into `sink_`?
##########
cpp/src/arrow/util/alp/alp_sampler_internal.h:
##########
@@ -34,7 +34,7 @@ namespace arrow::util::alp {
template <AlpFloatingType T>
class ARROW_EXPORT AlpSampler {
public:
- AlpSampler() = default;
+ explicit AlpSampler(int64_t sampling_window_size);
Review Comment:
I'm not sure I understand how this relates to `kSampleWindowSize`.
--
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]
