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:
Done
--
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]
