prtkgaur commented on code in PR #48345:
URL: https://github.com/apache/arrow/pull/48345#discussion_r3921390281
##########
cpp/src/parquet/decoder.cc:
##########
@@ -2372,6 +2375,130 @@ 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;
+
+ // TODO: support incremental decode. Partial reads currently decode the
entire
+ // page into `decoded_buffer_` on first call and copy out the requested
range;
+ // a future revision should decode only the requested values, with state
+ // tracking for cross-call resumption.
+ 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);
Review Comment:
Fixed. `SetData` opens the `VectorReader` and takes `num_values_` from the
ALP header, using the page's `num_values` only as an upper bound — a payload
declaring more values than the page has level slots is rejected. There's a
comment recording why, since "use the header, not the argument" reads like a
mistake otherwise. The test that was missing this is your `encoding_test.cc`
comment below.
##########
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) {
Review Comment:
Real bug, fixed. `FlushValues` no longer short-circuits on an empty sink —
it runs the normal path with a zero count, producing a 7-byte header with no
offset section. The reader accepts that: `Open` returns early when there are no
vectors and `num_values_` comes out 0.
Two tests write an all-null optional V2 page and read it back; the writer
path had no coverage at all before. I checked they fail without the fix.
##########
cpp/src/parquet/arrow/arrow_reader_writer_test.cc:
##########
@@ -5456,6 +5458,244 @@ TEST(TestArrowReadDeltaEncoding, OptionalColumn) {
#endif
+// ALP encoding correctness tests — reads ALP-encoded parquet files and
verifies
+// values match expected CSV data bit-exactly. Uses std::ifstream for CSV
parsing
+// (no ARROW_CSV dependency).
+class TestArrowReadAlpEncoding : public ::testing::Test {
+ public:
+ void ReadTableFromParquetFile(const std::string& file_name,
+ std::shared_ptr<Table>* out) {
+ auto file = test::get_data_file(file_name);
+ auto pool = ::arrow::default_memory_pool();
+ std::unique_ptr<FileReader> parquet_reader;
+ ASSERT_OK(FileReader::Make(pool, ParquetFileReader::OpenFile(file, false),
Review Comment:
These tests have since moved to `parquet/arrow/arrow_encoding_test.cc`, per
your other comment, so a rebase now picks up a different file than the one you
were on. If you're still hitting a compile error there, paste it and I'll chase
it — I'd rather not close this on the move having happened to fix it.
##########
cpp/src/parquet/column_writer_test.cc:
##########
@@ -2475,5 +2475,89 @@ TYPED_TEST(TestBloomFilterWriter, Basic) {
}
}
+// ----------------------------------------------------------------------
+// ALP Encoding Tests for Float/Double Columns
+// ----------------------------------------------------------------------
+
+using TestFloatValuesWriter = TestPrimitiveWriter<FloatType>;
+using TestDoubleValuesWriter = TestPrimitiveWriter<DoubleType>;
+
+TEST_F(TestFloatValuesWriter, RequiredAlpEncoding) {
Review Comment:
Added, and it found the bug you flagged in the same review — the all-null
page was writing a zero-byte payload, which is your `FlushValues` comment below.
The helper writes an optional column with V2 pages and leading nulls, so the
definition levels and the value buffer disagree by a known amount. Four tests
use it: float and double with a few nulls, and both with every value null,
which is the empty-buffer case.
##########
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)
Review Comment:
Right. It's a `static constexpr` now, with a comment saying why it's fixed:
a reader takes the vector size from the page header, so nothing depends on the
writer varying it. Same change as your `vector_size_` suggestion on the member.
##########
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
Review Comment:
Fair, and trimmed — the block above `AlpEncoder` went from 33 lines to 7.
What's left is the two TODOs and one fact each: `Put` buffers the page, so
working memory scales with it, and the PLAIN fallback needs a ratio estimate
the sampler computes but doesn't expose. I kept the paper's break-even numbers,
since they're the evidence the fallback matters; the argument about where that
decision belongs is down to one clause.
##########
cpp/src/parquet/arrow/arrow_reader_writer_test.cc:
##########
@@ -6079,5 +6319,323 @@ TEST(TestArrowReadWrite, AllNulls) {
ASSERT_TRUE(expected_table->Equals(*read_table));
}
+// ============================================================================
+// ALP Encoding File-Level Integration Tests
+// ============================================================================
+
+class ParquetAlpEncodingTest : public ::testing::Test {
Review Comment:
Done under exactly that name. `parquet/arrow/arrow_encoding_test.cc` holds
both ALP fixtures and `arrow_reader_writer_test.cc` lost 538 lines. It links
into the existing `arrow-reader-writer-test` target, the pattern already used
there for `arrow_statistics_test.cc`, so no new binary — and it was a pure
move: all 22 ALP tests are there under the same names.
Set up for the next encoding too: the only ALP-specific things in it are the
two fixtures, over a file-local roundtrip helper.
##########
cpp/src/parquet/decoder.cc:
##########
@@ -2372,6 +2375,130 @@ 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;
+
+ // TODO: support incremental decode. Partial reads currently decode the
entire
+ // page into `decoded_buffer_` on first call and copy out the requested
range;
+ // a future revision should decode only the requested values, with state
+ // tracking for cross-call resumption.
+ 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;
+ if (num_values > 0 && len <= 0) {
+ throw ParquetException("ALP SetData: num_values=" +
std::to_string(num_values) +
+ " but len=" + std::to_string(len));
+ }
+ needs_decode_ = (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_) {
+ PARQUET_THROW_NOT_OK(::arrow::util::alp::AlpCodec<T>::Decode(
+ this->num_values_, this->data_, this->len_, buffer));
+
+ const int decoded = this->num_values_;
+ this->num_values_ = 0;
+ needs_decode_ = false;
+ return decoded;
+ }
+
+ // Slow path: partial read - decode to intermediate buffer.
AlpCodec::Decode
+ // always starts from the beginning of the page and keeps no resumption
+ // state, so a partial read decodes the whole page once and then serves
this
+ // and later calls out of `decoded_buffer_`.
+ if (needs_decode_) {
+ decoded_buffer_.resize(this->num_values_);
+ PARQUET_THROW_NOT_OK(::arrow::util::alp::AlpCodec<T>::Decode(
+ this->num_values_, this->data_, this->len_, decoded_buffer_.data()));
+ 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_);
+ PARQUET_THROW_NOT_OK(::arrow::util::alp::AlpCodec<T>::Decode(
+ this->num_values_, this->data_, this->len_, decoded_buffer_.data()));
+ needs_decode_ = false;
+ }
+
+ if (null_count == 0) {
+ // Fast path: no nulls
+ PARQUET_THROW_NOT_OK(builder->AppendValues(decoded_buffer_.data() +
current_offset_,
+ values_to_decode));
+ current_offset_ += values_to_decode;
+ this->num_values_ -= values_to_decode;
+ return values_to_decode;
+ } else {
+ // Slow path: with nulls
+ int value_idx = 0;
+ for (int i = 0; i < num_values; ++i) {
+ if (::arrow::bit_util::GetBit(valid_bits, valid_bits_offset + i)) {
+ PARQUET_THROW_NOT_OK(
+ builder->Append(decoded_buffer_[current_offset_ + value_idx]));
+ ++value_idx;
+ } else {
+ PARQUET_THROW_NOT_OK(builder->AppendNull());
+ }
+ }
+ 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>::DictAccumulator* builder)
override {
+ ParquetException::NYI("DecodeArrow to DictAccumulator for ALP");
+ }
+
+ private:
+ std::vector<T> decoded_buffer_;
Review Comment:
Both parts done, and the buffer got much smaller on the way. `AlpDecoder`
takes the pool and the scratch is a pool-backed `ResizableBuffer`, grown on
first use. Because of the incremental-decode change it holds one vector rather
than a page, so a 1 MB page now costs 8 KB of tracked scratch instead of 1 MB
of untracked — and vector-aligned batches skip it entirely.
##########
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(); }
Review Comment:
Comment added, and it says which direction each way: it over-reports for a
column ALP compresses and under-reports for one it doesn't. I couldn't find
simple math either — the size depends on the per-vector bit width and the
exception count, both of which come out of the encode itself. The sampler
already computes an estimate for its own decisions, so exposing that is the
real fix rather than guessing a formula.
##########
cpp/src/parquet/decoder.cc:
##########
@@ -2372,6 +2375,130 @@ 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;
+
+ // TODO: support incremental decode. Partial reads currently decode the
entire
+ // page into `decoded_buffer_` on first call and copy out the requested
range;
+ // a future revision should decode only the requested values, with state
+ // tracking for cross-call resumption.
+ 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;
+ if (num_values > 0 && len <= 0) {
+ throw ParquetException("ALP SetData: num_values=" +
std::to_string(num_values) +
+ " but len=" + std::to_string(len));
+ }
+ needs_decode_ = (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_) {
+ PARQUET_THROW_NOT_OK(::arrow::util::alp::AlpCodec<T>::Decode(
+ this->num_values_, this->data_, this->len_, buffer));
+
+ const int decoded = this->num_values_;
+ this->num_values_ = 0;
+ needs_decode_ = false;
+ return decoded;
+ }
+
+ // Slow path: partial read - decode to intermediate buffer.
AlpCodec::Decode
+ // always starts from the beginning of the page and keeps no resumption
+ // state, so a partial read decodes the whole page once and then serves
this
+ // and later calls out of `decoded_buffer_`.
+ if (needs_decode_) {
+ decoded_buffer_.resize(this->num_values_);
+ PARQUET_THROW_NOT_OK(::arrow::util::alp::AlpCodec<T>::Decode(
+ this->num_values_, this->data_, this->len_, decoded_buffer_.data()));
+ 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_);
+ PARQUET_THROW_NOT_OK(::arrow::util::alp::AlpCodec<T>::Decode(
+ this->num_values_, this->data_, this->len_, decoded_buffer_.data()));
+ needs_decode_ = false;
+ }
+
+ if (null_count == 0) {
+ // Fast path: no nulls
+ PARQUET_THROW_NOT_OK(builder->AppendValues(decoded_buffer_.data() +
current_offset_,
+ values_to_decode));
+ current_offset_ += values_to_decode;
+ this->num_values_ -= values_to_decode;
+ return values_to_decode;
+ } else {
+ // Slow path: with nulls
Review Comment:
Following it now. `DecodeArrow` reserves once, decodes into builder storage
packed to the right, then either `UnsafeAdvance` when there are no nulls or
`SpacedExpandLeftward` followed by `UnsafeAdvance`. No per-value builder calls
remain in the ALP path.
##########
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:
Both, and I went one step further: a `const` member is still a per-encoder
word and still reads as configurable, so it's a `static constexpr`, with a
comment saying why it's fixed — a reader takes the vector size from the page
header, so nothing depends on the writer varying it. The power-of-two check
went with it, since a constant can't fail it.
The codec's `Encode` still takes `vector_size` and its tests exercise 8
through 32768, so that's still covered even though the writer always uses one
size. If you'd rather it be a writer property, that's small on top.
##########
cpp/src/parquet/decoder.cc:
##########
@@ -35,6 +35,9 @@
#include "arrow/array/builder_dict.h"
#include "arrow/array/builder_primitive.h"
#include "arrow/type_traits.h"
+#include "arrow/util/alp/alp.h"
Review Comment:
`decoder.cc` is down to one, `alp_codec_internal.h`; it had been pulling in
three. `encoder.cc` keeps two, since it also needs
`AlpConstants::kAlpVectorSize` for the writer's fixed vector size. Moving
`AlpMode` into `alp_codec.cc` removed the rest.
##########
cpp/src/parquet/encoding_test.cc:
##########
@@ -2660,4 +2663,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(),
Review Comment:
Both. The spaced round-trip helper passes the level count with nulls
included, matching `ColumnReader`, and asserts `values_left()` is zero after
`DecodeSpaced`. There's a comment naming the production caller so the next
person doesn't simplify it back. You were right that the old form hid the count
bug: with the non-null count passed in, the decoder's wrong reading of the
argument happened to be correct.
##########
cpp/src/parquet/decoder.cc:
##########
@@ -2372,6 +2375,130 @@ 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;
+
+ // TODO: support incremental decode. Partial reads currently decode the
entire
Review Comment:
Agreed, and done — it removed code rather than adding it. `VectorReader`
validates the header and the whole offset chain once in `Open` and then decodes
any single vector on demand, so `Decode` is `Open` plus a loop and both paths
share one validator instead of having one each.
A vector entered at its first value and read to its end decodes straight
into the caller's buffer; anything else goes through one vector of scratch. So
the decoder holds 4 or 8 KB rather than a page, pool-backed and allocated on
first use — which also answers your scratch-buffer comment.
`TestAlpEncoding.BatchedDecode` covers six batch plans over a 2000-value page.
##########
cpp/src/parquet/decoder.cc:
##########
@@ -2372,6 +2375,130 @@ 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;
+
+ // TODO: support incremental decode. Partial reads currently decode the
entire
+ // page into `decoded_buffer_` on first call and copy out the requested
range;
+ // a future revision should decode only the requested values, with state
+ // tracking for cross-call resumption.
+ 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;
+ if (num_values > 0 && len <= 0) {
+ throw ParquetException("ALP SetData: num_values=" +
std::to_string(num_values) +
+ " but len=" + std::to_string(len));
+ }
+ needs_decode_ = (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_) {
+ PARQUET_THROW_NOT_OK(::arrow::util::alp::AlpCodec<T>::Decode(
+ this->num_values_, this->data_, this->len_, buffer));
+
+ const int decoded = this->num_values_;
+ this->num_values_ = 0;
+ needs_decode_ = false;
+ return decoded;
+ }
+
+ // Slow path: partial read - decode to intermediate buffer.
AlpCodec::Decode
+ // always starts from the beginning of the page and keeps no resumption
+ // state, so a partial read decodes the whole page once and then serves
this
+ // and later calls out of `decoded_buffer_`.
+ if (needs_decode_) {
+ decoded_buffer_.resize(this->num_values_);
+ PARQUET_THROW_NOT_OK(::arrow::util::alp::AlpCodec<T>::Decode(
+ this->num_values_, this->data_, this->len_, decoded_buffer_.data()));
+ 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_);
+ PARQUET_THROW_NOT_OK(::arrow::util::alp::AlpCodec<T>::Decode(
+ this->num_values_, this->data_, this->len_, decoded_buffer_.data()));
+ needs_decode_ = false;
+ }
+
+ if (null_count == 0) {
+ // Fast path: no nulls
+ PARQUET_THROW_NOT_OK(builder->AppendValues(decoded_buffer_.data() +
current_offset_,
+ values_to_decode));
+ current_offset_ += values_to_decode;
+ this->num_values_ -= values_to_decode;
+ return values_to_decode;
+ } else {
+ // Slow path: with nulls
+ int value_idx = 0;
+ for (int i = 0; i < num_values; ++i) {
+ if (::arrow::bit_util::GetBit(valid_bits, valid_bits_offset + i)) {
+ PARQUET_THROW_NOT_OK(
+ builder->Append(decoded_buffer_[current_offset_ + value_idx]));
+ ++value_idx;
+ } else {
+ PARQUET_THROW_NOT_OK(builder->AppendNull());
+ }
+ }
+ 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>::DictAccumulator* builder)
override {
+ ParquetException::NYI("DecodeArrow to DictAccumulator for ALP");
+ }
+
+ private:
+ std::vector<T> decoded_buffer_;
+ size_t current_offset_;
+ bool needs_decode_;
Review Comment:
Consolidated to one: the inherited `num_values_`. `needs_decode_` and
`current_offset_` are gone, along with a `scratch_filled_` flag that my first
attempt at this introduced. Position is derived as `total_values_ -
num_values_`, with `total_values_` set once by `SetData` and not touched again,
and `Decode` and `DecodeArrow` each decrement `num_values_` exactly once, so
the two paths can't disagree.
--
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]
