prtkgaur commented on code in PR #48345:
URL: https://github.com/apache/arrow/pull/48345#discussion_r3921391206
##########
cpp/src/parquet/encoder.cc:
##########
@@ -997,6 +1000,137 @@ class ByteStreamSplitEncoder<FLBAType> : public
ByteStreamSplitEncoderBase<FLBAT
}
};
+// ----------------------------------------------------------------------
+// ----------------------------------------------------------------------
+// ALP encoder (Adaptive Lossless floating-Point)
+
+// TODO: support incremental encoding. Today `Put` only appends raw input
+// to `sink_`, and `FlushValues` runs the entire ALP pipeline (sample +
+// preset selection + per-vector compression) on the whole buffer in one
+// shot. A future revision should encode complete vectors as `Put` calls
+// fill them, holding only a partial-vector tail across calls, so the
+// encoder can produce output progressively and use bounded memory.
+//
+// TODO: fall back to PLAIN when ALP is not paying for itself. ALP always
+// emits ALP-encoded pages, so a column whose values never compress (every
+// value an exception, e.g. random doubles or NaN) pays the per-vector
+// metadata and exception overhead and lands larger than PLAIN. This is not
+// hypothetical: on the encoding_alp_benchmark datasets, msg_sp encodes to
+// 113% of its plain size, and poi_longitude, num_brain and num_control are
+// all within 8% of break-even.
+//
+// The decision belongs in ColumnWriterImpl, not here: `encoding_` is const,
+// so this encoder cannot relabel its own page, and the choice depends on the
+// page compressor (ALP at 113% of raw may still beat PLAIN+ZSTD), which this
+// layer knows nothing about. The mechanism already exists — mirror
+// `FallbackToPlainEncoding()` in column_writer.cc, which swaps
+// `current_encoder_` for a PLAIN encoder and updates `encoding_`. Parquet
+// records encoding per page, so mixing PLAIN and ALP pages in one column
+// chunk needs no format change.
+//
+// What is missing on this side is a way for the writer to know: AlpCodec
+// should expose the ratio it achieved or expects to achieve. The sampler
+// already computes an estimate in AlpCompression<T>::EstimateCompressedSize,
+// which is currently private. Deciding from that estimate before encoding
+// avoids encode-then-discard; a sticky post-hoc check is still worth keeping
+// as a safety net, since the sampler only inspects a subsample and can be
+// fooled by a column that changes character partway through.
+//
+// This needs to be resolved before ALP is enabled by default.
+template <typename DType>
+class AlpEncoder : public EncoderImpl, virtual public TypedEncoder<DType> {
+ public:
+ using T = typename DType::c_type;
+ using ArrowType = typename EncodingTraits<DType>::ArrowType;
+ using TypedEncoder<DType>::Put;
+
+ explicit AlpEncoder(
+ const ColumnDescriptor* descr,
+ ::arrow::MemoryPool* pool = ::arrow::default_memory_pool(),
+ int32_t vector_size = ::arrow::util::alp::AlpConstants::kAlpVectorSize)
+ : EncoderImpl(descr, Encoding::ALP, pool), sink_{pool},
vector_size_(vector_size) {
+ static_assert(std::is_same<T, float>::value || std::is_same<T,
double>::value,
+ "ALP only supports float and double types");
+ if (vector_size_ <= 0 ||
!std::has_single_bit(static_cast<uint32_t>(vector_size_))) {
+ throw ParquetException("ALP vector_size must be a positive power of 2,
got " +
+ std::to_string(vector_size_));
+ }
+ constexpr int32_t kMinVectorSize =
+ 1 << ::arrow::util::alp::AlpConstants::kMinLogVectorSize;
+ constexpr int32_t kMaxVectorSize =
+ 1 << ::arrow::util::alp::AlpConstants::kMaxLogVectorSize;
+ if (vector_size_ < kMinVectorSize || vector_size_ > kMaxVectorSize) {
+ throw ParquetException(
+ "ALP vector_size must be in [" + std::to_string(kMinVectorSize) + ",
" +
+ std::to_string(kMaxVectorSize) + "], got " +
std::to_string(vector_size_));
+ }
+ }
+
+ int64_t EstimatedDataEncodedSize() override { return sink_.length(); }
+
+ std::shared_ptr<Buffer> FlushValues() override {
+ if (sink_.length() == 0) {
+ // Empty buffer case
+ PARQUET_ASSIGN_OR_THROW(auto buf, sink_.Finish());
+ return buf;
+ }
+
+ // Call AlpCodec::Encode() - it handles sampling, preset selection, and
compression
+ const int64_t num_elements = sink_.length() /
static_cast<int64_t>(sizeof(T));
+ PARQUET_ASSIGN_OR_THROW(int64_t comp_size,
+
::arrow::util::alp::AlpCodec<T>::GetMaxCompressedSize(
+ num_elements, vector_size_));
+
+ PARQUET_ASSIGN_OR_THROW(auto compressed_buffer,
::arrow::AllocateResizableBuffer(
+ comp_size,
this->memory_pool()));
+
+ PARQUET_THROW_NOT_OK(::arrow::util::alp::AlpCodec<T>::Encode(
+ reinterpret_cast<const T*>(sink_.data()), num_elements, vector_size_,
+ compressed_buffer->mutable_data(), &comp_size));
+
+ PARQUET_THROW_NOT_OK(compressed_buffer->Resize(comp_size));
+ sink_.Reset();
+
+ return std::shared_ptr<Buffer>(std::move(compressed_buffer));
+ }
+
+ void Put(const T* buffer, int num_values) override {
+ if (num_values > 0) {
+ PARQUET_THROW_NOT_OK(sink_.Append(reinterpret_cast<const
uint8_t*>(buffer),
+ num_values *
static_cast<int64_t>(sizeof(T))));
+ }
+ }
+
+ void PutSpaced(const T* src, int num_values, const uint8_t* valid_bits,
+ int64_t valid_bits_offset) override {
+ if (valid_bits != NULLPTR) {
+ PARQUET_ASSIGN_OR_THROW(auto buffer, ::arrow::AllocateBuffer(num_values
* sizeof(T),
+
this->memory_pool()));
+ T* data = buffer->template mutable_data_as<T>();
+ const int num_valid_values = ::arrow::util::internal::SpacedCompress<T>(
+ src, num_values, valid_bits, valid_bits_offset, data);
+ Put(data, num_valid_values);
+ } else {
+ Put(src, num_values);
+ }
+ }
+
+ void Put(const ::arrow::Array& values) override {
+ if (values.type_id() != ArrowType::type_id) {
+ throw ParquetException(std::string() + "direct put from " +
+ values.type()->ToString() + " not supported");
+ }
+ const auto& data = *values.data();
+ this->PutSpaced(data.GetValues<typename ArrowType::c_type>(1),
+ static_cast<int>(data.length), data.GetValues<uint8_t>(0,
0),
+ data.offset);
+ }
+
+ private:
+ ::arrow::BufferBuilder sink_;
+ int32_t vector_size_;
Review Comment:
The member is gone instead — it's a `static constexpr int32_t kVectorSize`
on the encoder now, since nothing sets it per instance.
##########
cpp/src/parquet/encoding_test.cc:
##########
@@ -2660,4 +2595,407 @@ TEST(DeltaByteArrayEncodingAdHoc, ArrowDirectPut) {
}
}
+// ----------------------------------------------------------------------
+// ALP encoding tests for float/double
+
+template <typename Type>
+class TestAlpEncoding : public TestEncodingBase<Type> {
+ public:
+ using c_type = typename Type::c_type;
+ static constexpr int TYPE = Type::type_num;
+ static constexpr size_t kNumRoundTrips = 3;
+
+ void CheckRoundtrip() override {
+ auto encoder =
+ MakeTypedEncoder<Type>(Encoding::ALP, /*use_dictionary=*/false,
descr_.get());
+ auto decoder = MakeTypedDecoder<Type>(Encoding::ALP, descr_.get());
+
+ for (size_t i = 0; i < kNumRoundTrips; ++i) {
+ encoder->Put(draws_, num_values_);
+ encode_buffer_ = encoder->FlushValues();
+
+ decoder->SetData(num_values_, encode_buffer_->data(),
+ static_cast<int>(encode_buffer_->size()));
+ int values_decoded = decoder->Decode(decode_buf_, num_values_);
+ ASSERT_EQ(num_values_, values_decoded);
+
+ // Use memcmp for bit-exact comparison (important for -0.0, NaN bit
patterns)
+ ASSERT_EQ(0, std::memcmp(draws_, decode_buf_, num_values_ *
sizeof(c_type)));
+ }
+ }
+
+ void CheckRoundtripSpaced(const uint8_t* valid_bits,
+ int64_t valid_bits_offset) override {
+ auto encoder =
+ MakeTypedEncoder<Type>(Encoding::ALP, /*use_dictionary=*/false,
descr_.get());
+ auto decoder = MakeTypedDecoder<Type>(Encoding::ALP, descr_.get());
+
+ int null_count = 0;
+ for (auto i = 0; i < num_values_; i++) {
+ if (!bit_util::GetBit(valid_bits, valid_bits_offset + i)) {
+ null_count++;
+ }
+ }
+
+ for (size_t i = 0; i < kNumRoundTrips; ++i) {
+ encoder->PutSpaced(draws_, num_values_, valid_bits, valid_bits_offset);
+ encode_buffer_ = encoder->FlushValues();
+
+ decoder->SetData(num_values_ - null_count, encode_buffer_->data(),
+ static_cast<int>(encode_buffer_->size()));
+ auto values_decoded = decoder->DecodeSpaced(decode_buf_, num_values_,
null_count,
+ valid_bits,
valid_bits_offset);
+ ASSERT_EQ(num_values_, values_decoded);
+
+ // Verify only valid values
+ for (int j = 0; j < num_values_; ++j) {
+ if (bit_util::GetBit(valid_bits, valid_bits_offset + j)) {
+ ASSERT_EQ(0, std::memcmp(&draws_[j], &decode_buf_[j],
sizeof(c_type))) << j;
+ }
+ }
+ }
+ }
+
+ void InitDataWithSpecialValues(int nvalues, int repeats) {
+ num_values_ = nvalues * repeats;
+ this->input_bytes_.resize(num_values_ * sizeof(c_type));
+ this->output_bytes_.resize(num_values_ * sizeof(c_type));
+ draws_ = reinterpret_cast<c_type*>(this->input_bytes_.data());
+ decode_buf_ = reinterpret_cast<c_type*>(this->output_bytes_.data());
+
+ // Fill with mix of normal and special values
+ for (int i = 0; i < nvalues; ++i) {
+ if (i % 20 == 0) {
+ draws_[i] = std::numeric_limits<c_type>::quiet_NaN();
+ } else if (i % 20 == 5) {
+ draws_[i] = std::numeric_limits<c_type>::infinity();
+ } else if (i % 20 == 10) {
+ draws_[i] = -std::numeric_limits<c_type>::infinity();
+ } else if (i % 20 == 15) {
+ draws_[i] = static_cast<c_type>(-0.0);
+ } else {
+ draws_[i] = static_cast<c_type>(i) * static_cast<c_type>(0.123);
+ }
+ }
+
+ // Repeat pattern
+ for (int j = 1; j < repeats; ++j) {
+ for (int i = 0; i < nvalues; ++i) {
+ draws_[nvalues * j + i] = draws_[i];
+ }
+ }
+ }
+
+ void InitDataDecimalPattern(int nvalues, int repeats) {
+ num_values_ = nvalues * repeats;
+ this->input_bytes_.resize(num_values_ * sizeof(c_type));
+ this->output_bytes_.resize(num_values_ * sizeof(c_type));
+ draws_ = reinterpret_cast<c_type*>(this->input_bytes_.data());
+ decode_buf_ = reinterpret_cast<c_type*>(this->output_bytes_.data());
+
+ // Decimal-like values that ALP compresses well
+ for (int i = 0; i < nvalues; ++i) {
+ draws_[i] = static_cast<c_type>(100.0 + i * 0.01);
+ }
+
+ for (int j = 1; j < repeats; ++j) {
+ for (int i = 0; i < nvalues; ++i) {
+ draws_[nvalues * j + i] = draws_[i];
+ }
+ }
+ }
+
+ void ExecuteSpecialValues(int nvalues, int repeats) {
+ InitDataWithSpecialValues(nvalues, repeats);
+ CheckRoundtrip();
+ }
+
+ void ExecuteDecimalPattern(int nvalues, int repeats) {
+ InitDataDecimalPattern(nvalues, repeats);
+ CheckRoundtrip();
+ }
+
+ protected:
+ USING_BASE_MEMBERS();
+};
+
+using AlpEncodedTypes = ::testing::Types<FloatType, DoubleType>;
+TYPED_TEST_SUITE(TestAlpEncoding, AlpEncodedTypes);
+
+TYPED_TEST(TestAlpEncoding, BasicRoundTrip) {
+ // Test various sizes including edge cases
+ for (int values = 1; values < 32; ++values) {
+ ASSERT_NO_FATAL_FAILURE(this->Execute(values, 1));
+ }
+
+ // Test exactly vector size (1024)
+ ASSERT_NO_FATAL_FAILURE(this->Execute(1024, 1));
+
+ // Test just under and over vector size
+ ASSERT_NO_FATAL_FAILURE(this->Execute(1023, 1));
+ ASSERT_NO_FATAL_FAILURE(this->Execute(1025, 1));
+
+ // Test multiple vectors
+ ASSERT_NO_FATAL_FAILURE(this->Execute(2048, 1));
+ ASSERT_NO_FATAL_FAILURE(this->Execute(3000, 1));
+}
+
+TYPED_TEST(TestAlpEncoding, RoundTripWithRepeats) {
+ // Test with repeated patterns
+ ASSERT_NO_FATAL_FAILURE(this->Execute(100, 10));
+ ASSERT_NO_FATAL_FAILURE(this->Execute(1024, 3));
+}
+
+TYPED_TEST(TestAlpEncoding, SpecialValues) {
+ // Test NaN, Inf, -Inf, -0.0 (these become exceptions in ALP)
+ ASSERT_NO_FATAL_FAILURE(this->ExecuteSpecialValues(100, 1));
+ ASSERT_NO_FATAL_FAILURE(this->ExecuteSpecialValues(1024, 1));
+ ASSERT_NO_FATAL_FAILURE(this->ExecuteSpecialValues(2000, 1));
+}
+
+TYPED_TEST(TestAlpEncoding, DecimalPatterns) {
+ // Test decimal-like values that ALP compresses efficiently
+ ASSERT_NO_FATAL_FAILURE(this->ExecuteDecimalPattern(100, 1));
+ ASSERT_NO_FATAL_FAILURE(this->ExecuteDecimalPattern(1024, 1));
+ ASSERT_NO_FATAL_FAILURE(this->ExecuteDecimalPattern(5000, 1));
+}
+
+TYPED_TEST(TestAlpEncoding, SpacedRoundTrip) {
+ // Test with null values at various probabilities
+ for (double null_prob : {0.0, 0.1, 0.5, 0.9}) {
+ ASSERT_NO_FATAL_FAILURE(this->ExecuteSpaced(100, 1, 0, null_prob));
+ ASSERT_NO_FATAL_FAILURE(this->ExecuteSpaced(1024, 1, 0, null_prob));
+ ASSERT_NO_FATAL_FAILURE(this->ExecuteSpaced(2000, 1, 0, null_prob));
+ }
+
+ // Test with offset
+ ASSERT_NO_FATAL_FAILURE(this->ExecuteSpaced(1024, 1, 7, 0.3));
+ ASSERT_NO_FATAL_FAILURE(this->ExecuteSpaced(1024, 1, 64, 0.5));
+}
+
+TYPED_TEST(TestAlpEncoding, LargeDataset) {
+ // Test with large dataset (multiple pages worth)
+ ASSERT_NO_FATAL_FAILURE(this->Execute(100000, 1));
+}
+
+TYPED_TEST(TestAlpEncoding, RandomData) {
+ using c_type = typename TypeParam::c_type;
+ ::arrow::random::RandomArrayGenerator rag(42);
+
+ // Generate random float/double array
+ std::shared_ptr<::arrow::Array> arr;
+ if constexpr (std::is_same_v<c_type, float>) {
+ arr = rag.Float32(10000, -1000.0f, 1000.0f);
Review Comment:
Widened — that generator spans ±1e30 now instead of ±1000, so values land on
both sides of the ALP encodable window and the exception fallback actually gets
exercised. Beyond that there are `SpecialValues`, `ConstantValues`,
`AllExceptions` and `BoundaryValues`, plus `RandomAndExtremes` in the codec's
own test pushing on the encodable/exception boundary — all parameterized over
float and double.
--
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]
